diff --git a/internal/providers/deepseek/deepseek_test.go b/internal/providers/deepseek/deepseek_test.go index aac8a5f43..83cc28b80 100644 --- a/internal/providers/deepseek/deepseek_test.go +++ b/internal/providers/deepseek/deepseek_test.go @@ -152,6 +152,54 @@ func TestResponses_TranslatesToChatCompletions(t *testing.T) { } } +func TestResponses_ReplaysReasoningContentForToolCall(t *testing.T) { + var gotBody map[string]any + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if err := json.NewDecoder(r.Body).Decode(&gotBody); err != nil { + http.Error(w, "decode error", http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{ + "id":"chatcmpl-deepseek", + "created":1, + "model":"deepseek-v4-pro", + "choices":[{"index":0,"message":{"role":"assistant","content":"done"},"finish_reason":"stop"}], + "usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2} + }`)) + })) + defer server.Close() + + var req core.ResponsesRequest + if err := json.Unmarshal([]byte(`{ + "model":"deepseek-v4-pro", + "input":[ + {"type":"reasoning","summary":[],"content":[{"type":"reasoning_text","text":"Need the weather."}]}, + {"type":"function_call","call_id":"call_1","name":"lookup","arguments":"{}"}, + {"type":"function_call_output","call_id":"call_1","output":"sunny"} + ] + }`), &req); err != nil { + t.Fatalf("json.Unmarshal() error = %v", err) + } + + provider := NewWithHTTPClient("deepseek-key", server.URL, server.Client(), llmclient.Hooks{}) + if _, err := provider.Responses(context.Background(), &req); err != nil { + t.Fatalf("Responses() error = %v", err) + } + + messages, _ := gotBody["messages"].([]any) + if len(messages) != 2 { + t.Fatalf("messages = %#v, want assistant call and tool result", gotBody["messages"]) + } + assistant, _ := messages[0].(map[string]any) + if assistant["role"] != "assistant" || assistant["reasoning_content"] != "Need the weather." { + t.Fatalf("assistant = %#v", assistant) + } + if calls, _ := assistant["tool_calls"].([]any); len(calls) != 1 { + t.Fatalf("assistant tool_calls = %#v", assistant["tool_calls"]) + } +} + func TestStreamResponses_TranslatesToChatCompletions(t *testing.T) { var gotPath string var gotBody map[string]any diff --git a/internal/providers/responses_adapter_test.go b/internal/providers/responses_adapter_test.go index c878b85c0..c3a1c0229 100644 --- a/internal/providers/responses_adapter_test.go +++ b/internal/providers/responses_adapter_test.go @@ -731,7 +731,7 @@ func TestConvertResponsesRequestToChat_RejectsUnknownInputItemTypes(t *testing.T var req core.ResponsesRequest if err := json.Unmarshal([]byte(`{ "model":"test-model", - "input":[{"type":"reasoning","id":"rs_123","summary":[]}] + "input":[{"type":"computer_call","id":"cc_123"}] }`), &req); err != nil { t.Fatalf("json.Unmarshal() error = %v", err) } @@ -740,8 +740,97 @@ func TestConvertResponsesRequestToChat_RejectsUnknownInputItemTypes(t *testing.T if err == nil { t.Fatal("expected error, got nil") } - if !strings.Contains(err.Error(), `unsupported input item type "reasoning"`) { - t.Fatalf("error = %v, want unsupported reasoning item", err) + if !strings.Contains(err.Error(), `unsupported input item type "computer_call"`) { + t.Fatalf("error = %v, want unsupported computer_call item", err) + } +} + +// Reasoning from an ordinary assistant turn is accepted but omitted because +// chat providers do not need it on the following user turn. +func TestConvertResponsesRequestToChat_DropsReasoningWithoutToolCall(t *testing.T) { + var req core.ResponsesRequest + if err := json.Unmarshal([]byte(`{ + "model":"test-model", + "input":[ + {"type":"message","role":"user","content":"hello"}, + {"type":"reasoning","id":"rs_123","summary":[{"type":"summary_text","text":"thinking..."}]}, + {"type":"message","role":"assistant","content":"hi there"} + ] + }`), &req); err != nil { + t.Fatalf("json.Unmarshal() error = %v", err) + } + + chatReq, err := ConvertResponsesRequestToChat(&req) + if err != nil { + t.Fatalf("ConvertResponsesRequestToChat() error = %v", err) + } + if len(chatReq.Messages) != 2 { + t.Fatalf("Messages = %#v, want exactly the user and assistant messages (reasoning dropped)", chatReq.Messages) + } + if chatReq.Messages[0].Role != "user" || chatReq.Messages[1].Role != "assistant" { + t.Fatalf("Messages = %#v, want [user, assistant]", chatReq.Messages) + } + if got := chatReq.Messages[1].ExtraFields.Lookup("reasoning_content"); got != nil { + t.Fatalf("reasoning_content = %s, want omitted without a tool call", got) + } +} + +// DeepSeek requires reasoning_content to be replayed on the assistant message +// that made a tool call. Codex echoes Responses output items back as input, so +// the reasoning item and function-call item must be reassembled here. +func TestConvertResponsesRequestToChat_ReplaysReasoningWithToolCall(t *testing.T) { + var req core.ResponsesRequest + if err := json.Unmarshal([]byte(`{ + "model":"deepseek-v4-pro", + "input":[ + {"type":"message","role":"user","content":"weather?"}, + {"type":"reasoning","id":"rs_123","summary":[],"content":[{"type":"reasoning_text","text":"Need to check the weather."}]}, + {"type":"message","id":"msg_123","role":"assistant","content":"I'll check."}, + {"type":"function_call","call_id":"call_123","name":"lookup_weather","arguments":"{\"city\":\"Warsaw\"}"}, + {"type":"function_call_output","call_id":"call_123","output":"sunny"} + ] + }`), &req); err != nil { + t.Fatalf("json.Unmarshal() error = %v", err) + } + + chatReq, err := ConvertResponsesRequestToChat(&req) + if err != nil { + t.Fatalf("ConvertResponsesRequestToChat() error = %v", err) + } + if len(chatReq.Messages) != 3 { + t.Fatalf("Messages = %#v, want user, assistant tool call, and tool result", chatReq.Messages) + } + assistant := chatReq.Messages[1] + if assistant.Role != "assistant" || core.ExtractTextContent(assistant.Content) != "I'll check." || len(assistant.ToolCalls) != 1 { + t.Fatalf("assistant message = %#v, want merged text and tool call", assistant) + } + var reasoning string + if err := json.Unmarshal(assistant.ExtraFields.Lookup("reasoning_content"), &reasoning); err != nil { + t.Fatalf("reasoning_content decode error = %v", err) + } + if reasoning != "Need to check the weather." { + t.Fatalf("reasoning_content = %q", reasoning) + } + if chatReq.Messages[2].Role != "tool" || chatReq.Messages[2].ToolCallID != "call_123" { + t.Fatalf("tool message = %#v", chatReq.Messages[2]) + } +} + +func TestConvertResponsesRequestToChat_NormalizesDeveloperRole(t *testing.T) { + tests := map[string]any{ + "typed": []core.ResponsesInputElement{{Type: "message", Role: "developer", Content: "Be concise."}}, + "map": []any{map[string]any{"type": "message", "role": "developer", "content": "Be concise."}}, + } + for name, input := range tests { + t.Run(name, func(t *testing.T) { + chatReq, err := ConvertResponsesRequestToChat(&core.ResponsesRequest{Model: "test-model", Input: input}) + if err != nil { + t.Fatalf("ConvertResponsesRequestToChat() error = %v", err) + } + if len(chatReq.Messages) != 1 || chatReq.Messages[0].Role != "system" { + t.Fatalf("Messages = %#v, want one system message", chatReq.Messages) + } + }) } } @@ -1051,6 +1140,35 @@ func TestConvertChatResponseToResponses(t *testing.T) { } } +func TestConvertChatResponseToResponses_PreservesRawReasoning(t *testing.T) { + resp := &core.ChatResponse{ + ID: "chatcmpl-reasoning", + Model: "deepseek-v4-pro", + Created: 1, + Choices: []core.Choice{{ + Message: core.ResponseMessage{ + Role: "assistant", + Content: "done", + ExtraFields: core.UnknownJSONFieldsFromMap(map[string]json.RawMessage{ + "reasoning_content": json.RawMessage(`"raw trace"`), + }), + }, + }}, + } + + result := ConvertChatResponseToResponses(resp) + if len(result.Output) != 2 || result.Output[0].Type != "reasoning" || result.Output[1].Type != "message" { + t.Fatalf("Output = %#v, want reasoning then message", result.Output) + } + reasoning := result.Output[0] + if len(reasoning.Content) != 1 || reasoning.Content[0].Type != "reasoning_text" || reasoning.Content[0].Text != "raw trace" { + t.Fatalf("reasoning content = %#v", reasoning.Content) + } + if reasoning.ExtraFields.Lookup("summary") == nil { + t.Fatal("reasoning summary array missing") + } +} + func TestConvertChatResponseToResponses_PreservesStructuredAssistantContent(t *testing.T) { resp := &core.ChatResponse{ ID: "chatcmpl-structured", diff --git a/internal/providers/responses_converter.go b/internal/providers/responses_converter.go index 5db16f81a..9a788cea4 100644 --- a/internal/providers/responses_converter.go +++ b/internal/providers/responses_converter.go @@ -12,6 +12,7 @@ import ( "github.com/google/uuid" + "github.com/enterpilot/gomodel/internal/core" "github.com/enterpilot/gomodel/internal/streaming" ) @@ -19,20 +20,21 @@ import ( // and converts it to Responses API format. // Used by providers that have OpenAI-compatible streaming (Groq, Gemini, etc.) type OpenAIResponsesStreamConverter struct { - reader io.ReadCloser - model string - provider string - responseID string - createdAt int64 - output *ResponsesOutputEventState - toolCalls map[int]*ResponsesOutputToolCallState - buffer streaming.StreamBuffer - lineBuffer streaming.StreamBuffer - readBuf []byte - closed bool - sentCreate bool - sentDone bool - cachedUsage json.RawMessage // Stores usage from final chunk for inclusion in response.completed + reader io.ReadCloser + model string + provider string + responseID string + createdAt int64 + output *ResponsesOutputEventState + toolCalls map[int]*ResponsesOutputToolCallState + assistantOutputIndex int + buffer streaming.StreamBuffer + lineBuffer streaming.StreamBuffer + readBuf []byte + closed bool + sentCreate bool + sentDone bool + cachedUsage json.RawMessage // Stores usage from final chunk for inclusion in response.completed } // NewOpenAIResponsesStreamConverter creates a new converter that transforms @@ -60,13 +62,25 @@ type openAIStreamChunk struct { Usage json.RawMessage `json:"usage"` Choices []struct { Delta struct { - Content string `json:"content"` - ToolCalls []openAIChunkToolCall `json:"tool_calls"` + Content string `json:"content"` + ReasoningContent string `json:"reasoning_content"` + ToolCalls []openAIChunkToolCall `json:"tool_calls"` } `json:"delta"` FinishReason string `json:"finish_reason"` } `json:"choices"` } +// responsesStreamUsage is the conservative Responses API representation of a +// Chat Completions usage object. Keeping this local avoids the generic unknown- +// field preservation path on every streamed response. +type responsesStreamUsage struct { + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + TotalTokens int `json:"total_tokens"` + InputTokensDetails *core.PromptTokensDetails `json:"input_tokens_details,omitempty"` + OutputTokensDetails *core.CompletionTokensDetails `json:"output_tokens_details,omitempty"` +} + type openAIChunkToolCall struct { Index *int `json:"index"` ID string `json:"id"` @@ -80,6 +94,9 @@ func (sc *OpenAIResponsesStreamConverter) ensureToolCallState(index int) *Respon state := sc.toolCalls[index] if state == nil { outputIndex := index + if sc.output.ReasoningReserved() { + outputIndex++ + } if sc.output.AssistantReserved() { outputIndex++ } @@ -89,13 +106,56 @@ func (sc *OpenAIResponsesStreamConverter) ensureToolCallState(index int) *Respon return state } +// reasoningOutputIndex is always 0: reasoning content precedes any visible +// text or tool call in a reasoning model's stream, so the reasoning item (if +// any) always claims the first output slot. +const reasoningOutputIndex = 0 + +func (sc *OpenAIResponsesStreamConverter) reserveReasoningOutput() { + if sc.output.ReasoningReserved() { + return + } + sc.output.ReserveReasoning() + for _, state := range sc.toolCalls { + if state != nil && !state.Started { + state.OutputIndex++ + } + } +} + +func (sc *OpenAIResponsesStreamConverter) outputAlreadyStarted() bool { + if sc.output.AssistantStarted() { + return true + } + for _, state := range sc.toolCalls { + if state != nil && state.Started { + return true + } + } + return false +} + func (sc *OpenAIResponsesStreamConverter) reserveAssistantOutput() { if sc.output.AssistantReserved() { return } + + // Items that have already been emitted cannot move. Place the assistant + // after them, then shift only pending tool calls that would otherwise + // occupy the same or a later slot. + outputIndex := 0 + if sc.output.ReasoningReserved() { + outputIndex++ + } + for _, state := range sc.toolCalls { + if state != nil && state.Started && state.OutputIndex >= outputIndex { + outputIndex = state.OutputIndex + 1 + } + } + sc.assistantOutputIndex = outputIndex sc.output.ReserveAssistant() for _, state := range sc.toolCalls { - if state != nil && !state.Started { + if state != nil && !state.Started && state.OutputIndex >= outputIndex { state.OutputIndex++ } } @@ -137,8 +197,9 @@ func (sc *OpenAIResponsesStreamConverter) completePendingToolCalls() string { func (sc *OpenAIResponsesStreamConverter) handleToolCallDeltas(toolCalls []openAIChunkToolCall) string { var out bytes.Buffer + out.WriteString(sc.output.CompleteReasoningOutput(reasoningOutputIndex)) if sc.output.AssistantStarted() && !sc.output.AssistantDone() { - out.WriteString(sc.output.CompleteAssistantOutput(0)) + out.WriteString(sc.output.CompleteAssistantOutput(sc.assistantOutputIndex)) } for _, toolCall := range toolCalls { @@ -209,6 +270,9 @@ func (sc *OpenAIResponsesStreamConverter) processChunk(data []byte) { } choice := &chunk.Choices[0] + if choice.Delta.ReasoningContent != "" { + sc.appendReasoningDelta(choice.Delta.ReasoningContent) + } if choice.Delta.Content != "" { sc.appendTextDelta(choice.Delta.Content) } @@ -250,6 +314,9 @@ func (sc *OpenAIResponsesStreamConverter) processChunkTolerant(data []byte) { return } if delta, ok := choice["delta"].(map[string]any); ok { + if reasoning, ok := delta["reasoning_content"].(string); ok && reasoning != "" { + sc.appendReasoningDelta(reasoning) + } if content, ok := delta["content"].(string); ok && content != "" { sc.appendTextDelta(content) } @@ -297,10 +364,25 @@ func normalizeToolCallIndex(value any) (int, bool) { } } +// appendReasoningDelta records raw provider reasoning and starts the reasoning +// item before emitting its reasoning_text delta. +func (sc *OpenAIResponsesStreamConverter) appendReasoningDelta(content string) { + // Output indexes cannot be rewritten after an item has been emitted. Some + // OpenAI-compatible providers send a stray late reasoning delta; dropping + // that extension is safer than producing two items at index 0 or reopening + // an item after response.output_item.done. + if sc.output.ReasoningDone() || (!sc.output.ReasoningReserved() && sc.outputAlreadyStarted()) { + return + } + sc.reserveReasoningOutput() + sc.buffer.AppendString(sc.output.AppendReasoningDelta(reasoningOutputIndex, content)) +} + // appendTextDelta records assistant text and emits its output_text.delta event. func (sc *OpenAIResponsesStreamConverter) appendTextDelta(content string) { + sc.buffer.AppendString(sc.output.CompleteReasoningOutput(reasoningOutputIndex)) sc.reserveAssistantOutput() - sc.buffer.AppendString(sc.output.StartAssistantOutput(0)) + sc.buffer.AppendString(sc.output.StartAssistantOutput(sc.assistantOutputIndex)) sc.output.AppendAssistantText(content) jsonData, err := json.Marshal(struct { Type string `json:"type"` @@ -322,7 +404,8 @@ func (sc *OpenAIResponsesStreamConverter) appendCompletedEvents() { return } sc.sentDone = true - sc.buffer.AppendString(sc.output.CompleteAssistantOutput(0)) + sc.buffer.AppendString(sc.output.CompleteReasoningOutput(reasoningOutputIndex)) + sc.buffer.AppendString(sc.output.CompleteAssistantOutput(sc.assistantOutputIndex)) sc.buffer.AppendString(sc.completePendingToolCalls()) responseData := map[string]any{ "id": sc.responseID, @@ -332,9 +415,14 @@ func (sc *OpenAIResponsesStreamConverter) appendCompletedEvents() { "provider": sc.provider, "created_at": sc.createdAt, } - // Include usage data if captured from OpenAI stream + // Include usage data if captured from OpenAI stream, renamed from Chat + // Completions field names (prompt_tokens/completion_tokens) to the + // Responses API's (input_tokens/output_tokens) — clients like Codex + // require the latter and fail to parse response.completed without them. if sc.cachedUsage != nil { - responseData["usage"] = sc.cachedUsage + if usage, ok := chatUsageToResponsesUsage(sc.cachedUsage); ok { + responseData["usage"] = usage + } } doneEvent := map[string]any{ "type": "response.completed", @@ -403,6 +491,35 @@ func (sc *OpenAIResponsesStreamConverter) appendFailedEvents(raw json.RawMessage sc.buffer.AppendString("\n\ndata: [DONE]\n\n") } +// chatUsageToResponsesUsage renames a valid Chat Completions usage object into +// the conservative Responses API shape. Malformed usage is omitted rather than +// making clients reject an otherwise successful response.completed event. +func chatUsageToResponsesUsage(raw json.RawMessage) (responsesStreamUsage, bool) { + var chatUsage struct { + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + TotalTokens int `json:"total_tokens"` + PromptTokensDetails *core.PromptTokensDetails `json:"prompt_tokens_details"` + CompletionTokensDetails *core.CompletionTokensDetails `json:"completion_tokens_details"` + } + chatUsage.PromptTokens = -1 + chatUsage.CompletionTokens = -1 + chatUsage.TotalTokens = -1 + if err := json.Unmarshal(raw, &chatUsage); err != nil { + return responsesStreamUsage{}, false + } + if chatUsage.PromptTokens < 0 || chatUsage.CompletionTokens < 0 || chatUsage.TotalTokens < 0 { + return responsesStreamUsage{}, false + } + return responsesStreamUsage{ + InputTokens: chatUsage.PromptTokens, + OutputTokens: chatUsage.CompletionTokens, + TotalTokens: chatUsage.TotalTokens, + InputTokensDetails: chatUsage.PromptTokensDetails, + OutputTokensDetails: chatUsage.CompletionTokensDetails, + }, true +} + func (sc *OpenAIResponsesStreamConverter) Read(p []byte) (n int, err error) { if sc.closed { return 0, io.EOF diff --git a/internal/providers/responses_converter_test.go b/internal/providers/responses_converter_test.go index ab66275b5..a0d86d171 100644 --- a/internal/providers/responses_converter_test.go +++ b/internal/providers/responses_converter_test.go @@ -3,6 +3,7 @@ package providers import ( "encoding/json" "io" + "slices" "strings" "testing" ) @@ -79,6 +80,235 @@ data: [DONE] } } +// TestOpenAIResponsesStreamConverter_ReasoningContent covers DeepSeek-style +// reasoning_content deltas. reasoning_content is raw reasoning, so it must be +// exposed as reasoning_text rather than mislabeled as a readable summary. The +// item must be registered before its first delta and the assistant message must +// shift to output_index 1. +func TestOpenAIResponsesStreamConverter_ReasoningContent(t *testing.T) { + mockStream := `data: {"id":"chatcmpl-1","object":"chat.completion.chunk","created":1,"model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"role":"assistant","reasoning_content":""},"finish_reason":null}]} + +data: {"id":"chatcmpl-1","object":"chat.completion.chunk","created":1,"model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"reasoning_content":"Think"},"finish_reason":null}]} + +data: {"id":"chatcmpl-1","object":"chat.completion.chunk","created":1,"model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"reasoning_content":"ing..."},"finish_reason":null}]} + +data: {"id":"chatcmpl-1","object":"chat.completion.chunk","created":1,"model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"Hi"},"finish_reason":null}]} + +data: {"id":"chatcmpl-1","object":"chat.completion.chunk","created":1,"model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"!"},"finish_reason":"stop"}]} + +data: [DONE] +` + + reader := io.NopCloser(strings.NewReader(mockStream)) + converter := NewOpenAIResponsesStreamConverter(reader, "deepseek-v4-pro", "deepseek") + + raw, err := io.ReadAll(converter) + if err != nil { + t.Fatalf("failed to read from converter: %v", err) + } + rawStr := string(raw) + + events := parseTestSSEEvents(t, rawStr) + activeItems := map[string]bool{} + var reasoningItemID string + var messageOutputIndex float64 = -1 + var reasoningDeltas strings.Builder + sawReasoningDone := false + sawSummaryEvent := false + + for _, event := range events { + if event.Done { + continue + } + switch event.Name { + case "response.output_item.added": + item, _ := event.Payload["item"].(map[string]any) + id, _ := item["id"].(string) + activeItems[id] = true + if item["type"] == "reasoning" { + reasoningItemID = id + summary, ok := item["summary"].([]any) + if !ok || len(summary) != 0 { + t.Fatalf("reasoning output_item.added must carry an empty summary array, got %#v", item["summary"]) + } + if item["status"] != "in_progress" { + t.Fatalf("reasoning output_item.added status = %#v, want in_progress", item["status"]) + } + if idx, _ := event.Payload["output_index"].(float64); idx != 0 { + t.Fatalf("reasoning output_index = %v, want 0", event.Payload["output_index"]) + } + } + if item["type"] == "message" { + messageOutputIndex, _ = event.Payload["output_index"].(float64) + } + case "response.output_item.done": + item, _ := event.Payload["item"].(map[string]any) + id, _ := item["id"].(string) + if item["type"] == "reasoning" { + content, _ := item["content"].([]any) + if len(content) != 1 { + t.Fatalf("completed reasoning content = %#v, want one reasoning_text part", item["content"]) + } + part, _ := content[0].(map[string]any) + if part["type"] != "reasoning_text" || part["text"] != "Thinking..." { + t.Fatalf("completed reasoning part = %#v", part) + } + } + delete(activeItems, id) + case "response.reasoning_text.delta": + itemID, _ := event.Payload["item_id"].(string) + if !activeItems[itemID] { + t.Fatalf("%s referenced item %q before its response.output_item.added", event.Name, itemID) + } + delta, _ := event.Payload["delta"].(string) + reasoningDeltas.WriteString(delta) + case "response.reasoning_text.done": + itemID, _ := event.Payload["item_id"].(string) + if !activeItems[itemID] { + t.Fatalf("%s referenced item %q after it closed", event.Name, itemID) + } + if event.Payload["text"] != "Thinking..." { + t.Fatalf("reasoning_text.done text = %#v", event.Payload["text"]) + } + sawReasoningDone = true + case "response.reasoning_summary_part.added", "response.reasoning_summary_text.delta", + "response.reasoning_summary_text.done", "response.reasoning_summary_part.done": + sawSummaryEvent = true + case "response.output_text.delta": + // Assistant text must only stream after the reasoning item closed. + if reasoningItemID != "" && activeItems[reasoningItemID] { + t.Fatalf("response.output_text.delta arrived while the reasoning item was still open") + } + } + } + + if reasoningItemID == "" { + t.Fatal("expected a reasoning output_item.added event") + } + if messageOutputIndex != 1 { + t.Fatalf("message output_index = %v, want 1 (after the reasoning item at index 0)", messageOutputIndex) + } + if reasoningDeltas.String() != "Thinking..." || !sawReasoningDone { + t.Fatalf("reasoning stream = %q, done=%v", reasoningDeltas.String(), sawReasoningDone) + } + if sawSummaryEvent { + t.Fatalf("raw reasoning_content must not emit reasoning summary events:\n%s", rawStr) + } + if len(activeItems) != 0 { + t.Fatalf("expected every output item to close by end of stream, still open: %#v", activeItems) + } +} + +func TestOpenAIResponsesStreamConverter_DropsLateReasoningWithoutCorruptingIndexes(t *testing.T) { + mockStream := `data: {"choices":[{"delta":{"content":"answer"},"finish_reason":null}]} + +data: {"choices":[{"delta":{"reasoning_content":"late trace"},"finish_reason":"stop"}]} + +data: [DONE] +` + + converter := NewOpenAIResponsesStreamConverter(io.NopCloser(strings.NewReader(mockStream)), "test-model", "mock") + raw, err := io.ReadAll(converter) + if err != nil { + t.Fatalf("ReadAll() error = %v", err) + } + + for _, event := range parseTestSSEEvents(t, string(raw)) { + if strings.HasPrefix(event.Name, "response.reasoning_") { + t.Fatalf("late reasoning produced %s:\n%s", event.Name, raw) + } + if event.Name != "response.output_item.added" && event.Name != "response.output_item.done" { + continue + } + item, _ := event.Payload["item"].(map[string]any) + if item["type"] == "message" && event.Payload["output_index"] != float64(0) { + t.Fatalf("assistant %s output_index = %#v, want 0", event.Name, event.Payload["output_index"]) + } + } +} + +func TestOpenAIResponsesStreamConverter_ReasoningToolOutputOrder(t *testing.T) { + tests := []struct { + name string + mockStream string + wantIndexes map[string]float64 + wantSequence []string + }{ + { + name: "reasoning then tool call", + mockStream: `data: {"choices":[{"delta":{"reasoning_content":"Need the weather."},"finish_reason":null}]} + +data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"lookup_weather","arguments":"{\"city\":\"Warsaw\"}"}}]},"finish_reason":null}]} + +data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]} + +data: [DONE] +`, + wantIndexes: map[string]float64{"reasoning": 0, "function_call": 1}, + wantSequence: []string{"reasoning", "function_call"}, + }, + { + name: "assistant after started tool call", + mockStream: `data: {"choices":[{"delta":{"reasoning_content":"Need a tool."},"finish_reason":null}]} + +data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"lookup","arguments":"{}"}}]},"finish_reason":null}]} + +data: {"choices":[{"delta":{"content":"Tool selected."},"finish_reason":"stop"}]} + +data: [DONE] +`, + wantIndexes: map[string]float64{"reasoning": 0, "function_call": 1, "message": 2}, + wantSequence: []string{"reasoning", "function_call", "message"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + converter := NewOpenAIResponsesStreamConverter(io.NopCloser(strings.NewReader(tt.mockStream)), "test-model", "mock") + raw, err := io.ReadAll(converter) + if err != nil { + t.Fatalf("ReadAll() error = %v", err) + } + + addedIndexes := make(map[string]float64) + var addedSequence []string + reasoningDone := false + reasoningDoneBeforeTool := false + for _, event := range parseTestSSEEvents(t, string(raw)) { + item, _ := event.Payload["item"].(map[string]any) + itemType, _ := item["type"].(string) + switch event.Name { + case "response.output_item.done": + if itemType == "reasoning" { + reasoningDone = event.Payload["output_index"] == float64(0) + } + case "response.output_item.added": + addedIndexes[itemType], _ = event.Payload["output_index"].(float64) + addedSequence = append(addedSequence, itemType) + if itemType == "function_call" { + reasoningDoneBeforeTool = reasoningDone + } + } + } + + if len(addedIndexes) != len(tt.wantIndexes) { + t.Fatalf("output indexes = %#v, want %#v", addedIndexes, tt.wantIndexes) + } + for itemType, wantIndex := range tt.wantIndexes { + if addedIndexes[itemType] != wantIndex { + t.Fatalf("%s output_index = %v, want %v", itemType, addedIndexes[itemType], wantIndex) + } + } + if !slices.Equal(addedSequence, tt.wantSequence) { + t.Fatalf("output item sequence = %#v, want %#v", addedSequence, tt.wantSequence) + } + if !reasoningDoneBeforeTool { + t.Fatal("expected the reasoning item to close before the function_call item opened") + } + }) + } +} + func TestOpenAIResponsesStreamConverter_WithTextBeforeToolCall(t *testing.T) { mockStream := `data: {"id":"chatcmpl-123","object":"chat.completion.chunk","created":1677652288,"model":"test-model","choices":[{"index":0,"delta":{"content":"I'll check that for you."},"finish_reason":null}]} @@ -245,7 +475,7 @@ func TestOpenAIResponsesStreamConverter_TolerantChunkFallback(t *testing.T) { // The second chunk carries a float tool-call index (Python-style encoders) // alongside junk entries (non-object, index missing) that must be skipped // without discarding the valid call. - mockStream := `data: {"choices":[{"delta":{"content":[{"type":"text","text":"ignored"}]},"finish_reason":null}],"usage":{"prompt_tokens":3,"completion_tokens":4,"total_tokens":7}} + mockStream := `data: {"choices":[{"delta":{"content":[{"type":"text","text":"ignored"}]},"finish_reason":null}],"usage":{"prompt_tokens":3,"completion_tokens":4,"total_tokens":7,"prompt_tokens_details":{"cached_tokens":2},"completion_tokens_details":{"reasoning_tokens":1}}} data: {"choices":[{"delta":{"tool_calls":["junk",{"id":"call_no_index"},{"index":0.0,"id":"call_f","type":"function","function":{"name":"lookup","arguments":"{}"}}]},"finish_reason":null}]} @@ -288,39 +518,60 @@ data: [DONE] if usage["total_tokens"] != float64(7) { t.Fatalf("usage total_tokens = %v, want 7", usage["total_tokens"]) } + if usage["input_tokens"] != float64(3) || usage["output_tokens"] != float64(4) { + t.Fatalf("Responses usage counts = %#v", usage) + } + inputDetails, _ := usage["input_tokens_details"].(map[string]any) + outputDetails, _ := usage["output_tokens_details"].(map[string]any) + if inputDetails["cached_tokens"] != float64(2) || outputDetails["reasoning_tokens"] != float64(1) { + t.Fatalf("Responses usage details = %#v", usage) + } + if _, present := usage["prompt_tokens"]; present { + t.Fatalf("usage retained Chat field names: %#v", usage) + } if !foundToolAdded { t.Fatal("expected function_call output item from float-index tool call delta") } } -// TestOpenAIResponsesStreamConverter_DropsNonObjectUsage ensures off-spec -// non-object usage values never leak into the response.completed payload. -func TestOpenAIResponsesStreamConverter_DropsNonObjectUsage(t *testing.T) { - mockStream := `data: {"choices":[{"delta":{"content":"hi"},"finish_reason":null}],"usage":"n/a"} +func TestOpenAIResponsesStreamConverter_DropsInvalidUsage(t *testing.T) { + tests := []struct { + name string + usage string + }{ + {name: "non-object", usage: `"n/a"`}, + {name: "malformed required field", usage: `{"prompt_tokens":"unknown","completion_tokens":1,"total_tokens":1}`}, + {name: "required field nested", usage: `{"metadata":{"prompt_tokens":3},"completion_tokens":1,"total_tokens":1}`}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mockStream := `data: {"choices":[{"delta":{"content":"hi"},"finish_reason":"stop"}],"usage":` + tt.usage + `} data: [DONE] ` + converter := NewOpenAIResponsesStreamConverter(io.NopCloser(strings.NewReader(mockStream)), "test-model", "groq") + raw, err := io.ReadAll(converter) + if err != nil { + t.Fatalf("ReadAll() error = %v", err) + } - converter := NewOpenAIResponsesStreamConverter(io.NopCloser(strings.NewReader(mockStream)), "test-model", "groq") - raw, err := io.ReadAll(converter) - if err != nil { - t.Fatalf("failed to read from converter: %v", err) - } - - for _, event := range parseTestSSEEvents(t, string(raw)) { - if event.Done || event.Name != "response.completed" { - continue - } - response, _ := event.Payload["response"].(map[string]any) - if response == nil { - t.Fatal("response.completed missing response object") - } - if usage, present := response["usage"]; present { - t.Fatalf("response.completed usage = %#v, want omitted for non-object usage", usage) - } - return + for _, event := range parseTestSSEEvents(t, string(raw)) { + if event.Done || event.Name != "response.completed" { + continue + } + response, _ := event.Payload["response"].(map[string]any) + if response == nil { + t.Fatal("response.completed missing response object") + } + if usage, present := response["usage"]; present { + t.Fatalf("invalid usage leaked into response.completed: %#v", usage) + } + return + } + t.Fatal("expected response.completed event") + }) } - t.Fatal("expected response.completed event") } func TestOpenAIResponsesStreamConverter_PropagatesStreamError(t *testing.T) { diff --git a/internal/providers/responses_input.go b/internal/providers/responses_input.go index 7b09c4bb1..3554e6fae 100644 --- a/internal/providers/responses_input.go +++ b/internal/providers/responses_input.go @@ -38,24 +38,59 @@ func ConvertResponsesInputToMessages(input any) ([]core.Message, error) { func convertResponsesInputItems(items []any) ([]core.Message, error) { messages := make([]core.Message, 0, len(items)) var pendingAssistant *core.Message + var pendingReasoning string - flushPendingAssistant := func() { + flushPendingAssistant := func() error { if pendingAssistant == nil { - return + return nil + } + // Chat reasoning extensions associate reasoning_content with the + // assistant tool-call message. DeepSeek requires it on the following + // tool-result request, while reasoning from ordinary assistant turns is + // intentionally omitted because it is not part of their next-turn + // context. + if pendingReasoning != "" && len(pendingAssistant.ToolCalls) > 0 { + raw, err := json.Marshal(pendingReasoning) + if err != nil { + return err + } + extra, err := core.MergeUnknownJSONFields(pendingAssistant.ExtraFields, map[string]json.RawMessage{ + "reasoning_content": raw, + }) + if err != nil { + return err + } + pendingAssistant.ExtraFields = extra + pendingReasoning = "" } messages = append(messages, *pendingAssistant) pendingAssistant = nil + return nil } for i, item := range items { + if reasoning, ok := responsesInputReasoningText(item); ok { + if err := flushPendingAssistant(); err != nil { + return nil, err + } + pendingReasoning = "" + if reasoning != "" { + pendingReasoning = reasoning + } + continue + } + msg, itemType, err := convertResponsesInputItem(item, i) if err != nil { return nil, err } if msg.Role == "assistant" { - if itemType == "message" { - flushPendingAssistant() + if itemType == "message" && pendingAssistant != nil { + if err := flushPendingAssistant(); err != nil { + return nil, err + } + pendingReasoning = "" } if pendingAssistant == nil { assistant := cloneResponsesMessage(msg) @@ -63,21 +98,102 @@ func convertResponsesInputItems(items []any) ([]core.Message, error) { } else if canMergeAssistantMessages(*pendingAssistant, msg) { mergeAssistantMessage(pendingAssistant, msg) } else { - flushPendingAssistant() + if err := flushPendingAssistant(); err != nil { + return nil, err + } assistant := cloneResponsesMessage(msg) pendingAssistant = &assistant } continue } - flushPendingAssistant() + if err := flushPendingAssistant(); err != nil { + return nil, err + } + pendingReasoning = "" messages = append(messages, msg) } - flushPendingAssistant() + if err := flushPendingAssistant(); err != nil { + return nil, err + } return messages, nil } +// responsesInputReasoningText recognizes a Responses reasoning item and +// extracts raw reasoning content. Summary text is accepted as a compatibility +// fallback for older gateways that mislabeled reasoning_content as a summary. +// Encrypted-only reasoning stays opaque and is deliberately omitted. +func responsesInputReasoningText(item any) (string, bool) { + var raw json.RawMessage + switch typed := item.(type) { + case core.ResponsesInputElement: + if typed.Type != "reasoning" { + return "", false + } + raw = typed.Raw + if len(raw) == 0 { + encoded, err := json.Marshal(typed) + if err != nil { + return "", true + } + raw = encoded + } + case map[string]any: + itemType, _ := typed["type"].(string) + if itemType != "reasoning" { + return "", false + } + encoded, err := json.Marshal(typed) + if err != nil { + return "", true + } + raw = encoded + default: + return "", false + } + + var payload struct { + Content []responsesReasoningPart `json:"content"` + Summary []responsesReasoningPart `json:"summary"` + } + if err := json.Unmarshal(raw, &payload); err != nil { + return "", true + } + if text := reasoningTextParts(payload.Content, "reasoning_text"); text != "" { + return text, true + } + return reasoningTextParts(payload.Summary, "summary_text"), true +} + +type responsesReasoningPart struct { + Type string `json:"type"` + Text string `json:"text"` +} + +func reasoningTextParts(parts []responsesReasoningPart, partType string) string { + texts := make([]string, 0, len(parts)) + for _, part := range parts { + if part.Type == partType && part.Text != "" { + texts = append(texts, part.Text) + } + } + return strings.Join(texts, "\n\n") +} + +// normalizeChatTranslationRole maps "developer" onto "system". "developer" is +// OpenAI's newer alias for "system", understood only by OpenAI's own +// reasoning models; chat-translated providers (DeepSeek, Anthropic, Gemini, +// ...) speak the classic chat-completions roles, so it must be normalized +// the same way the other chat-only providers already do (see +// cohere/gemini/bedrock chat translation). +func normalizeChatTranslationRole(role string) string { + if role == "developer" { + return "system" + } + return role +} + func convertResponsesInputItem(item any, index int) (core.Message, string, error) { switch typed := item.(type) { case core.ResponsesInputElement: @@ -136,6 +252,7 @@ func convertResponsesInputElement(item core.ResponsesInputElement, index int) (c if role == "" { return core.Message{}, "", core.NewInvalidRequestError(fmt.Sprintf("invalid responses input item at index %d: role is required", index), nil) } + role = normalizeChatTranslationRole(role) content, ok := ConvertResponsesContentToChatContent(item.Content) if !ok { return core.Message{}, "", core.NewInvalidRequestError(fmt.Sprintf("invalid responses input item at index %d: unsupported content", index), nil) @@ -145,6 +262,9 @@ func convertResponsesInputElement(item core.ResponsesInputElement, index int) (c Content: content, ExtraFields: core.CloneUnknownJSONFields(item.ExtraFields), }, "message", nil + case "reasoning": + // Recognized by responsesInputReasoningText before item conversion. + return core.Message{}, "reasoning", nil default: return core.Message{}, "", core.NewInvalidRequestError(fmt.Sprintf("invalid responses input item at index %d: unsupported input item type %q for chat-translated providers", index, item.Type), nil) } @@ -195,6 +315,9 @@ func convertResponsesInputMap(item map[string]any, index int) (core.Message, str ExtraFields: core.UnknownJSONFieldsFromMap(rawJSONMapFromUnknownKeys(item, "type", "call_id", "status", "output")), }, "function_call_output", nil case "", "message": + case "reasoning": + // Recognized by responsesInputReasoningText before item conversion. + return core.Message{}, "reasoning", nil default: return core.Message{}, "", core.NewInvalidRequestError(fmt.Sprintf("invalid responses input item at index %d: unsupported input item type %q for chat-translated providers", index, itemType), nil) } @@ -204,6 +327,7 @@ func convertResponsesInputMap(item map[string]any, index int) (core.Message, str if role == "" { return core.Message{}, "", core.NewInvalidRequestError(fmt.Sprintf("invalid responses input item at index %d: role is required", index), nil) } + role = normalizeChatTranslationRole(role) content, ok := ConvertResponsesContentToChatContent(item["content"]) if !ok { @@ -236,13 +360,19 @@ func cloneResponsesMessage(msg core.Message) core.Message { } func canMergeAssistantMessages(current, next core.Message) bool { + // A Responses message followed by function_call items represents one Chat + // assistant message. Metadata on the message remains on the merged message; + // function-call metadata is already carried by each ToolCall. + if isAssistantToolCallOnlyMessage(next) && next.ExtraFields.IsEmpty() { + return true + } if !current.ExtraFields.IsEmpty() || !next.ExtraFields.IsEmpty() { return false } if !core.HasStructuredContent(current.Content) && !core.HasStructuredContent(next.Content) { return true } - return isAssistantToolCallOnlyMessage(next) + return false } func mergeAssistantMessage(dst *core.Message, src core.Message) { diff --git a/internal/providers/responses_output.go b/internal/providers/responses_output.go index c4a6d90af..be1435a2a 100644 --- a/internal/providers/responses_output.go +++ b/internal/providers/responses_output.go @@ -111,7 +111,21 @@ func buildResponsesContentItemsFromParts(parts []core.ContentPart) []core.Respon // BuildResponsesOutputItems converts a response message into Responses API output items. func BuildResponsesOutputItems(msg core.ResponseMessage) []core.ResponsesOutputItem { - output := make([]core.ResponsesOutputItem, 0, len(msg.ToolCalls)+1) + reasoningContent := responseMessageReasoningContent(msg) + output := make([]core.ResponsesOutputItem, 0, len(msg.ToolCalls)+2) + if reasoningContent != "" { + output = append(output, core.ResponsesOutputItem{ + ID: "rs_" + uuid.New().String(), + Type: "reasoning", + Status: "completed", + Content: []core.ResponsesContentItem{ + {Type: "reasoning_text", Text: reasoningContent}, + }, + ExtraFields: core.UnknownJSONFieldsFromMap(map[string]json.RawMessage{ + "summary": json.RawMessage(`[]`), + }), + }) + } contentItems := buildResponsesMessageContent(msg.Content) if len(contentItems) > 0 || len(msg.ToolCalls) == 0 { if len(contentItems) == 0 { @@ -145,6 +159,18 @@ func BuildResponsesOutputItems(msg core.ResponseMessage) []core.ResponsesOutputI return output } +func responseMessageReasoningContent(msg core.ResponseMessage) string { + raw := msg.ExtraFields.Lookup("reasoning_content") + if len(raw) == 0 { + return "" + } + var content string + if err := json.Unmarshal(raw, &content); err != nil { + return "" + } + return content +} + // ConvertChatResponseToResponses converts a ChatResponse to a ResponsesResponse. func ConvertChatResponseToResponses(resp *core.ChatResponse) *core.ResponsesResponse { var output []core.ResponsesOutputItem diff --git a/internal/providers/responses_output_state.go b/internal/providers/responses_output_state.go index 28bab0eae..5b4c00481 100644 --- a/internal/providers/responses_output_state.go +++ b/internal/providers/responses_output_state.go @@ -29,6 +29,12 @@ type ResponsesOutputEventState struct { assistantDone bool assistantMessageID string assistantText strings.Builder + + reasoningReserved bool + reasoningStarted bool + reasoningDone bool + reasoningItemID string + reasoningText strings.Builder } // NewResponsesOutputEventState creates a new Responses output-item state manager. @@ -128,6 +134,97 @@ func (s *ResponsesOutputEventState) CompleteAssistantOutput(outputIndex int) str }) } +// ReserveReasoning marks that a reasoning output item occupies index 0, +// ahead of the assistant message and any tool calls. +func (s *ResponsesOutputEventState) ReserveReasoning() { + s.reasoningReserved = true +} + +// ReasoningReserved reports whether a reasoning output item has been reserved. +func (s *ResponsesOutputEventState) ReasoningReserved() bool { + return s.reasoningReserved +} + +// ReasoningDone reports whether the reasoning output item has been completed. +func (s *ResponsesOutputEventState) ReasoningDone() bool { + return s.reasoningDone +} + +// ReasoningItem renders a raw reasoning output item. Provider +// reasoning_content is chain-of-thought text, not an OpenAI-generated summary, +// so it belongs in content while summary remains empty. +func (s *ResponsesOutputEventState) ReasoningItem(status string, includeContent bool) map[string]any { + item := map[string]any{ + "id": s.reasoningItemID, + "type": "reasoning", + "status": status, + "summary": []map[string]any{}, + } + if includeContent { + item["content"] = []map[string]any{ + {"type": "reasoning_text", "text": s.reasoningText.String()}, + } + } + return item +} + +// StartReasoningOutput emits the reasoning output_item.added event once before +// any reasoning_text delta references the item. +func (s *ResponsesOutputEventState) StartReasoningOutput(outputIndex int) string { + if s.reasoningStarted { + return "" + } + s.reasoningStarted = true + if s.reasoningItemID == "" { + s.reasoningItemID = "rs_" + uuid.New().String() + } + return s.WriteEvent("response.output_item.added", map[string]any{ + "type": "response.output_item.added", + "item": s.ReasoningItem("in_progress", false), + "output_index": outputIndex, + }) +} + +// AppendReasoningDelta starts the reasoning item if needed and emits a raw +// reasoning-text delta. +func (s *ResponsesOutputEventState) AppendReasoningDelta(outputIndex int, delta string) string { + var b strings.Builder + b.WriteString(s.StartReasoningOutput(outputIndex)) + s.reasoningText.WriteString(delta) + b.WriteString(s.WriteEvent("response.reasoning_text.delta", map[string]any{ + "type": "response.reasoning_text.delta", + "item_id": s.reasoningItemID, + "output_index": outputIndex, + "content_index": 0, + "delta": delta, + })) + return b.String() +} + +// CompleteReasoningOutput emits the raw reasoning completion and +// output_item.done events once. +func (s *ResponsesOutputEventState) CompleteReasoningOutput(outputIndex int) string { + if !s.reasoningReserved || s.reasoningDone { + return "" + } + s.reasoningDone = true + var b strings.Builder + b.WriteString(s.StartReasoningOutput(outputIndex)) + b.WriteString(s.WriteEvent("response.reasoning_text.done", map[string]any{ + "type": "response.reasoning_text.done", + "item_id": s.reasoningItemID, + "output_index": outputIndex, + "content_index": 0, + "text": s.reasoningText.String(), + })) + b.WriteString(s.WriteEvent("response.output_item.done", map[string]any{ + "type": "response.output_item.done", + "item": s.ReasoningItem("completed", true), + "output_index": outputIndex, + })) + return b.String() +} + // ToolCallArguments returns the serialized argument payload for a function_call item. func (s *ResponsesOutputEventState) ToolCallArguments(state *ResponsesOutputToolCallState) string { if state == nil { diff --git a/tests/contract/testdata/golden/groq/responses_stream.golden.json b/tests/contract/testdata/golden/groq/responses_stream.golden.json index d099ce4bb..0bd23687b 100644 --- a/tests/contract/testdata/golden/groq/responses_stream.golden.json +++ b/tests/contract/testdata/golden/groq/responses_stream.golden.json @@ -87,12 +87,8 @@ "provider": "groq", "status": "completed", "usage": { - "completion_time": 0.027554055, - "completion_tokens": 4, - "prompt_time": 0.003210785, - "prompt_tokens": 38, - "queue_time": 0.090321431, - "total_time": 0.03076484, + "input_tokens": 38, + "output_tokens": 4, "total_tokens": 42 } },