From 2b49d662cc5a237997ebe7705b9e966311371458 Mon Sep 17 00:00:00 2001 From: wucm667 Date: Thu, 25 Jun 2026 16:18:11 +0800 Subject: [PATCH] fix(openai): dedupe passthrough function call args --- ..._gateway_passthrough_function_args_test.go | 258 ++++++++++++++++++ .../service/openai_gateway_service.go | 94 ++++++- 2 files changed, 348 insertions(+), 4 deletions(-) create mode 100644 backend/internal/service/openai_gateway_passthrough_function_args_test.go diff --git a/backend/internal/service/openai_gateway_passthrough_function_args_test.go b/backend/internal/service/openai_gateway_passthrough_function_args_test.go new file mode 100644 index 0000000000..85de28143d --- /dev/null +++ b/backend/internal/service/openai_gateway_passthrough_function_args_test.go @@ -0,0 +1,258 @@ +package service + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/pkg/openai_compat" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +func TestHandleStreamingResponsePassthroughDeduplicatesFunctionCallArguments(t *testing.T) { + gin.SetMode(gin.TestMode) + + argsA := `{"cmd":"echo hi","meta":{"nested":[1,{"ok":true}],"quote":"a}b"}}` + argsB := `{"path":"/tmp/file","patch":{"ops":[{"op":"replace","value":{"lines":["x","y"]}}]}}` + upstreamBody := strings.Join([]string{ + passthroughSSEData(`{"type":"response.created","response":{"id":"resp_passthrough_args","model":"gpt-5.4"}}`), + passthroughSSEData(`{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_a","call_id":"call_a","name":"exec_command","arguments":"","status":"in_progress"}}`), + passthroughSSEData(functionArgsDeltaJSON(0, "fc_a", "call_a", "exec_command", `{"cmd":`)), + passthroughSSEData(functionArgsDeltaJSON(0, "fc_a", "call_a", "exec_command", `"echo hi","meta":{"nested":[1,{"ok":true}],"quote":"a}b"}}`)), + passthroughSSEData(functionArgsDoneJSON(0, "fc_a", "call_a", "exec_command", argsA+argsA)), + passthroughSSEData(outputItemDoneJSON(0, "fc_a", "call_a", "exec_command", argsA+argsA)), + passthroughSSEData(`{"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","id":"fc_b","call_id":"call_b","name":"apply_patch","arguments":"","status":"in_progress"}}`), + passthroughSSEData(functionArgsDeltaJSON(1, "fc_b", "call_b", "apply_patch", `{"path":"/tmp/file",`)), + passthroughSSEData(functionArgsDeltaJSON(1, "fc_b", "call_b", "apply_patch", `"patch":{"ops":[{"op":"replace","value":{"lines":["x","y"]}}]}}`)), + passthroughSSEData(functionArgsDoneJSON(1, "fc_b", "call_b", "apply_patch", argsB+argsB)), + passthroughSSEData(outputItemDoneJSON(1, "fc_b", "call_b", "apply_patch", argsB+argsB)), + passthroughSSEData(completedWithFunctionCallsJSON(argsA+argsA, argsB+argsB)), + "data: [DONE]\n\n", + }, "") + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil) + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + } + + svc := &OpenAIGatewayService{} + result, err := svc.handleStreamingResponsePassthrough(context.Background(), resp, c, &Account{ID: 1}, time.Now(), "gpt-5.4", "gpt-5.4") + require.NoError(t, err) + require.NotNil(t, result) + + events := collectSSEDataPayloads(t, rec.Body.String()) + require.Equal(t, argsA, accumulateFunctionArgumentDeltas(events, "call_a")) + require.Equal(t, argsB, accumulateFunctionArgumentDeltas(events, "call_b")) + + require.Equal(t, argsA, gjson.Get(findSSEEvent(t, events, "response.function_call_arguments.done", "call_a"), "arguments").String()) + require.Equal(t, argsB, gjson.Get(findSSEEvent(t, events, "response.function_call_arguments.done", "call_b"), "arguments").String()) + require.Equal(t, argsA, gjson.Get(findSSEEvent(t, events, "response.output_item.done", "call_a"), "item.arguments").String()) + require.Equal(t, argsB, gjson.Get(findSSEEvent(t, events, "response.output_item.done", "call_b"), "item.arguments").String()) + + completed := findSSEEvent(t, events, "response.completed", "") + require.Equal(t, argsA, gjson.Get(completed, "response.output.0.arguments").String()) + require.Equal(t, argsB, gjson.Get(completed, "response.output.1.arguments").String()) + requireJSONArgument(t, gjson.Get(completed, "response.output.0.arguments").String()) + requireJSONArgument(t, gjson.Get(completed, "response.output.1.arguments").String()) +} + +func TestForwardResponsesChatCompletionsFallbackKeepsFunctionArgumentsSingle(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","input":"run a command","stream":true}`) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(string(body))) + c.Request.Header.Set("Content-Type", "application/json") + + upstreamBody := strings.Join([]string{ + passthroughSSEData(chatToolCallChunkJSON(true, "")), + "", + passthroughSSEData(chatToolCallChunkJSON(false, `{"cmd":"echo hi"}`)), + "", + `data: {"id":"chatcmpl_tool","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":3,"completion_tokens":2,"total_tokens":5}}`, + "", + "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_fallback_tool_args"}}, + Body: io.NopCloser(strings.NewReader(upstreamBody)), + }} + account := passthroughArgsFallbackAccount() + account.Extra = map[string]any{ + openai_compat.ExtraKeyResponsesMode: string(openai_compat.ResponsesSupportModeForceChatCompletions), + } + svc := &OpenAIGatewayService{ + cfg: passthroughArgsTestConfig(), + httpUpstream: upstream, + } + + result, err := svc.Forward(context.Background(), c, account, body) + require.NoError(t, err) + require.NotNil(t, result) + + const wantArgs = `{"cmd":"echo hi"}` + events := collectSSEDataPayloads(t, rec.Body.String()) + require.Equal(t, wantArgs, accumulateFunctionArgumentDeltas(events, "chatcmpl-tool-a")) + require.Equal(t, wantArgs, gjson.Get(findSSEEvent(t, events, "response.function_call_arguments.done", "chatcmpl-tool-a"), "arguments").String()) + require.Equal(t, wantArgs, gjson.Get(findSSEEvent(t, events, "response.output_item.done", "chatcmpl-tool-a"), "item.arguments").String()) +} + +func passthroughSSEData(payload string) string { + return "data: " + payload + "\n\n" +} + +func functionArgsDeltaJSON(outputIndex int, itemID, callID, name, delta string) string { + return fmt.Sprintf( + `{"type":"response.function_call_arguments.delta","output_index":%d,"item_id":%s,"call_id":%s,"name":%s,"delta":%s}`, + outputIndex, + strconv.Quote(itemID), + strconv.Quote(callID), + strconv.Quote(name), + strconv.Quote(delta), + ) +} + +func functionArgsDoneJSON(outputIndex int, itemID, callID, name, arguments string) string { + return fmt.Sprintf( + `{"type":"response.function_call_arguments.done","output_index":%d,"item_id":%s,"call_id":%s,"name":%s,"arguments":%s}`, + outputIndex, + strconv.Quote(itemID), + strconv.Quote(callID), + strconv.Quote(name), + strconv.Quote(arguments), + ) +} + +func outputItemDoneJSON(outputIndex int, itemID, callID, name, arguments string) string { + return fmt.Sprintf( + `{"type":"response.output_item.done","output_index":%d,"item":{"type":"function_call","id":%s,"call_id":%s,"name":%s,"arguments":%s,"status":"completed"}}`, + outputIndex, + strconv.Quote(itemID), + strconv.Quote(callID), + strconv.Quote(name), + strconv.Quote(arguments), + ) +} + +func completedWithFunctionCallsJSON(argsA, argsB string) string { + return fmt.Sprintf( + `{"type":"response.completed","response":{"id":"resp_passthrough_args","status":"completed","output":[{"type":"function_call","id":"fc_a","call_id":"call_a","name":"exec_command","arguments":%s,"status":"completed"},{"type":"function_call","id":"fc_b","call_id":"call_b","name":"apply_patch","arguments":%s,"status":"completed"}],"usage":{"input_tokens":2,"output_tokens":3,"total_tokens":5}}}`, + strconv.Quote(argsA), + strconv.Quote(argsB), + ) +} + +func chatToolCallChunkJSON(includeIdentity bool, arguments string) string { + identity := "" + functionFields := make([]string, 0, 2) + if includeIdentity { + identity = `"id":"chatcmpl-tool-a","type":"function",` + functionFields = append(functionFields, `"name":"exec_command"`) + } + if includeIdentity || arguments != "" { + functionFields = append(functionFields, `"arguments":`+strconv.Quote(arguments)) + } + return fmt.Sprintf( + `{"id":"chatcmpl_tool","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,%s"function":{%s}}]},"finish_reason":null}]}`, + identity, + strings.Join(functionFields, ","), + ) +} + +func passthroughArgsTestConfig() *config.Config { + return &config.Config{ + Security: config.SecurityConfig{ + URLAllowlist: config.URLAllowlistConfig{ + Enabled: false, + AllowInsecureHTTP: true, + }, + }, + } +} + +func passthroughArgsFallbackAccount() *Account { + return &Account{ + ID: 102, + Name: "passthrough-args-openai-apikey", + Platform: PlatformOpenAI, + Type: AccountTypeAPIKey, + Concurrency: 1, + Credentials: map[string]any{ + "api_key": "sk-test", + "base_url": "http://upstream.example", + }, + } +} + +func collectSSEDataPayloads(t *testing.T, body string) []string { + t.Helper() + scanner := bufio.NewScanner(strings.NewReader(body)) + var events []string + for scanner.Scan() { + data, ok := extractOpenAISSEDataLine(scanner.Text()) + if !ok { + continue + } + if strings.TrimSpace(data) == "[DONE]" { + continue + } + require.True(t, gjson.Valid(data), "invalid SSE data payload: %s", data) + events = append(events, data) + } + require.NoError(t, scanner.Err()) + return events +} + +func findSSEEvent(t *testing.T, events []string, eventType, callID string) string { + t.Helper() + for _, event := range events { + if gjson.Get(event, "type").String() != eventType { + continue + } + if callID == "" || + gjson.Get(event, "call_id").String() == callID || + gjson.Get(event, "item.call_id").String() == callID { + return event + } + } + t.Fatalf("missing event type=%s call_id=%s in %d events", eventType, callID, len(events)) + return "" +} + +func accumulateFunctionArgumentDeltas(events []string, callID string) string { + var b strings.Builder + for _, event := range events { + if gjson.Get(event, "type").String() != "response.function_call_arguments.delta" { + continue + } + if gjson.Get(event, "call_id").String() != callID { + continue + } + _, _ = b.WriteString(gjson.Get(event, "delta").String()) + } + return b.String() +} + +func requireJSONArgument(t *testing.T, arguments string) { + t.Helper() + var decoded any + require.NoError(t, json.Unmarshal([]byte(arguments), &decoded)) +} diff --git a/backend/internal/service/openai_gateway_service.go b/backend/internal/service/openai_gateway_service.go index 2bed9e0394..933a1c179d 100644 --- a/backend/internal/service/openai_gateway_service.go +++ b/backend/internal/service/openai_gateway_service.go @@ -3869,6 +3869,11 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( trimmedData = strings.TrimSpace(replacedData) } } + if normalizedData, normalized := normalizeOpenAIResponsesFunctionCallArguments(dataBytes); normalized { + dataBytes = normalizedData + trimmedData = strings.TrimSpace(string(normalizedData)) + line = "data: " + string(normalizedData) + } eventType := strings.TrimSpace(gjson.Get(trimmedData, "type").String()) if eventType == "response.failed" { failedMessage = extractOpenAISSEErrorMessage(dataBytes) @@ -5150,11 +5155,92 @@ func (s *OpenAIGatewayService) correctToolCallsInResponseBody(body []byte) []byt return body } - corrected, changed := s.toolCorrector.CorrectToolCallsInSSEBytes(body) - if changed { - return corrected + updated := body + if s != nil && s.toolCorrector != nil { + if corrected, changed := s.toolCorrector.CorrectToolCallsInSSEBytes(updated); changed { + updated = corrected + } } - return body + if normalized, changed := normalizeOpenAIResponsesFunctionCallArguments(updated); changed { + updated = normalized + } + return updated +} + +func normalizeOpenAIResponsesFunctionCallArguments(data []byte) ([]byte, bool) { + if len(bytes.TrimSpace(data)) == 0 || !bytes.Contains(data, []byte(`"arguments"`)) { + return data, false + } + if !gjson.ValidBytes(data) { + return data, false + } + + updated := data + changed := false + setDedupedArgument := func(path string) { + arg := gjson.GetBytes(updated, path) + if !arg.Exists() || arg.Type != gjson.String { + return + } + deduped, ok := dedupeRepeatedJSONArgumentString(arg.Str) + if !ok { + return + } + next, err := sjson.SetBytes(updated, path, deduped) + if err != nil { + return + } + updated = next + changed = true + } + + eventType := strings.TrimSpace(gjson.GetBytes(updated, "type").String()) + if eventType == "response.function_call_arguments.done" { + setDedupedArgument("arguments") + } + if itemType := strings.TrimSpace(gjson.GetBytes(updated, "item.type").String()); isResponsesFunctionCallItemType(itemType) { + setDedupedArgument("item.arguments") + } + dedupeResponsesFunctionCallOutputArguments(updated, "response.output", setDedupedArgument) + dedupeResponsesFunctionCallOutputArguments(updated, "output", setDedupedArgument) + + return updated, changed +} + +func dedupeResponsesFunctionCallOutputArguments(data []byte, outputPath string, setDedupedArgument func(string)) { + output := gjson.GetBytes(data, outputPath) + if !output.Exists() || !output.IsArray() { + return + } + for i, item := range output.Array() { + if !isResponsesFunctionCallItemType(strings.TrimSpace(item.Get("type").String())) { + continue + } + setDedupedArgument(outputPath + "." + strconv.Itoa(i) + ".arguments") + } +} + +func isResponsesFunctionCallItemType(itemType string) bool { + return itemType == "function_call" || itemType == "custom_tool_call" +} + +func dedupeRepeatedJSONArgumentString(arguments string) (string, bool) { + if len(arguments) == 0 || len(arguments)%2 != 0 { + return "", false + } + halfLen := len(arguments) / 2 + first := arguments[:halfLen] + if first != arguments[halfLen:] { + return "", false + } + trimmed := strings.TrimSpace(first) + if trimmed == "" || (!strings.HasPrefix(trimmed, "{") && !strings.HasPrefix(trimmed, "[")) { + return "", false + } + if !json.Valid([]byte(first)) { + return "", false + } + return first, true } func (s *OpenAIGatewayService) parseSSEUsage(data string, usage *OpenAIUsage) {