diff --git a/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go b/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go index 2b9d69eae7..4b3331f301 100644 --- a/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go +++ b/backend/internal/pkg/apicompat/chatcompletions_responses_bridge.go @@ -279,6 +279,7 @@ func buildChatMessagesFromItems(messages []ChatMessage, rawItems []json.RawMessa // a user-side item ends the turn and clears it. var lastTurnReasoning string mediaByCallID := make(toolOutputMediaByCallID) + invalidFunctionCallIDs := make(map[string]struct{}) reasoningForAssistant := func() string { if pendingReasoning != "" { @@ -331,6 +332,20 @@ func buildChatMessagesFromItems(messages []ChatMessage, rawItems []json.RawMessa if strings.TrimSpace(arguments) == "" { arguments = "{}" } + callID := rawString(item["call_id"]) + if !json.Valid([]byte(arguments)) { + // A previous streamed turn can leave a truncated function_call in + // Codex history (for example after an upstream SSE parse failure or + // an output-limit interruption). Do not forward that item to a + // Chat Completions provider, which rejects the entire request. Its + // matching output is skipped below as well, allowing the next user + // turn to self-heal instead of repeatedly replaying the poison. + if callID != "" { + invalidFunctionCallIDs[callID] = struct{}{} + } + pendingReasoning = "" + continue + } name := rawString(item["name"]) // namespace 子工具的历史调用带 namespace 字段，需与请求方向的摊平 // 命名（namespaceChildrenToChatTools）保持一致。 @@ -338,7 +353,7 @@ func buildChatMessagesFromItems(messages []ChatMessage, rawItems []json.RawMessa name = flattenNamespaceToolName(ns, name) } toolCall := ChatToolCall{ - ID: rawString(item["call_id"]), + ID: callID, Type: "function", Function: ChatFunctionCall{ Name: name, @@ -388,6 +403,10 @@ func buildChatMessagesFromItems(messages []ChatMessage, rawItems []json.RawMessa case "function_call_output", "custom_tool_call_output", "tool_search_output": outputRaw := bytesTrimSpace(item["output"]) callID := rawString(item["call_id"]) + if _, skipped := invalidFunctionCallIDs[callID]; skipped { + pendingReasoning = "" + continue + } delete(mediaByCallID, callID) outputText, media, rewritten := extractToolOutputMedia(outputRaw) @@ -1406,6 +1425,36 @@ func NewChatCompletionsToResponsesStreamState(model string) *ChatCompletionsToRe } } +// ValidateToolCallArguments checks the accumulated function-call arguments +// before the stream is finalized. A tool call that ended because the upstream +// hit its output limit, or whose SSE chunk was lost, must not be emitted as a +// completed Responses item: Codex will persist it and replay it on the next +// turn, where a Chat Completions provider rejects the whole request. +func (state *ChatCompletionsToResponsesStreamState) ValidateToolCallArguments() error { + if state == nil { + return nil + } + if state.FinishReason == "length" && len(state.ToolCalls) > 0 { + return fmt.Errorf("tool call stream ended at max output length") + } + for idx, toolCall := range state.ToolCalls { + if toolCall == nil { + continue + } + if state.toolIsCustom[idx] || state.toolIsToolSearch[idx] { + continue + } + arguments := strings.TrimSpace(toolCall.Function.Arguments) + if arguments == "" { + continue + } + if !json.Valid([]byte(arguments)) { + return fmt.Errorf("tool call %q (%s) arguments are invalid JSON", toolCall.ID, toolCall.Function.Name) + } + } + return nil +} + func (state *ChatCompletionsToResponsesStreamState) allocOutputIndex() int { idx := state.nextOutputIndex state.nextOutputIndex++ diff --git a/backend/internal/pkg/apicompat/chatcompletions_responses_bridge_test.go b/backend/internal/pkg/apicompat/chatcompletions_responses_bridge_test.go index 2e770b4158..9dc0448f97 100644 --- a/backend/internal/pkg/apicompat/chatcompletions_responses_bridge_test.go +++ b/backend/internal/pkg/apicompat/chatcompletions_responses_bridge_test.go @@ -17,6 +17,26 @@ func TestResponsesInputToChatMessages_DeveloperRoleMapsToSystem(t *testing.T) { assert.JSONEq(t, `"follow project instructions"`, string(messages[0].Content)) } +func TestResponsesInputToChatMessages_SkipsInvalidHistoricalFunctionCall(t *testing.T) { + input := json.RawMessage(`[ + {"type":"function_call","call_id":"call_bad","name":"exec_command","arguments":"{\"cmd\": \"ssh root@HOST"}, + {"type":"function_call_output","call_id":"call_bad","output":"failed to parse function arguments"}, + {"type":"function_call","call_id":"call_ok","name":"exec_command","arguments":"{}"}, + {"type":"function_call_output","call_id":"call_ok","output":"ok"}, + {"role":"user","content":"continue"} + ]`) + + messages, err := responsesInputToChatMessages("", input) + require.NoError(t, err) + require.Len(t, messages, 3) + require.Equal(t, "assistant", messages[0].Role) + require.Len(t, messages[0].ToolCalls, 1) + require.Equal(t, "call_ok", messages[0].ToolCalls[0].ID) + require.Equal(t, "tool", messages[1].Role) + require.Equal(t, "call_ok", messages[1].ToolCallID) + require.Equal(t, "user", messages[2].Role) +} + func TestResponsesInputToChatMessages_KeepsChatCompletionRoles(t *testing.T) { input := json.RawMessage(`[ {"role":"system","content":"system message"}, diff --git a/backend/internal/pkg/apicompat/chatcompletions_responses_stream_lifecycle_test.go b/backend/internal/pkg/apicompat/chatcompletions_responses_stream_lifecycle_test.go index baa4e4b2a5..ebf44cb661 100644 --- a/backend/internal/pkg/apicompat/chatcompletions_responses_stream_lifecycle_test.go +++ b/backend/internal/pkg/apicompat/chatcompletions_responses_stream_lifecycle_test.go @@ -214,6 +214,64 @@ func TestStream_ToolCallArgumentsInFirstChunkNotDoubled(t *testing.T) { require.Equal(t, `{"cmd":"ls"}`, argsDelta.String()) } +func TestStream_InvalidToolArgumentsAreRejectedBeforeFinalize(t *testing.T) { + idx := 0 + state := NewChatCompletionsToResponsesStreamState("deepseek-v4-flash") + chunk := &ChatCompletionsChunk{ + Choices: []ChatChunkChoice{ + { + Index: 0, + Delta: ChatDelta{ + ToolCalls: []ChatToolCall{ + { + Index: &idx, + ID: "call_bad", + Type: "function", + Function: ChatFunctionCall{ + Name: "exec_command", + Arguments: `{"cmd": "ssh root@HOST`, + }, + }, + }, + }, + }, + }, + } + ChatCompletionsChunkToResponsesEvents(chunk, state) + + err := state.ValidateToolCallArguments() + require.ErrorContains(t, err, "invalid JSON") +} + +func TestStream_ToolCallAtOutputLimitIsRejectedBeforeFinalize(t *testing.T) { + idx := 0 + state := NewChatCompletionsToResponsesStreamState("deepseek-v4-flash") + chunk := &ChatCompletionsChunk{ + Choices: []ChatChunkChoice{ + { + Index: 0, + Delta: ChatDelta{ + ToolCalls: []ChatToolCall{ + { + Index: &idx, + ID: "call_at_limit", + Type: "function", + Function: ChatFunctionCall{ + Name: "exec_command", + Arguments: `{}`, + }, + }, + }, + }, + }, + }, + } + ChatCompletionsChunkToResponsesEvents(chunk, state) + state.FinishReason = "length" + + require.ErrorContains(t, state.ValidateToolCallArguments(), "max output length") +} + // TestStream_SSEWireComplete drives the full stream through SSE encoding and // asserts the function_call events carry complete fields on the wire. func TestStream_SSEWireComplete(t *testing.T) { diff --git a/backend/internal/service/openai_gateway_cc_pipeline.go b/backend/internal/service/openai_gateway_cc_pipeline.go index ce9dd54d37..6551e2076b 100644 --- a/backend/internal/service/openai_gateway_cc_pipeline.go +++ b/backend/internal/service/openai_gateway_cc_pipeline.go @@ -281,7 +281,13 @@ func (s *OpenAIGatewayService) scanCCStream( zap.Error(err), zap.String("request_id", requestID), ) - continue + // A malformed chunk may contain the middle or the tail of a tool-call + // argument. Skipping it and finalizing the accumulated state would turn + // a truncated function call into a completed Responses item, which then + // poisons Codex's next-turn history. Treat the stream as incomplete so + // callers skip finalization and surface an upstream-stream error. + st.Err = fmt.Errorf("malformed chat stream chunk: %w", err) + break } if st.FirstTokenMs == nil && !isOpenAIChatUsageOnlyStreamChunk(payload) && chatChunkStartsResponsesOutput(&chunk) { ms := int(time.Since(startTime).Milliseconds()) diff --git a/backend/internal/service/openai_gateway_responses_chat_fallback.go b/backend/internal/service/openai_gateway_responses_chat_fallback.go index 084f9e5020..d9b7c65b64 100644 --- a/backend/internal/service/openai_gateway_responses_chat_fallback.go +++ b/backend/internal/service/openai_gateway_responses_chat_fallback.go @@ -231,6 +231,20 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsResponses( FirstTokenMs: scan.FirstTokenMs, }, fmt.Errorf("stream usage incomplete: %w", scan.Err) } + if err := state.ValidateToolCallArguments(); err != nil { + return &OpenAIForwardResult{ + RequestID: requestID, + Usage: scan.Usage, + Model: originalModel, + BillingModel: billingModel, + UpstreamModel: upstreamModel, + ReasoningEffort: reasoningEffort, + ServiceTier: serviceTier, + Stream: true, + Duration: time.Since(startTime), + FirstTokenMs: scan.FirstTokenMs, + }, fmt.Errorf("invalid tool call arguments from upstream: %w", err) + } finalEvents := apicompat.FinalizeChatCompletionsResponsesStream(state) s.cacheReasoningItemsFromEvents(finalEvents) diff --git a/backend/internal/service/openai_gateway_responses_chat_fallback_test.go b/backend/internal/service/openai_gateway_responses_chat_fallback_test.go index 15952f3a0f..c47ea0d4f4 100644 --- a/backend/internal/service/openai_gateway_responses_chat_fallback_test.go +++ b/backend/internal/service/openai_gateway_responses_chat_fallback_test.go @@ -148,6 +148,76 @@ func TestForwardResponses_ForceChatCompletionsRoutesStreamingToChatCompletions(t require.NotNil(t, result.FirstTokenMs) } +func TestForwardResponses_ChatFallbackDoesNotFinalizeAfterMalformedStreamChunk(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"deepseek-v4-flash","input":"run the command","stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"id":"chatcmpl_bad_stream","object":"chat.completion.chunk","model":"deepseek-v4-flash","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_bad","type":"function","function":{"name":"exec_command","arguments":"{}"}}]},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_bad_stream","object":"chat.completion.chunk","choices":[`, + "", + "data: [DONE]", + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_bad_stream"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.Forward(context.Background(), c, forceChatResponsesFallbackAccount(), body) + require.ErrorContains(t, err, "stream usage incomplete") + require.NotNil(t, result) + require.NotContains(t, rec.Body.String(), "response.function_call_arguments.done") + require.NotContains(t, rec.Body.String(), "response.output_item.done") + require.NotContains(t, rec.Body.String(), "data: [DONE]") +} + +func TestForwardResponses_ChatFallbackDoesNotCompleteToolCallAtOutputLimit(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"deepseek-v4-flash","input":"run the command","stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + `data: {"id":"chatcmpl_length_tool","object":"chat.completion.chunk","model":"deepseek-v4-flash","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_length","type":"function","function":{"name":"exec_command","arguments":"{}"}}]},"finish_reason":null}]}`, + "", + `data: {"id":"chatcmpl_length_tool","object":"chat.completion.chunk","model":"deepseek-v4-flash","choices":[{"index":0,"delta":{},"finish_reason":"length"}],"usage":{"prompt_tokens":4,"completion_tokens":6492,"total_tokens":6496}}`, + "", + "data: [DONE]", + "", + }, "\n") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_length_tool"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + }} + svc := &OpenAIGatewayService{ + cfg: rawChatCompletionsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.Forward(context.Background(), c, forceChatResponsesFallbackAccount(), body) + require.ErrorContains(t, err, "max output length") + require.NotNil(t, result) + require.NotContains(t, rec.Body.String(), "response.function_call_arguments.done") + require.NotContains(t, rec.Body.String(), "response.output_item.done") + require.NotContains(t, rec.Body.String(), "data: [DONE]") +} + func TestForwardResponses_DeepSeekReasoningOnlyStreamProducesVisibleText(t *testing.T) { gin.SetMode(gin.TestMode)