diff --git a/backend/internal/pkg/apicompat/responses_namespace.go b/backend/internal/pkg/apicompat/responses_namespace.go new file mode 100644 index 0000000000..a5549760c5 --- /dev/null +++ b/backend/internal/pkg/apicompat/responses_namespace.go @@ -0,0 +1,212 @@ +package apicompat + +import ( + "bytes" + "encoding/json" + "fmt" + "strings" +) + +// ResponsesNamespaceName identifies a function child in a Responses namespace. +// It aliases the chat bridge mapping so both native and bridged paths share one +// namespace identity contract. +type ResponsesNamespaceName = NamespacedToolName + +// FlattenResponsesNamespaces converts Codex private namespace declarations into +// public Responses function tools and rewrites namespace-qualified request calls. +func FlattenResponsesNamespaces(req map[string]any) (map[string]ResponsesNamespaceName, bool, error) { + return FlattenResponsesNamespacesExcept(req, nil) +} + +// FlattenResponsesNamespacesExcept is FlattenResponsesNamespaces with a set of +// service-owned namespace names that must remain native in the request. +func FlattenResponsesNamespacesExcept(req map[string]any, preserved map[string]bool) (map[string]ResponsesNamespaceName, bool, error) { + if req == nil { + return nil, false, nil + } + tools, ok := req["tools"].([]any) + if !ok || len(tools) == 0 { + return nil, false, nil + } + + topLevel := make(map[string]bool) + for _, raw := range tools { + tool, ok := raw.(map[string]any) + if !ok { + continue + } + typ := strings.TrimSpace(stringValue(tool["type"])) + name := strings.TrimSpace(stringValue(tool["name"])) + if (typ == "function" || typ == "custom") && name != "" { + topLevel[name] = true + } + } + + names := make(map[string]ResponsesNamespaceName) + for _, raw := range tools { + tool, ok := raw.(map[string]any) + if !ok || strings.TrimSpace(stringValue(tool["type"])) != "namespace" { + continue + } + namespace := strings.TrimSpace(stringValue(tool["name"])) + if namespace == "" || preserved[namespace] { + continue + } + for _, rawChild := range namespaceChildren(tool) { + child, ok := rawChild.(map[string]any) + if !ok || strings.TrimSpace(stringValue(child["type"])) != "function" { + continue + } + name := strings.TrimSpace(stringValue(child["name"])) + if name == "" { + continue + } + flat := flattenNamespaceToolName(namespace, name) + entry := ResponsesNamespaceName{Namespace: namespace, Name: name} + if topLevel[flat] { + return nil, false, fmt.Errorf("namespace tool %q/%q flattens to %q which conflicts with a top-level tool of the same name; this upstream cannot disambiguate them, rename one of the tools", namespace, name, flat) + } + if prev, exists := names[flat]; exists && prev != entry { + return nil, false, fmt.Errorf("namespace tools %q/%q and %q/%q both flatten to %q; this upstream cannot disambiguate them, rename one of the tools", prev.Namespace, prev.Name, namespace, name, flat) + } + names[flat] = entry + } + } + if len(names) == 0 { + return nil, false, nil + } + + flattened := make([]any, 0, len(tools)+len(names)) + seen := make(map[string]bool) + for _, raw := range tools { + tool, ok := raw.(map[string]any) + if !ok || strings.TrimSpace(stringValue(tool["type"])) != "namespace" { + flattened = append(flattened, raw) + continue + } + namespace := strings.TrimSpace(stringValue(tool["name"])) + if preserved[namespace] { + flattened = append(flattened, raw) + continue + } + for _, rawChild := range namespaceChildren(tool) { + child, ok := rawChild.(map[string]any) + if !ok || strings.TrimSpace(stringValue(child["type"])) != "function" { + continue + } + name := strings.TrimSpace(stringValue(child["name"])) + flat := flattenNamespaceToolName(namespace, name) + if name == "" || seen[flat] { + continue + } + seen[flat] = true + flatChild := make(map[string]any, len(child)) + for key, value := range child { + flatChild[key] = value + } + flatChild["name"] = flat + flattened = append(flattened, flatChild) + } + } + req["tools"] = flattened + rewriteNamespaceQualifiedCalls(req["input"], names) + if choice, ok := req["tool_choice"].(map[string]any); ok { + choiceNamespace := strings.TrimSpace(stringValue(choice["name"])) + if strings.TrimSpace(stringValue(choice["type"])) == "namespace" && !preserved[choiceNamespace] { + req["tool_choice"] = "auto" + } else { + rewriteNamespaceQualifiedCall(choice, names) + } + } + return names, true, nil +} + +// RestoreResponsesNamespaceCalls restores flattened function calls in a JSON +// Responses payload to the namespace/name identity expected by Codex. +func RestoreResponsesNamespaceCalls(payload []byte, names map[string]ResponsesNamespaceName) ([]byte, bool, error) { + if len(payload) == 0 || len(names) == 0 { + return payload, false, nil + } + var value any + if err := json.Unmarshal(payload, &value); err != nil { + return payload, false, err + } + changed := restoreResponsesNamespaceValue(value, names) + if !changed { + return payload, false, nil + } + var rebuilt bytes.Buffer + encoder := json.NewEncoder(&rebuilt) + encoder.SetEscapeHTML(false) + if err := encoder.Encode(value); err != nil { + return payload, false, err + } + return bytes.TrimSuffix(rebuilt.Bytes(), []byte("\n")), true, nil +} + +func namespaceChildren(tool map[string]any) []any { + if children, ok := tool["tools"].([]any); ok && len(children) > 0 { + return children + } + children, _ := tool["children"].([]any) + return children +} + +func rewriteNamespaceQualifiedCalls(value any, names map[string]ResponsesNamespaceName) { + switch typed := value.(type) { + case []any: + for _, item := range typed { + rewriteNamespaceQualifiedCalls(item, names) + } + case map[string]any: + if strings.TrimSpace(stringValue(typed["type"])) == "function_call" { + rewriteNamespaceQualifiedCall(typed, names) + } + for _, child := range typed { + rewriteNamespaceQualifiedCalls(child, names) + } + } +} + +func rewriteNamespaceQualifiedCall(item map[string]any, names map[string]ResponsesNamespaceName) bool { + namespace := strings.TrimSpace(stringValue(item["namespace"])) + name := strings.TrimSpace(stringValue(item["name"])) + if namespace == "" || name == "" { + return false + } + flat := flattenNamespaceToolName(namespace, name) + entry, ok := names[flat] + if !ok || entry.Namespace != namespace || entry.Name != name { + return false + } + item["name"] = flat + delete(item, "namespace") + return true +} + +func restoreResponsesNamespaceValue(value any, names map[string]ResponsesNamespaceName) bool { + changed := false + switch typed := value.(type) { + case []any: + for _, item := range typed { + changed = restoreResponsesNamespaceValue(item, names) || changed + } + case map[string]any: + if strings.TrimSpace(stringValue(typed["type"])) == "function_call" { + if entry, ok := names[strings.TrimSpace(stringValue(typed["name"]))]; ok { + typed["name"] = entry.Name + typed["namespace"] = entry.Namespace + changed = true + } + } + for _, child := range typed { + changed = restoreResponsesNamespaceValue(child, names) || changed + } + } + return changed +} + +func stringValue(value any) string { + text, _ := value.(string) + return text +} diff --git a/backend/internal/pkg/apicompat/responses_namespace_test.go b/backend/internal/pkg/apicompat/responses_namespace_test.go new file mode 100644 index 0000000000..4c8e313c5d --- /dev/null +++ b/backend/internal/pkg/apicompat/responses_namespace_test.go @@ -0,0 +1,164 @@ +package apicompat + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestFlattenResponsesNamespaces_RewritesDeclarationHistoryAndChoice(t *testing.T) { + req := map[string]any{ + "model": "gpt-5.5", + "tools": []any{ + map[string]any{"type": "function", "name": "plain", "description": "keep"}, + map[string]any{ + "type": "namespace", + "name": "collaboration", + "tools": []any{ + map[string]any{"type": "function", "name": "spawn_agent", "description": "spawn", "parameters": map[string]any{"type": "object"}}, + }, + }, + }, + "tool_choice": map[string]any{"type": "function", "name": "spawn_agent", "namespace": "collaboration"}, + "input": []any{ + map[string]any{"type": "function_call", "call_id": "call_1", "name": "spawn_agent", "namespace": "collaboration", "arguments": "{}"}, + map[string]any{"type": "message", "role": "user", "content": "hi", "name": "spawn_agent", "namespace": "collaboration"}, + }, + } + + names, changed, err := FlattenResponsesNamespaces(req) + require.NoError(t, err) + require.True(t, changed) + require.Equal(t, ResponsesNamespaceName{Namespace: "collaboration", Name: "spawn_agent"}, names["collaboration__spawn_agent"]) + + tools, ok := req["tools"].([]any) + require.True(t, ok) + require.Len(t, tools, 2) + plainTool, ok := tools[0].(map[string]any) + require.True(t, ok) + require.Equal(t, "plain", plainTool["name"]) + flatTool, ok := tools[1].(map[string]any) + require.True(t, ok) + require.Equal(t, "collaboration__spawn_agent", flatTool["name"]) + require.Equal(t, "spawn", flatTool["description"]) + + choice, ok := req["tool_choice"].(map[string]any) + require.True(t, ok) + require.Equal(t, "collaboration__spawn_agent", choice["name"]) + require.NotContains(t, choice, "namespace") + + input, ok := req["input"].([]any) + require.True(t, ok) + require.Len(t, input, 2) + call, ok := input[0].(map[string]any) + require.True(t, ok) + require.Equal(t, "collaboration__spawn_agent", call["name"]) + require.NotContains(t, call, "namespace") + message, ok := input[1].(map[string]any) + require.True(t, ok) + require.Equal(t, "spawn_agent", message["name"]) + require.Equal(t, "collaboration", message["namespace"]) + require.Equal(t, "gpt-5.5", req["model"]) +} + +func TestFlattenResponsesNamespaces_RejectsFlatNameCollision(t *testing.T) { + req := map[string]any{"tools": []any{ + map[string]any{"type": "function", "name": "collaboration__spawn_agent"}, + map[string]any{"type": "namespace", "name": "collaboration", "tools": []any{ + map[string]any{"type": "function", "name": "spawn_agent"}, + }}, + }} + + _, _, err := FlattenResponsesNamespaces(req) + require.ErrorContains(t, err, "conflicts with a top-level tool") +} + +func TestFlattenResponsesNamespaces_NamespaceGroupChoiceFallsBackToAuto(t *testing.T) { + req := map[string]any{ + "tools": []any{map[string]any{ + "type": "namespace", "name": "collaboration", "tools": []any{ + map[string]any{"type": "function", "name": "spawn_agent"}, + map[string]any{"type": "function", "name": "send_message"}, + }, + }}, + "tool_choice": map[string]any{"type": "namespace", "name": "collaboration"}, + } + + _, changed, err := FlattenResponsesNamespaces(req) + require.NoError(t, err) + require.True(t, changed) + require.Equal(t, "auto", req["tool_choice"]) +} + +func TestFlattenResponsesNamespacesExcept_PreservesBuiltInNamespaceAndChoice(t *testing.T) { + req := map[string]any{ + "tools": []any{ + map[string]any{"type": "namespace", "name": "image_gen", "tools": []any{ + map[string]any{"type": "function", "name": "imagegen"}, + }}, + map[string]any{"type": "namespace", "name": "collaboration", "tools": []any{ + map[string]any{"type": "function", "name": "spawn_agent"}, + }}, + }, + "tool_choice": map[string]any{"type": "namespace", "name": "image_gen"}, + } + + names, changed, err := FlattenResponsesNamespacesExcept(req, map[string]bool{"image_gen": true}) + require.NoError(t, err) + require.True(t, changed) + require.Contains(t, names, "collaboration__spawn_agent") + tools, ok := req["tools"].([]any) + require.True(t, ok) + require.Len(t, tools, 2) + preservedTool, ok := tools[0].(map[string]any) + require.True(t, ok) + require.Equal(t, "namespace", preservedTool["type"]) + require.Equal(t, "image_gen", preservedTool["name"]) + flatTool, ok := tools[1].(map[string]any) + require.True(t, ok) + require.Equal(t, "function", flatTool["type"]) + require.Equal(t, "collaboration__spawn_agent", flatTool["name"]) + require.Equal(t, map[string]any{"type": "namespace", "name": "image_gen"}, req["tool_choice"]) +} + +func TestFlattenResponsesNamespaces_RejectsNamespaceCollision(t *testing.T) { + req := map[string]any{"tools": []any{ + map[string]any{"type": "namespace", "name": "a", "tools": []any{ + map[string]any{"type": "function", "name": "b__c"}, + }}, + map[string]any{"type": "namespace", "name": "a__b", "tools": []any{ + map[string]any{"type": "function", "name": "c"}, + }}, + }} + + _, _, err := FlattenResponsesNamespaces(req) + require.ErrorContains(t, err, "both flatten") +} + +func TestRestoreResponsesNamespaceCalls_RewritesOnlyFunctionCalls(t *testing.T) { + payload := []byte(`{"type":"response.completed","response":{"output":[{"type":"function_call","name":"collaboration__spawn_agent","call_id":"call_1","arguments":"{}","extra":"keep"},{"type":"function_call","name":"plain","arguments":"{}"},{"type":"message","name":"collaboration__spawn_agent","content":"&value"}]}}`) + names := map[string]ResponsesNamespaceName{ + "collaboration__spawn_agent": {Namespace: "collaboration", Name: "spawn_agent"}, + } + + got, changed, err := RestoreResponsesNamespaceCalls(payload, names) + require.NoError(t, err) + require.True(t, changed) + require.JSONEq(t, `{"type":"response.completed","response":{"output":[{"type":"function_call","name":"spawn_agent","namespace":"collaboration","call_id":"call_1","arguments":"{}","extra":"keep"},{"type":"function_call","name":"plain","arguments":"{}"},{"type":"message","name":"collaboration__spawn_agent","content":"&value"}]}}`, string(got)) + require.Contains(t, string(got), "&value") + require.NotContains(t, string(got), `\u003c`) +} + +func TestRestoreResponsesNamespaceCalls_RewritesLifecycleItems(t *testing.T) { + for _, eventType := range []string{"response.output_item.added", "response.output_item.done"} { + t.Run(eventType, func(t *testing.T) { + payload := []byte(`{"type":"` + eventType + `","item":{"type":"function_call","name":"collaboration__spawn_agent","arguments":"{}"}}`) + got, changed, err := RestoreResponsesNamespaceCalls(payload, map[string]ResponsesNamespaceName{ + "collaboration__spawn_agent": {Namespace: "collaboration", Name: "spawn_agent"}, + }) + require.NoError(t, err) + require.True(t, changed) + require.JSONEq(t, `{"type":"`+eventType+`","item":{"type":"function_call","name":"spawn_agent","namespace":"collaboration","arguments":"{}"}}`, string(got)) + }) + } +} diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index b7c850d0c8..6c7d45946a 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -42,6 +42,20 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if normalized { body = normalizedBody } + wsDecision := s.getOpenAIWSProtocolResolver().Resolve(account) + // 仅允许 WS 入站请求走 WS 上游,避免出现 HTTP -> WS 协议混用。 + wsDecision = resolveOpenAIWSDecisionByClientTransport(wsDecision, GetOpenAIClientTransport(c)) + passthroughEnabled := account.IsOpenAIPassthroughEnabled() + if shouldFlattenOpenAIResponsesNamespaces(account, wsDecision.Transport, passthroughEnabled) { + body, err = flattenOpenAIResponsesNamespaces(c, body) + if err != nil { + setOpsUpstreamError(c, http.StatusBadRequest, err.Error(), "") + c.JSON(http.StatusBadRequest, gin.H{"error": gin.H{ + "type": "invalid_request_error", "message": err.Error(), "param": "tools", + }}) + return nil, err + } + } originalBody := body requestView := newOpenAIRequestView(body) @@ -64,10 +78,6 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if isCodexCLI { codexImageGenerationExplicitToolPolicy = account.CodexImageGenerationExplicitToolPolicy() } - wsDecision := s.getOpenAIWSProtocolResolver().Resolve(account) - clientTransport := GetOpenAIClientTransport(c) - // 仅允许 WS 入站请求走 WS 上游,避免出现 HTTP -> WS 协议混用。 - wsDecision = resolveOpenAIWSDecisionByClientTransport(wsDecision, clientTransport) if c != nil { c.Set("openai_ws_transport_decision", string(wsDecision.Transport)) c.Set("openai_ws_transport_reason", wsDecision.Reason) @@ -96,7 +106,6 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco } return nil, errors.New("openai ws v1 is temporarily unsupported; use ws v2") } - passthroughEnabled := account.IsOpenAIPassthroughEnabled() if passthroughEnabled { if isCodexCLI && codexImageGenerationExplicitToolPolicy == codexImageGenerationExplicitToolPolicyStrip { strippedBody, changed, stripErr := stripOpenAIImageGenerationToolsFromRawPayload(body) diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index 8283b54a19..0a19a9d283 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -919,6 +919,17 @@ func (s *OpenAIGatewayService) handleStreamingResponsePassthrough( trimmedData = strings.TrimSpace(string(normalizedData)) line = "data: " + string(normalizedData) } + if trimmedData != "[DONE]" { + restoredData, restoreErr := restoreOpenAIResponsesNamespacePayload(c, dataBytes) + if restoreErr != nil { + return resultWithUsage(), fmt.Errorf("restore OpenAI passthrough namespace response: %w", restoreErr) + } + if !bytes.Equal(restoredData, dataBytes) { + dataBytes = restoredData + trimmedData = strings.TrimSpace(string(restoredData)) + line = "data: " + string(restoredData) + } + } eventType := strings.TrimSpace(gjson.Get(trimmedData, "type").String()) if eventType == "response.failed" { failedMessage = extractOpenAISSEErrorMessage(dataBytes) @@ -1099,6 +1110,10 @@ func (s *OpenAIGatewayService) handleNonStreamingResponsePassthrough( if originalModel != "" && mappedModel != "" && originalModel != mappedModel { body = s.replaceModelInResponseBody(body, mappedModel, originalModel) } + body, err = restoreOpenAIResponsesNamespacePayload(c, body) + if err != nil { + return nil, fmt.Errorf("restore OpenAI passthrough namespace response: %w", err) + } if !writeOpenAICompactSSEBridge(c, resp.StatusCode, body) { c.Data(resp.StatusCode, contentType, body) } @@ -1140,6 +1155,11 @@ func (s *OpenAIGatewayService) handlePassthroughSSEToJSON(resp *http.Response, c } // Correct tool calls in final response body = s.correctToolCallsInResponseBody(body) + restoredBody, restoreErr := restoreOpenAIResponsesNamespacePayload(c, body) + if restoreErr != nil { + return nil, fmt.Errorf("restore OpenAI passthrough namespace response: %w", restoreErr) + } + body = restoredBody } else { terminalType, terminalPayload, terminalOK := extractOpenAISSETerminalEvent(bodyText) if terminalOK && terminalType == "response.failed" { diff --git a/backend/internal/service/openai_gateway_response_handling.go b/backend/internal/service/openai_gateway_response_handling.go index 845428b641..1d70245eb0 100644 --- a/backend/internal/service/openai_gateway_response_handling.go +++ b/backend/internal/service/openai_gateway_response_handling.go @@ -310,6 +310,17 @@ func (s *OpenAIGatewayService) handleStreamingResponse(ctx context.Context, resp line = "data: " + data eventType = strings.TrimSpace(gjson.GetBytes(dataBytes, "type").String()) } + restoredData, restoreErr := restoreOpenAIResponsesNamespacePayload(c, dataBytes) + if restoreErr != nil { + streamEarlyErr = fmt.Errorf("restore OpenAI namespace response: %w", restoreErr) + return + } + if !bytes.Equal(restoredData, dataBytes) { + dataBytes = restoredData + data = string(restoredData) + line = "data: " + data + eventType = strings.TrimSpace(gjson.GetBytes(dataBytes, "type").String()) + } if sanitizedData, sanitized := sanitizeOpenAIResponseFailedEventForClient( dataBytes, eventType, @@ -855,7 +866,10 @@ func (s *OpenAIGatewayService) handleNonStreamingResponse(ctx context.Context, r if originalModel != mappedModel { body = s.replaceModelInResponseBody(body, mappedModel, originalModel) } - + body, err = restoreOpenAIResponsesNamespacePayload(c, body) + if err != nil { + return nil, fmt.Errorf("restore OpenAI namespace response: %w", err) + } responseheaders.WriteFilteredHeaders(c.Writer.Header(), resp.Header, s.responseHeaderFilter) contentType := "application/json" @@ -925,6 +939,11 @@ func (s *OpenAIGatewayService) handleSSEToJSON(resp *http.Response, c *gin.Conte } // Correct tool calls in final response body = s.correctToolCallsInResponseBody(body) + restoredBody, restoreErr := restoreOpenAIResponsesNamespacePayload(c, body) + if restoreErr != nil { + return nil, fmt.Errorf("restore OpenAI namespace response: %w", restoreErr) + } + body = restoredBody } else { terminalType, terminalPayload, terminalOK := extractOpenAISSETerminalEvent(bodyText) if terminalOK && terminalType == "response.failed" { diff --git a/backend/internal/service/openai_oauth_passthrough_test.go b/backend/internal/service/openai_oauth_passthrough_test.go index 60790b9e4c..0aa536f94f 100644 --- a/backend/internal/service/openai_oauth_passthrough_test.go +++ b/backend/internal/service/openai_oauth_passthrough_test.go @@ -14,6 +14,7 @@ import ( "time" "github.com/Wei-Shaw/sub2api/internal/config" + "github.com/Wei-Shaw/sub2api/internal/pkg/apicompat" "github.com/Wei-Shaw/sub2api/internal/pkg/logger" "github.com/Wei-Shaw/sub2api/internal/pkg/openai_compat" "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint" @@ -422,6 +423,194 @@ func TestOpenAIGatewayService_OAuthPassthrough_StreamKeepsToolNameAndBodyNormali require.NotContains(t, body, "\"name\":\"edit\"") } +func TestOpenAIGatewayService_OAuthPassthrough_NamespaceRequestAndStreamResponse(t *testing.T) { + gin.SetMode(gin.TestMode) + + 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") + + originalBody := []byte(`{ + "model":"gpt-5.5", + "stream":true, + "instructions":"local-test-instructions", + "tools":[ + {"type":"function","name":"plain","description":"keep","parameters":{"type":"object"}}, + {"type":"namespace","name":"collaboration","tools":[{"type":"function","name":"spawn_agent","description":"spawn","parameters":{"type":"object"}}]} + ], + "tool_choice":{"type":"function","name":"spawn_agent","namespace":"collaboration"}, + "input":[{"type":"function_call","call_id":"call_old","name":"spawn_agent","namespace":"collaboration","arguments":"{}"}] + }`) + + upstreamSSE := strings.Join([]string{ + `data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_1","name":"collaboration__spawn_agent","arguments":""}}`, + "", + `data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_1","name":"collaboration__spawn_agent","arguments":"{}"}}`, + "", + `data: {"type":"response.completed","response":{"id":"resp_1","status":"completed","output":[{"type":"function_call","id":"fc_1","call_id":"call_1","name":"collaboration__spawn_agent","arguments":"{}"}],"usage":{"input_tokens":2,"output_tokens":1,"total_tokens":3}}}`, + "", + "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_namespace"}}, + Body: io.NopCloser(strings.NewReader(upstreamSSE)), + }} + svc := &OpenAIGatewayService{cfg: &config.Config{}, httpUpstream: upstream} + account := &Account{ + ID: 123, Name: "acc", Platform: PlatformOpenAI, Type: AccountTypeOAuth, Concurrency: 1, + Credentials: map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-acc"}, + Extra: map[string]any{"openai_passthrough": true}, Status: StatusActive, Schedulable: true, RateMultiplier: f64p(1), + } + + result, err := svc.Forward(context.Background(), c, account, originalBody) + require.NoError(t, err) + require.NotNil(t, result) + + require.Len(t, gjson.GetBytes(upstream.lastBody, "tools").Array(), 2) + require.Equal(t, "plain", gjson.GetBytes(upstream.lastBody, "tools.0.name").String()) + require.Equal(t, "function", gjson.GetBytes(upstream.lastBody, "tools.1.type").String()) + require.Equal(t, "collaboration__spawn_agent", gjson.GetBytes(upstream.lastBody, "tools.1.name").String()) + require.False(t, gjson.GetBytes(upstream.lastBody, "tools.1.tools").Exists()) + require.Equal(t, "collaboration__spawn_agent", gjson.GetBytes(upstream.lastBody, "tool_choice.name").String()) + require.False(t, gjson.GetBytes(upstream.lastBody, "tool_choice.namespace").Exists()) + require.Equal(t, "collaboration__spawn_agent", gjson.GetBytes(upstream.lastBody, "input.0.name").String()) + require.False(t, gjson.GetBytes(upstream.lastBody, "input.0.namespace").Exists()) + + downstream := rec.Body.String() + require.NotContains(t, downstream, "collaboration__spawn_agent") + require.Contains(t, downstream, `"name":"spawn_agent"`) + require.Contains(t, downstream, `"namespace":"collaboration"`) +} + +func TestOpenAIGatewayService_NativeOAuth_NamespaceRequestAndStreamResponse(t *testing.T) { + gin.SetMode(gin.TestMode) + 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") + body := []byte(`{ + "model":"gpt-5.5","stream":true,"instructions":"test", + "tools":[{"type":"namespace","name":"collaboration","tools":[{"type":"function","name":"spawn_agent","parameters":{"type":"object"}}]}], + "input":[{"type":"function_call","call_id":"call_old","name":"spawn_agent","namespace":"collaboration","arguments":"{}"}] + }`) + upstreamSSE := strings.Join([]string{ + `data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_1","name":"collaboration__spawn_agent","arguments":""}}`, + "", + `data: {"type":"response.completed","response":{"id":"resp_1","status":"completed","output":[{"type":"function_call","id":"fc_1","call_id":"call_1","name":"collaboration__spawn_agent","arguments":"{}"}],"usage":{"input_tokens":2,"output_tokens":1,"total_tokens":3}}}`, + "", + "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_native_namespace"}}, + Body: io.NopCloser(strings.NewReader(upstreamSSE)), + }} + svc := &OpenAIGatewayService{cfg: &config.Config{}, httpUpstream: upstream} + account := &Account{ + ID: 124, Name: "native", Platform: PlatformOpenAI, Type: AccountTypeOAuth, Concurrency: 1, + Credentials: map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-acc"}, + Status: StatusActive, Schedulable: true, RateMultiplier: f64p(1), + } + + result, err := svc.Forward(context.Background(), c, account, body) + require.NoError(t, err) + require.NotNil(t, result) + require.Equal(t, "function", gjson.GetBytes(upstream.lastBody, "tools.0.type").String()) + require.Equal(t, "collaboration__spawn_agent", gjson.GetBytes(upstream.lastBody, "tools.0.name").String()) + require.Equal(t, "collaboration__spawn_agent", gjson.GetBytes(upstream.lastBody, "input.0.name").String()) + require.False(t, gjson.GetBytes(upstream.lastBody, "input.0.namespace").Exists()) + require.NotContains(t, rec.Body.String(), "collaboration__spawn_agent") + require.Contains(t, rec.Body.String(), `"name":"spawn_agent"`) + require.Contains(t, rec.Body.String(), `"namespace":"collaboration"`) +} + +func TestOpenAIGatewayService_NativeOAuth_NamespaceNonStreamingResponse(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(nil)) + setOpenAIResponsesNamespaceNames(c, map[string]apicompat.ResponsesNamespaceName{ + "collaboration__spawn_agent": {Namespace: "collaboration", Name: "spawn_agent"}, + }) + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{ + "id":"resp_1","output":[{"type":"function_call","name":"collaboration__spawn_agent","call_id":"call_1","arguments":"{}"}], + "usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2} + }`)), + } + + result, err := (&OpenAIGatewayService{cfg: &config.Config{}}).handleNonStreamingResponse( + context.Background(), resp, c, &Account{Type: AccountTypeOAuth}, "gpt-5.5", "gpt-5.5", + ) + require.NoError(t, err) + require.NotNil(t, result) + require.NotContains(t, rec.Body.String(), "collaboration__spawn_agent") + require.Contains(t, rec.Body.String(), `"name":"spawn_agent"`) + require.Contains(t, rec.Body.String(), `"namespace":"collaboration"`) +} + +func TestOpenAIGatewayService_OAuthPassthrough_NamespaceNonStreamingResponse(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(nil)) + resp := &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"id":"resp_1","output":[{"type":"function_call","name":"collaboration__spawn_agent","call_id":"call_1","arguments":"{}"}],"usage":{"input_tokens":1,"output_tokens":1}}`)), + } + names := map[string]apicompat.ResponsesNamespaceName{ + "collaboration__spawn_agent": {Namespace: "collaboration", Name: "spawn_agent"}, + } + setOpenAIResponsesNamespaceNames(c, names) + + result, err := (&OpenAIGatewayService{cfg: &config.Config{}}).handleNonStreamingResponsePassthrough( + context.Background(), resp, c, "gpt-5.5", "", + ) + require.NoError(t, err) + require.NotNil(t, result) + require.NotContains(t, rec.Body.String(), "collaboration__spawn_agent") + require.Contains(t, rec.Body.String(), `"name":"spawn_agent"`) + require.Contains(t, rec.Body.String(), `"namespace":"collaboration"`) +} + +func TestOpenAIGatewayService_OAuthPassthrough_NamespaceCollisionReturnsBadRequest(t *testing.T) { + gin.SetMode(gin.TestMode) + 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") + body := []byte(`{ + "model":"gpt-5.5","stream":true,"instructions":"test", + "tools":[ + {"type":"function","name":"collaboration__spawn_agent","parameters":{"type":"object"}}, + {"type":"namespace","name":"collaboration","tools":[{"type":"function","name":"spawn_agent","parameters":{"type":"object"}}]} + ],"input":"hi" + }`) + upstream := &httpUpstreamRecorder{} + svc := &OpenAIGatewayService{cfg: &config.Config{}, httpUpstream: upstream} + account := &Account{ + ID: 123, Name: "acc", Platform: PlatformOpenAI, Type: AccountTypeOAuth, Concurrency: 1, + Credentials: map[string]any{"access_token": "oauth-token", "chatgpt_account_id": "chatgpt-acc"}, + Extra: map[string]any{"openai_passthrough": true}, Status: StatusActive, Schedulable: true, RateMultiplier: f64p(1), + } + + result, err := svc.Forward(context.Background(), c, account, body) + require.Error(t, err) + require.Nil(t, result) + require.Nil(t, upstream.lastReq) + require.Equal(t, http.StatusBadRequest, rec.Code) + require.Equal(t, "invalid_request_error", gjson.Get(rec.Body.String(), "error.type").String()) + require.Equal(t, "tools", gjson.Get(rec.Body.String(), "error.param").String()) + require.Contains(t, gjson.Get(rec.Body.String(), "error.message").String(), "conflicts with a top-level tool") +} + func TestOpenAIGatewayService_OAuthPassthrough_CompactUsesJSONAndKeepsNonStreaming(t *testing.T) { gin.SetMode(gin.TestMode) diff --git a/backend/internal/service/openai_responses_namespace.go b/backend/internal/service/openai_responses_namespace.go new file mode 100644 index 0000000000..7d71b64814 --- /dev/null +++ b/backend/internal/service/openai_responses_namespace.go @@ -0,0 +1,83 @@ +package service + +import ( + "bytes" + "encoding/json" + "fmt" + + "github.com/Wei-Shaw/sub2api/internal/pkg/apicompat" + "github.com/gin-gonic/gin" +) + +const openAIResponsesNamespaceNamesContextKey = "openai_responses_namespace_names" + +// shouldFlattenOpenAIResponsesNamespaces 判定原生 Responses 转发前是否摊平 +// Codex namespace 工具。WSv2 上游原生支持 namespace,且 WS 出口 +// (openai_ws_forwarder_v2)原样转发上游事件、不经 HTTP 回程还原,摊平后的 +// 平名无法还原会破坏客户端工具匹配,因此实际走 WSv2 分支的请求保持 namespace +// 原样。透传账号先于 WSv2 分支经 HTTP 转发返回,仍需摊平。 +func shouldFlattenOpenAIResponsesNamespaces(account *Account, transport OpenAIUpstreamTransport, passthroughEnabled bool) bool { + if account == nil || account.Type != AccountTypeOAuth { + return false + } + if transport == OpenAIUpstreamTransportResponsesWebsocketV2 && !passthroughEnabled { + return false + } + return true +} + +func flattenOpenAIResponsesNamespaces(c *gin.Context, body []byte) ([]byte, error) { + if !bytes.Contains(body, []byte(`"namespace"`)) { + return body, nil + } + var requestBody map[string]any + if err := json.Unmarshal(body, &requestBody); err != nil { + return body, fmt.Errorf("decode OpenAI namespace body: %w", err) + } + names, changed, err := apicompat.FlattenResponsesNamespacesExcept(requestBody, map[string]bool{"image_gen": true}) + if err != nil { + return body, err + } + if !changed { + return body, nil + } + rebuilt, err := marshalOpenAIUpstreamJSON(requestBody) + if err != nil { + return body, fmt.Errorf("encode OpenAI namespace body: %w", err) + } + setOpenAIResponsesNamespaceNames(c, names) + return rebuilt, nil +} + +func setOpenAIResponsesNamespaceNames(c *gin.Context, names map[string]apicompat.ResponsesNamespaceName) { + if c != nil && len(names) > 0 { + c.Set(openAIResponsesNamespaceNamesContextKey, names) + } +} + +func openAIResponsesNamespaceNames(c *gin.Context) map[string]apicompat.ResponsesNamespaceName { + if c == nil { + return nil + } + value, ok := c.Get(openAIResponsesNamespaceNamesContextKey) + if !ok { + return nil + } + names, _ := value.(map[string]apicompat.ResponsesNamespaceName) + return names +} + +func restoreOpenAIResponsesNamespacePayload(c *gin.Context, payload []byte) ([]byte, error) { + names := openAIResponsesNamespaceNames(c) + if len(names) == 0 || !json.Valid(payload) { + return payload, nil + } + restored, changed, err := apicompat.RestoreResponsesNamespaceCalls(payload, names) + if err != nil { + return payload, err + } + if changed { + return restored, nil + } + return payload, nil +} diff --git a/backend/internal/service/openai_responses_namespace_test.go b/backend/internal/service/openai_responses_namespace_test.go new file mode 100644 index 0000000000..7aa1392260 --- /dev/null +++ b/backend/internal/service/openai_responses_namespace_test.go @@ -0,0 +1,34 @@ +package service + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestShouldFlattenOpenAIResponsesNamespaces(t *testing.T) { + oauth := &Account{Type: AccountTypeOAuth} + apiKey := &Account{Type: AccountTypeAPIKey} + + tests := []struct { + name string + account *Account + transport OpenAIUpstreamTransport + passthroughEnabled bool + want bool + }{ + {name: "oauth_http", account: oauth, transport: OpenAIUpstreamTransportHTTPSSE, want: true}, + {name: "oauth_http_passthrough", account: oauth, transport: OpenAIUpstreamTransportHTTPSSE, passthroughEnabled: true, want: true}, + // WSv2 出口原样转发上游事件、不做回程还原,摊平会让客户端收到无法匹配的平名。 + {name: "oauth_wsv2", account: oauth, transport: OpenAIUpstreamTransportResponsesWebsocketV2, want: false}, + // 透传账号先于 WSv2 分支经 HTTP 转发返回,仍需摊平。 + {name: "oauth_wsv2_passthrough", account: oauth, transport: OpenAIUpstreamTransportResponsesWebsocketV2, passthroughEnabled: true, want: true}, + {name: "apikey_http", account: apiKey, transport: OpenAIUpstreamTransportHTTPSSE, want: false}, + {name: "nil_account", account: nil, transport: OpenAIUpstreamTransportHTTPSSE, want: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.want, shouldFlattenOpenAIResponsesNamespaces(tt.account, tt.transport, tt.passthroughEnabled)) + }) + } +}