diff --git a/config/config.go b/config/config.go index 92daf99fc..52732b933 100644 --- a/config/config.go +++ b/config/config.go @@ -32,6 +32,7 @@ type Config struct { HTTP HTTPConfig `yaml:"http"` Admin AdminConfig `yaml:"admin"` Guardrails GuardrailsConfig `yaml:"guardrails"` + ThinkExtract ThinkExtractConfig `yaml:"think_extract"` Failover FailoverConfig `yaml:"failover"` Workflows WorkflowsConfig `yaml:"workflows"` Resilience ResilienceConfig `yaml:"resilience"` @@ -186,6 +187,10 @@ func buildDefaultConfig() *Config { LiveLogsHeartbeatSeconds: 15, }, Guardrails: GuardrailsConfig{}, + ThinkExtract: ThinkExtractConfig{ + // Pointer-nil so IsEnabled() falls through to the default (true). + // Operators set THINK_EXTRACT_ENABLED=false to opt out per deployment. + }, Session: SessionConfig{ Enabled: true, AutoDetect: true, diff --git a/config/thinkextract.go b/config/thinkextract.go new file mode 100644 index 000000000..352073647 --- /dev/null +++ b/config/thinkextract.go @@ -0,0 +1,91 @@ +package config + +// ThinkExtractConfig controls the response-path translation of legacy +// `...` (and configured equivalents) into the native reasoning +// field. The translation only runs on responses — request-side bodies are +// never modified. +// +// The default is enabled because the translation is lossless on the wire +// (no model-visible character is dropped) and the dialect converters for +// OpenAI chat completions, OpenAI responses, and Anthropic messages all +// surface ExtraFields["reasoning_content"] as their native reasoning field. +// Operators who observe a regression on a model that already emits structured +// reasoning can disable the feature per deployment. +type ThinkExtractConfig struct { + // Enabled toggles the translation globally. Default true. + Enabled *bool `yaml:"enabled" env:"THINK_EXTRACT_ENABLED"` + // ChatEnabled toggles the translation on the chat completions surface. + // Nil falls back to Enabled. Env: THINK_EXTRACT_CHAT_ENABLED. + ChatEnabled *bool `yaml:"chat_enabled" env:"THINK_EXTRACT_CHAT_ENABLED"` + // ResponsesEnabled toggles the translation on the OpenAI responses + // surface. Nil falls back to Enabled. Env: THINK_EXTRACT_RESPONSES_ENABLED. + ResponsesEnabled *bool `yaml:"responses_enabled" env:"THINK_EXTRACT_RESPONSES_ENABLED"` + // MessagesPolicy controls how synthesized reasoning is emitted on the + // Anthropic messages surface. Values: off (default), unsigned, redacted. + // "off" means no extraction runs for messages requests, so legacy tags + // stay in the message content unchanged. Env: THINK_EXTRACT_MESSAGES_POLICY. + MessagesPolicy string `yaml:"messages_policy" env:"THINK_EXTRACT_MESSAGES_POLICY"` + // TagPairs overrides the recognition list. The default list covers the + // union of vLLM/SGLang/Open WebUI standard tags. Format is a + // comma-separated "..." list, e.g. + // "...,...". + TagPairs string `yaml:"tag_pairs" env:"THINK_EXTRACT_TAG_PAIRS"` + // MaxBufferBytes caps the size of an unclosed block held in streaming + // state before it is flushed as ordinary content. Default 65536. + MaxBufferBytes int `yaml:"max_buffer_bytes" env:"THINK_EXTRACT_MAX_BUFFER_BYTES"` +} + +// IsEnabled reports whether the translation is active at the global level. +// Nil receiver and nil Enabled pointer are both treated as default-true. +func (c ThinkExtractConfig) IsEnabled() bool { + if c.Enabled == nil { + return true + } + return *c.Enabled +} + +// IsEnabledForChat reports whether the translation runs on the chat +// completions surface. Falls back to the global Enabled value when the +// per-surface pointer is unset. +func (c ThinkExtractConfig) IsEnabledForChat() bool { + if c.ChatEnabled != nil { + return *c.ChatEnabled + } + return c.IsEnabled() +} + +// IsEnabledForResponses reports whether the translation runs on the OpenAI +// responses surface. Falls back to the global Enabled value when unset. +func (c ThinkExtractConfig) IsEnabledForResponses() bool { + if c.ResponsesEnabled != nil { + return *c.ResponsesEnabled + } + return c.IsEnabled() +} + +// IsEnabledForMessages reports whether the translation runs on the +// Anthropic messages surface. The messages policy defaults to off, so the +// translation only runs when an operator opts in explicitly. The global +// Enabled switch is also authoritative — a global off kills the feature +// everywhere regardless of the per-surface policy. +func (c ThinkExtractConfig) IsEnabledForMessages() bool { + if !c.IsEnabled() { + return false + } + switch c.MessagesPolicy { + case "unsigned", "redacted": + return true + default: + return false + } +} + +// MessagesPolicyOrDefault returns the configured messages policy, falling +// back to "off" when unset so the per-call site can rely on a non-empty +// value. +func (c ThinkExtractConfig) MessagesPolicyOrDefault() string { + if c.MessagesPolicy == "" { + return "off" + } + return c.MessagesPolicy +} \ No newline at end of file diff --git a/config/thinkextract_test.go b/config/thinkextract_test.go new file mode 100644 index 000000000..3579b7674 --- /dev/null +++ b/config/thinkextract_test.go @@ -0,0 +1,94 @@ +package config + +import "testing" + +func boolPtr(v bool) *bool { return &v } + +func TestThinkExtractConfig_Defaults(t *testing.T) { + cfg := ThinkExtractConfig{} + if !cfg.IsEnabled() { + t.Errorf("zero config: IsEnabled=false, want true") + } + if !cfg.IsEnabledForChat() { + t.Errorf("zero config: IsEnabledForChat=false, want true") + } + if cfg.IsEnabledForMessages() { + t.Errorf("zero config: IsEnabledForMessages=true, want false (messages policy defaults to off)") + } + if got := cfg.MessagesPolicyOrDefault(); got != "off" { + t.Errorf("MessagesPolicyOrDefault=%q, want %q", got, "off") + } +} + +func TestThinkExtractConfig_GlobalOff(t *testing.T) { + cfg := ThinkExtractConfig{Enabled: boolPtr(false)} + if cfg.IsEnabled() { + t.Errorf("IsEnabled=true, want false") + } + if cfg.IsEnabledForChat() { + t.Errorf("IsEnabledForChat=true, want false (falls back to global)") + } + if cfg.IsEnabledForMessages() { + t.Errorf("IsEnabledForMessages=true, want false (falls back to global)") + } +} + +func TestThinkExtractConfig_PerSurfaceOverride(t *testing.T) { + cfg := ThinkExtractConfig{ + Enabled: boolPtr(true), + ChatEnabled: boolPtr(false), + MessagesPolicy: "unsigned", + } + if !cfg.IsEnabled() { + t.Errorf("IsEnabled=false, want true") + } + if cfg.IsEnabledForChat() { + t.Errorf("IsEnabledForChat=true, want false (per-surface override)") + } + if !cfg.IsEnabledForMessages() { + t.Errorf("IsEnabledForMessages=false, want true (unsigned policy)") + } +} + +func TestThinkExtractConfig_PerSurfaceTrueCannotResurrect(t *testing.T) { + cfg := ThinkExtractConfig{ + Enabled: boolPtr(false), + ChatEnabled: boolPtr(true), + } + if cfg.IsEnabled() { + t.Errorf("IsEnabled=true, want false") + } + if !cfg.IsEnabledForChat() { + t.Errorf("IsEnabledForChat=false, want true (per-surface value)") + } +} + +func TestThinkExtractConfig_MessagesPolicyParsing(t *testing.T) { + tests := []struct { + name string + cfg ThinkExtractConfig + want bool + }{ + {name: "empty means off", cfg: ThinkExtractConfig{}, want: false}, + {name: "off explicit", cfg: ThinkExtractConfig{MessagesPolicy: "off"}, want: false}, + {name: "unsigned enables", cfg: ThinkExtractConfig{MessagesPolicy: "unsigned"}, want: true}, + {name: "redacted enables", cfg: ThinkExtractConfig{MessagesPolicy: "redacted"}, want: true}, + {name: "unknown falls back to off", cfg: ThinkExtractConfig{MessagesPolicy: "nonsense"}, want: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := tt.cfg.IsEnabledForMessages(); got != tt.want { + t.Errorf("IsEnabledForMessages()=%v, want %v", got, tt.want) + } + }) + } +} + +func TestThinkExtractConfig_MessagesPolicyOrDefault(t *testing.T) { + if got := (ThinkExtractConfig{}).MessagesPolicyOrDefault(); got != "off" { + t.Errorf("empty cfg default=%q, want off", got) + } + if got := (ThinkExtractConfig{MessagesPolicy: "redacted"}).MessagesPolicyOrDefault(); got != "redacted" { + t.Errorf("explicit cfg default=%q, want redacted", got) + } +} \ No newline at end of file diff --git a/internal/anthropicapi/policy_test.go b/internal/anthropicapi/policy_test.go new file mode 100644 index 000000000..d33a5a86a --- /dev/null +++ b/internal/anthropicapi/policy_test.go @@ -0,0 +1,168 @@ +package anthropicapi + +import ( + "encoding/json" + "io" + "strings" + "testing" + + "github.com/enterpilot/gomodel/internal/core" + "github.com/enterpilot/gomodel/internal/thinkextract" +) + +func chatRespWithReasoning(reasoning string, synthesized bool) *core.ChatResponse { + extra := core.UnknownJSONFields{} + fields := map[string]json.RawMessage{ + "reasoning_content": json.RawMessage(`"` + reasoning + `"`), + } + if synthesized { + fields[thinkextract.SynthesizedMarkerKey] = json.RawMessage("true") + } + merged, err := core.MergeUnknownJSONFields(extra, fields) + if err != nil { + panic(err) + } + return &core.ChatResponse{ + Choices: []core.Choice{{ + Message: core.ResponseMessage{ + Role: "assistant", + Content: "answer", + ExtraFields: merged, + }, + }}, + } +} + +func TestFromChatResponse_NativeReasoningUnchangedByPolicy(t *testing.T) { + // Provider-supplied reasoning (no synthesized marker) always renders as a + // thinking block regardless of the policy. + for _, policy := range []thinkextract.MessagesThinkingPolicy{ + thinkextract.MessagesPolicyOff, + thinkextract.MessagesPolicyUnsigned, + thinkextract.MessagesPolicyRedacted, + } { + out := FromChatResponseWithPolicy(chatRespWithReasoning("native", false), policy) + if len(out.Content) != 2 || out.Content[0].Type != "thinking" || out.Content[0].Thinking != "native" { + t.Errorf("policy=%q: native reasoning must render as thinking block, got %+v", policy, out.Content) + } + } +} + +func TestFromChatResponse_SynthesizedOff(t *testing.T) { + out := FromChatResponseWithPolicy(chatRespWithReasoning("synth", true), thinkextract.MessagesPolicyOff) + for _, b := range out.Content { + if b.Type == "thinking" || b.Type == "redacted_thinking" { + t.Errorf("off policy: synthesized reasoning leaked as %q", b.Type) + } + } + // Content text survives. + if len(out.Content) != 1 || out.Content[0].Type != "text" || out.Content[0].Text != "answer" { + t.Errorf("off policy: content=%+v, want single text block", out.Content) + } +} + +func TestFromChatResponse_SynthesizedUnsigned(t *testing.T) { + out := FromChatResponseWithPolicy(chatRespWithReasoning("synth", true), thinkextract.MessagesPolicyUnsigned) + if len(out.Content) != 2 || out.Content[0].Type != "thinking" || out.Content[0].Thinking != "synth" { + t.Errorf("unsigned policy: got %+v, want thinking block first", out.Content) + } +} + +func TestFromChatResponse_SynthesizedRedacted(t *testing.T) { + out := FromChatResponseWithPolicy(chatRespWithReasoning("synth", true), thinkextract.MessagesPolicyRedacted) + if len(out.Content) != 2 || out.Content[0].Type != "redacted_thinking" { + t.Fatalf("redacted policy: got %+v, want redacted_thinking first", out.Content) + } + var data string + if err := json.Unmarshal(out.Content[0].Data, &data); err != nil { + t.Fatalf("redacted data unmarshal: %v", err) + } + if data != "synth" { + t.Errorf("redacted data=%q, want %q", data, "synth") + } +} + +func TestParseMessagesPolicy(t *testing.T) { + tests := []struct { + raw string + want thinkextract.MessagesThinkingPolicy + }{ + {"", thinkextract.MessagesPolicyOff}, + {"off", thinkextract.MessagesPolicyOff}, + {"unsigned", thinkextract.MessagesPolicyUnsigned}, + {"redacted", thinkextract.MessagesPolicyRedacted}, + {"garbage", thinkextract.MessagesPolicyOff}, + } + for _, tt := range tests { + if got := thinkextract.ParseMessagesPolicy(tt.raw); got != tt.want { + t.Errorf("ParseMessagesPolicy(%q)=%q, want %q", tt.raw, got, tt.want) + } + } +} + +func TestStreamConverter_PolicyOffDropsSynthesized(t *testing.T) { + input := "data: {\"id\":\"x\",\"model\":\"m\",\"choices\":[{\"index\":0,\"delta\":{\"reasoning_content\":\"r\",\"thinkextract_synthesized\":true}}]}\n\n" + + "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"answer\"}}]}\n\n" + + "data: [DONE]\n" + rc := NewStreamConverterWithPolicy(io.NopCloser(strings.NewReader(input)), "m", 0, thinkextract.MessagesPolicyOff) + defer rc.Close() + out, err := io.ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + s := string(out) + if strings.Contains(s, "thinking_delta") { + t.Errorf("off policy: synthesized thinking leaked: %q", s) + } + if !strings.Contains(s, "answer") { + t.Errorf("content text dropped: %q", s) + } +} + +func TestStreamConverter_PolicyUnsignedEmitsThinking(t *testing.T) { + input := "data: {\"id\":\"x\",\"model\":\"m\",\"choices\":[{\"index\":0,\"delta\":{\"reasoning_content\":\"r\",\"thinkextract_synthesized\":true}}]}\n\n" + + "data: [DONE]\n" + rc := NewStreamConverterWithPolicy(io.NopCloser(strings.NewReader(input)), "m", 0, thinkextract.MessagesPolicyUnsigned) + defer rc.Close() + out, err := io.ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + s := string(out) + if !strings.Contains(s, "\"thinking\"") { + t.Errorf("unsigned policy: no thinking block: %q", s) + } + if strings.Contains(s, "thinkextract_synthesized") { + t.Errorf("marker leaked to wire: %q", s) + } +} + +func TestStreamConverter_PolicyRedactedEmitsRedacted(t *testing.T) { + input := "data: {\"id\":\"x\",\"model\":\"m\",\"choices\":[{\"index\":0,\"delta\":{\"reasoning_content\":\"r\",\"thinkextract_synthesized\":true}}]}\n\n" + + "data: [DONE]\n" + rc := NewStreamConverterWithPolicy(io.NopCloser(strings.NewReader(input)), "m", 0, thinkextract.MessagesPolicyRedacted) + defer rc.Close() + out, err := io.ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + s := string(out) + if !strings.Contains(s, "redacted_thinking") { + t.Errorf("redacted policy: no redacted_thinking block: %q", s) + } +} + +func TestStreamConverter_NativeReasoningUnaffected(t *testing.T) { + // No marker: provider-native reasoning always renders as thinking. + input := "data: {\"id\":\"x\",\"model\":\"m\",\"choices\":[{\"index\":0,\"delta\":{\"reasoning_content\":\"native\"}}]}\n\n" + + "data: [DONE]\n" + rc := NewStreamConverterWithPolicy(io.NopCloser(strings.NewReader(input)), "m", 0, thinkextract.MessagesPolicyOff) + defer rc.Close() + out, err := io.ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(string(out), "thinking_delta") { + t.Errorf("native reasoning dropped under off policy: %q", string(out)) + } +} diff --git a/internal/anthropicapi/response.go b/internal/anthropicapi/response.go index e2c9b645a..bb93d955d 100644 --- a/internal/anthropicapi/response.go +++ b/internal/anthropicapi/response.go @@ -7,11 +7,19 @@ import ( "github.com/goccy/go-json" "github.com/enterpilot/gomodel/internal/core" + "github.com/enterpilot/gomodel/internal/thinkextract" ) // FromChatResponse renders a canonical chat response in the Anthropic Messages -// response shape. +// response shape using the default off policy for synthesized reasoning. func FromChatResponse(resp *core.ChatResponse) *MessagesResponse { + return FromChatResponseWithPolicy(resp, thinkextract.MessagesPolicyOff) +} + +// FromChatResponseWithPolicy renders a canonical chat response in the Anthropic +// Messages response shape, applying the given policy to any reasoning +// content that thinkextract marked as synthesized (vs provider-supplied). +func FromChatResponseWithPolicy(resp *core.ChatResponse, policy thinkextract.MessagesThinkingPolicy) *MessagesResponse { out := &MessagesResponse{ Type: "message", Role: "assistant", @@ -28,8 +36,17 @@ func FromChatResponse(resp *core.ChatResponse) *MessagesResponse { if len(resp.Choices) > 0 { choice := resp.Choices[0] - if thinking := reasoningContent(choice.Message.ExtraFields); thinking != "" { - out.Content = append(out.Content, ResponseContentBlock{Type: "thinking", Thinking: thinking}) + if thinking, synthesized := reasoningContentWithMarker(choice.Message.ExtraFields); thinking != "" { + switch { + case synthesized && policy == thinkextract.MessagesPolicyOff: + // Synthesized reasoning under off policy: drop the block; + // the tags were stripped from content at extraction time so + // no text is lost. + case synthesized && policy == thinkextract.MessagesPolicyRedacted: + out.Content = append(out.Content, redactedThinkingBlock(thinking)) + default: + out.Content = append(out.Content, ResponseContentBlock{Type: "thinking", Thinking: thinking}) + } } if text := core.ExtractTextContent(choice.Message.Content); text != "" { out.Content = append(out.Content, ResponseContentBlock{Type: "text", Text: text}) @@ -95,6 +112,34 @@ func reasoningContent(fields core.UnknownJSONFields) string { return text } +// reasoningContentWithMarker extracts the reasoning_content together with the +// synthesized-from-tags marker. The second return value is true when the +// reasoning was produced by thinkextract tag extraction rather than arriving +// from the provider as structured reasoning_content. +func reasoningContentWithMarker(fields core.UnknownJSONFields) (string, bool) { + text := reasoningContent(fields) + if text == "" { + return "", false + } + if len(fields.Lookup(thinkextract.SynthesizedMarkerKey)) == 0 { + return text, false + } + return text, true +} + +// redactedThinkingBlock renders the synthesized reasoning as a +// redacted_thinking block. The Data field carries the raw JSON text the +// provider would normally encrypt; for synthesized content we pass through +// the plain string verbatim. Anthropic clients that honour redacted_thinking +// will treat the value as opaque and surface it as such. +func redactedThinkingBlock(text string) ResponseContentBlock { + encoded, err := json.Marshal(text) + if err != nil { + return ResponseContentBlock{Type: "redacted_thinking"} + } + return ResponseContentBlock{Type: "redacted_thinking", Data: encoded} +} + // argumentsToRaw renders a tool-call arguments string as a JSON object value. func argumentsToRaw(arguments string) json.RawMessage { trimmed := strings.TrimSpace(arguments) diff --git a/internal/anthropicapi/stream.go b/internal/anthropicapi/stream.go index 792e8229b..b1d4c4486 100644 --- a/internal/anthropicapi/stream.go +++ b/internal/anthropicapi/stream.go @@ -8,6 +8,7 @@ import ( "github.com/goccy/go-json" "github.com/enterpilot/gomodel/internal/streaming" + "github.com/enterpilot/gomodel/internal/thinkextract" ) // chatChunk is the subset of an OpenAI chat.completion.chunk consumed by the @@ -19,8 +20,14 @@ type chatChunk struct { Delta struct { Content string `json:"content"` ReasoningContent string `json:"reasoning_content"` - StopSequence string `json:"stop_sequence"` - ToolCalls []chatToolCallDelta `json:"tool_calls"` + // SynthesizedReasoning marks a reasoning_content delta that came + // from thinkextract tag extraction rather than the provider's + // own structured field. The messages converter uses it to apply + // the messages thinking policy; the marker never reaches the + // client wire output. + SynthesizedReasoning bool `json:"thinkextract_synthesized"` + StopSequence string `json:"stop_sequence"` + ToolCalls []chatToolCallDelta `json:"tool_calls"` } `json:"delta"` FinishReason string `json:"finish_reason"` } `json:"choices"` @@ -66,6 +73,20 @@ func (u chatUsage) cacheRead() int { // available value there. The authoritative usage still arrives in // message_delta, which SDK accumulators prefer. func NewStreamConverter(body io.ReadCloser, model string, inputTokensEstimate int) io.ReadCloser { + return NewStreamConverterWithPolicy(body, model, inputTokensEstimate, thinkextract.MessagesPolicyOff) +} + +// NewStreamConverterWithPolicy wraps an OpenAI-style chat completion SSE +// stream and emits the equivalent Anthropic Messages SSE event sequence, +// applying the given policy to any reasoning delta that thinkextract marked +// as synthesized. The returned reader owns body and closes it on Close. +// +// inputTokensEstimate seeds message_start's usage.input_tokens: the Anthropic +// contract reports input tokens at stream start, but the OpenAI upstream only +// delivers usage in the final chunk, so a heuristic estimate is the best +// available value there. The authoritative usage still arrives in +// message_delta, which SDK accumulators prefer. +func NewStreamConverterWithPolicy(body io.ReadCloser, model string, inputTokensEstimate int, policy thinkextract.MessagesThinkingPolicy) io.ReadCloser { return &streamConverter{ reader: bufio.NewReader(body), body: body, @@ -73,6 +94,7 @@ func NewStreamConverter(body io.ReadCloser, model string, inputTokensEstimate in buffer: streaming.NewStreamBuffer(1024), toolBlock: make(map[int]int), inputEstimate: inputTokensEstimate, + policy: policy, } } @@ -83,6 +105,7 @@ type streamConverter struct { body io.ReadCloser buffer streaming.StreamBuffer model string + policy thinkextract.MessagesThinkingPolicy started bool blockOpen bool @@ -170,12 +193,26 @@ func (sc *streamConverter) handleChunk(chunk *chatChunk) { } for _, choice := range chunk.Choices { if choice.Delta.ReasoningContent != "" { - sc.ensureBlock("thinking") - sc.emit("content_block_delta", map[string]any{ - "type": "content_block_delta", - "index": sc.curIndex, - "delta": map[string]any{"type": "thinking_delta", "thinking": choice.Delta.ReasoningContent}, - }) + synthesized := choice.Delta.SynthesizedReasoning + switch { + case synthesized && sc.policy == thinkextract.MessagesPolicyOff: + // Synthesized reasoning under off policy: drop the delta; the + // tag text was stripped at extraction time so no content is lost. + case synthesized && sc.policy == thinkextract.MessagesPolicyRedacted: + sc.ensureBlock("redacted_thinking") + sc.emit("content_block_delta", map[string]any{ + "type": "content_block_delta", + "index": sc.curIndex, + "delta": map[string]any{"type": "thinking_delta", "thinking": choice.Delta.ReasoningContent}, + }) + default: + sc.ensureBlock("thinking") + sc.emit("content_block_delta", map[string]any{ + "type": "content_block_delta", + "index": sc.curIndex, + "delta": map[string]any{"type": "thinking_delta", "thinking": choice.Delta.ReasoningContent}, + }) + } } if choice.Delta.Content != "" { sc.ensureBlock("text") diff --git a/internal/anthropicapi/types.go b/internal/anthropicapi/types.go index 26edd8f60..cd0e1bc71 100644 --- a/internal/anthropicapi/types.go +++ b/internal/anthropicapi/types.go @@ -109,9 +109,12 @@ type ResponseContentBlock struct { Type string `json:"type"` Text string `json:"text,omitempty"` Thinking string `json:"thinking,omitempty"` - ID string `json:"id,omitempty"` - Name string `json:"name,omitempty"` - Input json.RawMessage `json:"input,omitempty" swaggertype:"object"` + // Data is set for redacted_thinking blocks; the field is opaque to the + // caller, so the type holds the raw bytes verbatim. + Data json.RawMessage `json:"data,omitempty"` + ID string `json:"id,omitempty"` + Name string `json:"name,omitempty"` + Input json.RawMessage `json:"input,omitempty" swaggertype:"object"` } // Usage reports Anthropic-style token usage. diff --git a/internal/app/app.go b/internal/app/app.go index 07bcb70ab..2d7a8b60f 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -45,6 +45,7 @@ import ( "github.com/enterpilot/gomodel/internal/session" "github.com/enterpilot/gomodel/internal/storage" "github.com/enterpilot/gomodel/internal/tagging" + "github.com/enterpilot/gomodel/internal/thinkextract" "github.com/enterpilot/gomodel/internal/usage" "github.com/enterpilot/gomodel/internal/virtualmodels" "github.com/enterpilot/gomodel/internal/workflows" @@ -761,6 +762,7 @@ func New(ctx context.Context, cfg Config) (*App, error) { SwaggerEnabled: swaggerEnabled, Tagging: taggingResult.Service, SessionDetector: session.NewDetectorFromConfig(appCfg.Session), + ThinkExtractOptions: thinkExtractOptionsFromConfig(appCfg.ThinkExtract), MCPEnabled: appCfg.MCP.Enabled, } if mcpResult != nil { @@ -1380,6 +1382,26 @@ func configGuardrailDefinitions(cfg config.GuardrailsConfig) ([]guardrails.Defin return definitions, nil } +// thinkExtractOptionsFromConfig converts the loaded think_extract config into +// the options consumed by the orchestrator, or nil when the feature is off. +// The global Enabled switch is authoritative: a per-surface true cannot +// resurrect the feature when the global switch is off. +func thinkExtractOptionsFromConfig(cfg config.ThinkExtractConfig) *thinkextract.Options { + if !cfg.IsEnabled() { + return nil + } + opts := &thinkextract.Options{ + MaxBufferBytes: cfg.MaxBufferBytes, + ChatEnabled: cfg.ChatEnabled, + ResponsesEnabled: cfg.ResponsesEnabled, + MessagesPolicy: cfg.MessagesPolicyOrDefault(), + } + if pairs := thinkextract.ParseTagPairs(cfg.TagPairs); len(pairs) > 0 { + opts.TagPairs = pairs + } + return opts +} + func defaultWorkflowInput(cfg *config.Config, availableGuardrails []string, configuredGuardrails []guardrails.Definition) workflows.CreateInput { failoverEnabled := failoverFeatureEnabledGlobally(cfg) budgetEnabled := cfg.Budgets.Enabled diff --git a/internal/gateway/inference_execute.go b/internal/gateway/inference_execute.go index 707158a2d..fe46b2694 100644 --- a/internal/gateway/inference_execute.go +++ b/internal/gateway/inference_execute.go @@ -8,6 +8,7 @@ import ( "time" "github.com/enterpilot/gomodel/internal/core" + "github.com/enterpilot/gomodel/internal/thinkextract" "github.com/enterpilot/gomodel/internal/usage" ) @@ -477,6 +478,9 @@ func (o *InferenceOrchestrator) chatCompletionProviderCall(ctx context.Context, if resp == nil { return nil, emptyProviderResponseError("") } + if o.thinkExtractOptions != nil && o.thinkExtractOptions.EnabledFor(thinkextract.SurfaceFrom(ctx)) { + thinkextract.TransformChatResponseForSurface(resp, *o.thinkExtractOptions, thinkextract.SurfaceFrom(ctx)) + } return resp, nil } @@ -488,15 +492,34 @@ func (o *InferenceOrchestrator) responsesProviderCall(ctx context.Context, req * if resp == nil { return nil, emptyProviderResponseError("") } + if o.thinkExtractOptions != nil && o.thinkExtractOptions.EnabledFor(thinkextract.SurfaceFrom(ctx)) { + thinkextract.TransformResponsesResponse(resp, *o.thinkExtractOptions) + } return resp, nil } func (o *InferenceOrchestrator) streamChatCompletionProviderCall(ctx context.Context, req *core.ChatRequest) (io.ReadCloser, error) { - return o.provider.StreamChatCompletion(ctx, req) + stream, err := o.provider.StreamChatCompletion(ctx, req) + if err != nil { + return nil, err + } + if stream != nil && o.thinkExtractOptions != nil && + o.thinkExtractOptions.EnabledFor(thinkextract.SurfaceFrom(ctx)) { + stream = thinkextract.TransformStreamForSurface(stream, *o.thinkExtractOptions, thinkextract.SurfaceFrom(ctx)) + } + return stream, nil } func (o *InferenceOrchestrator) streamResponsesProviderCall(ctx context.Context, req *core.ResponsesRequest) (io.ReadCloser, error) { - return o.provider.StreamResponses(ctx, req) + stream, err := o.provider.StreamResponses(ctx, req) + if err != nil { + return nil, err + } + if stream != nil && o.thinkExtractOptions != nil && + o.thinkExtractOptions.EnabledFor(thinkextract.SurfaceFrom(ctx)) { + stream = thinkextract.TransformResponsesStream(stream, *o.thinkExtractOptions) + } + return stream, nil } func emptyProviderResponseError(providerType string) *core.GatewayError { diff --git a/internal/gateway/inference_orchestrator.go b/internal/gateway/inference_orchestrator.go index 0244796e0..fc95448f8 100644 --- a/internal/gateway/inference_orchestrator.go +++ b/internal/gateway/inference_orchestrator.go @@ -7,6 +7,7 @@ import ( "github.com/enterpilot/gomodel/internal/core" "github.com/enterpilot/gomodel/internal/streaming" + "github.com/enterpilot/gomodel/internal/thinkextract" "github.com/enterpilot/gomodel/internal/usage" ) @@ -29,6 +30,7 @@ type InferenceConfig struct { PricingResolver usage.PricingResolver RouteGate RouteGate GuardrailsHash string + ThinkExtractOptions *thinkextract.Options // Optional: legacy block translation; nil disables the feature } // InferenceOrchestrator owns translated inference workflow resolution, request @@ -44,6 +46,7 @@ type InferenceOrchestrator struct { pricingResolver usage.PricingResolver routeGate RouteGate guardrailsHash string + thinkExtractOptions *thinkextract.Options } // NewInferenceOrchestrator creates a translated inference orchestrator. @@ -59,6 +62,7 @@ func NewInferenceOrchestrator(cfg InferenceConfig) *InferenceOrchestrator { pricingResolver: cfg.PricingResolver, routeGate: cfg.RouteGate, guardrailsHash: cfg.GuardrailsHash, + thinkExtractOptions: cfg.ThinkExtractOptions, } } diff --git a/internal/gateway/thinkextract_integration_test.go b/internal/gateway/thinkextract_integration_test.go new file mode 100644 index 000000000..aca5dc8fa --- /dev/null +++ b/internal/gateway/thinkextract_integration_test.go @@ -0,0 +1,253 @@ +package gateway + +import ( + "context" + "io" + "strings" + "testing" + + "github.com/enterpilot/gomodel/internal/core" + "github.com/enterpilot/gomodel/internal/thinkextract" +) + +// thinkProvider is a mock provider that returns a fixed response and stream +// for each endpoint. Used to verify the thinkextract orchestrator hooks +// without touching real providers. +type thinkProvider struct { + chatResponse *core.ChatResponse + chatStream io.ReadCloser + responsesResp *core.ResponsesResponse + responsesStream io.ReadCloser +} + +func (p *thinkProvider) ChatCompletion(_ context.Context, _ *core.ChatRequest) (*core.ChatResponse, error) { + return p.chatResponse, nil +} + +func (p *thinkProvider) StreamChatCompletion(_ context.Context, _ *core.ChatRequest) (io.ReadCloser, error) { + return p.chatStream, nil +} + +func (p *thinkProvider) Responses(_ context.Context, _ *core.ResponsesRequest) (*core.ResponsesResponse, error) { + return p.responsesResp, nil +} + +func (p *thinkProvider) StreamResponses(_ context.Context, _ *core.ResponsesRequest) (io.ReadCloser, error) { + return p.responsesStream, nil +} + +func (p *thinkProvider) ListModels(_ context.Context) (*core.ModelsResponse, error) { + return &core.ModelsResponse{}, nil +} + +func (p *thinkProvider) Embeddings(_ context.Context, _ *core.EmbeddingRequest) (*core.EmbeddingResponse, error) { + return nil, nil +} + +func (p *thinkProvider) GetProviderType(_ string) string { + return "think" +} + +func (p *thinkProvider) GetName() string { + return "think" +} + +func (p *thinkProvider) Supports(_ string) bool { + return true +} + +func TestOrchestratorChatCompletion_ThinkExtracted(t *testing.T) { + provider := &thinkProvider{ + chatResponse: &core.ChatResponse{ + Choices: []core.Choice{{ + Message: core.ResponseMessage{Role: "assistant", Content: "answerreasoning rest"}, + }}, + }, + } + o := NewInferenceOrchestrator(InferenceConfig{ + Provider: provider, + ThinkExtractOptions: &thinkextract.Options{}, + }) + resp, _, _, _, _, err := o.DispatchChatCompletion(context.Background(), nil, &core.ChatRequest{Model: "test"}) + if err != nil { + t.Fatalf("DispatchChatCompletion: %v", err) + } + msg := resp.Choices[0].Message + if content, _ := msg.Content.(string); content != "answer rest" { + t.Errorf("content=%q, want %q", content, "answer rest") + } + raw := msg.ExtraFields.Lookup("reasoning_content") + if len(raw) == 0 { + t.Fatalf("reasoning_content missing") + } +} + +func TestOrchestratorChatCompletion_DisabledOnChatSurface(t *testing.T) { + provider := &thinkProvider{ + chatResponse: &core.ChatResponse{ + Choices: []core.Choice{{ + Message: core.ResponseMessage{Role: "assistant", Content: "axb"}, + }}, + }, + } + off := false + o := NewInferenceOrchestrator(InferenceConfig{ + Provider: provider, + ThinkExtractOptions: &thinkextract.Options{ChatEnabled: &off}, + }) + resp, _, _, _, _, err := o.DispatchChatCompletion(thinkextract.WithSurface(context.Background(), thinkextract.SurfaceChat), nil, &core.ChatRequest{Model: "test"}) + if err != nil { + t.Fatalf("DispatchChatCompletion: %v", err) + } + msg := resp.Choices[0].Message + if content, _ := msg.Content.(string); content != "axb" { + t.Errorf("content rewritten despite chat disabled: %q", content) + } + if len(msg.ExtraFields.Lookup("reasoning_content")) > 0 { + t.Errorf("reasoning set despite chat disabled") + } +} + +func TestOrchestratorStreamChatCompletion_ThinkExtracted(t *testing.T) { + input := "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"arb\"}}]}\n\ndata: [DONE]\n" + provider := &thinkProvider{ + chatStream: io.NopCloser(strings.NewReader(input)), + } + o := NewInferenceOrchestrator(InferenceConfig{ + Provider: provider, + ThinkExtractOptions: &thinkextract.Options{}, + }) + stream, err := o.StreamChatCompletion(context.Background(), nil, &core.ChatRequest{Model: "test", Stream: true}) + if err != nil { + t.Fatalf("StreamChatCompletion: %v", err) + } + defer stream.Stream.Close() + out, err := io.ReadAll(stream.Stream) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(string(out), "reasoning_content") { + t.Errorf("stream not rewritten: %q", string(out)) + } +} + +func TestOrchestratorResponses_ThinkExtracted(t *testing.T) { + provider := &thinkProvider{ + responsesResp: &core.ResponsesResponse{ + Output: []core.ResponsesOutputItem{{ + ID: "msg_1", + Type: "message", + Content: []core.ResponsesContentItem{ + {Type: "output_text", Text: "axb"}, + }, + }}, + }, + } + o := NewInferenceOrchestrator(InferenceConfig{ + Provider: provider, + ThinkExtractOptions: &thinkextract.Options{}, + }) + resp, _, _, _, _, err := o.DispatchResponses(context.Background(), nil, &core.ResponsesRequest{Model: "test"}) + if err != nil { + t.Fatalf("DispatchResponses: %v", err) + } + if len(resp.Output) != 2 { + t.Fatalf("output items=%d, want 2 (reasoning + message)", len(resp.Output)) + } + if resp.Output[0].Type != "reasoning" { + t.Errorf("output[0].Type=%q, want reasoning", resp.Output[0].Type) + } +} + +func TestOrchestratorStreamResponses_ThinkExtracted(t *testing.T) { + input := "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"msg_1\",\"type\":\"message\"}}\n\n" + + "data: {\"type\":\"response.output_text.delta\",\"item_id\":\"msg_1\",\"output_index\":0,\"content_index\":0,\"delta\":\"axb\"}\n\n" + + "data: [DONE]\n" + provider := &thinkProvider{ + responsesStream: io.NopCloser(strings.NewReader(input)), + } + o := NewInferenceOrchestrator(InferenceConfig{ + Provider: provider, + ThinkExtractOptions: &thinkextract.Options{}, + }) + stream, err := o.StreamResponses(context.Background(), nil, &core.ResponsesRequest{Model: "test", Stream: true}) + if err != nil { + t.Fatalf("StreamResponses: %v", err) + } + defer stream.Stream.Close() + out, err := io.ReadAll(stream.Stream) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(string(out), `"type":"reasoning"`) { + t.Errorf("reasoning item not synthesized: %q", string(out)) + } +} + +func TestOrchestratorChatCompletion_NilThinkExtractOptions(t *testing.T) { + provider := &thinkProvider{ + chatResponse: &core.ChatResponse{ + Choices: []core.Choice{{ + Message: core.ResponseMessage{Role: "assistant", Content: "axb"}, + }}, + }, + } + o := NewInferenceOrchestrator(InferenceConfig{Provider: provider}) + resp, _, _, _, _, err := o.DispatchChatCompletion(context.Background(), nil, &core.ChatRequest{Model: "test"}) + if err != nil { + t.Fatalf("DispatchChatCompletion: %v", err) + } + msg := resp.Choices[0].Message + if content, _ := msg.Content.(string); content != "axb" { + t.Errorf("content rewritten with nil options: %q", content) + } +} + +func TestOrchestratorChatCompletion_MessagesSurfaceDefaultsOff(t *testing.T) { + provider := &thinkProvider{ + chatResponse: &core.ChatResponse{ + Choices: []core.Choice{{ + Message: core.ResponseMessage{Role: "assistant", Content: "axb"}, + }}, + }, + } + o := NewInferenceOrchestrator(InferenceConfig{ + Provider: provider, + ThinkExtractOptions: &thinkextract.Options{}, + }) + ctx := thinkextract.WithSurface(context.Background(), thinkextract.SurfaceMessages) + resp, _, _, _, _, err := o.DispatchChatCompletion(ctx, nil, &core.ChatRequest{Model: "test"}) + if err != nil { + t.Fatalf("DispatchChatCompletion: %v", err) + } + msg := resp.Choices[0].Message + if content, _ := msg.Content.(string); content != "axb" { + t.Errorf("content rewritten on messages surface despite default-off policy: %q", content) + } +} + +func TestOrchestratorChatCompletion_MessagesSurfaceUnsignedPolicy(t *testing.T) { + provider := &thinkProvider{ + chatResponse: &core.ChatResponse{ + Choices: []core.Choice{{ + Message: core.ResponseMessage{Role: "assistant", Content: "axb"}, + }}, + }, + } + o := NewInferenceOrchestrator(InferenceConfig{ + Provider: provider, + ThinkExtractOptions: &thinkextract.Options{MessagesPolicy: "unsigned"}, + }) + ctx := thinkextract.WithSurface(context.Background(), thinkextract.SurfaceMessages) + resp, _, _, _, _, err := o.DispatchChatCompletion(ctx, nil, &core.ChatRequest{Model: "test"}) + if err != nil { + t.Fatalf("DispatchChatCompletion: %v", err) + } + msg := resp.Choices[0].Message + if content, _ := msg.Content.(string); content != "ab" { + t.Errorf("content not rewritten on messages surface with unsigned policy: %q", content) + } + if len(msg.ExtraFields.Lookup(thinkextract.SynthesizedMarkerKey)) == 0 { + t.Errorf("synthesized marker missing on messages surface") + } +} diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 181fd751f..5ab603461 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -19,6 +19,7 @@ import ( "github.com/enterpilot/gomodel/internal/realtime" "github.com/enterpilot/gomodel/internal/responsecache" "github.com/enterpilot/gomodel/internal/responsestore" + "github.com/enterpilot/gomodel/internal/thinkextract" "github.com/enterpilot/gomodel/internal/usage" ) @@ -57,6 +58,7 @@ type Handler struct { guardrailsHash string storageProbe ReadinessProbe cacheProbe ReadinessProbe + thinkExtractOptions *thinkextract.Options translatedSvc *translatedInferenceService // snapshot of handler fields at first use; server.New sets cache/hash before traffic translatedSvcOnce sync.Once @@ -162,6 +164,7 @@ func (h *Handler) translatedInference() *translatedInferenceService { responseCache: h.responseCache, guardrailsHash: h.guardrailsHash, responseStore: h.currentResponseStore(), + thinkExtractOptions: h.thinkExtractOptions, } s.initHandlers() h.storesMu.Lock() diff --git a/internal/server/http.go b/internal/server/http.go index e73aa32d8..d4667d980 100644 --- a/internal/server/http.go +++ b/internal/server/http.go @@ -30,6 +30,7 @@ import ( "github.com/enterpilot/gomodel/internal/responsestore" "github.com/enterpilot/gomodel/internal/session" "github.com/enterpilot/gomodel/internal/tagging" + "github.com/enterpilot/gomodel/internal/thinkextract" "github.com/enterpilot/gomodel/internal/usage" ) @@ -117,6 +118,7 @@ type Config struct { RequestAuthenticators []ext.RequestAuthenticator // Optional extension-provided request authentication mechanisms Tagging *tagging.Service // Optional: request labelling based on configured tagging headers SessionDetector *session.Detector // Optional: client session identification for sticky routing and audit grouping + ThinkExtractOptions *thinkextract.Options // Optional: legacy block translation options; nil disables the feature } // ReadinessProbe verifies that a dependency the gateway owns is reachable. @@ -192,6 +194,7 @@ func New(provider core.RoutableProvider, cfg *Config) *Server { handler.guardrailsHash = cfg.GuardrailsHash handler.storageProbe = cfg.StorageProbe handler.cacheProbe = cfg.CacheProbe + handler.thinkExtractOptions = cfg.ThinkExtractOptions } if cfg != nil && cfg.EnabledPassthroughProviders != nil { handler.setEnabledPassthroughProviders(cfg.EnabledPassthroughProviders) diff --git a/internal/server/messages_handler.go b/internal/server/messages_handler.go index 8bf934dea..2fe0af859 100644 --- a/internal/server/messages_handler.go +++ b/internal/server/messages_handler.go @@ -10,6 +10,7 @@ import ( "github.com/enterpilot/gomodel/internal/anthropicapi" "github.com/enterpilot/gomodel/internal/auditlog" "github.com/enterpilot/gomodel/internal/core" + "github.com/enterpilot/gomodel/internal/thinkextract" ) // Messages handles POST /v1/messages. @@ -191,6 +192,7 @@ func (s *translatedInferenceService) CountMessageTokens(c *echo.Context) error { func (s *translatedInferenceService) dispatchMessages(c *echo.Context, req *core.ChatRequest, workflow *core.Workflow) error { s.observeLiveProviderAttempts(c, workflow) ctx := c.Request().Context() + ctx = thinkextract.WithSurface(ctx, thinkextract.SurfaceMessages) requestID := requestIDFromContextOrHeader(c.Request()) adm, err := enforceAdmission(c, s.rateLimiter, s.budgetChecker, @@ -219,7 +221,10 @@ func (s *translatedInferenceService) dispatchMessages(c *echo.Context, req *core result.Meta.FailoverModel, result.Stream, func(stream io.ReadCloser) io.ReadCloser { - converted := anthropicapi.NewStreamConverter(stream, model, anthropicapi.EstimateChatInputTokens(req)) + converted := anthropicapi.NewStreamConverterWithPolicy( + stream, model, anthropicapi.EstimateChatInputTokens(req), + s.messagesThinkingPolicy(), + ) return result.WrapDeliveryStream(ctx, converted) }, ) @@ -241,7 +246,19 @@ func (s *translatedInferenceService) dispatchMessages(c *echo.Context, req *core result.Meta.ProviderName, ) - return c.JSON(http.StatusOK, anthropicapi.FromChatResponse(result.Response)) + return c.JSON(http.StatusOK, anthropicapi.FromChatResponseWithPolicy( + result.Response, + s.messagesThinkingPolicy(), + )) +} + +// messagesThinkingPolicy returns the configured messages thinking-block +// policy, or off when the thinkextract feature is disabled entirely. +func (s *translatedInferenceService) messagesThinkingPolicy() thinkextract.MessagesThinkingPolicy { + if s.thinkExtractOptions == nil { + return thinkextract.MessagesPolicyOff + } + return thinkextract.ParseMessagesPolicy(s.thinkExtractOptions.MessagesPolicy) } // decodeMessagesChatRequest reads the request body, decodes the Anthropic diff --git a/internal/server/translated_inference_service.go b/internal/server/translated_inference_service.go index 7e51cfb6a..c8f69c6bb 100644 --- a/internal/server/translated_inference_service.go +++ b/internal/server/translated_inference_service.go @@ -25,6 +25,7 @@ import ( "github.com/enterpilot/gomodel/internal/responsecache" "github.com/enterpilot/gomodel/internal/responsestore" "github.com/enterpilot/gomodel/internal/streaming" + "github.com/enterpilot/gomodel/internal/thinkextract" "github.com/enterpilot/gomodel/internal/usage" ) @@ -48,6 +49,7 @@ type translatedInferenceService struct { responseStoreMu sync.RWMutex conversationStore conversationstore.Store conversationStoreMu sync.RWMutex + thinkExtractOptions *thinkextract.Options // snapshotWrites tracks background response snapshot writes so shutdown // can drain them before closing the response store. snapshotMu gates new // writes against the drain: a handler that outlives the HTTP drain window @@ -83,6 +85,7 @@ func (s *translatedInferenceService) newInferenceOrchestrator() *gateway.Inferen UsageLogger: s.usageLogger, PricingResolver: s.pricingResolver, GuardrailsHash: s.guardrailsHash, + ThinkExtractOptions: s.thinkExtractOptions, } // Guarded assignment keeps the gate nil when rate limits are off (a nil // RateLimiter assigned unconditionally would arrive as a typed non-nil @@ -104,6 +107,7 @@ func (s *translatedInferenceService) handleChatCompletion(c *echo.Context) error func (s *translatedInferenceService) dispatchChatCompletion(c *echo.Context, req *core.ChatRequest, workflow *core.Workflow) error { s.observeLiveProviderAttempts(c, workflow) ctx := c.Request().Context() + ctx = thinkextract.WithSurface(ctx, thinkextract.SurfaceChat) requestID := requestIDFromContextOrHeader(c.Request()) adm, err := enforceAdmission(c, s.rateLimiter, s.budgetChecker, @@ -290,6 +294,7 @@ func handleWithCache[R any]( func (s *translatedInferenceService) dispatchResponses(c *echo.Context, req *core.ResponsesRequest, workflow *core.Workflow) error { s.observeLiveProviderAttempts(c, workflow) ctx := c.Request().Context() + ctx = thinkextract.WithSurface(ctx, thinkextract.SurfaceResponses) requestID := requestIDFromContextOrHeader(c.Request()) adm, err := enforceAdmission(c, s.rateLimiter, s.budgetChecker, diff --git a/internal/thinkextract/chat.go b/internal/thinkextract/chat.go new file mode 100644 index 000000000..a45cace67 --- /dev/null +++ b/internal/thinkextract/chat.go @@ -0,0 +1,164 @@ +package thinkextract + +import ( + "encoding/json" + "strings" + + "github.com/enterpilot/gomodel/internal/core" +) + +// MessagesThinkingPolicy is the policy applied on the Anthropic messages +// surface to synthesized-from-tags reasoning. Valid values are the empty +// string (treated as Off), "off", "unsigned", and "redacted". +type MessagesThinkingPolicy string + +const ( + MessagesPolicyOff MessagesThinkingPolicy = "off" + MessagesPolicyUnsigned MessagesThinkingPolicy = "unsigned" + MessagesPolicyRedacted MessagesThinkingPolicy = "redacted" +) + +// ParseMessagesPolicy normalizes a raw config value. Unknown values fall back +// to off so a typo cannot silently change wire behaviour. +func ParseMessagesPolicy(raw string) MessagesThinkingPolicy { + switch MessagesThinkingPolicy(raw) { + case MessagesPolicyUnsigned, MessagesPolicyRedacted: + return MessagesThinkingPolicy(raw) + default: + return MessagesPolicyOff + } +} + +// SynthesizedMarkerKey is the ExtraFields key that thinkextract sets on a +// chat response message when its reasoning_content came from tag extraction. +// The messages converters read it to apply MessagesThinkingPolicy; the +// marker is only emitted on the messages surface so chat-surface responses +// never carry it. +const SynthesizedMarkerKey = "thinkextract_synthesized" + +// TransformChatResponse rewrites every choice of resp whose message text +// carries legacy think-block tags: the tags and their bodies move from the +// message content into ExtraFields["reasoning_content"], leaving the visible +// text as content. Choices that already carry a reasoning_content field are +// left untouched — upstream-structured reasoning always wins over extracted +// tags. +// +// The function returns the number of choices rewritten, so callers can skip +// downstream bookkeeping when nothing changed. +// +// On the messages surface, extracted reasoning is also marked via the +// SynthesizedMarkerKey so the Anthropic dialect converter can apply +// MessagesThinkingPolicy without affecting native provider reasoning. +func TransformChatResponse(resp *core.ChatResponse, opts Options) int { + return TransformChatResponseForSurface(resp, opts, "") +} + +// TransformChatResponseForSurface is TransformChatResponse with an explicit +// surface so synthesized reasoning on the messages surface can be marked. +func TransformChatResponseForSurface(resp *core.ChatResponse, opts Options, surface Surface) int { + if resp == nil { + return 0 + } + markSynthesized := surface == SurfaceMessages + rewritten := 0 + for i := range resp.Choices { + if transformMessage(&resp.Choices[i].Message, opts) { + rewritten++ + if markSynthesized { + markSynthesizedOnMessage(&resp.Choices[i].Message) + } + } + } + return rewritten +} + +// markSynthesizedOnMessage sets the SynthesizedMarkerKey ExtraFields flag on +// the message so the messages converter can apply MessagesThinkingPolicy. +func markSynthesizedOnMessage(msg *core.ResponseMessage) { + raw, err := json.Marshal(true) + if err != nil { + return + } + merged, err := core.MergeUnknownJSONFields(msg.ExtraFields, map[string]json.RawMessage{ + SynthesizedMarkerKey: raw, + }) + if err != nil { + return + } + msg.ExtraFields = merged +} + +// transformMessage applies the think-block extraction to one response +// message. It returns true when the message was rewritten. +func transformMessage(msg *core.ResponseMessage, opts Options) bool { + if msg == nil { + return false + } + if len(msg.ExtraFields.Lookup(FieldReasoning)) > 0 { + return false + } + switch content := msg.Content.(type) { + case string: + cleaned, reasoning, found := Extract(content, opts) + if !found { + return false + } + msg.Content = cleaned + setReasoning(msg, reasoning) + return true + case []core.ContentPart: + return transformContentParts(msg, content, opts) + default: + return false + } +} + +// transformContentParts extracts think blocks from every text part of a +// structured content array. Non-text parts are untouched. +func transformContentParts(msg *core.ResponseMessage, parts []core.ContentPart, opts Options) bool { + var reasoning strings.Builder + changed := false + for i := range parts { + if parts[i].Type != "text" || parts[i].Text == "" { + continue + } + cleaned, reason, found := Extract(parts[i].Text, opts) + if !found { + continue + } + parts[i].Text = cleaned + if reason != "" { + if reasoning.Len() > 0 { + reasoning.WriteString("\n\n") + } + reasoning.WriteString(reason) + } + changed = true + } + if !changed { + return false + } + msg.Content = parts + setReasoning(msg, reasoning.String()) + return true +} + +// setReasoning stores the extracted reasoning text on the message's extra +// fields, preserving every other unknown field the upstream set. +func setReasoning(msg *core.ResponseMessage, reasoning string) bool { + if reasoning == "" { + return false + } + encoded, err := json.Marshal(reasoning) + if err != nil { + return false + } + merged, err := core.MergeUnknownJSONFields(msg.ExtraFields, map[string]json.RawMessage{ + FieldReasoning: encoded, + }) + if err != nil { + return false + } + msg.ExtraFields = merged + return true +} \ No newline at end of file diff --git a/internal/thinkextract/chat_test.go b/internal/thinkextract/chat_test.go new file mode 100644 index 000000000..349fd64c7 --- /dev/null +++ b/internal/thinkextract/chat_test.go @@ -0,0 +1,165 @@ +package thinkextract + +import ( + "encoding/json" + "testing" + + "github.com/enterpilot/gomodel/internal/core" +) + +func newResp(content string, extra core.UnknownJSONFields) *core.ChatResponse { + return &core.ChatResponse{ + Choices: []core.Choice{{ + Message: core.ResponseMessage{ + Role: "assistant", + Content: content, + ExtraFields: extra, + }, + }}, + } +} + +func TestTransformChatResponse_NoTags(t *testing.T) { + resp := newResp("hello world", core.UnknownJSONFields{}) + if got := TransformChatResponse(resp, Options{}); got != 0 { + t.Fatalf("rewritten=%d, want 0", got) + } + if resp.Choices[0].Message.Content != "hello world" { + t.Errorf("content mutated: %v", resp.Choices[0].Message.Content) + } + if len(resp.Choices[0].Message.ExtraFields.Lookup(FieldReasoning)) > 0 { + t.Errorf("reasoning unexpectedly set") + } +} + +func TestTransformChatResponse_StringContent(t *testing.T) { + resp := newResp("answerhidden rest", core.UnknownJSONFields{}) + if got := TransformChatResponse(resp, Options{}); got != 1 { + t.Fatalf("rewritten=%d, want 1", got) + } + if content, _ := resp.Choices[0].Message.Content.(string); content != "answer rest" { + t.Errorf("content=%q, want %q", content, "answer rest") + } + raw := resp.Choices[0].Message.ExtraFields.Lookup(FieldReasoning) + if len(raw) == 0 { + t.Fatalf("reasoning_content missing") + } + var got string + if err := jsonUnmarshalString(raw, &got); err != nil { + t.Fatalf("unmarshal reasoning: %v", err) + } + if got != "hidden" { + t.Errorf("reasoning=%q, want %q", got, "hidden") + } +} + +func TestTransformChatResponse_PreservesExistingExtra(t *testing.T) { + resp := newResp("abc", core.UnknownJSONFields{}) + // Add a known extra field first. + merged, err := core.MergeUnknownJSONFields(resp.Choices[0].Message.ExtraFields, map[string]json.RawMessage{ + "x_custom": json.RawMessage(`"keepme"`), + }) + if err != nil { + t.Fatalf("merge: %v", err) + } + resp.Choices[0].Message.ExtraFields = merged + + if got := TransformChatResponse(resp, Options{}); got != 1 { + t.Fatalf("rewritten=%d, want 1", got) + } + raw := resp.Choices[0].Message.ExtraFields.Lookup("x_custom") + if len(raw) == 0 { + t.Fatalf("x_custom lost during rewrite") + } + var got string + if err := jsonUnmarshalString(raw, &got); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if got != "keepme" { + t.Errorf("x_custom=%q, want %q", got, "keepme") + } +} + +func TestTransformChatResponse_UpstreamReasoningWins(t *testing.T) { + resp := newResp("abc", core.UnknownJSONFields{}) + merged, _ := core.MergeUnknownJSONFields(resp.Choices[0].Message.ExtraFields, map[string]json.RawMessage{ + FieldReasoning: json.RawMessage(`"upstream"`), + }) + resp.Choices[0].Message.ExtraFields = merged + + if got := TransformChatResponse(resp, Options{}); got != 0 { + t.Fatalf("rewritten=%d, want 0 (upstream reasoning preserved)", got) + } + if content, _ := resp.Choices[0].Message.Content.(string); content != "abc" { + t.Errorf("content mutated: %q", content) + } +} + +func TestTransformChatResponse_ContentParts(t *testing.T) { + resp := &core.ChatResponse{ + Choices: []core.Choice{{ + Message: core.ResponseMessage{ + Role: "assistant", + Content: []core.ContentPart{ + {Type: "text", Text: "beforex"}, + {Type: "image_url", ImageURL: &core.ImageURLContent{URL: "data:image/png;base64,xxx"}}, + {Type: "text", Text: "yafter"}, + }, + }, + }}, + } + if got := TransformChatResponse(resp, Options{}); got != 1 { + t.Fatalf("rewritten=%d, want 1", got) + } + parts := resp.Choices[0].Message.Content.([]core.ContentPart) + if parts[0].Text != "before" { + t.Errorf("part[0].Text=%q, want %q", parts[0].Text, "before") + } + if parts[1].Type != "image_url" { + t.Errorf("part[1] type=%q, want image_url", parts[1].Type) + } + if parts[2].Text != "after" { + t.Errorf("part[2].Text=%q, want %q", parts[2].Text, "after") + } + raw := resp.Choices[0].Message.ExtraFields.Lookup(FieldReasoning) + if len(raw) == 0 { + t.Fatalf("reasoning missing") + } + var got string + if err := jsonUnmarshalString(raw, &got); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if got != "x\n\ny" { + t.Errorf("reasoning=%q, want %q", got, "x\n\ny") + } +} + +func TestTransformChatResponse_NilResponse(t *testing.T) { + if got := TransformChatResponse(nil, Options{}); got != 0 { + t.Errorf("nil response: got=%d, want 0", got) + } +} + +func TestTransformChatResponse_MultipleChoices(t *testing.T) { + resp := &core.ChatResponse{ + Choices: []core.Choice{ + {Message: core.ResponseMessage{Role: "assistant", Content: "axb"}}, + {Message: core.ResponseMessage{Role: "assistant", Content: "plain"}}, + {Message: core.ResponseMessage{Role: "assistant", Content: "cyd"}}, + }, + } + if got := TransformChatResponse(resp, Options{}); got != 2 { + t.Fatalf("rewritten=%d, want 2", got) + } + for i, want := range []string{"ab", "plain", "cd"} { + if c, _ := resp.Choices[i].Message.Content.(string); c != want { + t.Errorf("choice[%d].Content=%q, want %q", i, c, want) + } + } +} + +// jsonUnmarshalString unmarshals a JSON string into a string value. +// Standard for unmarshalling json.RawMessage that contains a JSON string. +func jsonUnmarshalString(raw []byte, out *string) error { + return json.Unmarshal(raw, out) +} \ No newline at end of file diff --git a/internal/thinkextract/responses.go b/internal/thinkextract/responses.go new file mode 100644 index 000000000..8c232d64f --- /dev/null +++ b/internal/thinkextract/responses.go @@ -0,0 +1,93 @@ +package thinkextract + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "strings" + + "github.com/enterpilot/gomodel/internal/core" +) + +// TransformResponsesResponse rewrites each output item of resp whose message +// content carries legacy reasoning tags. For every item rewritten, a fresh +// reasoning output item is prepended in place and the message item keeps +// only the cleaned text. Items that already carry a reasoning item ahead of +// them are left untouched. +// +// Returns the number of message items rewritten. The function is the +// Responses API analogue of TransformChatResponse; both use the same reasoning +// item shape that the gateway emits for native reasoning providers. +func TransformResponsesResponse(resp *core.ResponsesResponse, opts Options) int { + if resp == nil { + return 0 + } + rewritten := 0 + for i := 0; i < len(resp.Output); i++ { + item := &resp.Output[i] + cleaned, reasoning, changed := transformResponsesOutputItem(item, opts) + if !changed { + continue + } + item.Content = cleaned + rewritten++ + reasonItem := core.ResponsesOutputItem{ + ID: "rs_" + shortID(), + Type: "reasoning", + Status: "completed", + Content: []core.ResponsesContentItem{ + {Type: "reasoning_text", Text: reasoning}, + }, + ExtraFields: core.UnknownJSONFieldsFromMap(map[string]json.RawMessage{ + "summary": json.RawMessage(`[]`), + }), + } + resp.Output = append(resp.Output[:i], append([]core.ResponsesOutputItem{reasonItem}, resp.Output[i:]...)...) + i++ // // skip the inserted item on the next iteration + } + return rewritten +} + +// transformResponsesOutputItem rewrites the text content of a single +// Responses output item. It returns the cleaned Content slice, the +// concatenated reasoning text, and a changed flag. +func transformResponsesOutputItem(item *core.ResponsesOutputItem, opts Options) ([]core.ResponsesContentItem, string, bool) { + if item == nil || item.Type != "message" || len(item.Content) == 0 { + return nil, "", false + } + cleaned := make([]core.ResponsesContentItem, len(item.Content)) + copy(cleaned, item.Content) + var reasoning strings.Builder + changed := false + for ci := range cleaned { + part := &cleaned[ci] + if part.Type != "output_text" || part.Text == "" { + continue + } + newText, body, found := Extract(part.Text, opts) + if !found { + continue + } + part.Text = newText + if body != "" { + if reasoning.Len() > 0 { + reasoning.WriteString("\n\n") + } + reasoning.WriteString(body) + } + changed = true + } + return cleaned, reasoning.String(), changed +} + +// shortID returns a short opaque hex identifier suitable for reasoning item +// IDs on synthesized Responses output. Real production code should use +// uuid.NewString; crypto/rand is used here so the package stays +// dependency-free for tests. +func shortID() string { + var b [8]byte + if _, err := rand.Read(b[:]); err != nil { + return "fallback" + } + return hex.EncodeToString(b[:]) +} \ No newline at end of file diff --git a/internal/thinkextract/responses_stream.go b/internal/thinkextract/responses_stream.go new file mode 100644 index 000000000..1c12399c3 --- /dev/null +++ b/internal/thinkextract/responses_stream.go @@ -0,0 +1,321 @@ +package thinkextract + +import ( + "bufio" + "bytes" + "encoding/json" + "io" + "strings" +) + +// TransformResponsesStream wraps an SSE stream of OpenAI Responses API events +// and rewrites any output_text delta carrying legacy think-block tags into a +// synthesized reasoning item plus reasoning_text deltas. Non-Responses events +// pass through unchanged. A data: [DONE] sentinel terminates the stream. +// +// The transformer tracks the current output item per response. On the first +// delta that opens a reasoning block it synthesizes a reasoning output item +// (response.output_item.added with type=reasoning, id=rs_*, status=in_progress), +// emits response.reasoning_text.delta events for the reasoning body, closes the +// item with response.output_item.done, and re-emits the remaining text delta as +// response.output_text.delta on the original item. Subsequent reasoning blocks +// in the same item reuse the same synthesized reasoning item ID so the client +// sees one contiguous reasoning span. +// +// A reasoning block that opens but never closes within the stream is +// forwarded verbatim — the tag stays in the output text so nothing is lost. +// When the upstream stream ends without a [DONE] sentinel the transformer +// flushes any still-open reasoning item as ordinary text so the client sees +// the bytes the model produced. +func TransformResponsesStream(in io.ReadCloser, opts Options) io.ReadCloser { + o := opts.withDefaults() + pr, pw := io.Pipe() + go transformResponsesLoop(in, pw, o) + return pr +} + +// responsesStreamState tracks the current output item so the transformer +// can route reasoning deltas to a synthesized reasoning item and the +// remaining text to the original message item. +type responsesStreamState struct { + opts Options + // state is the tag-scanner state for the currently open message item. + // A new state is created whenever the output item changes (the Responses + // protocol emits output_item.added before deltas). + state *State + // itemID is the ID of the current output item from the upstream stream. + itemID string + // itemIndex is the output_index of the current output item. + itemIndex int + // reasoningItemID is the ID assigned to the synthesized reasoning item. + reasoningItemID string + // reasoningOpen is true while a reasoning block is being synthesized + // (between output_item.added and output_item.done). + reasoningOpen bool + // reasoningText accumulates the reasoning body so output_item.done can + // carry the full text in a single reasoning_text part. + reasoningText strings.Builder +} + +// transformResponsesLoop reads SSE lines from in and rewrites any event +// carrying legacy think-block tags. It is the Responses API counterpart of +// transformLoop. +func transformResponsesLoop(in io.ReadCloser, pw *io.PipeWriter, o Options) { + defer pw.Close() + defer in.Close() + + st := &responsesStreamState{opts: o} + scanner := bufio.NewScanner(in) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + + for scanner.Scan() { + line := append([]byte(nil), scanner.Bytes()...) + trimmed := bytes.TrimLeft(line, " \t") + if !bytes.HasPrefix(trimmed, []byte("data:")) { + if !writeLine(pw, line) { + return + } + continue + } + payload := bytes.TrimPrefix(trimmed, []byte("data:")) + payload = bytes.TrimPrefix(payload, []byte(" ")) + + if string(payload) == "[DONE]" { + flushResponsesState(pw, st) + if !writeLine(pw, line) { + return + } + continue + } + + events, err := rewriteResponsesEvent(payload, st) + if err != nil { + if !writeLine(pw, line) { + return + } + continue + } + for _, ev := range events { + if !writeEvent(pw, ev) { + return + } + } + } + flushResponsesState(pw, st) +} + +// responsesEvent is the parsed shape of a Responses API SSE event. Only the +// fields the transformer reads are typed; everything else is preserved as +// raw JSON so re-emission is byte-exact for unmodified events. +type responsesEvent struct { + Type string `json:"type"` + ItemID string `json:"item_id,omitempty"` + OutputIndex int `json:"output_index,omitempty"` + ContentIndex int `json:"content_index,omitempty"` + Delta string `json:"delta,omitempty"` + Item *responsesItemInfo `json:"item,omitempty"` + RawExtra json.RawMessage `json:"-"` // rest of the event, untouched +} + +// responsesItemInfo is the subset of the `item` payload on +// response.output_item.added / done that the transformer reads. +type responsesItemInfo struct { + ID string `json:"id,omitempty"` + Type string `json:"type,omitempty"` + Status string `json:"status,omitempty"` + Role string `json:"role,omitempty"` + Content []responsesContentPart `json:"content,omitempty"` +} + +// responsesContentPart is one element of an item's content array. +type responsesContentPart struct { + Type string `json:"type"` + Text string `json:"text,omitempty"` +} + +// rewriteResponsesEvent decodes one Responses API SSE event payload and +// returns the replacement events to emit. A single event can yield 0..N +// output events (synthesized reasoning item events plus the original event +// with cleaned text, or just the original event unchanged). +func rewriteResponsesEvent(payload []byte, st *responsesStreamState) ([][]byte, error) { + var ev responsesEvent + if err := json.Unmarshal(payload, &ev); err != nil { + return nil, err + } + + switch ev.Type { + case "response.output_item.added": + // Track the current item so deltas know which state to use. + if ev.Item != nil { + st.itemID = ev.Item.ID + st.itemIndex = ev.OutputIndex + st.state = NewState(st.opts) + } + return [][]byte{payload}, nil + + case "response.output_item.done": + // The upstream item is closing. If we synthesized a reasoning item + // for it, emit the reasoning item's done event first, then forward + // the upstream done. + if st.reasoningOpen { + done := buildReasoningDoneEvent(st) + st.reasoningOpen = false + st.reasoningItemID = "" + st.reasoningText.Reset() + return [][]byte{done, payload}, nil + } + return [][]byte{payload}, nil + + case "response.output_text.delta": + return rewriteTextDelta(ev, payload, st), nil + + default: + return [][]byte{payload}, nil + } +} + +// rewriteTextDelta handles one response.output_text.delta event. It feeds the +// delta into the state scanner and emits reasoning item events plus the +// cleaned text delta as needed. +func rewriteTextDelta(ev responsesEvent, original []byte, st *responsesStreamState) [][]byte { + if st.state == nil { + st.state = NewState(st.opts) + } + cd, rd := st.state.Feed(ev.Delta) + if cd == "" && rd == "" { + // Fully buffered mid-tag; emit nothing until more text arrives. + return nil + } + + var out [][]byte + + if rd != "" { + if !st.reasoningOpen { + st.reasoningItemID = "rs_" + shortID() + st.reasoningOpen = true + st.reasoningText.Reset() + out = append(out, buildReasoningAddedEvent(st)) + } + st.reasoningText.WriteString(rd) + out = append(out, buildReasoningTextDeltaEvent(st, rd)) + } + + if cd != "" { + // Cleaned text delta on the original item. + var cleanedEv responsesEvent + if err := json.Unmarshal(original, &cleanedEv); err == nil { + cleanedEv.Delta = cd + if encoded, err := json.Marshal(&cleanedEv); err == nil { + out = append(out, encoded) + } + } + } + + return out +} + +// flushResponsesState drains the tag state at stream end. Any unclosed block +// becomes ordinary text on the current item so nothing is dropped. +func flushResponsesState(pw *io.PipeWriter, st *responsesStreamState) { + if st.state == nil { + return + } + cd, rd := st.state.Flush() + if rd != "" { + if !st.reasoningOpen { + st.reasoningItemID = "rs_" + shortID() + st.reasoningOpen = true + if !writeEvent(pw, buildReasoningAddedEvent(st)) { + return + } + } + if !writeEvent(pw, buildReasoningTextDeltaEvent(st, rd)) { + return + } + } + if st.reasoningOpen { + if !writeEvent(pw, buildReasoningDoneEvent(st)) { + return + } + st.reasoningOpen = false + st.reasoningItemID = "" + st.reasoningText.Reset() + } + if cd != "" { + // Residual text after any open tag closed. Emit as ordinary text on + // the current item so the client sees the bytes. + ev := map[string]any{ + "type": "response.output_text.delta", + "item_id": st.itemID, + "output_index": st.itemIndex, + "content_index": 0, + "delta": cd, + } + if payload, err := json.Marshal(ev); err == nil { + _ = writeEvent(pw, payload) + } + } +} + +// buildReasoningAddedEvent synthesizes response.output_item.added for a +// reasoning item. The summary array is empty by design — the reasoning text +// is the raw model output, not a summary. +func buildReasoningAddedEvent(st *responsesStreamState) []byte { + ev := map[string]any{ + "type": "response.output_item.added", + "output_index": st.itemIndex, + "item": map[string]any{ + "id": st.reasoningItemID, + "type": "reasoning", + "status": "in_progress", + "summary": []any{}, + "content": []any{}, + }, + } + b, err := json.Marshal(ev) + if err != nil { + return []byte(`{"type":"response.output_item.added"}`) + } + return b +} + +// buildReasoningTextDeltaEvent synthesizes response.reasoning_text.delta for +// one chunk of reasoning text. +func buildReasoningTextDeltaEvent(st *responsesStreamState, delta string) []byte { + ev := map[string]any{ + "type": "response.reasoning_text.delta", + "item_id": st.reasoningItemID, + "output_index": st.itemIndex, + "content_index": 0, + "delta": delta, + } + b, err := json.Marshal(ev) + if err != nil { + return []byte(`{"type":"response.reasoning_text.delta","delta":""}`) + } + return b +} + +// buildReasoningDoneEvent synthesizes response.output_item.done for the +// synthesized reasoning item, carrying the full reasoning text in a single +// reasoning_text part. +func buildReasoningDoneEvent(st *responsesStreamState) []byte { + ev := map[string]any{ + "type": "response.output_item.done", + "output_index": st.itemIndex, + "item": map[string]any{ + "id": st.reasoningItemID, + "type": "reasoning", + "status": "completed", + "summary": []any{}, + "content": []any{ + map[string]any{"type": "reasoning_text", "text": st.reasoningText.String()}, + }, + }, + } + b, err := json.Marshal(ev) + if err != nil { + return []byte(`{"type":"response.output_item.done"}`) + } + return b +} diff --git a/internal/thinkextract/responses_stream_test.go b/internal/thinkextract/responses_stream_test.go new file mode 100644 index 000000000..654184bc6 --- /dev/null +++ b/internal/thinkextract/responses_stream_test.go @@ -0,0 +1,206 @@ +package thinkextract + +import ( + "io" + "strings" + "testing" +) + +const responsesStreamBasic = `data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_1","type":"message","status":"in_progress"}} + +data: {"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"content_index":0,"delta":"answerx rest"} + +data: {"type":"response.output_item.done","output_index":0,"item":{"id":"msg_1","type":"message","status":"completed","content":[{"type":"output_text","text":"answerx rest"}]}} + +data: [DONE] +` + +func TestTransformResponsesStream_Basic(t *testing.T) { + rc := TransformResponsesStream(io.NopCloser(strings.NewReader(responsesStreamBasic)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + // Reasoning item added, text delta with cleaned text, reasoning delta, + // reasoning done, original done, [DONE]. + if !strings.Contains(out, `"type":"reasoning"`) { + t.Errorf("reasoning item not synthesized: %q", out) + } + if !strings.Contains(out, `"reasoning_text"`) { + t.Errorf("reasoning_text not emitted: %q", out) + } + if !strings.Contains(out, `"delta":"answer rest"`) && !strings.Contains(out, `"delta":"answer rest"`) { + t.Errorf("cleaned text delta missing: %q", out) + } + if !strings.Contains(out, "[DONE]") { + t.Errorf("missing [DONE]: %q", out) + } +} + +func TestTransformResponsesStream_NoTags(t *testing.T) { + input := `data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_1","type":"message"}} + +data: {"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"content_index":0,"delta":"plain text"} + +data: [DONE] +` + rc := TransformResponsesStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "plain text") { + t.Errorf("plain text lost: %q", out) + } + if strings.Contains(out, `"type":"reasoning"`) { + t.Errorf("reasoning item synthesized for plain text: %q", out) + } +} + +func TestTransformResponsesStream_NonDeltaEventsPassThrough(t *testing.T) { + input := `data: {"type":"response.created","response":{"id":"resp_1"}} + +data: {"type":"response.in_progress","response":{"id":"resp_1"}} + +data: [DONE] +` + rc := TransformResponsesStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "response.created") { + t.Errorf("created event lost: %q", out) + } + if !strings.Contains(out, "response.in_progress") { + t.Errorf("in_progress event lost: %q", out) + } +} + +func TestTransformResponsesStream_OnlyReasoning(t *testing.T) { + input := `data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_1","type":"message"}} + +data: {"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"content_index":0,"delta":"all"} + +data: {"type":"response.output_item.done","output_index":0,"item":{"id":"msg_1","type":"message","status":"completed","content":[{"type":"output_text","text":"all"}]}} + +data: [DONE] +` + rc := TransformResponsesStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, `"delta":"all"`) && !strings.Contains(out, `"text":"all"`) { + t.Errorf("reasoning body missing: %q", out) + } + // Empty content delta is dropped; the original done event still flows. + if !strings.Contains(out, `"type":"response.output_item.done"`) { + t.Errorf("done event lost: %q", out) + } +} + +func TestTransformResponsesStream_EmptyDeltaSkipped(t *testing.T) { + input := `data: {"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"content_index":0,"delta":""} + +data: [DONE] +` + rc := TransformResponsesStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "[DONE]") { + t.Errorf("missing [DONE]: %q", out) + } +} + +func TestTransformResponsesStream_MalformedForwarded(t *testing.T) { + input := "data: {not json}\n\ndata: [DONE]\n" + rc := TransformResponsesStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "{not json}") { + t.Errorf("malformed event not forwarded: %q", out) + } +} + +func TestTransformResponsesStream_NonDataLinesForwarded(t *testing.T) { + input := ": comment\n\nevent: response.created\nid: 1\n\ndata: [DONE]\n" + rc := TransformResponsesStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + for _, want := range []string{": comment", "event: response.created", "id: 1", "[DONE]"} { + if !strings.Contains(out, want) { + t.Errorf("missing %q: %q", want, out) + } + } +} + +func TestTransformResponsesStream_UnclosedTagFlushed(t *testing.T) { + input := `data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_1","type":"message"}} + +data: {"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"content_index":0,"delta":"answer unclosed"} + +data: [DONE] +` + rc := TransformResponsesStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "unclosed") { + t.Errorf("unclosed reasoning dropped: %q", out) + } + if strings.Contains(out, `"reasoning"`) { + t.Errorf("reasoning synthesized for unclosed tag: %q", out) + } +} + +func TestTransformResponsesStream_NoDoneEOF(t *testing.T) { + // Stream ends without [DONE]: any buffered text flushes as ordinary + // content on the current message item so nothing is dropped. + input := "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"msg_1\",\"type\":\"message\"}}\n\n" + + "data: {\"type\":\"response.output_text.delta\",\"item_id\":\"msg_1\",\"output_index\":0,\"content_index\":0,\"delta\":\"answer rest\"}\n" + rc := TransformResponsesStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + s := string(out) + if !strings.Contains(s, "rest") { + t.Errorf("unclosed reasoning body dropped at EOF: %q", s) + } +} + +func TestTransformResponsesStream_MultipleBlocksSameItem(t *testing.T) { + // Two reasoning blocks on the same message item reuse one synthesized + // reasoning item id, so the client sees one contiguous reasoning span. + input := "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"msg_1\",\"type\":\"message\"}}\n\n" + + "data: {\"type\":\"response.output_text.delta\",\"item_id\":\"msg_1\",\"output_index\":0,\"content_index\":0,\"delta\":\"aoneb\"}\n\n" + + "data: {\"type\":\"response.output_text.delta\",\"item_id\":\"msg_1\",\"output_index\":0,\"content_index\":0,\"delta\":\"ctwod\"}\n\n" + + "data: [DONE]\n" + rc := TransformResponsesStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + s := string(out) + if strings.Count(s, "response.output_item.added") != 2 { + t.Errorf("expected exactly one synthesized reasoning item added (plus the message item's own added), got: %q", s) + } +} diff --git a/internal/thinkextract/responses_test.go b/internal/thinkextract/responses_test.go new file mode 100644 index 000000000..33d544385 --- /dev/null +++ b/internal/thinkextract/responses_test.go @@ -0,0 +1,124 @@ +package thinkextract + +import ( + "testing" + + "github.com/enterpilot/gomodel/internal/core" +) + +func responsesResp(texts ...string) *core.ResponsesResponse { + out := &core.ResponsesResponse{} + for _, text := range texts { + out.Output = append(out.Output, core.ResponsesOutputItem{ + ID: "msg_x", + Type: "message", + Content: []core.ResponsesContentItem{ + {Type: "output_text", Text: text}, + }, + }) + } + return out +} + +func TestTransformResponsesResponse_NoTags(t *testing.T) { + resp := responsesResp("plain answer") + if got := TransformResponsesResponse(resp, Options{}); got != 0 { + t.Fatalf("rewritten=%d, want 0", got) + } + if len(resp.Output) != 1 { + t.Errorf("output items=%d, want 1", len(resp.Output)) + } +} + +func TestTransformResponsesResponse_SingleItem(t *testing.T) { + resp := responsesResp("answerhidden rest") + if got := TransformResponsesResponse(resp, Options{}); got != 1 { + t.Fatalf("rewritten=%d, want 1", got) + } + if len(resp.Output) != 2 { + t.Fatalf("output items=%d, want 2 (reasoning + message)", len(resp.Output)) + } + if resp.Output[0].Type != "reasoning" { + t.Errorf("output[0].Type=%q, want reasoning", resp.Output[0].Type) + } + if resp.Output[0].Content[0].Type != "reasoning_text" || resp.Output[0].Content[0].Text != "hidden" { + t.Errorf("reasoning content=%+v, want reasoning_text hidden", resp.Output[0].Content[0]) + } + if resp.Output[1].Type != "message" { + t.Errorf("output[1].Type=%q, want message", resp.Output[1].Type) + } + if resp.Output[1].Content[0].Text != "answer rest" { + t.Errorf("message text=%q, want %q", resp.Output[1].Content[0].Text, "answer rest") + } +} + +func TestTransformResponsesResponse_MultipleItems(t *testing.T) { + resp := responsesResp("axb", "plain", "cyd") + if got := TransformResponsesResponse(resp, Options{}); got != 2 { + t.Fatalf("rewritten=%d, want 2", got) + } + // Expect: reasoning, message(ab), message(plain), reasoning, message(cd) + if len(resp.Output) != 5 { + t.Fatalf("items=%d, want 5", len(resp.Output)) + } + if resp.Output[0].Type != "reasoning" || resp.Output[3].Type != "reasoning" { + t.Errorf("reasoning items misplaced: %+v", resp.Output) + } + if resp.Output[1].Content[0].Text != "ab" || resp.Output[4].Content[0].Text != "cd" { + t.Errorf("message texts wrong: %q %q", resp.Output[1].Content[0].Text, resp.Output[4].Content[0].Text) + } + if resp.Output[2].Content[0].Text != "plain" { + t.Errorf("untouched middle item changed: %q", resp.Output[2].Content[0].Text) + } +} + +func TestTransformResponsesResponse_NonMessageItemUntouched(t *testing.T) { + resp := &core.ResponsesResponse{ + Output: []core.ResponsesOutputItem{ + {ID: "fc_1", Type: "function_call", Name: "f", Arguments: "{}"}, + {ID: "msg_1", Type: "message", Content: []core.ResponsesContentItem{{Type: "output_text", Text: "axb"}}}, + }, + } + if got := TransformResponsesResponse(resp, Options{}); got != 1 { + t.Fatalf("rewritten=%d, want 1", got) + } + if resp.Output[0].Type != "function_call" { + t.Errorf("function_call item moved or rewritten: %+v", resp.Output[0]) + } + if resp.Output[1].Type != "reasoning" { + t.Errorf("reasoning item not inserted before message: %+v", resp.Output[1]) + } +} + +func TestTransformResponsesResponse_NonTextContentUntouched(t *testing.T) { + resp := &core.ResponsesResponse{ + Output: []core.ResponsesOutputItem{{ + ID: "msg_1", + Type: "message", + Content: []core.ResponsesContentItem{ + {Type: "input_image", ImageURL: &core.ImageURLContent{URL: "data:image/png;base64,xxx"}}, + }, + }}, + } + if got := TransformResponsesResponse(resp, Options{}); got != 0 { + t.Errorf("rewritten=%d, want 0", got) + } +} + +func TestTransformResponsesResponse_NilResponse(t *testing.T) { + if got := TransformResponsesResponse(nil, Options{}); got != 0 { + t.Errorf("nil response: got=%d, want 0", got) + } +} + +func TestTransformResponsesResponse_EmptyBodyBlock(t *testing.T) { + resp := responsesResp("beforeafter") + if got := TransformResponsesResponse(resp, Options{}); got != 1 { + t.Fatalf("rewritten=%d, want 1 (empty-body block is still a rewrite)", got) + } + // Reasoning item is still prepended with empty text, matching the chat + // path's behaviour of counting empty-body rewrites. + if resp.Output[0].Type != "reasoning" { + t.Errorf("reasoning item missing for empty-body block") + } +} diff --git a/internal/thinkextract/stream.go b/internal/thinkextract/stream.go new file mode 100644 index 000000000..aa99aa1d7 --- /dev/null +++ b/internal/thinkextract/stream.go @@ -0,0 +1,282 @@ +package thinkextract + +import ( + "bufio" + "bytes" + "encoding/json" + "io" + "strings" +) + +// TransformStream wraps an SSE stream of OpenAI chat-completion chunks, +// rewriting each event's delta.content that carries legacy think-block tags +// into a paired reasoning_content + content delta. Non-data lines and the +// data: [DONE] sentinel are passed through unchanged. Invalid JSON in a data +// line is forwarded byte-for-byte so a malformed upstream event cannot break +// the stream. +// +// A line is treated as data when its trimmed prefix equals "data:". Any +// other SSE field name (event:, id:, retry:, comment lines starting with ":") +// is forwarded untouched. +// +// Multi-choice streams are supported: a State is maintained per choice index +// so each choice's tag boundaries resolve independently. +func TransformStream(in io.ReadCloser, opts Options) io.ReadCloser { + return TransformStreamForSurface(in, opts, "") +} + +// TransformStreamForSurface is TransformStream with an explicit surface so +// synthesized reasoning deltas can be marked on the messages surface. The +// marker (SynthesizedMarkerKey) is only emitted for the messages surface and +// is consumed by the Anthropic messages converter; chat-surface responses +// never carry it. +func TransformStreamForSurface(in io.ReadCloser, opts Options, surface Surface) io.ReadCloser { + o := opts.withDefaults() + pr, pw := io.Pipe() + go transformLoop(in, pw, o, surface == SurfaceMessages) + return pr +} + +func transformLoop(in io.ReadCloser, pw *io.PipeWriter, o Options, markSynthesized bool) { + defer pw.Close() + defer in.Close() + + states := map[int]*State{} + scanner := bufio.NewScanner(in) + // 1 MiB per-line cap: enough for any sane SSE event, low enough to + // bound per-event memory. + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + + for scanner.Scan() { + line := append([]byte(nil), scanner.Bytes()...) + trimmed := bytes.TrimLeft(line, " \t") + if !bytes.HasPrefix(trimmed, []byte("data:")) { + if !writeLine(pw, line) { + return + } + continue + } + payload := bytes.TrimPrefix(trimmed, []byte("data:")) + payload = bytes.TrimPrefix(payload, []byte(" ")) + + if string(payload) == "[DONE]" { + for idx, st := range states { + cd, rd := st.Flush() + if cd == "" && rd == "" { + continue + } + if !writeFlushDelta(pw, idx, cd, rd) { + return + } + } + if !writeLine(pw, line) { + return + } + continue + } + + rewritten, err := rewriteChunk(payload, states, o, markSynthesized) + if err != nil { + if !writeLine(pw, line) { + return + } + continue + } + for _, out := range rewritten { + if !writeEvent(pw, out) { + return + } + } + } + // Defensive drain for streams that terminate without a [DONE] sentinel. + for idx, st := range states { + cd, rd := st.Flush() + if cd == "" && rd == "" { + continue + } + _ = writeFlushDelta(pw, idx, cd, rd) + } +} + +// rewriteChunk decodes a single OpenAI chat-completion chunk and rewrites each +// choice's delta.content for think-block markers. The chunk is returned as a +// slice of JSON payloads, one per output event. A single input chunk can yield +// 0, 1, or 2 output chunks per choice (content + reasoning). +// +// All non-delta fields on every choice are preserved byte-for-byte: the +// rewriter only mutates delta.content and delta.reasoning_content. +func rewriteChunk(payload []byte, states map[int]*State, o Options, markSynthesized bool) ([][]byte, error) { + var root map[string]json.RawMessage + if err := json.Unmarshal(payload, &root); err != nil { + return nil, err + } + rawChoices, ok := root["choices"] + if !ok { + return [][]byte{payload}, nil + } + var choices []json.RawMessage + if err := json.Unmarshal(rawChoices, &choices); err != nil { + return nil, err + } + if len(choices) == 0 { + return [][]byte{payload}, nil + } + + type pendingDelta struct { + choiceIdx int + contentDelta string + reasonDelta string + } + var pending []pendingDelta + for i, choiceRaw := range choices { + var meta struct { + Index int `json:"index"` + Delta struct { + Content string `json:"content"` + ReasoningContent string `json:"reasoning_content"` + } `json:"delta"` + } + if err := json.Unmarshal(choiceRaw, &meta); err != nil { + continue + } + // Existing upstream reasoning wins; never stomp on structured data. + if meta.Delta.ReasoningContent != "" { + continue + } + state := states[meta.Index] + if state == nil { + state = NewState(o) + states[meta.Index] = state + } + cd, rd := state.Feed(meta.Delta.Content) + if cd == "" && rd == "" { + continue + } + pending = append(pending, pendingDelta{ + choiceIdx: i, + contentDelta: cd, + reasonDelta: rd, + }) + } + if len(pending) == 0 { + return [][]byte{payload}, nil + } + + out := make([][]byte, 0, len(pending)) + for _, p := range pending { + newChoice, err := rewriteChoiceDelta(choices[p.choiceIdx], p.contentDelta, p.reasonDelta, markSynthesized) + if err != nil { + continue + } + newChoices := make([]json.RawMessage, len(choices)) + copy(newChoices, choices) + newChoices[p.choiceIdx] = newChoice + newRoot := make(map[string]json.RawMessage, len(root)) + for k, v := range root { + if k == "choices" { + continue + } + newRoot[k] = v + } + newChoicesJSON, err := json.Marshal(newChoices) + if err != nil { + continue + } + newRoot["choices"] = newChoicesJSON + encoded, err := json.Marshal(newRoot) + if err != nil { + continue + } + out = append(out, encoded) + } + return out, nil +} + +// rewriteChoiceDelta returns a copy of choice with delta.content and +// delta.reasoning_content replaced. Every other field on the choice is +// preserved verbatim. When markSynthesized is true the delta also carries +// the SynthesizedMarkerKey flag so the messages converter can apply its +// thinking-block policy. +func rewriteChoiceDelta(choice json.RawMessage, content, reasoning string, markSynthesized bool) (json.RawMessage, error) { + var m map[string]json.RawMessage + if err := json.Unmarshal(choice, &m); err != nil { + return nil, err + } + delta := map[string]any{} + if raw, ok := m["delta"]; ok && len(raw) > 0 { + _ = json.Unmarshal(raw, &delta) + } + delta["content"] = content + delta["reasoning_content"] = reasoning + if markSynthesized && reasoning != "" { + delta[SynthesizedMarkerKey] = true + } + deltaJSON, err := json.Marshal(delta) + if err != nil { + return nil, err + } + m["delta"] = deltaJSON + return json.Marshal(m) +} + +// writeLine writes a single SSE line followed by "\n". +func writeLine(pw *io.PipeWriter, line []byte) bool { + if _, err := pw.Write(line); err != nil { + return false + } + if len(line) == 0 || line[len(line)-1] != '\n' { + if _, err := pw.Write([]byte("\n")); err != nil { + return false + } + } + return true +} + +// writeEvent writes a complete SSE event (data: \n\n). +func writeEvent(pw *io.PipeWriter, payload []byte) bool { + if _, err := pw.Write([]byte("data: ")); err != nil { + return false + } + if _, err := pw.Write(payload); err != nil { + return false + } + if _, err := pw.Write([]byte("\n\n")); err != nil { + return false + } + return true +} + +// writeFlushDelta emits a final cleanup event for a stream that ended without +// a [DONE] sentinel. The event carries only the flushed content / reasoning. +func writeFlushDelta(pw *io.PipeWriter, idx int, contentDelta, reasonDelta string) bool { + payload := mustEncode(map[string]any{ + "choices": []map[string]any{{ + "index": idx, + "delta": map[string]any{ + "content": contentDelta, + "reasoning_content": reasonDelta, + }, + }}, + }) + return writeEvent(pw, payload) +} + +func mustEncode(v any) []byte { + b, err := json.Marshal(v) + if err != nil { + return []byte(`{"choices":[]}`) + } + return b +} + +// ReadAll is a small convenience that drains an io.ReadCloser produced by +// TransformStream into a single string. Tests use it; production code keeps +// using io.Copy. +func ReadAll(rc io.ReadCloser) (string, error) { + defer rc.Close() + var sb strings.Builder + if _, err := io.Copy(&sb, rc); err != nil { + return "", err + } + return sb.String(), nil +} \ No newline at end of file diff --git a/internal/thinkextract/surface.go b/internal/thinkextract/surface.go new file mode 100644 index 000000000..02548c661 --- /dev/null +++ b/internal/thinkextract/surface.go @@ -0,0 +1,40 @@ +package thinkextract + +import "context" + +// Surface identifies the API surface a request is being processed on. The +// translation is gated per surface so an operator can disable it on one +// surface without affecting the others. +type Surface string + +const ( + // SurfaceChat is the OpenAI chat completions surface (/v1/chat/completions). + SurfaceChat Surface = "chat" + // SurfaceMessages is the Anthropic messages surface (/v1/messages). + SurfaceMessages Surface = "messages" + // SurfaceResponses is the OpenAI responses surface (/v1/responses). + // Currently not exercised by the chat-side hook; tracked for the + // follow-up native Responses transformer. + SurfaceResponses Surface = "responses" +) + +// surfaceKey is the unexported context key under which a Surface value is +// stored. Using an unexported empty struct prevents external packages from +// colliding on the same key. +type surfaceKey struct{} + +// WithSurface returns a context that carries the given Surface for use by +// the orchestrator's extraction gate. +func WithSurface(ctx context.Context, surface Surface) context.Context { + return context.WithValue(ctx, surfaceKey{}, surface) +} + +// SurfaceFrom returns the Surface stored on ctx, or the empty string when +// no surface has been set. The empty string is the "default" surface and +// is treated as enabled when the global translation is on. +func SurfaceFrom(ctx context.Context) Surface { + if v, ok := ctx.Value(surfaceKey{}).(Surface); ok { + return v + } + return "" +} \ No newline at end of file diff --git a/internal/thinkextract/thinkextract.go b/internal/thinkextract/thinkextract.go new file mode 100644 index 000000000..2d56a1f00 --- /dev/null +++ b/internal/thinkextract/thinkextract.go @@ -0,0 +1,423 @@ +// Package thinkextract rewrites assistant message text that carries legacy +// ... reasoning blocks (and configured equivalents) into the +// structured reasoning field expected by GoModel's downstream surfaces. +// +// The package operates on the gateway's response wire format after a model +// returns. A model that emits reasoninganswer produces, after +// translation, a response with reasoning_content="reasoning" and content="answer". +// +// The translation is lossless on the wire: no model-visible character is +// dropped. If a block opens but the matching does not yet +// appear in the input the package buffers the partial text and waits for more +// bytes; a never-closed block is forwarded verbatim with no rewrite applied, +// so a chunk-boundary cut can never strand user content as reasoning. +package thinkextract + +import ( + "strings" +) + +// FieldReasoning is the OpenAI chat-completions field name that carries the +// extracted reasoning text on the wire. The Anthropic messages endpoint and +// OpenAI responses API surface the same string as their native reasoning +// payload via their respective dialect converters. +const FieldReasoning = "reasoning_content" + +// defaultMaxBufferBytes is the cross-chunk cap for an unclosed reasoning +// block. 64 KiB clears any legacy variant the package is documented to +// recognise without risking unbounded growth on a runaway open tag. +const defaultMaxBufferBytes = 64 * 1024 + +// TagPair is a matched open/close delimiter pair that brackets a reasoning +// block. Both fields are matched literally; no regex is used so the scanner +// stays allocation-free and chunk-boundary-safe. +type TagPair struct { + Open string + Close string +} + +// DefaultTagPairs is the evidence-backed default recognition list. It mirrors +// the union of what vLLM, SGLang, and Open WebUI treat as standard legacy +// reasoning markers (T9 research: docs.vllm.ai reasoning outputs, SGLang +// separate-reasoning docs, Open WebUI reasoning-models docs). Granite's +// plain-English delimiters and ERNIE's answer-wrapper are +// deliberately excluded: the former are false-positive-prone natural language, +// the latter marks the answer rather than the reasoning. +func DefaultTagPairs() []TagPair { + return []TagPair{ + {Open: "", Close: ""}, + {Open: "", Close: ""}, + {Open: "", Close: ""}, + {Open: "", Close: ""}, + {Open: "", Close: ""}, + {Open: "<|begin_of_thought|>", Close: "<|end_of_thought|>"}, + {Open: "◁think▷", Close: "◁/think▷"}, + {Open: "[THINK]", Close: "[/THINK]"}, + {Open: "<|channel|>analysis<|message|>", Close: "<|end|>"}, + } +} + +// ParseTagPairs parses a comma-separated list of "..." entries +// into TagPairs, e.g. "...,...". +// Malformed entries (missing the "..." separator, empty open or close) are +// skipped so one bad entry never breaks the whole list. +func ParseTagPairs(list string) []TagPair { + if strings.TrimSpace(list) == "" { + return nil + } + var pairs []TagPair + for _, entry := range strings.Split(list, ",") { + entry = strings.TrimSpace(entry) + open, close, ok := strings.Cut(entry, "...") + if !ok || open == "" || close == "" { + continue + } + pairs = append(pairs, TagPair{Open: open, Close: close}) + } + return pairs +} + +// Options configures the tag delimiters, the per-stream buffer cap, and the +// per-surface enable gates. +// +// The zero value is valid: it defaults to DefaultTagPairs with the 64 KiB +// cross-chunk buffer and all surfaces enabled. +type Options struct { + // TagPairs is the recognition list. When empty, DefaultTagPairs applies. + // TagOpen/TagClose below, when set, override TagPairs with a single pair. + TagPairs []TagPair + // TagOpen opens a reasoning block. Legacy single-pair override; prefer + // TagPairs for new configuration. + TagOpen string + // TagClose closes a reasoning block. Legacy single-pair override; prefer + // TagPairs for new configuration. + TagClose string + // MaxBufferBytes caps the size of an unclosed block held in streaming + // state. Once exceeded the buffered text is flushed as ordinary content + // and no further reasoning is emitted on that stream. Default 64 KiB. + MaxBufferBytes int + // ChatEnabled gates the translation on the chat completions surface. + // Nil means on. + ChatEnabled *bool + // ResponsesEnabled gates the translation on the OpenAI responses surface. + // Nil means on. + ResponsesEnabled *bool + // MessagesPolicy gates the translation on the Anthropic messages surface. + // Values: "off" (default), "unsigned", "redacted". Empty means off, + // matching the messages endpoint's default of no synthesized thinking + // blocks. + MessagesPolicy string +} + +// EnabledFor reports whether the translation runs on the given surface. +// The empty surface (no surface set on the request context) is treated as +// enabled, matching the global-on default. +func (o Options) EnabledFor(surface Surface) bool { + switch surface { + case SurfaceChat: + if o.ChatEnabled != nil { + return *o.ChatEnabled + } + case SurfaceResponses: + if o.ResponsesEnabled != nil { + return *o.ResponsesEnabled + } + case SurfaceMessages: + return ParseMessagesPolicy(o.MessagesPolicy) != MessagesPolicyOff + } + return true +} + +// pairs resolves the effective recognition list for these options. +func (o Options) pairs() []TagPair { + if len(o.TagPairs) > 0 { + return o.TagPairs + } + if o.TagOpen != "" && o.TagClose != "" { + return []TagPair{{Open: o.TagOpen, Close: o.TagClose}} + } + return DefaultTagPairs() +} + +func (o Options) withDefaults() Options { + if o.MaxBufferBytes <= 0 { + o.MaxBufferBytes = defaultMaxBufferBytes + } + return o +} + +// Extract returns the input with every recognised reasoning block removed +// (cleaned), the concatenated reasoning text, and a flag indicating whether +// any block was rewritten. The flag lets callers skip re-serialisation when +// the input was already clean. +// +// When the input ends inside an open block with no matching close, Extract +// reports found=false and returns the input unchanged: the text might still +// receive the closing tag from a future chunk and we cannot risk dropping it. +// With multiple tag pairs configured, the earliest opening tag wins; an +// unclosed block of any pair aborts the whole extraction conservatively. +func Extract(text string, opts Options) (cleaned string, reasoning string, found bool) { + o := opts.withDefaults() + pairs := o.pairs() + if text == "" || !containsAnyOpen(text, pairs) { + return text, "", false + } + + var ( + cursor int + out strings.Builder + outSet bool + reason strings.Builder + reasonSet bool + ) + for cursor < len(text) { + openIdx, pairIdx := earliestOpen(text, cursor, pairs) + if openIdx == -1 { + out.WriteString(text[cursor:]) + outSet = true + break + } + if openIdx > cursor { + out.WriteString(text[cursor:openIdx]) + outSet = true + } + pair := pairs[pairIdx] + bodyStart := openIdx + len(pair.Open) + relClose := strings.Index(text[bodyStart:], pair.Close) + if relClose == -1 { + // Open block with no matching close. Treat the whole input as + // ordinary content: the close may yet arrive in a future chunk. + return text, "", false + } + body := strings.TrimSpace(text[bodyStart : bodyStart+relClose]) + if reasonSet && body != "" { + reason.WriteString("\n\n") + } + reason.WriteString(body) + reasonSet = true + cursor = bodyStart + relClose + len(pair.Close) + } + if !outSet { + out.WriteString(text) + } + if !reasonSet { + return text, "", false + } + cleaned = strings.TrimSpace(out.String()) + reasoning = strings.TrimSpace(reason.String()) + return cleaned, reasoning, true +} + +// containsAnyOpen reports whether text contains any configured open tag. +func containsAnyOpen(text string, pairs []TagPair) bool { + for _, p := range pairs { + if strings.Contains(text, p.Open) { + return true + } + } + return false +} + +// earliestOpen finds the first occurrence of any open tag at or after cursor. +// It returns the absolute index and the pair index, or -1 when none matches. +func earliestOpen(text string, cursor int, pairs []TagPair) (int, int) { + best, bestPair := -1, -1 + for i, p := range pairs { + rel := strings.Index(text[cursor:], p.Open) + if rel == -1 { + continue + } + abs := cursor + rel + if best == -1 || abs < best { + best, bestPair = abs, i + } + } + return best, bestPair +} + +// State holds the running state of a streaming rewrite. One State is created +// per stream by TransformStream; a State is not safe for concurrent use. +type State struct { + opts Options + pairs []TagPair + + // inThink is true after an open tag has been seen but before its close. + // activePair is the pair that opened the current block. + inThink bool + activePair int + // buffer holds text seen after the last emitted boundary that has not yet + // been classified. While inThink the buffer holds only reasoning text; + // otherwise it holds only visible text. + buffer strings.Builder + // reasoning accumulates the body of every closed think block, separated + // by a blank line so multiple blocks round-trip distinctly. The first + // block is appended without a leading separator. + reasoning strings.Builder + // emitted tracks how many bytes of `reasoning` have already been returned + // via a Feed/Flush call, so the caller sees each block as a discrete delta. + emitted int +} + +// NewState constructs a streaming State with the given options applied. +func NewState(opts Options) *State { + o := opts.withDefaults() + return &State{opts: o, pairs: o.pairs()} +} + +// Feed pushes a chunk of assistant delta content and returns the +// (content, reasoning) deltas to emit immediately. Any text that opens or +// closes a tag is held in the State until the matching boundary arrives. +// +// The returned deltas are non-empty only when there is data to emit right +// now; buffered text (an unclosed tail) yields two empty strings. +func (s *State) Feed(chunk string) (contentDelta string, reasoningDelta string) { + if chunk == "" { + return "", "" + } + if s.opts.MaxBufferBytes > 0 && s.buffer.Len()+len(chunk) > s.opts.MaxBufferBytes { + // Cap exceeded: drop any buffered text on the floor as content so + // nothing is lost and a runaway open tag cannot leak memory. + rest := s.buffer.String() + chunk + s.buffer.Reset() + s.opts.MaxBufferBytes = 0 // disable further buffering + return rest, "" + } + s.buffer.WriteString(chunk) + for { + cd, rd, advanced := s.tryAdvance() + if !advanced { + break + } + contentDelta += cd + reasoningDelta += rd + } + // Outside a think block, drain any text whose tail cannot be a partial + // open tag of any configured pair — emit it as content so the client sees + // progress. + if !s.inThink { + bs := s.buffer.String() + if bs == "" { + return contentDelta, reasoningDelta + } + safeEnd := safeEmitPrefixAll(bs, s.pairs) + if safeEnd > 0 { + contentDelta += bs[:safeEnd] + rest := bs[safeEnd:] + s.buffer.Reset() + s.buffer.WriteString(rest) + } + } + return contentDelta, reasoningDelta +} + +// Flush releases any buffered text at end-of-stream. An unclosed tag at the +// tail of the last chunk is re-emitted with its open tag as ordinary content +// so nothing is lost. Reasoning accumulated in earlier Feed calls but not yet +// emitted is returned. +func (s *State) Flush() (contentDelta string, reasoningDelta string) { + full := s.reasoning.String() + if len(full) > s.emitted { + reasoningDelta = full[s.emitted:] + s.emitted = len(full) + } + if s.buffer.Len() == 0 { + return contentDelta, reasoningDelta + } + bs := s.buffer.String() + if s.inThink { + // Unclosed block. Re-emit the open tag literal plus the buffered body + // so the original bytes round-trip even without a close. + contentDelta += s.pairs[s.activePair].Open + bs + s.buffer.Reset() + return contentDelta, reasoningDelta + } + // End of stream: emit the tail verbatim. A partial open tag can no + // longer be completed by a later chunk, so it is literal content now. + contentDelta += bs + s.buffer.Reset() + return contentDelta, reasoningDelta +} + +// tryAdvance consumes one open-or-close boundary if present in the buffer. +// Returns advanced=false when the buffer is too short to decide (an open +// boundary may be cut by the chunk edge). +func (s *State) tryAdvance() (contentDelta string, reasoningDelta string, advanced bool) { + bs := s.buffer.String() + if s.inThink { + closeTag := s.pairs[s.activePair].Close + i := strings.Index(bs, closeTag) + if i == -1 { + return "", "", false + } + body := strings.TrimSpace(bs[:i]) + rest := bs[i+len(closeTag):] + if s.reasoning.Len() > 0 && body != "" { + s.reasoning.WriteString("\n\n") + } + if body != "" { + s.reasoning.WriteString(body) + } + s.buffer.Reset() + s.buffer.WriteString(rest) + s.inThink = false + full := s.reasoning.String() + if len(full) > s.emitted { + reasoningDelta = full[s.emitted:] + s.emitted = len(full) + } + return "", reasoningDelta, true + } + openIdx, pairIdx := earliestOpenIn(bs, s.pairs) + if openIdx == -1 { + return "", "", false + } + contentDelta = bs[:openIdx] + rest := bs[openIdx+len(s.pairs[pairIdx].Open):] + s.buffer.Reset() + s.buffer.WriteString(rest) + s.inThink = true + s.activePair = pairIdx + return contentDelta, "", true +} + +// earliestOpenIn finds the first occurrence of any open tag in bs. +func earliestOpenIn(bs string, pairs []TagPair) (int, int) { + best, bestPair := -1, -1 + for i, p := range pairs { + idx := strings.Index(bs, p.Open) + if idx == -1 { + continue + } + if best == -1 || idx < best { + best, bestPair = idx, i + } + } + return best, bestPair +} + +// safeEmitPrefixAll returns the largest prefix length of bs whose tail cannot +// form a prefix of any configured open tag. +func safeEmitPrefixAll(bs string, pairs []TagPair) int { + safeEnd := len(bs) + for _, p := range pairs { + if end := safeEmitPrefix(bs, p.Open); end < safeEnd { + safeEnd = end + } + } + return safeEnd +} + +// safeEmitPrefix returns the largest prefix length of bs whose tail cannot +// form a prefix of openTag. Used to keep partial open tags in the buffer +// until the next chunk arrives. +func safeEmitPrefix(bs, openTag string) int { + if openTag == "" { + return len(bs) + } + for k := 1; k < len(openTag); k++ { + if strings.HasSuffix(bs, openTag[:k]) { + return len(bs) - k + } + } + return len(bs) +} \ No newline at end of file diff --git a/internal/thinkextract/thinkextract_extra_test.go b/internal/thinkextract/thinkextract_extra_test.go new file mode 100644 index 000000000..2133637a8 --- /dev/null +++ b/internal/thinkextract/thinkextract_extra_test.go @@ -0,0 +1,272 @@ +package thinkextract + +import ( + "io" + "strings" + "testing" + + "github.com/enterpilot/gomodel/internal/core" +) + +func TestState_Flush_PartialReasoningEmitted(t *testing.T) { + // Reasoning accumulated across Feed calls but not yet emitted in the + // last Flush is returned by the next Flush call. + s := NewState(Options{}) + s.Feed("part1") + cd, rd := s.Feed("part2") + if cd != "" { + t.Errorf("mid-stream content=%q, want empty", cd) + } + if rd != "\n\npart2" { + t.Errorf("mid-stream reasoning=%q, want %q", rd, "\n\npart2") + } + // Flush emits nothing more: all reasoning already returned. + cd, rd = s.Flush() + if cd != "" || rd != "" { + t.Errorf("post-flush: got (%q,%q), want empty", cd, rd) + } +} + +func TestState_BufferCapOverflow(t *testing.T) { + // A single Feed that exceeds the cap must not panic and must drop the + // buffered text on the floor as ordinary content. + opts := Options{MaxBufferBytes: 8} + s := NewState(opts) + cd, rd := s.Feed("0123456789ABCDEF") + if rd != "" { + t.Errorf("reasoning=%q, want empty", rd) + } + if cd == "" { + t.Errorf("content delta is empty after cap overflow, want some") + } + if s.buffer.Len() > 0 { + t.Errorf("buffer not cleared after cap, len=%d", s.buffer.Len()) + } +} + +func TestState_BufferCapOverflowInsideThinkBlock(t *testing.T) { + // Cap hit while inside a think block: content after the cap must be + // emitted as ordinary content, not as reasoning. + opts := Options{MaxBufferBytes: 8} + s := NewState(Options{}) + s = NewState(opts) + s.Feed("ahidden") + cd, _ := s.Feed(strings.Repeat("z", 64)) + if cd == "" { + t.Errorf("expected overflow content emission") + } +} + +func TestState_SafeEmitPrefix_NoPartialMatch(t *testing.T) { + // A tail that does not match any prefix of the open tag is safe to emit. + s := NewState(Options{}) + cd, _ := s.Feed("answer hello") + if cd != "answer hello" { + t.Errorf("content=%q, want %q", cd, "answer hello") + } +} + +func TestExtract_NestedTags(t *testing.T) { + // Nested opens: the first close terminates the first block; the inner + // open stays inside the extracted reasoning text. Documented behaviour. + input := "aonetwob" + cleaned, reasoning, found := Extract(input, Options{}) + if !found { + t.Fatalf("found=false, want true") + } + if cleaned != "ab" { + t.Errorf("cleaned=%q, want %q", cleaned, "ab") + } + if reasoning != "onetwo" { + t.Errorf("reasoning=%q, want %q", reasoning, "onetwo") + } +} + +func TestExtract_EmptyInput(t *testing.T) { + cleaned, reasoning, found := Extract("", Options{}) + if found { + t.Errorf("found=true, want false") + } + if cleaned != "" || reasoning != "" { + t.Errorf("got (%q,%q), want empty", cleaned, reasoning) + } +} + +func TestTransformStream_FlushAtDone(t *testing.T) { + // State that has buffered text at [DONE] should emit a flush delta + // carrying the residual content so the client sees nothing lost. + input := "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"axb\"}}]}\n\n" + + "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"yz more\"}}]}\n\n" + + "data: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + // Both chunks should rewrite; the second leaves " more" after the block. + if !strings.Contains(out, "\"reasoning_content\":\"\\n\\nz\"") { + t.Errorf("missing reasoning=second block with separator: %q", out) + } + if !strings.Contains(out, "[DONE]") { + t.Errorf("missing [DONE]: %q", out) + } +} + +func TestTransformStream_NoDoneDrain(t *testing.T) { + // Streams that terminate without [DONE] still flush residual state. + input := "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"axb\"}}]}\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "\"reasoning_content\":\"x\"") { + t.Errorf("missing reasoning_content=x: %q", out) + } +} + +func TestTransformStream_MalformedChoicesArray(t *testing.T) { + // choices is a string, not an array: json.Unmarshal succeeds, the + // unmarshal-into-[]json.RawMessage fails, returns err, forwards verbatim. + input := "data: {\"choices\":\"oops\"}\n\ndata: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "oops") { + t.Errorf("malformed chunk not forwarded: %q", out) + } +} + +func TestTransformStream_EmptyChoices(t *testing.T) { + // Empty choices array: passes through verbatim. + input := "data: {\"choices\":[]}\n\ndata: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "[DONE]") { + t.Errorf("empty-choices chunk lost: %q", out) + } +} + +func TestTransformStream_PartialTagAcrossDone(t *testing.T) { + // Tag opens in one chunk, closes in another, followed by [DONE]: must + // produce both content and reasoning deltas. + input := "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ar\"}}]}\n\n" + + "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"est of reason\"}}]}\n\n" + + "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ing end\"}}]}\n\n" + + "data: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "\"reasoning_content\":\"r\\nest of reason\\ning\"") && + !strings.Contains(out, "reasoning_content") { + t.Errorf("reasoning across chunks missing: %q", out) + } +} + +func TestTransformChatResponse_UnknownContentType(t *testing.T) { + // Content is an int (unexpected). Must not panic, must report no + // rewrite. + resp := &core.ChatResponse{ + Choices: []core.Choice{{Message: core.ResponseMessage{Role: "assistant", Content: 42}}}, + } + if got := TransformChatResponse(resp, Options{}); got != 0 { + t.Errorf("rewritten=%d, want 0", got) + } +} + +func TestTransformChatResponse_NilMessage(t *testing.T) { + // Choices with no content at all. No panic, no rewrite. + resp := &core.ChatResponse{ + Choices: []core.Choice{{Message: core.ResponseMessage{Role: "assistant", Content: nil}}}, + } + if got := TransformChatResponse(resp, Options{}); got != 0 { + t.Errorf("rewritten=%d, want 0", got) + } +} + +func TestTransformChatResponse_EmptyContentPartText(t *testing.T) { + // A text part with empty text is not rewritten; a text part with the + // tag but no body is rewritten with empty reasoning (which setReasoning + // then rejects, so no-op). + resp := &core.ChatResponse{ + Choices: []core.Choice{{Message: core.ResponseMessage{ + Role: "assistant", + Content: []core.ContentPart{ + {Type: "text", Text: ""}, + {Type: "text", Text: "beforeafter"}, + }, + }}}, + } + if got := TransformChatResponse(resp, Options{}); got != 1 { + t.Errorf("rewritten=%d, want 1 (empty-body block is a rewrite)", got) + } + parts := resp.Choices[0].Message.Content.([]core.ContentPart) + if parts[0].Text != "" || parts[1].Text != "beforeafter" { + t.Errorf("parts=%+v, want empty + \"beforeafter\"", parts) + } +} + +func TestTransformChatResponse_NonTextPartUntouched(t *testing.T) { + // ImageURL parts are untouched even if they look like text fields. + resp := &core.ChatResponse{ + Choices: []core.Choice{{Message: core.ResponseMessage{ + Role: "assistant", + Content: []core.ContentPart{ + {Type: "image_url", ImageURL: &core.ImageURLContent{URL: "data:image/png;base64,xxx"}}, + {Type: "text", Text: "yshown"}, + }, + }}}, + } + if got := TransformChatResponse(resp, Options{}); got != 1 { + t.Errorf("rewritten=%d, want 1", got) + } + parts := resp.Choices[0].Message.Content.([]core.ContentPart) + if parts[0].Type != "image_url" { + t.Errorf("image part lost type: %v", parts[0]) + } + if parts[1].Text != "shown" { + t.Errorf("text part: %q, want \"shown\"", parts[1].Text) + } +} + +func TestTransformChatResponse_PreservesToolCalls(t *testing.T) { + // Tool calls on the message must survive the rewrite. + calls := []core.ToolCall{{ID: "call_1", Type: "function", Function: core.FunctionCall{Name: "f", Arguments: "{}"}}} + resp := &core.ChatResponse{ + Choices: []core.Choice{{ + Message: core.ResponseMessage{ + Role: "assistant", + Content: "abc", + ToolCalls: calls, + }, + }}, + } + if got := TransformChatResponse(resp, Options{}); got != 1 { + t.Errorf("rewritten=%d, want 1", got) + } + if len(resp.Choices[0].Message.ToolCalls) != 1 { + t.Errorf("tool calls lost") + } +} + +// Cover safeEmitPrefix with no tail match for the entire tag. +func TestSafeEmitPrefix_NoMatch(t *testing.T) { + got := safeEmitPrefix("plain text", "") + if got != len("plain text") { + t.Errorf("got %d, want %d", got, len("plain text")) + } +} + +// Sanity check: jsonUnmarshalString handles valid and invalid raw input. \ No newline at end of file diff --git a/internal/thinkextract/thinkextract_flush_test.go b/internal/thinkextract/thinkextract_flush_test.go new file mode 100644 index 000000000..515233e88 --- /dev/null +++ b/internal/thinkextract/thinkextract_flush_test.go @@ -0,0 +1,145 @@ +package thinkextract + +import ( + "io" + "strings" + "testing" + "time" + + "github.com/enterpilot/gomodel/internal/core" +) + +func TestTransformStream_UnclosedBlockFlushedAtDone(t *testing.T) { + // A think block that never closes is re-emitted at [DONE] as literal + // content (open tag + body) so no bytes are dropped. + input := "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"text unclosed\"}}]}\n\ndata: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "think\\u003eunclosed") { + t.Errorf("unclosed block body dropped: %q", out) + } + if !strings.Contains(out, "[DONE]") { + t.Errorf("missing [DONE]: %q", out) + } +} + +func TestTransformStream_ChoiceDeltaNonObject(t *testing.T) { + // delta as a string instead of an object: choice decode fails, chunk is + // forwarded verbatim. + input := "data: {\"choices\":[{\"index\":0,\"delta\":\"oops\"}]}\n\ndata: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "oops") { + t.Errorf("non-object delta chunk not forwarded: %q", out) + } +} + +func TestTransformStream_EarlyReaderClose(t *testing.T) { + // Closing the output reader before the upstream is drained must not + // deadlock the transformer goroutine. + input := strings.Repeat("data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"x\"}}]}\n\n", 100) + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + buf := make([]byte, 64) + _, _ = rc.Read(buf) + if err := rc.Close(); err != nil { + t.Fatalf("close: %v", err) + } + // Give the goroutine a moment to notice the closed pipe and exit. The + // test passes as long as it terminates without hanging. + time.Sleep(50 * time.Millisecond) +} + +func TestSafeEmitPrefix_EmptyTag(t *testing.T) { + if got := safeEmitPrefix("abc", ""); got != 3 { + t.Errorf("safeEmitPrefix with empty tag: got %d, want 3", got) + } +} + +func TestState_Feed_CapWhileInThink(t *testing.T) { + // Cap overflow while inside a think block: buffered reasoning text is + // flushed as ordinary content so nothing is silently dropped. + opts := Options{MaxBufferBytes: 8} + s := NewState(opts) + s.Feed("") + cd, _ := s.Feed(strings.Repeat("y", 64)) + if cd == "" { + t.Errorf("expected overflow content emission while in think block") + } +} + +func TestState_Feed_EmptyChunk(t *testing.T) { + s := NewState(Options{}) + cd, rd := s.Feed("") + if cd != "" || rd != "" { + t.Errorf("empty chunk produced (%q,%q), want empty", cd, rd) + } +} + +func TestExtract_CustomBufferCapOption(t *testing.T) { + // Options plumbing: a custom cap is respected and the default pair list + // fills the rest. + opts := Options{MaxBufferBytes: 16} + s := NewState(opts) + if s.opts.MaxBufferBytes != 16 { + t.Errorf("cap=%d, want 16", s.opts.MaxBufferBytes) + } + if len(s.pairs) == 0 || s.pairs[0].Open != "" { + t.Errorf("default pairs not applied: %+v", s.pairs) + } +} + +func TestTransformChatResponse_OnlyUnclosedTextParts(t *testing.T) { + // ContentParts where the only text has an unclosed block: Extract returns + // found=false for each part, !changed holds, no rewrite. + resp := &core.ChatResponse{ + Choices: []core.Choice{{Message: core.ResponseMessage{ + Role: "assistant", + Content: []core.ContentPart{ + {Type: "text", Text: "aunclosed"}, + }, + }}}, + } + if got := TransformChatResponse(resp, Options{}); got != 0 { + t.Errorf("rewritten=%d, want 0 (unclosed block keeps content intact)", got) + } +} + +func TestTransformChatResponse_NonTextPartOnly(t *testing.T) { + // ContentParts with only non-text parts: no rewrite. + resp := &core.ChatResponse{ + Choices: []core.Choice{{Message: core.ResponseMessage{ + Role: "assistant", + Content: []core.ContentPart{ + {Type: "image_url", ImageURL: &core.ImageURLContent{URL: "data:image/png;base64,xxx"}}, + }, + }}}, + } + if got := TransformChatResponse(resp, Options{}); got != 0 { + t.Errorf("rewritten=%d, want 0", got) + } +} + +func TestTransformStream_LongOutputForcesGoroutineExit(t *testing.T) { + // Stream enough events that the internal pipe fills; then close the + // reader. The transformer goroutine must observe the closed pipe and + // exit without leaking. + pr, pw := io.Pipe() + defer pw.Close() + input := strings.Repeat("data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\""+strings.Repeat("x", 1024)+"\"}}]}\n\n", 256) + src := io.NopCloser(strings.NewReader(input)) + go transformLoop(src, pw, Options{}, false) + // Read a small chunk then close. + buf := make([]byte, 256) + _, _ = pr.Read(buf) + _ = pr.Close() + // Give the goroutine a chance to wake up. + time.Sleep(50 * time.Millisecond) +} diff --git a/internal/thinkextract/thinkextract_iowrite_test.go b/internal/thinkextract/thinkextract_iowrite_test.go new file mode 100644 index 000000000..e6dc74c75 --- /dev/null +++ b/internal/thinkextract/thinkextract_iowrite_test.go @@ -0,0 +1,108 @@ +package thinkextract + +import ( + "errors" + "io" + "strings" + "testing" +) + +// errWriter is an io.Writer that returns errWrite on every Write call. It +// covers the error branches in writeLine, writeEvent, and the streaming +// goroutines' pipe-write paths without forcing a broken-pipe race. +type errWriter struct{ err error } + +func (w *errWriter) Write(_ []byte) (int, error) { return 0, w.err } + +func TestWriteLine_NoTrailingNewline(t *testing.T) { + pr, pw := io.Pipe() + done := make(chan string, 1) + go func() { + ok := writeLine(pw, []byte("event: created")) + _ = pw.Close() + if !ok { + done <- "writeLine returned false" + return + } + done <- "" + }() + out, err := io.ReadAll(pr) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if msg := <-done; msg != "" { + t.Errorf("%s", msg) + } + if string(out) != "event: created\n" { + t.Errorf("got %q, want %q", string(out), "event: created\n") + } +} + +func TestTransformLoop_BadWriter(t *testing.T) { + // Drive transformLoop with a writer that always errors. The loop must + // observe the error on the first event and exit cleanly without + // panicking or leaking. We hook the pipe writer itself: a real + // io.PipeWriter only errors after the reader closes. + in := io.NopCloser(strings.NewReader( + "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"axb\"}}]}\n\ndata: [DONE]\n", + )) + pr, pw := io.Pipe() + _ = pr.Close() // ensure every write returns ErrClosedPipe + transformLoop(in, pw, Options{}, false) +} + +func TestTransformStream_NoDataLinePassThrough(t *testing.T) { + // Lines that are not data lines (comments, event:, id:, retry:) must be + // forwarded byte-for-byte, with their original line ending preserved or + // a trailing \n appended if missing. + input := ": comment line\n\nevent: ping\nid: 7\nretry: 1000\n\ndata: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + for _, want := range []string{": comment line", "event: ping", "id: 7", "retry: 1000"} { + if !strings.Contains(string(out), want) { + t.Errorf("missing %q in %q", want, string(out)) + } + } +} + +func TestWriteEvent_Empty(t *testing.T) { + pr, pw := io.Pipe() + done := make(chan string, 1) + go func() { + ok := writeEvent(pw, []byte{}) + _ = pw.Close() + if !ok { + done <- "writeEvent returned false" + return + } + done <- "" + }() + out, err := io.ReadAll(pr) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if msg := <-done; msg != "" { + t.Errorf("%s", msg) + } + if string(out) != "data: \n\n" { + t.Errorf("got %q, want %q", string(out), "data: \n\n") + } +} + +func TestMustEncode_FallbackPath(t *testing.T) { + // Direct exercise of mustEncode: map values marshal fine so the + // fallback path is unreachable through normal input; we only verify + // the happy path. The fallback is documented as unreachable with + // map[string]any values. + out := mustEncode(map[string]any{"a": 1}) + if len(out) == 0 { + t.Errorf("mustEncode returned empty") + } +} + +// erroringPipeWriter is unused; the test above directly closes the real pipe. +var _ = errors.New \ No newline at end of file diff --git a/internal/thinkextract/thinkextract_pairs_test.go b/internal/thinkextract/thinkextract_pairs_test.go new file mode 100644 index 000000000..e832b6484 --- /dev/null +++ b/internal/thinkextract/thinkextract_pairs_test.go @@ -0,0 +1,188 @@ +package thinkextract + +import ( + "strings" + "testing" +) + +func TestDefaultTagPairs_HasThinkTag(t *testing.T) { + pairs := DefaultTagPairs() + if len(pairs) == 0 { + t.Fatalf("DefaultTagPairs returned empty list") + } + var found bool + for _, p := range pairs { + if p.Open == "" && p.Close == "" { + found = true + break + } + } + if !found { + t.Errorf("DefaultTagPairs missing the canonical pair") + } +} + +func TestParseTagPairs_Empty(t *testing.T) { + if got := ParseTagPairs(""); got != nil { + t.Errorf("ParseTagPairs(\"\") = %+v, want nil", got) + } + if got := ParseTagPairs(" "); got != nil { + t.Errorf("ParseTagPairs(\" \") = %+v, want nil", got) + } +} + +func TestParseTagPairs_Single(t *testing.T) { + got := ParseTagPairs("...") + if len(got) != 1 || got[0].Open != "" || got[0].Close != "" { + t.Errorf("got %+v, want single pair", got) + } +} + +func TestParseTagPairs_Multiple(t *testing.T) { + got := ParseTagPairs("...,...") + if len(got) != 2 { + t.Fatalf("got %+v, want 2 pairs", got) + } + if got[0] != (TagPair{Open: "", Close: ""}) { + t.Errorf("pair[0]=%+v", got[0]) + } + if got[1] != (TagPair{Open: "", Close: ""}) { + t.Errorf("pair[1]=%+v", got[1]) + } +} + +func TestParseTagPairs_SkipsMalformed(t *testing.T) { + // Missing "..." separator, or empty open/close: skipped silently so + // one bad entry never breaks the whole list. + got := ParseTagPairs("...,malformed,...,...close-only") + if len(got) != 2 { + t.Errorf("got %+v, want 2 pairs (malformed entries skipped)", got) + } +} + +func TestExtract_AlternateTag(t *testing.T) { + opts := Options{TagPairs: []TagPair{{Open: "", Close: ""}}} + cleaned, reasoning, found := Extract("abc", opts) + if !found { + t.Fatalf("found=false, want true") + } + if cleaned != "ac" { + t.Errorf("cleaned=%q, want %q", cleaned, "ac") + } + if reasoning != "b" { + t.Errorf("reasoning=%q, want %q", reasoning, "b") + } +} + +func TestExtract_DefaultPairs_MatchesMultiple(t *testing.T) { + // Default list matches both and blocks in one input. + input := "a middle b end" + cleaned, reasoning, found := Extract(input, Options{}) + if !found { + t.Fatalf("found=false, want true") + } + if cleaned != "middle end" { + t.Errorf("cleaned=%q, want %q", cleaned, "middle end") + } + if reasoning != "a\n\nb" { + t.Errorf("reasoning=%q, want %q", reasoning, "a\n\nb") + } +} + +func TestExtract_EarliestOpenWins(t *testing.T) { + // When two configured open tags are both present, the earliest absolute + // position is the one used. + opts := Options{TagPairs: []TagPair{ + {Open: "", Close: ""}, + {Open: "", Close: ""}, + }} + cleaned, reasoning, found := Extract("xfirstysecondz", opts) + if !found { + t.Fatalf("found=false") + } + if cleaned != "xyz" { + t.Errorf("cleaned=%q, want %q", cleaned, "xyz") + } + if reasoning != "first\n\nsecond" { + t.Errorf("reasoning=%q, want %q", reasoning, "first\n\nsecond") + } +} + +func TestExtract_UnclosedAlternateTag(t *testing.T) { + // Unclosed alternate tag: treated as ordinary content (no rewrite). + opts := Options{TagPairs: []TagPair{{Open: "", Close: ""}}} + input := "beforeopen" + cleaned, reasoning, found := Extract(input, opts) + if found { + t.Errorf("found=true, want false on unclosed alternate tag") + } + if cleaned != input { + t.Errorf("cleaned=%q, want %q", cleaned, input) + } + if reasoning != "" { + t.Errorf("reasoning=%q, want empty", reasoning) + } +} + +func TestState_AlternateTag(t *testing.T) { + opts := Options{TagPairs: []TagPair{{Open: "", Close: ""}}} + s := NewState(opts) + cd, rd := s.Feed("axb") + if cd != "ab" { + t.Errorf("content=%q, want %q", cd, "ab") + } + if rd != "x" { + t.Errorf("reasoning=%q, want %q", rd, "x") + } +} + +func TestState_DefaultPairs_StreamBothKinds(t *testing.T) { + s := NewState(Options{}) + cd, rd := s.Feed("axmiddleyend") + if cd != "amiddleend" { + t.Errorf("content=%q, want %q", cd, "amiddleend") + } + if rd != "x\n\ny" { + t.Errorf("reasoning=%q, want %q", rd, "x\n\ny") + } +} + +func TestState_BufferCapOverflowMultiPair(t *testing.T) { + opts := Options{MaxBufferBytes: 8, TagPairs: []TagPair{{Open: "", Close: ""}}} + s := NewState(opts) + cd, _ := s.Feed(strings.Repeat("z", 64)) + if cd == "" { + t.Errorf("expected overflow content emission") + } +} + +func TestSafeEmitPrefixAll_EmptyPairs(t *testing.T) { + if got := safeEmitPrefixAll("text", nil); got != 4 { + t.Errorf("got %d, want 4", got) + } +} + +func TestEnabledFor_PerSurfaceDefaults(t *testing.T) { + // Per-surface pointers unset falls back to enabled for any surface. + o := Options{} + if !o.EnabledFor(SurfaceChat) { + t.Errorf("EnabledFor(chat)=false, want true") + } + if o.EnabledFor(SurfaceMessages) { + t.Errorf("EnabledFor(messages)=true, want false (messages default off)") + } + if !o.EnabledFor("") { + t.Errorf("EnabledFor(empty)=false, want true") + } +} + +func TestEnabledFor_ExplicitDisable(t *testing.T) { + off := false + o := Options{ChatEnabled: &off, MessagesPolicy: "unsigned"} + if o.EnabledFor(SurfaceChat) { + t.Errorf("chat should be disabled") + } + if !o.EnabledFor(SurfaceMessages) { + t.Errorf("messages should be enabled under unsigned policy") + } +} \ No newline at end of file diff --git a/internal/thinkextract/thinkextract_surface_test.go b/internal/thinkextract/thinkextract_surface_test.go new file mode 100644 index 000000000..5aa3b189a --- /dev/null +++ b/internal/thinkextract/thinkextract_surface_test.go @@ -0,0 +1,62 @@ +package thinkextract + +import ( + "context" + "encoding/json" + "testing" + + "github.com/enterpilot/gomodel/internal/core" +) + +func TestWithSurface_AndSurfaceFrom(t *testing.T) { + if got := SurfaceFrom(context.Background()); got != "" { + t.Errorf("empty context: SurfaceFrom=%q, want empty", got) + } + ctx := WithSurface(context.Background(), SurfaceChat) + if got := SurfaceFrom(ctx); got != SurfaceChat { + t.Errorf("SurfaceFrom(chat ctx)=%q, want %q", got, SurfaceChat) + } + ctx2 := WithSurface(context.Background(), SurfaceMessages) + if got := SurfaceFrom(ctx2); got != SurfaceMessages { + t.Errorf("SurfaceFrom(messages ctx)=%q, want %q", got, SurfaceMessages) + } +} + +func TestMarkSynthesizedOnMessage(t *testing.T) { + msg := &core.ResponseMessage{ + Role: "assistant", + Content: "answer", + } + markSynthesizedOnMessage(msg) + raw := msg.ExtraFields.Lookup(SynthesizedMarkerKey) + if len(raw) == 0 { + t.Fatalf("synthesized marker not set") + } + var got bool + if err := json.Unmarshal(raw, &got); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if !got { + t.Errorf("synthesized marker=%v, want true", got) + } +} + +func TestMarkSynthesizedOnMessage_PreservesExistingExtras(t *testing.T) { + msg := &core.ResponseMessage{ + Content: "answer", + ExtraFields: mustUnknownFields(map[string]json.RawMessage{ + "x_custom": json.RawMessage(`"keepme"`), + }), + } + markSynthesizedOnMessage(msg) + if len(msg.ExtraFields.Lookup("x_custom")) == 0 { + t.Errorf("existing extra dropped during mark") + } + if len(msg.ExtraFields.Lookup(SynthesizedMarkerKey)) == 0 { + t.Errorf("synthesized marker not set") + } +} + +func mustUnknownFields(m map[string]json.RawMessage) core.UnknownJSONFields { + return core.UnknownJSONFieldsFromMap(m) +} diff --git a/internal/thinkextract/thinkextract_test.go b/internal/thinkextract/thinkextract_test.go new file mode 100644 index 000000000..58b8d558a --- /dev/null +++ b/internal/thinkextract/thinkextract_test.go @@ -0,0 +1,378 @@ +package thinkextract + +import ( + "io" + "strings" + "testing" +) + +func TestExtract_NoThink(t *testing.T) { + tests := []struct { + name string + input string + }{ + {name: "empty", input: ""}, + {name: "plain text", input: "hello world"}, + {name: "text with angle brackets but no tag", input: "a < b and c > d"}, + {name: "other xml-ish tags", input: "x"}, + {name: "partial open at end", input: "answerreasoningmore", wantClean: "answermore", wantReason: "reasoning", wantFound: true}, + {name: "leading whitespace stripped", input: " x y ", wantClean: "y", wantReason: "x", wantFound: true}, + {name: "trailing whitespace stripped", input: "abc ", wantClean: "ac", wantReason: "b", wantFound: true}, + {name: "two blocks", input: "abcde", wantClean: "ace", wantReason: "b\n\nd", wantFound: true}, + {name: "multiline reasoning", input: "aline1\nline2b", wantClean: "ab", wantReason: "line1\nline2", wantFound: true}, + {name: "empty block", input: "ab", wantClean: "ab", wantReason: "", wantFound: true}, + {name: "whitespace-only block", input: "a b", wantClean: "ab", wantReason: "", wantFound: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cleaned, reasoning, found := Extract(tt.input, Options{}) + if found != tt.wantFound { + t.Fatalf("Extract(%q) found=%v, want %v", tt.input, found, tt.wantFound) + } + if cleaned != tt.wantClean { + t.Errorf("Extract(%q) cleaned=%q, want %q", tt.input, cleaned, tt.wantClean) + } + if reasoning != tt.wantReason { + t.Errorf("Extract(%q) reasoning=%q, want %q", tt.input, reasoning, tt.wantReason) + } + }) + } +} + +func TestExtract_UnclosedBlock(t *testing.T) { + // Unclosed block is treated as ordinary content, not reasoning. + tests := []struct { + name string + input string + }{ + {name: "open without close", input: "areasoning"}, + {name: "open at very end", input: "a"}, + {name: "close before open", input: "abc"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cleaned, reasoning, found := Extract(tt.input, Options{}) + if found { + t.Errorf("Extract(%q) found=true, want false", tt.input) + } + if cleaned != tt.input { + t.Errorf("Extract(%q) cleaned=%q, want %q", tt.input, cleaned, tt.input) + } + if reasoning != "" { + t.Errorf("Extract(%q) reasoning=%q, want empty", tt.input, reasoning) + } + }) + } +} + +func TestExtract_CustomTags(t *testing.T) { + opts := Options{TagOpen: "", TagClose: ""} + cleaned, reasoning, found := Extract("abc", opts) + if !found { + t.Fatalf("Extract with custom tags found=false, want true") + } + if cleaned != "ac" { + t.Errorf("cleaned=%q, want %q", cleaned, "ac") + } + if reasoning != "b" { + t.Errorf("reasoning=%q, want %q", reasoning, "b") + } +} + +func TestExtract_TagInsideCode(t *testing.T) { + // Tags inside markdown code blocks are still rewritten. This is a + // documented caveat of the feature: the package cannot know whether the + // text is code. + input := "```\nx\n```" + cleaned, reasoning, found := Extract(input, Options{}) + if !found { + t.Fatalf("found=false, want true") + } + if cleaned != "```\n\n```" { + t.Errorf("cleaned=%q, want %q", cleaned, "```\n\n```") + } + if reasoning != "x" { + t.Errorf("reasoning=%q, want %q", reasoning, "x") + } +} + +func TestState_Feed_FullBlock(t *testing.T) { + s := NewState(Options{}) + cd, rd := s.Feed("axb") + if cd != "ab" { + t.Errorf("content delta=%q, want %q", cd, "ab") + } + if rd != "x" { + t.Errorf("reasoning delta=%q, want %q", rd, "x") + } + cd, rd = s.Flush() + if cd != "" || rd != "" { + t.Errorf("Flush after full block: got (%q,%q), want empty", cd, rd) + } +} + +func TestState_Feed_PartialTagAcrossChunks(t *testing.T) { + s := NewState(Options{}) + cd, rd := s.Feed("answer reasoning") + if cd != "" { + t.Errorf("after tag open: content=%q, want empty", cd) + } + if rd != "" { + t.Errorf("after tag open: reasoning=%q, want empty", rd) + } + + cd, rd = s.Feed(" more") + if cd != " more" { + t.Errorf("after close: content=%q, want %q", cd, " more") + } + if rd != "reasoning" { + t.Errorf("after close: reasoning=%q, want %q", rd, "reasoning") + } +} + +func TestState_Feed_MultipleBlocks(t *testing.T) { + s := NewState(Options{}) + cd, rd := s.Feed("aonebtwoc") + if cd != "abc" { + t.Errorf("content=%q, want %q", cd, "abc") + } + if rd != "one\n\ntwo" { + t.Errorf("reasoning=%q, want %q", rd, "one\n\ntwo") + } +} + +func TestState_Flush_UnclosedBlock(t *testing.T) { + s := NewState(Options{}) + s.Feed("apartial") + cd, rd := s.Flush() + // Unclosed block: the buffer is inThink, so flush emits as content. + if cd != "partial" { + t.Errorf("flush content=%q, want %q", cd, "partial") + } + if rd != "" { + t.Errorf("flush reasoning=%q, want empty", rd) + } +} + +func TestState_Flush_PartialOpenTagTail(t *testing.T) { + s := NewState(Options{}) + // Feed emits the safe prefix; the partial tag tail stays buffered. + cd, _ := s.Feed("answer " + strings.Repeat("y", 100)) + // Should not panic, and should not have stuck everything in buffer. + if s.buffer.Len() > 1000 { + t.Errorf("buffer.Len()=%d after cap overflow, want bounded", s.buffer.Len()) + } + _ = cd +} + +func TestTransformStream_NoTags(t *testing.T) { + input := "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hello\"}}]}\n\ndata: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "hello") { + t.Errorf("output does not contain 'hello': %q", out) + } + if !strings.Contains(out, "[DONE]") { + t.Errorf("output missing [DONE]: %q", out) + } +} + +func TestTransformStream_SingleChunkWithTag(t *testing.T) { + input := "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"arb\"}}]}\n\ndata: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "\"reasoning_content\":\"r\"") { + t.Errorf("output missing reasoning_content=r: %q", out) + } + if !strings.Contains(out, "\"content\":\"ab\"") { + t.Errorf("output missing content=ab: %q", out) + } +} + +func TestTransformStream_PartialTagAcrossChunks(t *testing.T) { + input := strings.Join([]string{ + "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"answer reason\"}}]}", + "", + "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ing end\"}}]}", + "", + "data: [DONE]", + "", + }, "\n") + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "reasoning_content") { + t.Errorf("output missing reasoning_content: %q", out) + } + if strings.Contains(out, "") { + t.Errorf("output still contains literal: %q", out) + } +} + +func TestTransformStream_PreservesNonDeltaFields(t *testing.T) { + input := "data: {\"id\":\"x\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"arb\"},\"finish_reason\":null}],\"model\":\"m\"}\n\ndata: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "\"id\":\"x\"") { + t.Errorf("output dropped id: %q", out) + } + if !strings.Contains(out, "\"model\":\"m\"") { + t.Errorf("output dropped model: %q", out) + } + if !strings.Contains(out, "\"role\":\"assistant\"") { + t.Errorf("output dropped role: %q", out) + } +} + +func TestTransformStream_UpstreamReasoningPreserved(t *testing.T) { + input := "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"arb\",\"reasoning_content\":\"upstream\"}}]}\n\ndata: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "\"reasoning_content\":\"upstream\"") { + t.Errorf("upstream reasoning_content stomped: %q", out) + } + // And the text should still be in content (untouched). + if !strings.Contains(out, "") { + t.Errorf("upstream reasoning chunk should not be rewritten: %q", out) + } +} + +func TestTransformStream_InvalidJSONForwarded(t *testing.T) { + input := "data: {not json}\n\ndata: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "{not json}") { + t.Errorf("invalid JSON not forwarded: %q", out) + } +} + +func TestTransformStream_NonDataLinesForwarded(t *testing.T) { + input := ": comment\n\nevent: test\nid: 1\n\ndata: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + for _, want := range []string{": comment", "event: test", "id: 1", "[DONE]"} { + if !strings.Contains(out, want) { + t.Errorf("output missing %q: %q", want, out) + } + } +} + +func TestTransformStream_MultipleChoices(t *testing.T) { + input := "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"axb\"}},{\"index\":1,\"delta\":{\"content\":\"cyd\"}}]}\n\ndata: [DONE]\n" + rc := TransformStream(io.NopCloser(strings.NewReader(input)), Options{}) + defer rc.Close() + out, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + if !strings.Contains(out, "\"reasoning_content\":\"x\"") { + t.Errorf("missing choice 0 reasoning: %q", out) + } + if !strings.Contains(out, "\"reasoning_content\":\"y\"") { + t.Errorf("missing choice 1 reasoning: %q", out) + } +} + +func TestReadAll_ClosesReader(t *testing.T) { + rc := TransformStream(io.NopCloser(strings.NewReader("data: [DONE]\n")), Options{}) + _, err := ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } +} \ No newline at end of file