From 095b525360fc597e8af55e0ddd61bca76f474ddf Mon Sep 17 00:00:00 2001 From: shaw Date: Tue, 25 Aug 2026 12:14:52 +0800 Subject: [PATCH] fix(openai): pin Responses Lite parallel tool calls --- .../service/openai_codex_tool_names_test.go | 2 +- .../openai_compact_body_signal_test.go | 2 +- .../openai_gateway_apikey_item_id_test.go | 4 +- .../service/openai_gateway_forward.go | 2 +- .../service/openai_gateway_passthrough.go | 3 +- .../service/openai_gateway_request_body.go | 22 +-- ...nai_gateway_request_body_reasoning_test.go | 23 +-- .../openai_passthrough_normalization_test.go | 14 +- .../service/openai_responses_lite_tools.go | 24 --- .../openai_responses_lite_tools_test.go | 139 +++++++++++------- .../openai_responses_tool_schema_test.go | 4 +- .../service/openai_ws_forwarder_ingress.go | 5 +- ...penai_ws_forwarder_ingress_session_test.go | 3 +- .../service/openai_ws_http_bridge_test.go | 2 +- .../openai_ws_v2_passthrough_adapter.go | 8 +- ...openai_ws_v2_passthrough_lifecycle_test.go | 36 ++++- 16 files changed, 166 insertions(+), 127 deletions(-) diff --git a/backend/internal/service/openai_codex_tool_names_test.go b/backend/internal/service/openai_codex_tool_names_test.go index d6cb9eafa1..73b7164deb 100644 --- a/backend/internal/service/openai_codex_tool_names_test.go +++ b/backend/internal/service/openai_codex_tool_names_test.go @@ -91,7 +91,7 @@ func TestApplyCodexOAuthTransform_ReservedPythonNameIsOAuthOnly(t *testing.T) { require.Equal(t, codexPythonToolAlias, tool["name"]) apiKeyBody := []byte(`{"type":"response.create","tools":[{"type":"function","name":"python"}]}`) - normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(apiKeyBody, &Account{Platform: PlatformOpenAI, Type: AccountTypeAPIKey}) + normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(apiKeyBody, &Account{Platform: PlatformOpenAI, Type: AccountTypeAPIKey}, false) require.NoError(t, err) require.False(t, changed) require.JSONEq(t, string(apiKeyBody), string(normalized)) diff --git a/backend/internal/service/openai_compact_body_signal_test.go b/backend/internal/service/openai_compact_body_signal_test.go index db8b6b404b..16587df0fd 100644 --- a/backend/internal/service/openai_compact_body_signal_test.go +++ b/backend/internal/service/openai_compact_body_signal_test.go @@ -96,7 +96,7 @@ func TestWebSocketCompatibilityNormalizesTriggerAfterPairedOutputCleanup(t *test body := []byte(`{"type":"response.create","model":"gpt-5.4","input":[{"type":"compaction_trigger"},{"type":"function_call","call_id":"call_1","name":"lookup","arguments":"{}"},{"type":"function_call_output","call_id":"call_1","output":"ok"},{"type":"message","role":"user","content":"visible"}]}`) account := &Account{Platform: PlatformOpenAI, Type: AccountTypeOAuth} - normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, account) + normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, account, false) require.NoError(t, err) require.True(t, changed) items := gjson.GetBytes(normalized, "input").Array() diff --git a/backend/internal/service/openai_gateway_apikey_item_id_test.go b/backend/internal/service/openai_gateway_apikey_item_id_test.go index e3a93bebb3..d5be3cec49 100644 --- a/backend/internal/service/openai_gateway_apikey_item_id_test.go +++ b/backend/internal/service/openai_gateway_apikey_item_id_test.go @@ -271,7 +271,7 @@ func TestNormalizeOpenAIResponsesWebSocketCompatibilityBodyPreservesOpaqueRefere normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{ Platform: PlatformOpenAI, Type: accountType, - }) + }, false) require.NoError(t, err) require.True(t, changed) @@ -285,7 +285,7 @@ func TestNormalizeOpenAIResponsesWebSocketCompatibilityBodyPreservesOpaqueRefere second, changedAgain, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(normalized, &Account{ Platform: PlatformOpenAI, Type: accountType, - }) + }, false) require.NoError(t, err) require.False(t, changedAgain) require.JSONEq(t, string(normalized), string(second)) diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index 38d747ca33..6487385000 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -152,7 +152,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco return s.forwardResponsesViaNativeAnthropic(ctx, c, account, body, reqModel) } if account.IsOpenAIApiKey() { - if normalized, changed, normalizeErr := normalizeOpenAIParallelToolCallsWithoutTools(body); normalizeErr != nil { + if normalized, changed, normalizeErr := normalizeOpenAIParallelToolCallsWithoutTools(body, responsesLite); normalizeErr != nil { return nil, normalizeErr } else if changed { body = normalized diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index d5d0c75408..b87ce9df87 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -212,7 +212,8 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough( } } if account != nil && account.IsOpenAI() { - normalizedBody, normalized, normalizeErr := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, account) + responsesLite := isOpenAIResponsesLiteHeader(c.GetHeader(responsesLiteHeader)) || isOpenAIResponsesLiteWebSocketPayload(body) + normalizedBody, normalized, normalizeErr := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, account, responsesLite) if normalizeErr != nil { return nil, fmt.Errorf("normalize passthrough Responses compatibility: %w", normalizeErr) } diff --git a/backend/internal/service/openai_gateway_request_body.go b/backend/internal/service/openai_gateway_request_body.go index 8aa4364777..6272875cc5 100644 --- a/backend/internal/service/openai_gateway_request_body.go +++ b/backend/internal/service/openai_gateway_request_body.go @@ -349,7 +349,7 @@ func normalizeOpenAICompactRequestBody(body []byte) ([]byte, bool, error) { } normalized = next } - if next, removed, err := normalizeOpenAIParallelToolCallsWithoutTools(normalized); err != nil { + if next, removed, err := normalizeOpenAIParallelToolCallsWithoutTools(normalized, false); err != nil { return body, false, err } else if removed { normalized = next @@ -361,7 +361,10 @@ func normalizeOpenAICompactRequestBody(body []byte) ([]byte, bool, error) { return normalized, true, nil } -func normalizeOpenAIParallelToolCallsWithoutTools(body []byte) ([]byte, bool, error) { +func normalizeOpenAIParallelToolCallsWithoutTools(body []byte, responsesLite bool) ([]byte, bool, error) { + if responsesLite { + return body, false, nil + } parallel := gjson.GetBytes(body, "parallel_tool_calls") if !parallel.Exists() { return body, false, nil @@ -376,16 +379,7 @@ func normalizeOpenAIParallelToolCallsWithoutTools(body []byte) ([]byte, bool, er return normalized, true, nil } -// openAIRequestBodyHasTools is the []byte counterpart of openAIResponsesLiteHasTools: -// besides the top-level "tools" array it also recognizes the Responses Lite carrier. -// normalizeOpenAIResponsesLiteTools moves namespace tools into an input item of type -// "additional_tools" and drops the top-level "tools" key; the request still carries -// tools at that point. Looking only at the top level therefore misreads such a body as -// "no tools" and deletes the parallel_tool_calls:false that -// ensureOpenAIResponsesLiteParallelToolCalls had just pinned, and OpenAI falls back to -// its default of true and rejects the request with -// 400 unsupported_value: "X-OpenAI-Internal-Codex-Responses-Lite requires -// `parallel_tool_calls` to be false." +// openAIRequestBodyHasTools 同时识别顶层 tools 和 input[].additional_tools。 func openAIRequestBodyHasTools(body []byte) bool { if tools := gjson.GetBytes(body, "tools"); tools.IsArray() && len(tools.Array()) > 0 { return true @@ -948,7 +942,7 @@ func normalizeOpenAIResponseFormatSchemasBody(body []byte) ([]byte, bool, error) return normalized, true, nil } -func normalizeOpenAIResponsesWebSocketCompatibilityBody(body []byte, account *Account) ([]byte, bool, error) { +func normalizeOpenAIResponsesWebSocketCompatibilityBody(body []byte, account *Account, responsesLite bool) ([]byte, bool, error) { if account == nil || !account.IsOpenAI() { return body, false, nil } @@ -962,7 +956,7 @@ func normalizeOpenAIResponsesWebSocketCompatibilityBody(body []byte, account *Ac } } if account.IsOpenAIApiKey() { - if next, normalizedParallel, err := normalizeOpenAIParallelToolCallsWithoutTools(normalized); err != nil { + if next, normalizedParallel, err := normalizeOpenAIParallelToolCallsWithoutTools(normalized, responsesLite); err != nil { return body, false, err } else if normalizedParallel { normalized = next diff --git a/backend/internal/service/openai_gateway_request_body_reasoning_test.go b/backend/internal/service/openai_gateway_request_body_reasoning_test.go index 03397e751f..543d294880 100644 --- a/backend/internal/service/openai_gateway_request_body_reasoning_test.go +++ b/backend/internal/service/openai_gateway_request_body_reasoning_test.go @@ -259,34 +259,37 @@ func TestNormalizeOpenAIAPIKeyStoreFalseReasoningReplayRejectsEmptyEncryptedCont func TestNormalizeOpenAIParallelToolCallsWithoutTools(t *testing.T) { withTools := []byte(`{"tools":[{"type":"function","name":"lookup"}],"parallel_tool_calls":false}`) - normalized, changed, err := normalizeOpenAIParallelToolCallsWithoutTools(withTools) + normalized, changed, err := normalizeOpenAIParallelToolCallsWithoutTools(withTools, false) require.NoError(t, err) require.False(t, changed) require.Equal(t, string(withTools), string(normalized)) withoutTools := []byte(`{"input":"hi","parallel_tool_calls":true}`) - normalized, changed, err = normalizeOpenAIParallelToolCallsWithoutTools(withoutTools) + normalized, changed, err = normalizeOpenAIParallelToolCallsWithoutTools(withoutTools, false) require.NoError(t, err) require.True(t, changed) require.False(t, gjson.GetBytes(normalized, "parallel_tool_calls").Exists()) } -// A Responses Lite body that has already been through normalizeOpenAIResponsesLiteTools -// carries its tools in an input item of type "additional_tools" and no longer has a -// top-level "tools" key. It still has tools, so the parallel_tool_calls:false that -// ensureOpenAIResponsesLiteParallelToolCalls pinned must survive this normalization — -// otherwise OpenAI applies its default of true and rejects the request. +// Lite 工具迁移到 input[].additional_tools 后，仍应按有工具请求处理。 func TestNormalizeOpenAIParallelToolCallsWithoutTools_KeepsResponsesLiteAdditionalTools(t *testing.T) { liteBody := []byte(`{"input":[{"type":"message","role":"user","content":"hi"},{"type":"additional_tools","tools":[{"type":"function","name":"spawn_agent"}]}],"parallel_tool_calls":false}`) - normalized, changed, err := normalizeOpenAIParallelToolCallsWithoutTools(liteBody) + normalized, changed, err := normalizeOpenAIParallelToolCallsWithoutTools(liteBody, false) require.NoError(t, err) require.False(t, changed) require.Equal(t, gjson.False, gjson.GetBytes(normalized, "parallel_tool_calls").Type) - // An empty additional_tools item carries no tools, so the field is still dropped. + // 非 Lite 请求的空 additional_tools 不构成有效工具声明，字段仍需删除。 emptyLiteBody := []byte(`{"input":[{"type":"additional_tools","tools":[]}],"parallel_tool_calls":true}`) - normalized, changed, err = normalizeOpenAIParallelToolCallsWithoutTools(emptyLiteBody) + normalized, changed, err = normalizeOpenAIParallelToolCallsWithoutTools(emptyLiteBody, false) require.NoError(t, err) require.True(t, changed) require.False(t, gjson.GetBytes(normalized, "parallel_tool_calls").Exists()) + + // Lite 请求即使没有工具，也必须保留已经固定的 false。 + toolLessLiteBody := []byte(`{"input":"hi","parallel_tool_calls":false}`) + normalized, changed, err = normalizeOpenAIParallelToolCallsWithoutTools(toolLessLiteBody, true) + require.NoError(t, err) + require.False(t, changed) + require.Equal(t, gjson.False, gjson.GetBytes(normalized, "parallel_tool_calls").Type) } diff --git a/backend/internal/service/openai_passthrough_normalization_test.go b/backend/internal/service/openai_passthrough_normalization_test.go index ac5cb87c71..f86ecc6927 100644 --- a/backend/internal/service/openai_passthrough_normalization_test.go +++ b/backend/internal/service/openai_passthrough_normalization_test.go @@ -55,7 +55,7 @@ func TestNormalizeOpenAIOAuthResponsesCompatibilityBody_PreservesExplicitInput(t func TestNormalizeOpenAIResponsesWebSocketCompatibilityBody_OnlyStripsOAuthFields(t *testing.T) { body := []byte(`{"type":"response.create","prompt":"hello","commands":{},"truncation":"auto","stop_sequences":["END"],"chat_template_kwargs":{"enable_thinking":true}}`) - oauthBody, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: AccountTypeOAuth}) + oauthBody, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: AccountTypeOAuth}, false) require.NoError(t, err) require.True(t, changed) require.Equal(t, "hello", gjson.GetBytes(oauthBody, "input").String()) @@ -63,7 +63,7 @@ func TestNormalizeOpenAIResponsesWebSocketCompatibilityBody_OnlyStripsOAuthField require.False(t, gjson.GetBytes(oauthBody, field).Exists(), field) } - apiKeyBody, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: AccountTypeAPIKey}) + apiKeyBody, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: AccountTypeAPIKey}, false) require.NoError(t, err) require.False(t, changed) require.JSONEq(t, string(body), string(apiKeyBody)) @@ -81,7 +81,7 @@ func TestNormalizeOpenAIResponsesWebSocketCompatibilityBody_SanitizesNativeItemI if oauth { accountType = AccountTypeOAuth } - normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: accountType}) + normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: accountType}, false) require.NoError(t, err) require.True(t, changed) require.Equal(t, "response.create", gjson.GetBytes(normalized, "type").String()) @@ -105,7 +105,7 @@ func TestNormalizeOpenAIResponsesWebSocketCompatibilityBody_APIKeyStoreFalseRepl normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{ Platform: PlatformOpenAI, Type: AccountTypeAPIKey, - }) + }, false) require.NoError(t, err) require.True(t, changed) @@ -141,13 +141,13 @@ func TestNormalizeOpenAIResponsesReasoningMode(t *testing.T) { func TestNormalizeOpenAIResponsesWebSocketCompatibilityBody_ReasoningModeAccountScope(t *testing.T) { body := []byte(`{"type":"response.create","reasoning":{"mode":"pro"}}`) for _, accountType := range []string{AccountTypeOAuth, AccountTypeSetupToken} { - normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: accountType}) + normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: accountType}, false) require.NoError(t, err) require.True(t, changed) require.Equal(t, "max", gjson.GetBytes(normalized, "reasoning.effort").String()) require.False(t, gjson.GetBytes(normalized, "reasoning.mode").Exists()) } - apiKeyBody, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: AccountTypeAPIKey}) + apiKeyBody, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: AccountTypeAPIKey}, false) require.NoError(t, err) require.False(t, changed) require.JSONEq(t, string(body), string(apiKeyBody)) @@ -156,7 +156,7 @@ func TestNormalizeOpenAIResponsesWebSocketCompatibilityBody_ReasoningModeAccount func TestNormalizeOpenAIResponsesWebSocketCompatibilityBody_SanitizesToolSchemas(t *testing.T) { body := []byte(`{"type":"response.create","tools":[{"type":"function","name":"search","parameters":{"type":null,"properties":{"q":{"type":"string","pattern":"^(?=.*foo).+$"}}}}]}`) for _, accountType := range []string{AccountTypeAPIKey, AccountTypeOAuth} { - normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: accountType}) + normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{Platform: PlatformOpenAI, Type: accountType}, false) require.NoError(t, err) require.True(t, changed) require.Equal(t, "object", gjson.GetBytes(normalized, "tools.0.parameters.type").String()) diff --git a/backend/internal/service/openai_responses_lite_tools.go b/backend/internal/service/openai_responses_lite_tools.go index 752ed00561..7f6c7f58e5 100644 --- a/backend/internal/service/openai_responses_lite_tools.go +++ b/backend/internal/service/openai_responses_lite_tools.go @@ -106,13 +106,6 @@ func ensureOpenAIResponsesLiteParallelToolCalls(reqBody map[string]any, changed return false, newOpenAIResponsesLiteValidationError("parallel_tool_calls", "responses Lite requires parallel_tool_calls to be a boolean") } } - if !openAIResponsesLiteHasTools(reqBody) { - if exists { - delete(reqBody, "parallel_tool_calls") - return true, nil - } - return changed, nil - } if parallel == false { return changed, nil } @@ -120,23 +113,6 @@ func ensureOpenAIResponsesLiteParallelToolCalls(reqBody map[string]any, changed return true, nil } -func openAIResponsesLiteHasTools(reqBody map[string]any) bool { - if tools, ok := reqBody["tools"].([]any); ok && len(tools) > 0 { - return true - } - input, _ := reqBody["input"].([]any) - for _, rawItem := range input { - item, ok := rawItem.(map[string]any) - if !ok || strings.TrimSpace(firstNonEmptyString(item["type"])) != "additional_tools" { - continue - } - if tools, ok := item["tools"].([]any); ok && len(tools) > 0 { - return true - } - } - return false -} - func ensureOpenAIResponsesLiteReasoningContext(reqBody map[string]any) (bool, error) { rawReasoning, exists := reqBody["reasoning"] if !exists || rawReasoning == nil { diff --git a/backend/internal/service/openai_responses_lite_tools_test.go b/backend/internal/service/openai_responses_lite_tools_test.go index 3edff5c2da..4088882d24 100644 --- a/backend/internal/service/openai_responses_lite_tools_test.go +++ b/backend/internal/service/openai_responses_lite_tools_test.go @@ -201,25 +201,33 @@ func TestNormalizeOpenAIResponsesLiteTools_ForcesParallelToolCallsFalse(t *testi } } -func TestNormalizeOpenAIResponsesLiteTools_RemovesParallelToolCallsWithoutTools(t *testing.T) { - for _, value := range []bool{true, false} { - reqBody := map[string]any{ - "reasoning": map[string]any{"context": "all_turns"}, - "parallel_tool_calls": value, - } - - changed, err := normalizeOpenAIResponsesLiteTools(reqBody) - - require.NoError(t, err) - require.True(t, changed) - require.NotContains(t, reqBody, "parallel_tool_calls") +func TestNormalizeOpenAIResponsesLiteTools_PinsParallelToolCallsWithoutTools(t *testing.T) { + tests := []struct { + name string + parallel any + include bool + wantChanged bool + }{ + {name: "字段缺失", wantChanged: true}, + {name: "值为 true", parallel: true, include: true, wantChanged: true}, + {name: "值为 false", parallel: false, include: true, wantChanged: false}, } - reqBody := map[string]any{"reasoning": map[string]any{"context": "all_turns"}} - changed, err := normalizeOpenAIResponsesLiteTools(reqBody) - require.NoError(t, err) - require.False(t, changed) - require.NotContains(t, reqBody, "parallel_tool_calls") + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + reqBody := map[string]any{"reasoning": map[string]any{"context": "all_turns"}} + if tt.include { + reqBody["parallel_tool_calls"] = tt.parallel + } + + changed, err := normalizeOpenAIResponsesLiteTools(reqBody) + + require.NoError(t, err) + require.Equal(t, tt.wantChanged, changed) + require.Contains(t, reqBody, "parallel_tool_calls") + require.Equal(t, false, reqBody["parallel_tool_calls"]) + }) + } } func TestNormalizeOpenAIResponsesLiteTools_RejectsNonBooleanParallelToolCalls(t *testing.T) { @@ -490,49 +498,70 @@ func TestOpenAIGatewayServiceForward_NormalizesResponsesLiteToolsForOAuth(t *tes } } -func TestOpenAIGatewayServiceForward_RemovesParallelToolCallsForToollessResponsesLite(t *testing.T) { +func TestOpenAIGatewayServiceForward_PinsParallelToolCallsForToollessResponsesLite(t *testing.T) { gin.SetMode(gin.TestMode) - for _, passthrough := range []bool{false, true} { - name := "managed" - if passthrough { - name = "passthrough" - } - t.Run(name, func(t *testing.T) { - rec := httptest.NewRecorder() - c, _ := gin.CreateTestContext(rec) - c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(nil)) - c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.1") - c.Request.Header.Set(responsesLiteHeader, "true") - upstream := &httpUpstreamRecorder{resp: &http.Response{ - StatusCode: http.StatusOK, - Header: http.Header{"Content-Type": []string{"text/event-stream"}}, - Body: io.NopCloser(strings.NewReader( - "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_lite\",\"usage\":{\"input_tokens\":1,\"output_tokens\":1}}}\n\n" + - "data: [DONE]\n\n", - )), - }} - svc := &OpenAIGatewayService{cfg: &config.Config{}, httpUpstream: upstream} - account := &Account{ - ID: 502, Name: "responses-lite-no-tools", Platform: PlatformOpenAI, Type: AccountTypeOAuth, - Concurrency: 1, Status: StatusActive, Schedulable: true, RateMultiplier: f64p(1), - Credentials: map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-account"}, - Extra: map[string]any{"openai_passthrough": passthrough}, + accountCases := []struct { + name string + accountType string + credentials map[string]any + }{ + {name: "oauth", accountType: AccountTypeOAuth, credentials: map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-account"}}, + {name: "apikey", accountType: AccountTypeAPIKey, credentials: map[string]any{"api_key": "sk-test"}}, + } + parallelCases := []struct { + name string + field string + }{ + {name: "字段缺失"}, + {name: "值为 true", field: `,"parallel_tool_calls":true`}, + {name: "值为 false", field: `,"parallel_tool_calls":false`}, + } + + for _, accountCase := range accountCases { + for _, passthrough := range []bool{false, true} { + mode := "managed" + if passthrough { + mode = "passthrough" } - body := []byte(`{ - "model":"gpt-5.6-terra","stream":true,"instructions":"test", - "reasoning":{"effort":"high","context":"current_turn"}, - "parallel_tool_calls":true, - "input":[{"type":"message","role":"user","content":"hello"}] - }`) + for _, parallelCase := range parallelCases { + name := accountCase.name + "/" + mode + "/" + parallelCase.name + t.Run(name, func(t *testing.T) { + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(nil)) + c.Request.Header.Set("User-Agent", "codex_cli_rs/0.144.1") + c.Request.Header.Set(responsesLiteHeader, "true") + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader( + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_lite\",\"usage\":{\"input_tokens\":1,\"output_tokens\":1}}}\n\n" + + "data: [DONE]\n\n", + )), + }} + svc := &OpenAIGatewayService{cfg: &config.Config{}, httpUpstream: upstream} + account := &Account{ + ID: 502, Name: "responses-lite-no-tools", Platform: PlatformOpenAI, Type: accountCase.accountType, + Concurrency: 1, Status: StatusActive, Schedulable: true, RateMultiplier: f64p(1), + Credentials: accountCase.credentials, + Extra: map[string]any{"openai_passthrough": passthrough}, + } + body := []byte(`{ + "model":"gpt-5.6-terra","stream":true,"instructions":"test", + "reasoning":{"effort":"high","context":"current_turn"}, + "input":[{"type":"message","role":"user","content":"hello"}]` + parallelCase.field + ` + }`) - result, err := svc.Forward(context.Background(), c, account, body) + result, err := svc.Forward(context.Background(), c, account, body) - require.NoError(t, err) - require.NotNil(t, result) - require.Equal(t, "true", upstream.lastReq.Header.Get(responsesLiteHeader)) - require.False(t, gjson.GetBytes(upstream.lastBody, "parallel_tool_calls").Exists(), string(upstream.lastBody)) - }) + require.NoError(t, err) + require.NotNil(t, result) + require.Equal(t, "true", upstream.lastReq.Header.Get(responsesLiteHeader)) + require.Equal(t, gjson.False, gjson.GetBytes(upstream.lastBody, "parallel_tool_calls").Type, string(upstream.lastBody)) + }) + } + } } } diff --git a/backend/internal/service/openai_responses_tool_schema_test.go b/backend/internal/service/openai_responses_tool_schema_test.go index faaf9062de..e899e17098 100644 --- a/backend/internal/service/openai_responses_tool_schema_test.go +++ b/backend/internal/service/openai_responses_tool_schema_test.go @@ -497,7 +497,7 @@ func TestOpenAIResponsesToolSchemaPlatformGate_APIKeyAndOAuth(t *testing.T) { normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{ Platform: PlatformOpenAI, Type: accountType, - }) + }, false) require.NoError(t, err) require.True(t, changed) require.Equal(t, "object", gjson.GetBytes(normalized, "tools.0.parameters.type").String()) @@ -508,7 +508,7 @@ func TestOpenAIResponsesToolSchemaPlatformGate_APIKeyAndOAuth(t *testing.T) { normalized, changed, err := normalizeOpenAIResponsesWebSocketCompatibilityBody(body, &Account{ Platform: PlatformGrok, Type: AccountTypeAPIKey, - }) + }, false) require.NoError(t, err) require.False(t, changed) require.Equal(t, string(body), string(normalized)) diff --git a/backend/internal/service/openai_ws_forwarder_ingress.go b/backend/internal/service/openai_ws_forwarder_ingress.go index 5f9aed1f61..8dbb5b1abb 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress.go +++ b/backend/internal/service/openai_ws_forwarder_ingress.go @@ -262,7 +262,8 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( normalized = capped } } - if compatibilityBody, compatibilityChanged, compatibilityErr := normalizeOpenAIResponsesWebSocketCompatibilityBody(normalized, account); compatibilityErr != nil { + responsesLite := isOpenAIResponsesLiteWebSocketPayload(normalized) + if compatibilityBody, compatibilityChanged, compatibilityErr := normalizeOpenAIResponsesWebSocketCompatibilityBody(normalized, account, responsesLite); compatibilityErr != nil { return openAIWSClientPayload{}, NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, "invalid websocket request payload", compatibilityErr) } else if compatibilityChanged { normalized = compatibilityBody @@ -320,7 +321,7 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( if accountScoped { normalized = accountScopedPayload } - if isOpenAIResponsesLiteWebSocketPayload(normalized) { + if responsesLite { litePayload, _, liteErr := normalizeOpenAIResponsesLitePayloadForAccount(normalized, account) if liteErr != nil { return openAIWSClientPayload{}, NewOpenAIWSClientCloseError( diff --git a/backend/internal/service/openai_ws_forwarder_ingress_session_test.go b/backend/internal/service/openai_ws_forwarder_ingress_session_test.go index 2e09082d6e..b05a2efd10 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress_session_test.go +++ b/backend/internal/service/openai_ws_forwarder_ingress_session_test.go @@ -1364,8 +1364,7 @@ func TestOpenAIGatewayService_ProxyResponsesWebSocketFromClient_HTTPBridgeModeRe err = clientConn.Write(writeCtx, coderws.MessageText, []byte(`{ "type":"response.create","model":"gpt-5.1","stream":false, "parallel_tool_calls":true, - "client_metadata":{"ws_request_header_x_openai_internal_codex_responses_lite":"true"}, - "tools":[{"type":"function","name":"lookup","parameters":{"type":"object"}}] + "client_metadata":{"ws_request_header_x_openai_internal_codex_responses_lite":"true"} }`)) cancelWrite() require.NoError(t, err) diff --git a/backend/internal/service/openai_ws_http_bridge_test.go b/backend/internal/service/openai_ws_http_bridge_test.go index 752fdcca19..13752a55ef 100644 --- a/backend/internal/service/openai_ws_http_bridge_test.go +++ b/backend/internal/service/openai_ws_http_bridge_test.go @@ -1239,7 +1239,7 @@ func TestOpenAIWSHTTPBridgeRelaysSSEFramesAsWebSocketMessages(t *testing.T) { Concurrency: 1, Status: StatusActive, } - payload := []byte(`{"type":"response.create","generate":true,"model":"gpt-5","stream":true,"client_metadata":{"ws_request_header_x_openai_internal_codex_responses_lite":"true"},"input":"hi","tools":[{"type":"function","name":"lookup","parameters":{"type":"object"}}],"parallel_tool_calls":true}`) + payload := []byte(`{"type":"response.create","generate":true,"model":"gpt-5","stream":true,"client_metadata":{"ws_request_header_x_openai_internal_codex_responses_lite":"true"},"input":"hi","parallel_tool_calls":true}`) type bridgeResult struct { result *OpenAIForwardResult diff --git a/backend/internal/service/openai_ws_v2_passthrough_adapter.go b/backend/internal/service/openai_ws_v2_passthrough_adapter.go index d22e386e11..60e7b41416 100644 --- a/backend/internal/service/openai_ws_v2_passthrough_adapter.go +++ b/backend/internal/service/openai_ws_v2_passthrough_adapter.go @@ -735,7 +735,8 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( if capturedSessionModel != "" && capturedSessionModel != strings.TrimSpace(gjson.GetBytes(firstClientMessage, "model").String()) { firstClientMessage = s.ReplaceModelInBody(firstClientMessage, capturedSessionModel) } - if normalized, compatibilityChanged, normalizeErr := normalizeOpenAIResponsesWebSocketCompatibilityBody(firstClientMessage, account); normalizeErr != nil { + firstMessageResponsesLite := isOpenAIResponsesLiteWebSocketPayload(firstClientMessage) + if normalized, compatibilityChanged, normalizeErr := normalizeOpenAIResponsesWebSocketCompatibilityBody(firstClientMessage, account, firstMessageResponsesLite); normalizeErr != nil { return fmt.Errorf("normalize first websocket response.create: %w", normalizeErr) } else if compatibilityChanged { firstClientMessage = normalized @@ -984,8 +985,9 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( } }() } + responsesLite := isResponseCreate && isOpenAIResponsesLiteWebSocketPayload(payload) if isResponseCreate { - if normalized, compatibilityChanged, normalizeErr := normalizeOpenAIResponsesWebSocketCompatibilityBody(payload, account); normalizeErr != nil { + if normalized, compatibilityChanged, normalizeErr := normalizeOpenAIResponsesWebSocketCompatibilityBody(payload, account, responsesLite); normalizeErr != nil { return payload, nil, NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, "invalid websocket request payload", normalizeErr) } else if compatibilityChanged { payload = normalized @@ -1011,7 +1013,7 @@ func (s *OpenAIGatewayService) proxyResponsesWebSocketV2Passthrough( } } if isResponseCreate { - if isOpenAIResponsesLiteWebSocketPayload(payload) { + if responsesLite { litePayload, _, liteErr := normalizeOpenAIResponsesLitePayloadForAccount(payload, account) if liteErr != nil { return payload, nil, NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, liteErr.Error(), liteErr) diff --git a/backend/internal/service/openai_ws_v2_passthrough_lifecycle_test.go b/backend/internal/service/openai_ws_v2_passthrough_lifecycle_test.go index de3d5fdf27..ab7d704b04 100644 --- a/backend/internal/service/openai_ws_v2_passthrough_lifecycle_test.go +++ b/backend/internal/service/openai_ws_v2_passthrough_lifecycle_test.go @@ -190,13 +190,18 @@ func startPassthroughLifecycleServer( } func dialPassthroughLifecycleClient(t *testing.T, server *httptest.Server) *coderws.Conn { + t.Helper() + return dialPassthroughLifecycleClientWithPayload(t, server, `{"type":"response.create","model":"gpt-5.1","stream":false}`) +} + +func dialPassthroughLifecycleClientWithPayload(t *testing.T, server *httptest.Server, payload string) *coderws.Conn { t.Helper() dialCtx, cancelDial := context.WithTimeout(context.Background(), 3*time.Second) clientConn, _, err := coderws.Dial(dialCtx, "ws"+strings.TrimPrefix(server.URL, "http"), nil) cancelDial() require.NoError(t, err) writeCtx, cancelWrite := context.WithTimeout(context.Background(), 3*time.Second) - err = clientConn.Write(writeCtx, coderws.MessageText, []byte(`{"type":"response.create","model":"gpt-5.1","stream":false}`)) + err = clientConn.Write(writeCtx, coderws.MessageText, []byte(payload)) cancelWrite() require.NoError(t, err) return clientConn @@ -221,6 +226,35 @@ func requirePassthroughUpstreamWrite(t *testing.T, upstream *stagedPassthroughCo } } +func TestPassthroughLifecycle_ResponsesLiteFirstFramePinsParallelToolCalls(t *testing.T) { + gin.SetMode(gin.TestMode) + controlCtx, cancelControl := context.WithCancelCause(context.Background()) + defer cancelControl(context.Canceled) + upstream := newStagedPassthroughConn() + upstream.Send(`{"type":"response.completed","response":{"id":"resp_lite","model":"gpt-5.1","usage":{"input_tokens":1,"output_tokens":1}}}`) + server, serverErr := startPassthroughLifecycleServer(t, controlCtx, newPassthroughLifecycleService(passthroughLifecycleConfig(), upstream), passthroughLifecycleAccount()) + defer server.Close() + clientConn := dialPassthroughLifecycleClientWithPayload(t, server, `{ + "type":"response.create","model":"gpt-5.1","stream":false, + "parallel_tool_calls":true, + "client_metadata":{"ws_request_header_x_openai_internal_codex_responses_lite":"true"} + }`) + defer func() { _ = clientConn.CloseNow() }() + + upstreamBody := requirePassthroughUpstreamWrite(t, upstream, 3*time.Second) + require.Equal(t, gjson.False, gjson.GetBytes(upstreamBody, "parallel_tool_calls").Type, string(upstreamBody)) + + event, err := readPassthroughLifecycleFrame(t, clientConn, 3*time.Second) + require.NoError(t, err) + require.Equal(t, "response.completed", gjson.GetBytes(event, "type").String()) + require.NoError(t, clientConn.Close(coderws.StatusNormalClosure, "done")) + select { + case <-serverErr: + case <-time.After(3 * time.Second): + t.Fatal("Lite 首帧测试等待 passthrough 退出超时") + } +} + func TestOpenAIWSPassthroughTurnLifecycle_SerializesTerminalCommitAndNextTurn(t *testing.T) { clientFrameConn := &openAIWSClientFrameConn{interTurnStarted: make(chan struct{}, 1)} clientFrameConn.markTurnCompleted()