diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index a94d570078..8192ebb6dc 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "net/http" + "regexp" "strings" "sync/atomic" "time" @@ -431,6 +432,18 @@ func (s *OpenAIGatewayService) forwardAsChatCompletions( ) return s.forwardAsRawChatCompletions(ctx, c, account, body, defaultMappedModel) } + responsesMode, _ := account.Extra[openai_compat.ExtraKeyResponsesMode].(string) + if account.IsOpenAIApiKey() && + openai_compat.NormalizeResponsesSupportMode(responsesMode) == openai_compat.ResponsesSupportModeAuto && + openai_compat.ResolveResponsesSupport(account.Extra) == openai_compat.ResponsesSupportYes && + !isResponsesShape && + isConvertedResponsesInputStringRejection(resp.StatusCode, respBody) { + logger.L().Info("openai chat_completions: structured Responses input rejected, falling back to raw chat completions", + zap.Int64("account_id", account.ID), + zap.Int("upstream_status", resp.StatusCode), + ) + return s.forwardAsRawChatCompletions(ctx, c, account, body, defaultMappedModel) + } if foErr := s.failoverOpenAIUpstreamHTTPError(ctx, c, account, resp, respBody, upstreamMsg, upstreamModel); foErr != nil { return nil, foErr } @@ -483,6 +496,30 @@ func (s *OpenAIGatewayService) forwardAsChatCompletions( return result, handleErr } +var responsesInputStringValidationError = regexp.MustCompile(`['"]loc['"]\s*:\s*\(\s*['"]body['"]\s*,\s*['"]input['"]\s*,\s*['"]str['"]\s*\)\s*,\s*['"]msg['"]\s*:\s*['"]Input should be a valid string['"]`) + +func isConvertedResponsesInputStringRejection(status int, responseBody []byte) bool { + if status != http.StatusBadRequest { + return false + } + message := gjson.GetBytes(responseBody, "error.message").String() + if responsesInputStringValidationError.MatchString(message) { + return true + } + code := strings.ToLower(strings.TrimSpace(gjson.GetBytes(responseBody, "error.code").String())) + typ := strings.ToLower(strings.TrimSpace(gjson.GetBytes(responseBody, "error.type").String())) + if code != "invalid_type" && typ != "invalid_type" { + return false + } + param := strings.ToLower(strings.TrimSpace(gjson.GetBytes(responseBody, "error.param").String())) + if param != "input" { + return false + } + message = strings.ToLower(strings.TrimSpace(message)) + return strings.Contains(message, "expected a string") && + (strings.Contains(message, "got an array") || strings.Contains(message, "got an object")) +} + func normalizeResponsesRequestServiceTier(req *apicompat.ResponsesRequest) { if req == nil { return diff --git a/backend/internal/service/openai_gateway_chat_completions_test.go b/backend/internal/service/openai_gateway_chat_completions_test.go index 5ff9a55e6b..5d8a04831b 100644 --- a/backend/internal/service/openai_gateway_chat_completions_test.go +++ b/backend/internal/service/openai_gateway_chat_completions_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "fmt" "io" "net/http" "net/http/httptest" @@ -13,6 +14,7 @@ import ( "github.com/Wei-Shaw/sub2api/internal/config" "github.com/Wei-Shaw/sub2api/internal/pkg/apicompat" + "github.com/Wei-Shaw/sub2api/internal/pkg/openai_compat" "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" "github.com/tidwall/gjson" @@ -32,6 +34,290 @@ func (w *openAIChatFailingWriter) Write(p []byte) (int, error) { return w.ResponseWriter.Write(p) } +func TestForwardAsChatCompletions_ResponsesSupportedFallsBackWhenStructuredInputRequiresString(t *testing.T) { + for name, rejection := range map[string]string{ + "OpenAI invalid type": `{"error":{"type":"invalid_request_error","code":"invalid_type","param":"input","message":"Invalid type for 'input': expected a string, but got an array instead."}}`, + "vLLM validation": `{"error":{"message":"240 validation errors: ... 'loc':('body','input','str'), 'msg':'Input should be a valid string', 'input':[{'role':'user','content':'hi'}, {'role':'assistant','content':[{'type':'output_text'}]}, ResponseFunctionToolCall(type='function_call'), {'type':'function_call_output'}]"}}`, + } { + t.Run(name, func(t *testing.T) { + gin.SetMode(gin.TestMode) + + body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"},{"role":"assistant","content":[{"type":"text","text":"checking"},{"type":"thinking","thinking":"think"}],"tool_calls":[{"id":"call_1","type":"function","function":{"name":"lookup","arguments":"{}"}}]},{"role":"tool","tool_call_id":"call_1","content":"done"},{"role":"user","content":"what now"}],"stream":false}`) + originalBody := append([]byte(nil), body...) + logs, releaseLogs := captureStructuredLog(t) + defer releaseLogs() + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + c.Request.Header.Set("Content-Type", "application/json") + + upstream := &httpUpstreamRecorder{responses: []*http.Response{ + { + StatusCode: http.StatusBadRequest, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader( + rejection, + )), + }, + { + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid_raw_fallback"}}, + Body: io.NopCloser(strings.NewReader( + `{"id":"chatcmpl_fallback","object":"chat.completion","model":"gpt-5.4","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}`, + )), + }, + }} + svc := &OpenAIGatewayService{ + cfg: structuredInputRetryTestConfig(), + httpUpstream: upstream, + } + account := structuredInputRetryTestAccount() + + result, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "") + require.NoError(t, err) + require.NotNil(t, result) + require.Len(t, upstream.requests, 2) + require.Equal(t, "/v1/responses", upstream.requests[0].URL.Path) + require.True(t, gjson.GetBytes(upstream.bodies[0], "input").IsArray()) + require.Equal(t, "/v1/chat/completions", upstream.requests[1].URL.Path) + require.JSONEq(t, string(body), string(upstream.bodies[1])) + require.Equal(t, originalBody, body) + require.Equal(t, "/v1/chat/completions", result.UpstreamEndpoint) + require.Equal(t, "/v1/chat/completions", GetActualOpenAIUpstreamEndpoint(c)) + require.Equal(t, "ok", gjson.Get(rec.Body.String(), "choices.0.message.content").String()) + require.True(t, logs.ContainsMessage("structured Responses input rejected")) + logs.mu.Lock() + defer logs.mu.Unlock() + for _, event := range logs.events { + require.NotContains(t, event.Fields, "upstream_message") + require.NotContains(t, event.Fields, "upstream_body") + require.NotContains(t, event.Message, rejection) + } + }) + } +} + +func TestConvertedResponsesInputStringRejectionIsNarrow(t *testing.T) { + matching := []byte(`{"error":{"code":"invalid_type","param":"input","message":"Expected a string for input, but got an object instead."}}`) + require.True(t, isConvertedResponsesInputStringRejection(http.StatusBadRequest, matching)) + + for _, tc := range []struct { + name string + status int + body string + }{ + {name: "nested content validation", status: http.StatusBadRequest, body: `{"error":{"message":"'loc':('body','input',0,'content','str'), 'msg':'Input should be a valid string'"}}`}, + {name: "other field validation", status: http.StatusBadRequest, body: `{"error":{"message":"'loc':('body','model','str'), 'msg':'Input should be a valid string'"}}`}, + {name: "generic string validation", status: http.StatusBadRequest, body: `{"error":{"message":"Input should be a valid string"}}`}, + {name: "wrong status", status: http.StatusUnprocessableEntity, body: string(matching)}, + {name: "authentication", status: http.StatusUnauthorized, body: string(matching)}, + {name: "authorization", status: http.StatusForbidden, body: string(matching)}, + {name: "rate limit", status: http.StatusTooManyRequests, body: string(matching)}, + {name: "server error", status: http.StatusInternalServerError, body: string(matching)}, + {name: "wrong parameter", status: http.StatusBadRequest, body: `{"error":{"code":"invalid_type","param":"tools","message":"Expected a string, but got an array instead."}}`}, + {name: "wrong code", status: http.StatusBadRequest, body: `{"error":{"code":"invalid_request_error","param":"input","message":"Expected a string, but got an array instead."}}`}, + {name: "unrelated input error", status: http.StatusBadRequest, body: `{"error":{"code":"invalid_type","param":"input","message":"Input is too long."}}`}, + } { + t.Run(tc.name, func(t *testing.T) { + require.False(t, isConvertedResponsesInputStringRejection(tc.status, []byte(tc.body))) + }) + } +} + +func TestForwardAsChatCompletions_StructuredInputRejectionDoesNotRetryIneligibleRequests(t *testing.T) { + rejection := `{"error":{"code":"invalid_type","param":"input","message":"Expected a string, but got an array instead."}}` + for _, test := range []struct { + name string + mode string + unknown bool + platform string + accountType string + responsesBody bool + nativeIngress bool + status int + rejection string + }{ + {name: "forced Responses", mode: "force_responses"}, + {name: "forced Responses without probe", mode: "force_responses", unknown: true}, + {name: "unknown support", unknown: true}, + {name: "Responses shaped Chat ingress", responsesBody: true}, + {name: "native Responses ingress", responsesBody: true, nativeIngress: true}, + {name: "native CN Responses protocol", platform: PlatformDeepseek}, + {name: "OAuth", accountType: AccountTypeOAuth}, + {name: "authentication", status: http.StatusUnauthorized}, + {name: "authorization", status: http.StatusForbidden}, + {name: "rate limit", status: http.StatusTooManyRequests}, + {name: "unrelated validation", rejection: `{"error":{"code":"invalid_type","param":"tools","message":"Expected a string, but got an array instead."}}`}, + {name: "nested validation", rejection: `{"error":{"message":"'loc':('body','input',0,'content','str'), 'msg':'Input should be a valid string'"}}`}, + } { + t.Run(test.name, func(t *testing.T) { + body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}],"stream":false}`) + if test.responsesBody { + body = []byte(`{"model":"gpt-5.4","input":[{"role":"user","content":"hello"}],"stream":false}`) + } + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + account := structuredInputRetryTestAccount() + account.Extra = map[string]any{openai_compat.ExtraKeyResponsesSupported: true, openai_compat.ExtraKeyResponsesMode: test.mode} + if test.unknown { + delete(account.Extra, openai_compat.ExtraKeyResponsesSupported) + } + if test.platform != "" { + account.Platform = test.platform + account.Credentials["api_protocol"] = APIProtocolResponses + } + if test.accountType != "" { + account.Type = test.accountType + account.Credentials["access_token"] = "test-token" + } + status := test.status + if status == 0 { + status = http.StatusBadRequest + } + errorBody := rejection + if test.rejection != "" { + errorBody = test.rejection + } + upstream := &httpUpstreamRecorder{resp: &http.Response{ + StatusCode: status, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(errorBody)), + }} + svc := &OpenAIGatewayService{cfg: structuredInputRetryTestConfig(), httpUpstream: upstream} + var result *OpenAIForwardResult + var err error + if test.nativeIngress { + c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body)) + result, err = svc.Forward(context.Background(), c, account, body) + } else { + result, err = svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "") + } + require.Error(t, err) + require.Nil(t, result) + require.Len(t, upstream.requests, 1) + require.True(t, strings.HasSuffix(upstream.requests[0].URL.Path, "/responses")) + var failover *UpstreamFailoverError + if status == http.StatusUnauthorized || status == http.StatusForbidden || status == http.StatusTooManyRequests { + require.ErrorAs(t, err, &failover) + require.Equal(t, status, failover.StatusCode) + } else { + require.False(t, errors.As(err, &failover)) + require.True(t, c.Writer.Written()) + } + }) + } +} + +func TestForwardAsChatCompletions_StructuredInputRetryPreservesStreamingAndMapping(t *testing.T) { + for _, useAccountMapping := range []bool{false, true} { + t.Run(fmt.Sprintf("account_mapping=%t", useAccountMapping), func(t *testing.T) { + body := []byte(`{"model":"client-model","messages":[{"role":"user","content":"hello"}],"stream":true,"stream_options":{"include_usage":false},"vendor_option":"preserve"}`) + originalBody := append([]byte(nil), body...) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + account := structuredInputRetryTestAccount() + account.Extra = map[string]any{openai_compat.ExtraKeyResponsesSupported: true, openai_compat.ExtraKeyResponsesMode: "auto"} + mappedModel := "gpt-5.4" + if useAccountMapping { + account.Credentials["model_mapping"] = map[string]any{"client-model": mappedModel} + } + stream := "data: {\"id\":\"chatcmpl_retry\",\"object\":\"chat.completion.chunk\",\"model\":\"gpt-5.4\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\ndata: {\"choices\":[],\"usage\":{\"prompt_tokens\":9,\"completion_tokens\":4,\"total_tokens\":13}}\n\ndata: [DONE]\n\n" + upstream := &httpUpstreamRecorder{responses: []*http.Response{ + {StatusCode: http.StatusBadRequest, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"error":{"code":"invalid_type","param":"input","message":"Expected a string, but got an array instead."}}`))}, + {StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(stream))}, + }} + svc := &OpenAIGatewayService{cfg: structuredInputRetryTestConfig(), httpUpstream: upstream} + defaultMappedModel := mappedModel + if useAccountMapping { + defaultMappedModel = "" + } + result, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", defaultMappedModel) + require.NoError(t, err) + require.Len(t, upstream.requests, 2) + for _, outbound := range upstream.bodies { + require.Equal(t, mappedModel, gjson.GetBytes(outbound, "model").String()) + require.True(t, gjson.GetBytes(outbound, "stream").Bool()) + } + require.JSONEq(t, gjson.GetBytes(body, "messages").Raw, gjson.GetBytes(upstream.bodies[1], "messages").Raw) + require.Equal(t, "preserve", gjson.GetBytes(upstream.bodies[1], "vendor_option").String()) + require.True(t, gjson.GetBytes(upstream.bodies[1], "stream_options.include_usage").Bool()) + require.Equal(t, originalBody, body) + require.True(t, result.Stream) + require.Equal(t, mappedModel, result.BillingModel) + require.Equal(t, mappedModel, result.UpstreamModel) + require.Equal(t, 9, result.Usage.InputTokens) + require.Equal(t, 4, result.Usage.OutputTokens) + require.Equal(t, "/v1/chat/completions", result.UpstreamEndpoint) + require.Contains(t, rec.Header().Get("Content-Type"), "text/event-stream") + require.Contains(t, rec.Body.String(), "data: [DONE]") + }) + } +} + +func TestForwardAsChatCompletions_StructuredInputRetryIsBoundedAndPreservesRawErrors(t *testing.T) { + for _, status := range []int{http.StatusBadRequest, http.StatusUnauthorized, http.StatusTooManyRequests} { + t.Run(fmt.Sprintf("status=%d", status), func(t *testing.T) { + rejection := `{"error":{"code":"invalid_type","param":"input","message":"Expected a string, but got an array instead."}}` + response := func(status int) *http.Response { + return &http.Response{StatusCode: status, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(rejection))} + } + body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}],"stream":false}`) + account := structuredInputRetryTestAccount() + account.Extra = map[string]any{openai_compat.ExtraKeyResponsesSupported: true} + forward := func(retry bool) (*httptest.ResponseRecorder, error) { + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + upstream := &httpUpstreamRecorder{responses: []*http.Response{response(status)}} + if retry { + upstream.responses = append([]*http.Response{response(http.StatusBadRequest)}, upstream.responses...) + } + svc := &OpenAIGatewayService{cfg: structuredInputRetryTestConfig(), httpUpstream: upstream} + var result *OpenAIForwardResult + var err error + if retry { + result, err = svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "") + require.Len(t, upstream.requests, 2) + require.Equal(t, "/v1/chat/completions", upstream.requests[1].URL.Path) + } else { + result, err = svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") + require.Len(t, upstream.requests, 1) + } + require.Nil(t, result) + require.Error(t, err) + return rec, err + } + baseline, baselineErr := forward(false) + retried, retryErr := forward(true) + require.Equal(t, baseline.Code, retried.Code) + require.Equal(t, baseline.Body.String(), retried.Body.String()) + require.Equal(t, baselineErr.Error(), retryErr.Error()) + require.IsType(t, baselineErr, retryErr) + }) + } +} + +func structuredInputRetryTestConfig() *config.Config { + return &config.Config{Security: config.SecurityConfig{URLAllowlist: config.URLAllowlistConfig{ + Enabled: false, + AllowInsecureHTTP: true, + }}} +} + +func structuredInputRetryTestAccount() *Account { + return &Account{ + ID: 101, + Name: "raw-openai-apikey", + Platform: PlatformOpenAI, + Type: AccountTypeAPIKey, + Concurrency: 1, + Credentials: map[string]any{"api_key": "sk-test", "base_url": "http://upstream.example"}, + Extra: map[string]any{openai_compat.ExtraKeyResponsesSupported: true}, + } +} + type openAIChatStreamReadErrorCloser struct { payload []byte err error