diff --git a/internal/catalog/resolve.go b/internal/catalog/resolve.go index 9c24c8e1..364bf5f6 100644 --- a/internal/catalog/resolve.go +++ b/internal/catalog/resolve.go @@ -93,6 +93,15 @@ func (ic *IndexedCatalog) ResolveShort(short string) (ResolvedModel, error) { return ic.resolveFromMatches(short, matches) } + for key, model := range ic.Models { + if strings.EqualFold(key, short) || strings.EqualFold(model.Name, short) || strings.EqualFold(modelNameFromKey(key), short) { + matches = append(matches, key) + } + } + if len(matches) > 0 { + return ic.resolveFromMatches(short, matches) + } + return ResolvedModel{}, fmt.Errorf("unknown short model id: %q", short) } diff --git a/internal/catalog/resolve_test.go b/internal/catalog/resolve_test.go index 055829aa..a579173b 100644 --- a/internal/catalog/resolve_test.go +++ b/internal/catalog/resolve_test.go @@ -2,6 +2,7 @@ package catalog import ( "sort" + "strings" "testing" ) @@ -239,6 +240,26 @@ func TestResolveShort_Name(t *testing.T) { } } +func TestResolveShort_CaseInsensitiveCanonicalizesUniqueModel(t *testing.T) { + ic := newFixtureCatalog() + + got, err := ic.ResolveShort("DeepSeek-V4-Flash") + if err != nil { + t.Fatalf("ResolveShort unexpected error: %v", err) + } + if got.ModelID != "deepseek-v4-flash" || got.CanonicalName != "opencode-go/deepseek-v4-flash" { + t.Fatalf("ResolveShort = %+v, want canonical deepseek model", got) + } +} + +func TestResolveShort_CaseInsensitiveStillRejectsAmbiguity(t *testing.T) { + ic := newFixtureCatalog() + + if _, err := ic.ResolveShort("KIMI-K2.6"); err == nil || !strings.Contains(err.Error(), "ambiguous") { + t.Fatalf("ResolveShort error = %v, want ambiguity", err) + } +} + func TestResolveShort_DisabledProvider(t *testing.T) { ic := newFixtureCatalog() diff --git a/internal/config/model_registry.go b/internal/config/model_registry.go index 4d6fe8ab..f3fc31a4 100644 --- a/internal/config/model_registry.go +++ b/internal/config/model_registry.go @@ -1,5 +1,7 @@ package config +import "strings" + const DefaultContextMargin = 8192 // ModelMetadata describes a known model's capabilities (context window size, @@ -38,6 +40,28 @@ var modelMetadata = map[string]ModelMetadata{ "qwen3.5-plus": {ContextWindow: 1000000, MaxOutputTokens: 8192, Vision: true, SupportsTools: true}, } +// CanonicalModelID returns the registered spelling for a known model ID. +// Unknown and ambiguously cased custom IDs are preserved unchanged. +func CanonicalModelID(modelID string) string { + if _, ok := modelMetadata[modelID]; ok { + return modelID + } + match := "" + for known := range modelMetadata { + if !strings.EqualFold(known, modelID) { + continue + } + if match != "" { + return modelID + } + match = known + } + if match != "" { + return match + } + return modelID +} + // ResolveModelConfig fills in default capability values (context window, // max output tokens, vision, tool support) for a ModelConfig by consulting // the built-in modelMetadata registry. If the model is unknown or a field @@ -45,6 +69,7 @@ var modelMetadata = map[string]ModelMetadata{ // a ModelConfig so capacity filtering and scenario routing see accurate // per-model limits. func ResolveModelConfig(model ModelConfig) ModelConfig { + model.ModelID = CanonicalModelID(model.ModelID) if model.ModelRef == "" { if meta, ok := modelMetadata[model.ModelID]; ok { if model.ContextWindow == 0 { diff --git a/internal/config/model_registry_test.go b/internal/config/model_registry_test.go index 3d0f4719..d92c2e87 100644 --- a/internal/config/model_registry_test.go +++ b/internal/config/model_registry_test.go @@ -28,6 +28,26 @@ func TestResolveModelConfig(t *testing.T) { SupportsTools: boolPtr(true), }, }, + { + name: "known mixed-case model uses canonical ID and metadata", + input: ModelConfig{ModelID: "DeepSeek-V4-Pro"}, + expected: ModelConfig{ + ModelID: "deepseek-v4-pro", + ContextWindow: 1000000, + MaxOutputTokens: 8192, + ContextMargin: DefaultContextMargin, + SupportsTools: boolPtr(true), + }, + }, + { + name: "unknown custom model preserves case", + input: ModelConfig{ModelID: "Vendor-Custom-Pro"}, + expected: ModelConfig{ + ModelID: "Vendor-Custom-Pro", + ContextMargin: DefaultContextMargin, + SupportsTools: boolPtr(true), + }, + }, { name: "kimi-k3 gets hardcoded metadata (1M context, 131K output, vision)", input: ModelConfig{ diff --git a/internal/router/model_router.go b/internal/router/model_router.go index f12ff973..34cb9618 100644 --- a/internal/router/model_router.go +++ b/internal/router/model_router.go @@ -93,6 +93,10 @@ func (r *ModelRouter) resolveRequestedModel(cfg *config.Config, requestedModel s // Look up the requested model in config to inherit its settings primary, ok := cfg.Models[requestedModel] + canonicalRequestedModel := config.CanonicalModelID(requestedModel) + if !ok && canonicalRequestedModel != requestedModel { + primary, ok = cfg.Models[canonicalRequestedModel] + } if !ok { // Not in legacy config — try the catalog before falling back to the // legacy unknown-model behavior. Provider-qualified references that @@ -108,12 +112,12 @@ func (r *ModelRouter) resolveRequestedModel(cfg *config.Config, requestedModel s } else if providerQualified { return RouteResult{}, false, fmt.Errorf("model reference %q uses unknown provider %q: %w", requestedModel, sel.Provider, ErrUnknownProvider) } else { - primary = r.legacyUnknownModelConfig(cfg, requestedModel) + primary = r.legacyUnknownModelConfig(cfg, canonicalRequestedModel) } } else if providerQualified { return RouteResult{}, false, fmt.Errorf("model reference %q uses unknown provider %q: %w", requestedModel, sel.Provider, ErrUnknownProvider) } else { - primary = r.legacyUnknownModelConfig(cfg, requestedModel) + primary = r.legacyUnknownModelConfig(cfg, canonicalRequestedModel) } } primary = config.ResolveModelConfig(primary) diff --git a/internal/router/model_router_test.go b/internal/router/model_router_test.go index ad4c4967..78e438ed 100644 --- a/internal/router/model_router_test.go +++ b/internal/router/model_router_test.go @@ -505,6 +505,12 @@ func TestResolveRequestedModel(t *testing.T) { Temperature: 0.3, MaxTokens: 2048, }, + "deepseek-v4-pro": { + Provider: "opencode-go", + ModelID: "deepseek-v4-pro", + Temperature: 0.7, + MaxTokens: 8192, + }, }, Fallbacks: map[string][]config.ModelConfig{ "default": {{Provider: "opencode-go", ModelID: "qwen3.5-plus"}}, @@ -538,6 +544,13 @@ func TestResolveRequestedModel(t *testing.T) { wantModelID: "deepseek-v4-flash", wantModelRef: "deepseek-v4-flash", }, + { + name: "mixed-case known model resolves to configured canonical ID", + requestedModel: "DeepSeek-V4-Pro", + wantProvider: "opencode-go", + wantModelID: "deepseek-v4-pro", + wantModelRef: "", + }, { name: "config model takes precedence over catalog", requestedModel: "custom-model", diff --git a/internal/transformer/request_test.go b/internal/transformer/request_test.go index 40489848..a47f32e3 100644 --- a/internal/transformer/request_test.go +++ b/internal/transformer/request_test.go @@ -495,6 +495,13 @@ func TestTransformRequestThinkingDecisionMatrix(t *testing.T) { model: config.ModelConfig{ModelID: "qwen3.6-plus"}, wantThink: "", }, + { + name: "kimi ignores request and history thinking without explicit capability", + messages: thinkingHistory, + thinking: json.RawMessage(`{"type":"enabled","budget_tokens":4096}`), + model: config.ModelConfig{ModelID: "kimi-k2.6"}, + wantThink: "", + }, { name: "request disabled overrides explicit model thinking", messages: userOnly, diff --git a/internal/transformer/stream.go b/internal/transformer/stream.go index c97b3e7a..e669e5d5 100644 --- a/internal/transformer/stream.go +++ b/internal/transformer/stream.go @@ -26,6 +26,8 @@ var ErrStreamIdle = fmt.Errorf("upstream stream idle") var ErrEmptyStream = fmt.Errorf("upstream returned empty stream") +const thinkingSignaturePlaceholder = "proxy-thinking-placeholder" + // readBufPool pools read buffers for streaming operations. // sync.Pool reduces GC pressure under concurrent stream load by reusing // 4KB buffers across goroutines instead of allocating fresh ones per read. @@ -94,6 +96,9 @@ func (h *StreamHandler) EmitMessageResponse(w http.ResponseWriter, resp *types.M startBlock.Text = "" case "thinking": startBlock.Thinking = "" + if startBlock.Signature == "" { + startBlock.Signature = thinkingSignaturePlaceholder + } case "tool_use": startBlock.Input = json.RawMessage(`{}`) } @@ -125,6 +130,17 @@ func (h *StreamHandler) EmitMessageResponse(w http.ResponseWriter, resp *types.M return ErrClientDisconnected } } + signature := block.Signature + if signature == "" { + signature = thinkingSignaturePlaceholder + } + if err := writeSSEEvent(w, types.MessageEvent{ + Type: "content_block_delta", + Index: &idx, + Delta: &types.Delta{Type: "signature_delta", Signature: signature}, + }); err != nil { + return ErrClientDisconnected + } case "tool_use": if len(block.Input) > 0 { if err := writeSSEEvent(w, types.MessageEvent{ @@ -221,7 +237,8 @@ func (h *StreamHandler) ProxyStream( var lineBuf []byte contentStarted := false reasoningStarted := false - stopSent := false + terminalStopReason := "" + var terminalUsage *types.UsageInfo toolUseCount := 0 startedToolCalls := make(map[int]int) // maps OpenAI tool call index → Anthropic content block index decodeErrors := 0 // consecutive SSE decode failures @@ -255,7 +272,7 @@ func (h *StreamHandler) ProxyStream( b := (*readBuf)[i] if b == '\n' { // Process complete line - if err := h.processSSELine(w, flusher, lineBuf, &contentIndex, &contentStarted, &reasoningStarted, &stopSent, &toolUseCount, startedToolCalls, originalModel, &decodeErrors); err != nil { + if err := h.processSSELine(w, flusher, lineBuf, &contentIndex, &contentStarted, &reasoningStarted, &terminalStopReason, &terminalUsage, &toolUseCount, startedToolCalls, originalModel, &decodeErrors); err != nil { return err } lineBuf = lineBuf[:0] @@ -268,7 +285,7 @@ func (h *StreamHandler) ProxyStream( if err == io.EOF { // Process any remaining data in buffer if len(lineBuf) > 0 { - if err := h.processSSELine(w, flusher, lineBuf, &contentIndex, &contentStarted, &reasoningStarted, &stopSent, &toolUseCount, startedToolCalls, originalModel, &decodeErrors); err != nil { + if err := h.processSSELine(w, flusher, lineBuf, &contentIndex, &contentStarted, &reasoningStarted, &terminalStopReason, &terminalUsage, &toolUseCount, startedToolCalls, originalModel, &decodeErrors); err != nil { return err } } @@ -288,17 +305,17 @@ func (h *StreamHandler) ProxyStream( } } - // Close any open content block (text or reasoning) - if contentStarted || reasoningStarted { - stopEvent := types.MessageEvent{ - Type: "content_block_stop", - Index: &contentIndex, + // Close any open content block (text or reasoning). + if reasoningStarted { + if err := writeThinkingBlockStop(w, contentIndex); err != nil { + return ErrClientDisconnected } - if err := writeSSEEvent(w, stopEvent); err != nil { + reasoningStarted = false + } else if contentStarted { + if err := writeContentBlockStop(w, contentIndex); err != nil { return ErrClientDisconnected } contentStarted = false - reasoningStarted = false } // Send stop events for any tool blocks not yet closed (e.g. upstream @@ -322,25 +339,24 @@ func (h *StreamHandler) ProxyStream( } } - // Send message_delta if not already sent. - // If tool calls were in progress when the stream ended, - // the stop reason should be "tool_use" rather than "end_turn". - if !stopSent { - stopReason := "end_turn" + // Anthropic expects one terminal message_delta containing both stop_reason + // and usage. OpenAI-compatible providers commonly send those in separate + // chunks, so both are retained until the upstream stream ends. + if terminalStopReason == "" { + terminalStopReason = "end_turn" if len(startedToolCalls) > 0 { - stopReason = "tool_use" - } - msgDelta := types.MessageEvent{ - Type: "message_delta", - Delta: &types.Delta{ - StopReason: stopReason, - }, - Usage: usageInfoToAnthropic(nil), + terminalStopReason = "tool_use" } - if err := writeSSEEvent(w, msgDelta); err != nil { - return ErrClientDisconnected - } - stopSent = true + } + msgDelta := types.MessageEvent{ + Type: "message_delta", + Delta: &types.Delta{ + StopReason: terminalStopReason, + }, + Usage: usageInfoToAnthropic(terminalUsage), + } + if err := writeSSEEvent(w, msgDelta); err != nil { + return ErrClientDisconnected } // Send message_stop event to signal stream completion. @@ -364,7 +380,8 @@ func (h *StreamHandler) processSSELine( contentIndex *int, contentStarted *bool, reasoningStarted *bool, - stopSent *bool, + terminalStopReason *string, + terminalUsage **types.UsageInfo, toolUseCount *int, startedToolCalls map[int]int, originalModel string, @@ -425,7 +442,7 @@ func (h *StreamHandler) processSSELine( if !*contentStarted { // If reasoning was already started, close it first if *reasoningStarted { - if err := writeContentBlockStop(w, *contentIndex); err != nil { + if err := writeThinkingBlockStop(w, *contentIndex); err != nil { return ErrClientDisconnected } *contentIndex++ @@ -480,27 +497,11 @@ func (h *StreamHandler) processSSELine( return nil } *decodeErrors = 0 + if chunk.Usage != nil { + *terminalUsage = chunk.Usage + } if len(chunk.Choices) == 0 { - if chunk.Usage != nil { - if *stopSent { - // Stop reason already sent — emit usage-only message_delta (no duplicate stop_reason). - event := types.MessageEvent{ - Type: "message_delta", - Delta: &types.Delta{}, - Usage: usageInfoToAnthropic(chunk.Usage), - } - if err := writeSSEEvent(w, event); err != nil { - return ErrClientDisconnected - } - flusher.Flush() - } else { - if err := h.sendUsageDelta(w, flusher, chunk.Usage); err != nil { - return err - } - *stopSent = true - } - } return nil } @@ -525,7 +526,7 @@ func (h *StreamHandler) processSSELine( startEvent := types.MessageEvent{ Type: "content_block_start", Index: contentIndex, - ContentBlock: &types.ContentBlock{Type: "thinking", Thinking: ""}, + ContentBlock: &types.ContentBlock{Type: "thinking", Thinking: "", Signature: thinkingSignaturePlaceholder}, } if err := writeSSEEvent(w, startEvent); err != nil { return ErrClientDisconnected @@ -552,11 +553,7 @@ func (h *StreamHandler) processSSELine( if !*contentStarted { // If reasoning was already started, close it first if *reasoningStarted { - stopEvent := types.MessageEvent{ - Type: "content_block_stop", - Index: contentIndex, - } - if err := writeSSEEvent(w, stopEvent); err != nil { + if err := writeThinkingBlockStop(w, *contentIndex); err != nil { return ErrClientDisconnected } *contentIndex++ @@ -610,11 +607,13 @@ func (h *StreamHandler) processSSELine( // advance contentIndex — the close itself clears the flags. hadStartedBlock := *contentStarted || *reasoningStarted if hadStartedBlock { - stopEvent := types.MessageEvent{ - Type: "content_block_stop", - Index: contentIndex, + var err error + if *reasoningStarted { + err = writeThinkingBlockStop(w, *contentIndex) + } else { + err = writeContentBlockStop(w, *contentIndex) } - if err := writeSSEEvent(w, stopEvent); err != nil { + if err != nil { return ErrClientDisconnected } *contentStarted = false @@ -675,16 +674,16 @@ func (h *StreamHandler) processSSELine( // Handle finish reason if choice.FinishReason != "" { // Close any open content block (reasoning or text) - if *contentStarted || *reasoningStarted { - stopEvent := types.MessageEvent{ - Type: "content_block_stop", - Index: contentIndex, + if *reasoningStarted { + if err := writeThinkingBlockStop(w, *contentIndex); err != nil { + return ErrClientDisconnected } - if err := writeSSEEvent(w, stopEvent); err != nil { + *reasoningStarted = false + } else if *contentStarted { + if err := writeContentBlockStop(w, *contentIndex); err != nil { return ErrClientDisconnected } *contentStarted = false - *reasoningStarted = false } // Close any open tool_use blocks in ascending index order. @@ -717,38 +716,13 @@ func (h *StreamHandler) processSSELine( } *toolUseCount = 0 - msgDelta := types.MessageEvent{ - Type: "message_delta", - Delta: &types.Delta{ - StopReason: h.responseTransformer.mapFinishReason(choice.FinishReason), - }, - Usage: usageInfoToAnthropic(chunk.Usage), - } - if err := writeSSEEvent(w, msgDelta); err != nil { - return ErrClientDisconnected - } - *stopSent = true + *terminalStopReason = h.responseTransformer.mapFinishReason(choice.FinishReason) flusher.Flush() } return nil } -func (h *StreamHandler) sendUsageDelta(w http.ResponseWriter, flusher http.Flusher, usage *types.UsageInfo) error { - event := types.MessageEvent{ - Type: "message_delta", - Delta: &types.Delta{ - StopReason: "end_turn", - }, - Usage: usageInfoToAnthropic(usage), - } - if err := writeSSEEvent(w, event); err != nil { - return ErrClientDisconnected - } - flusher.Flush() - return nil -} - func usageInfoToAnthropic(usage *types.UsageInfo) *types.Usage { if usage == nil { return &types.Usage{ @@ -778,6 +752,17 @@ func writeContentBlockStop(w http.ResponseWriter, index int) error { }) } +func writeThinkingBlockStop(w http.ResponseWriter, index int) error { + if err := writeSSEEvent(w, types.MessageEvent{ + Type: "content_block_delta", + Index: &index, + Delta: &types.Delta{Type: "signature_delta", Signature: thinkingSignaturePlaceholder}, + }); err != nil { + return err + } + return writeContentBlockStop(w, index) +} + // writeSSEEvent writes a single SSE event to the HTTP response writer. // Format: "event: \ndata: \n\n" func writeSSEEvent(w http.ResponseWriter, event types.MessageEvent) error { diff --git a/internal/transformer/stream_test.go b/internal/transformer/stream_test.go index fd143b20..225659f7 100644 --- a/internal/transformer/stream_test.go +++ b/internal/transformer/stream_test.go @@ -118,9 +118,10 @@ func TestProxyStream_ReasoningContentFastPath(t *testing.T) { events := parseSSEEvents(t, w.buf.String()) - // Expected: message_start, content_block_start, 2x content_block_delta, content_block_stop, message_delta, message_stop - if len(events) != 7 { - t.Fatalf("expected 7 events, got %d: %+v", len(events), events) + // Expected: message_start, content_block_start, 2x thinking_delta, + // signature_delta, content_block_stop, message_delta, message_stop. + if len(events) != 8 { + t.Fatalf("expected 8 events, got %d: %+v", len(events), events) } if events[0].Type != "message_start" { @@ -147,14 +148,57 @@ func TestProxyStream_ReasoningContentFastPath(t *testing.T) { if got := events[3].Delta.Thinking; got != " step by step" { t.Errorf("event[3].Delta.Thinking = %q, want %q", got, " step by step") } + if got := events[1].ContentBlock.Signature; got != thinkingSignaturePlaceholder { + t.Errorf("event[1].ContentBlock.Signature = %q, want %q", got, thinkingSignaturePlaceholder) + } + if events[4].Type != "content_block_delta" || events[4].Delta == nil || events[4].Delta.Type != "signature_delta" { + t.Errorf("event[4] = %+v, want signature_delta", events[4]) + } + if got := events[4].Delta.Signature; got != thinkingSignaturePlaceholder { + t.Errorf("event[4].Delta.Signature = %q, want %q", got, thinkingSignaturePlaceholder) + } + if events[5].Type != "content_block_stop" { + t.Errorf("event[5].Type = %q, want content_block_stop", events[5].Type) + } + if events[6].Type != "message_delta" { + t.Errorf("event[6].Type = %q, want message_delta", events[6].Type) + } + if events[7].Type != "message_stop" { + t.Errorf("event[7].Type = %q, want message_stop", events[7].Type) + } +} + +func TestProxyStream_ReasoningSignatureAndMergedMessageDelta(t *testing.T) { + handler := NewStreamHandler() + w := newMockResponseWriter() + body := sseLines( + `{"choices":[{"delta":{"reasoning_content":"Thinking..."}}]}`, + `{"choices":[{"delta":{},"finish_reason":"stop"}]}`, + `{"choices":[],"usage":{"prompt_tokens":100,"completion_tokens":20,"total_tokens":120}}`, + ) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + if err := handler.ProxyStream(w, body, "kimi-k2.6", ctx, 0, cancel); err != nil { + t.Fatalf("ProxyStream error: %v", err) + } + + events := parseSSEEvents(t, w.buf.String()) + if len(events) != 7 { + t.Fatalf("expected 7 events, got %d: %+v", len(events), events) + } + if events[3].Delta == nil || events[3].Delta.Type != "signature_delta" || events[3].Delta.Signature == "" { + t.Fatalf("event[3] = %+v, want non-empty signature_delta", events[3]) + } if events[4].Type != "content_block_stop" { - t.Errorf("event[4].Type = %q, want content_block_stop", events[4].Type) + t.Fatalf("event[4] = %+v, want content_block_stop", events[4]) } - if events[5].Type != "message_delta" { - t.Errorf("event[5].Type = %q, want message_delta", events[5].Type) + if events[5].Type != "message_delta" || events[5].Delta == nil || events[5].Delta.StopReason != "end_turn" { + t.Fatalf("event[5] = %+v, want terminal message_delta", events[5]) } - if events[6].Type != "message_stop" { - t.Errorf("event[6].Type = %q, want message_stop", events[6].Type) + if events[5].Usage == nil || events[5].Usage.InputTokens != 100 || events[5].Usage.OutputTokens != 20 { + t.Fatalf("event[5].Usage = %+v, want input=100 output=20", events[5].Usage) } } @@ -177,23 +221,24 @@ func TestProxyStream_ReasoningThenText(t *testing.T) { events := parseSSEEvents(t, w.buf.String()) - // Expected: message_start, content_block_start(thinking, idx=0), thinking_delta, content_block_stop(idx=0), - // content_block_start(text, idx=1), text_delta x2, content_block_stop(idx=1), message_delta, message_stop - if len(events) != 10 { - t.Fatalf("expected 10 events, got %d: %+v", len(events), events) + // Expected: message_start, content_block_start(thinking, idx=0), thinking_delta, + // signature_delta, content_block_stop(idx=0), content_block_start(text, idx=1), + // text_delta x2, content_block_stop(idx=1), message_delta, message_stop. + if len(events) != 11 { + t.Fatalf("expected 11 events, got %d: %+v", len(events), events) } // Verify indexes if got := *events[1].Index; got != 0 { t.Errorf("thinking start index = %d, want 0", got) } - if got := *events[3].Index; got != 0 { + if got := *events[4].Index; got != 0 { t.Errorf("thinking stop index = %d, want 0", got) } - if got := *events[4].Index; got != 1 { + if got := *events[5].Index; got != 1 { t.Errorf("text start index = %d, want 1", got) } - if got := *events[7].Index; got != 1 { + if got := *events[8].Index; got != 1 { t.Errorf("text stop index = %d, want 1", got) } @@ -204,11 +249,14 @@ func TestProxyStream_ReasoningThenText(t *testing.T) { if got := events[2].Delta.Type; got != "thinking_delta" { t.Errorf("event[2].Delta.Type = %q, want thinking_delta", got) } - if events[4].ContentBlock == nil || events[4].ContentBlock.Type != "text" { - t.Errorf("event[4].ContentBlock = %+v, want text block", events[4].ContentBlock) + if events[3].Delta == nil || events[3].Delta.Type != "signature_delta" { + t.Errorf("event[3] = %+v, want signature_delta", events[3]) } - if got := events[5].Delta.Type; got != "text_delta" { - t.Errorf("event[5].Delta.Type = %q, want text_delta", got) + if events[5].ContentBlock == nil || events[5].ContentBlock.Type != "text" { + t.Errorf("event[5].ContentBlock = %+v, want text block", events[5].ContentBlock) + } + if got := events[6].Delta.Type; got != "text_delta" { + t.Errorf("event[6].Delta.Type = %q, want text_delta", got) } } @@ -362,10 +410,8 @@ func TestProxyStream_PartialCacheTokensStreaming(t *testing.T) { } } -// TestProxyStream_NoDuplicateMessageDelta verifies that when finish_reason and -// usage arrive in separate chunks, only ONE message_delta with a stop_reason -// is emitted. Usage may arrive in a separate message_delta (without stop_reason) -// if the upstream sends them in separate chunks. +// TestProxyStream_NoDuplicateMessageDelta verifies that finish_reason and usage +// arriving in separate chunks are merged into exactly one message_delta. func TestProxyStream_NoDuplicateMessageDelta(t *testing.T) { handler := NewStreamHandler() w := newMockResponseWriter() @@ -384,25 +430,21 @@ func TestProxyStream_NoDuplicateMessageDelta(t *testing.T) { events := parseSSEEvents(t, w.buf.String()) - // Count message_delta events with a stop_reason - var stopDeltas []types.MessageEvent + var messageDeltas []types.MessageEvent for _, ev := range events { - if ev.Type == "message_delta" && ev.Delta != nil && ev.Delta.StopReason != "" { - stopDeltas = append(stopDeltas, ev) + if ev.Type == "message_delta" { + messageDeltas = append(messageDeltas, ev) } } - if len(stopDeltas) != 1 { - t.Fatalf("expected exactly 1 message_delta with stop_reason, got %d: %+v", len(stopDeltas), stopDeltas) + if len(messageDeltas) != 1 { + t.Fatalf("expected exactly 1 message_delta, got %d: %+v", len(messageDeltas), messageDeltas) } - - // Verify usage is somewhere in the stream - var totalUsage *types.Usage - for _, ev := range events { - if ev.Usage != nil { - totalUsage = ev.Usage - } + if messageDeltas[0].Delta == nil || messageDeltas[0].Delta.StopReason != "end_turn" { + t.Fatalf("message_delta = %+v, want stop_reason=end_turn", messageDeltas[0]) } + + totalUsage := messageDeltas[0].Usage if totalUsage == nil { t.Fatalf("no usage found in stream: %+v", events) return @@ -431,9 +473,9 @@ func TestProxyStream_ReasoningJSONFallback(t *testing.T) { events := parseSSEEvents(t, w.buf.String()) - // Expected: message_start, content_block_start, content_block_delta, content_block_stop, message_delta, message_stop - if len(events) != 6 { - t.Fatalf("expected 6 events, got %d: %+v", len(events), events) + // Expected: message_start, thinking start/delta/signature/stop, message_delta, message_stop. + if len(events) != 7 { + t.Fatalf("expected 7 events, got %d: %+v", len(events), events) } if events[1].Type != "content_block_start" || events[1].ContentBlock == nil || events[1].ContentBlock.Type != "thinking" { @@ -445,6 +487,9 @@ func TestProxyStream_ReasoningJSONFallback(t *testing.T) { if events[2].Delta.Thinking != "Reasoning via JSON" { t.Errorf("event[2].Delta.Thinking = %q, want %q", events[2].Delta.Thinking, "Reasoning via JSON") } + if events[3].Delta == nil || events[3].Delta.Type != "signature_delta" { + t.Errorf("event[3] = %+v, want signature_delta", events[3]) + } } func TestProxyStream_EmptyReasoningContentSkipped(t *testing.T) { @@ -499,11 +544,11 @@ func TestProxyStream_ReasoningAndContentInSameChunk(t *testing.T) { events := parseSSEEvents(t, w.buf.String()) - // message_start + thinking_start + thinking_delta + thinking_stop + + // message_start + thinking_start + thinking_delta + signature_delta + thinking_stop + // text_start + text_delta("Hello") + text_delta(" world") + text_stop + - // message_delta + message_stop = 10 - if len(events) != 10 { - t.Fatalf("expected 10 events, got %d: %+v", len(events), events) + // message_delta + message_stop = 11 + if len(events) != 11 { + t.Fatalf("expected 11 events, got %d: %+v", len(events), events) } // Block 0: thinking @@ -516,28 +561,31 @@ func TestProxyStream_ReasoningAndContentInSameChunk(t *testing.T) { if events[2].Delta.Thinking != "Thinking..." { t.Errorf("event[2].Delta.Thinking = %q, want %q", events[2].Delta.Thinking, "Thinking...") } - if events[3].Type != "content_block_stop" { - t.Errorf("event[3].Type = %q, want content_block_stop", events[3].Type) + if events[3].Delta == nil || events[3].Delta.Type != "signature_delta" { + t.Errorf("event[3] = %+v, want signature_delta", events[3]) + } + if events[4].Type != "content_block_stop" { + t.Errorf("event[4].Type = %q, want content_block_stop", events[4].Type) } // Block 1: text - if events[4].Type != "content_block_start" || events[4].ContentBlock == nil || events[4].ContentBlock.Type != "text" { - t.Errorf("event[4] = %+v, want content_block_start(text)", events[4]) - } - if events[5].Type != "content_block_delta" || events[5].Delta.Type != "text_delta" { - t.Errorf("event[5] = %+v, want content_block_delta(text_delta)", events[5]) - } - if events[5].Delta.Text != "Hello" { - t.Errorf("event[5].Delta.Text = %q, want Hello", events[5].Delta.Text) + if events[5].Type != "content_block_start" || events[5].ContentBlock == nil || events[5].ContentBlock.Type != "text" { + t.Errorf("event[5] = %+v, want content_block_start(text)", events[5]) } if events[6].Type != "content_block_delta" || events[6].Delta.Type != "text_delta" { t.Errorf("event[6] = %+v, want content_block_delta(text_delta)", events[6]) } - if events[6].Delta.Text != " world" { - t.Errorf("event[6].Delta.Text = %q, want \" world\"", events[6].Delta.Text) + if events[6].Delta.Text != "Hello" { + t.Errorf("event[6].Delta.Text = %q, want Hello", events[6].Delta.Text) } - if events[7].Type != "content_block_stop" { - t.Errorf("event[7].Type = %q, want content_block_stop", events[7].Type) + if events[7].Type != "content_block_delta" || events[7].Delta.Type != "text_delta" { + t.Errorf("event[7] = %+v, want content_block_delta(text_delta)", events[7]) + } + if events[7].Delta.Text != " world" { + t.Errorf("event[7].Delta.Text = %q, want \" world\"", events[7].Delta.Text) + } + if events[8].Type != "content_block_stop" { + t.Errorf("event[8].Type = %q, want content_block_stop", events[8].Type) } } @@ -567,11 +615,11 @@ func TestProxyStream_ReasoningBeforeContentFastPathRegression(t *testing.T) { events := parseSSEEvents(t, w.buf.String()) - // message_start + thinking_start + thinking_delta + thinking_stop + + // message_start + thinking_start + thinking_delta + signature_delta + thinking_stop + // text_start + text_delta("Hello") + text_delta(" world") + text_stop + - // message_delta + message_stop = 10 - if len(events) != 10 { - t.Fatalf("expected 10 events, got %d: %+v", len(events), events) + // message_delta + message_stop = 11 + if len(events) != 11 { + t.Fatalf("expected 11 events, got %d: %+v", len(events), events) } // Block 0: thinking (must NOT be lost) @@ -584,13 +632,16 @@ func TestProxyStream_ReasoningBeforeContentFastPathRegression(t *testing.T) { if events[2].Delta.Thinking != "Thinking..." { t.Errorf("event[2].Delta.Thinking = %q, want %q", events[2].Delta.Thinking, "Thinking...") } + if events[3].Delta == nil || events[3].Delta.Type != "signature_delta" { + t.Errorf("event[3] = %+v, want signature_delta", events[3]) + } // Block 1: text - if events[4].Type != "content_block_start" || events[4].ContentBlock == nil || events[4].ContentBlock.Type != "text" { - t.Errorf("event[4] = %+v, want content_block_start(text)", events[4]) + if events[5].Type != "content_block_start" || events[5].ContentBlock == nil || events[5].ContentBlock.Type != "text" { + t.Errorf("event[5] = %+v, want content_block_start(text)", events[5]) } - if events[5].Delta.Text != "Hello" { - t.Errorf("event[5].Delta.Text = %q, want Hello", events[5].Delta.Text) + if events[6].Delta.Text != "Hello" { + t.Errorf("event[6].Delta.Text = %q, want Hello", events[6].Delta.Text) } } @@ -890,11 +941,14 @@ func TestProxyStream_MixedReasoningAndToolCall(t *testing.T) { if events[1].Type != "content_block_start" || events[1].ContentBlock == nil || events[1].ContentBlock.Type != "thinking" { t.Errorf("event[1] = %+v, want content_block_start(thinking)", events[1]) } - if events[3].Type != "content_block_stop" || events[3].Index == nil || *events[3].Index != 0 { - t.Errorf("event[3] = %+v, want content_block_stop(index=0)", events[3]) + if events[3].Delta == nil || events[3].Delta.Type != "signature_delta" { + t.Errorf("event[3] = %+v, want signature_delta", events[3]) } - if events[4].Type != "content_block_start" || events[4].ContentBlock == nil || events[4].ContentBlock.Type != "tool_use" { - t.Errorf("event[4] = %+v, want content_block_start(tool_use)", events[4]) + if events[4].Type != "content_block_stop" || events[4].Index == nil || *events[4].Index != 0 { + t.Errorf("event[4] = %+v, want content_block_stop(index=0)", events[4]) + } + if events[5].Type != "content_block_start" || events[5].ContentBlock == nil || events[5].ContentBlock.Type != "tool_use" { + t.Errorf("event[5] = %+v, want content_block_start(tool_use)", events[5]) } var stopIndices []int diff --git a/pkg/types/anthropic.go b/pkg/types/anthropic.go index 4d72e35a..8295e950 100644 --- a/pkg/types/anthropic.go +++ b/pkg/types/anthropic.go @@ -285,6 +285,7 @@ type Delta struct { Type string `json:"type,omitempty"` Text string `json:"text,omitempty"` Thinking string `json:"thinking,omitempty"` + Signature string `json:"signature,omitempty"` PartialJSON string `json:"partial_json,omitempty"` StopReason string `json:"stop_reason,omitempty"` }