fix(apicompat): reject malformed tool-call arguments

This commit is contained in:
lbyxiaolizi committed 2026-08-19 18:48:35 +08:00
1 parent e943f817b1
commit e2d9ce0cad
6 files changed
+219 -2

No files matched your search

@@ -279,6 +279,7 @@ func buildChatMessagesFromItems(messages []ChatMessage, rawItems []json.RawMessa
// a user-side item ends the turn and clears it. // a user-side item ends the turn and clears it.
var lastTurnReasoning string var lastTurnReasoning string
mediaByCallID := make(toolOutputMediaByCallID) mediaByCallID := make(toolOutputMediaByCallID)
invalidFunctionCallIDs := make(map[string]struct{})
reasoningForAssistant := func() string { reasoningForAssistant := func() string {
if pendingReasoning != "" { if pendingReasoning != "" {
@@ -331,6 +332,20 @@ func buildChatMessagesFromItems(messages []ChatMessage, rawItems []json.RawMessa
if strings.TrimSpace(arguments) == "" { if strings.TrimSpace(arguments) == "" {
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"]) name := rawString(item["name"])
// namespace 子工具的历史调用带 namespace 字段,需与请求方向的摊平 // namespace 子工具的历史调用带 namespace 字段,需与请求方向的摊平
// 命名(namespaceChildrenToChatTools)保持一致。 // 命名(namespaceChildrenToChatTools)保持一致。
@@ -338,7 +353,7 @@ func buildChatMessagesFromItems(messages []ChatMessage, rawItems []json.RawMessa
name = flattenNamespaceToolName(ns, name) name = flattenNamespaceToolName(ns, name)
} }
toolCall := ChatToolCall{ toolCall := ChatToolCall{
ID: rawString(item["call_id"]), ID: callID,
Type: "function", Type: "function",
Function: ChatFunctionCall{ Function: ChatFunctionCall{
Name: name, Name: name,
@@ -388,6 +403,10 @@ func buildChatMessagesFromItems(messages []ChatMessage, rawItems []json.RawMessa
case "function_call_output", "custom_tool_call_output", "tool_search_output": case "function_call_output", "custom_tool_call_output", "tool_search_output":
outputRaw := bytesTrimSpace(item["output"]) outputRaw := bytesTrimSpace(item["output"])
callID := rawString(item["call_id"]) callID := rawString(item["call_id"])
if _, skipped := invalidFunctionCallIDs[callID]; skipped {
pendingReasoning = ""
continue
}
delete(mediaByCallID, callID) delete(mediaByCallID, callID)
outputText, media, rewritten := extractToolOutputMedia(outputRaw) 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 { func (state *ChatCompletionsToResponsesStreamState) allocOutputIndex() int {
idx := state.nextOutputIndex idx := state.nextOutputIndex
state.nextOutputIndex++ state.nextOutputIndex++
@@ -17,6 +17,26 @@ func TestResponsesInputToChatMessages_DeveloperRoleMapsToSystem(t *testing.T) {
assert.JSONEq(t, `"follow project instructions"`, string(messages[0].Content)) 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) { func TestResponsesInputToChatMessages_KeepsChatCompletionRoles(t *testing.T) {
input := json.RawMessage(`[ input := json.RawMessage(`[
{"role":"system","content":"system message"}, {"role":"system","content":"system message"},
@@ -214,6 +214,64 @@ func TestStream_ToolCallArgumentsInFirstChunkNotDoubled(t *testing.T) {
require.Equal(t, `{"cmd":"ls"}`, argsDelta.String()) 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 // TestStream_SSEWireComplete drives the full stream through SSE encoding and
// asserts the function_call events carry complete fields on the wire. // asserts the function_call events carry complete fields on the wire.
func TestStream_SSEWireComplete(t *testing.T) { func TestStream_SSEWireComplete(t *testing.T) {
@@ -281,7 +281,13 @@ func (s *OpenAIGatewayService) scanCCStream(
zap.Error(err), zap.Error(err),
zap.String("request_id", requestID), 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) { if st.FirstTokenMs == nil && !isOpenAIChatUsageOnlyStreamChunk(payload) && chatChunkStartsResponsesOutput(&chunk) {
ms := int(time.Since(startTime).Milliseconds()) ms := int(time.Since(startTime).Milliseconds())
@@ -231,6 +231,20 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsResponses(
FirstTokenMs: scan.FirstTokenMs, FirstTokenMs: scan.FirstTokenMs,
}, fmt.Errorf("stream usage incomplete: %w", scan.Err) }, 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) finalEvents := apicompat.FinalizeChatCompletionsResponsesStream(state)
s.cacheReasoningItemsFromEvents(finalEvents) s.cacheReasoningItemsFromEvents(finalEvents)
@@ -148,6 +148,76 @@ func TestForwardResponses_ForceChatCompletionsRoutesStreamingToChatCompletions(t
require.NotNil(t, result.FirstTokenMs) 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) { func TestForwardResponses_DeepSeekReasoningOnlyStreamProducesVisibleText(t *testing.T) {
gin.SetMode(gin.TestMode) gin.SetMode(gin.TestMode)