mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-09-24 16:05:44 +08:00
fix(openai): retry explicitly rejected Responses fields
This commit is contained in:
王鹏
@@ -391,9 +391,8 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
|
||||
if maxOutputTokens.Exists() {
|
||||
switch account.Platform {
|
||||
case PlatformOpenAI:
|
||||
if account.Type == AccountTypeAPIKey {
|
||||
markPatchDelete("max_output_tokens")
|
||||
}
|
||||
// Preserve Responses-native output limits unless the selected upstream
|
||||
// explicitly rejects the field in the bounded HTTP retry loop below.
|
||||
case PlatformAnthropic:
|
||||
decoded, decodeErr := ensureReqBody()
|
||||
if decodeErr != nil {
|
||||
@@ -744,6 +743,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
|
||||
|
||||
httpInvalidEncryptedContentRetryTried := false
|
||||
agentTaskRecoveryTried := false
|
||||
rejectedFieldRetryState := newOpenAIResponsesRejectedFieldRetryState(body)
|
||||
for {
|
||||
// Build upstream request
|
||||
upstreamCtx, releaseUpstreamCtx := detachUpstreamContext(ctx)
|
||||
@@ -830,11 +830,21 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
|
||||
return nil, fmt.Errorf("serialize invalid_encrypted_content retry body: %w", err)
|
||||
}
|
||||
httpInvalidEncryptedContentRetryTried = true
|
||||
rejectedFieldRetryState.remember(body)
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Retrying non-WSv2 request once after invalid_encrypted_content (account: %s)", account.Name)
|
||||
continue
|
||||
}
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Skip non-WSv2 invalid_encrypted_content retry because encrypted reasoning items are missing (account: %s)", account.Name)
|
||||
}
|
||||
if retryBody, reason, changed, retryErr := normalizeOpenAIResponsesRejectedFieldRetryBody(resp.StatusCode, body, respBody); retryErr != nil {
|
||||
return nil, fmt.Errorf("normalize rejected Responses field retry body: %w", retryErr)
|
||||
} else if changed && rejectedFieldRetryState.Allow(retryBody) {
|
||||
body = retryBody
|
||||
requestView = newOpenAIRequestView(body)
|
||||
reqBody = nil
|
||||
logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Retrying non-WSv2 request after %s (account: %s)", reason, account.Name)
|
||||
continue
|
||||
}
|
||||
if s.shouldFailoverOpenAIUpstreamResponse(resp.StatusCode, upstreamMsg, respBody) {
|
||||
upstreamDetail := ""
|
||||
if s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody {
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/tidwall/gjson"
|
||||
"github.com/tidwall/sjson"
|
||||
)
|
||||
|
||||
const maxOpenAIResponsesRejectedFieldRetries = 6
|
||||
|
||||
var (
|
||||
openAIResponsesRejectedNamespaceParamPattern = regexp.MustCompile(`(?i)^input\[(\d+)\]\.namespace$`)
|
||||
openAIResponsesRejectedMessageParamPattern = regexp.MustCompile(`(?i)(?:unknown|unsupported)[ _-]+parameter\s*(?::|=|is)?\s*["']?(max_output_tokens|input\[\d+\]\.namespace)(?:["']|\b)`)
|
||||
)
|
||||
|
||||
type openAIResponsesRejectedFieldRetryState struct {
|
||||
attempts int
|
||||
seenBodyHashes map[[sha256.Size]byte]struct{}
|
||||
}
|
||||
|
||||
func newOpenAIResponsesRejectedFieldRetryState(initialBody []byte) *openAIResponsesRejectedFieldRetryState {
|
||||
state := &openAIResponsesRejectedFieldRetryState{
|
||||
seenBodyHashes: make(map[[sha256.Size]byte]struct{}, maxOpenAIResponsesRejectedFieldRetries+1),
|
||||
}
|
||||
state.remember(initialBody)
|
||||
return state
|
||||
}
|
||||
|
||||
func (s *openAIResponsesRejectedFieldRetryState) Allow(nextBody []byte) bool {
|
||||
if s == nil || len(nextBody) == 0 || s.attempts >= maxOpenAIResponsesRejectedFieldRetries {
|
||||
return false
|
||||
}
|
||||
bodyHash := sha256.Sum256(nextBody)
|
||||
if _, seen := s.seenBodyHashes[bodyHash]; seen {
|
||||
return false
|
||||
}
|
||||
s.seenBodyHashes[bodyHash] = struct{}{}
|
||||
s.attempts++
|
||||
return true
|
||||
}
|
||||
|
||||
func (s *openAIResponsesRejectedFieldRetryState) remember(body []byte) {
|
||||
if s == nil || len(body) == 0 {
|
||||
return
|
||||
}
|
||||
if s.seenBodyHashes == nil {
|
||||
s.seenBodyHashes = make(map[[sha256.Size]byte]struct{}, maxOpenAIResponsesRejectedFieldRetries+1)
|
||||
}
|
||||
s.seenBodyHashes[sha256.Sum256(body)] = struct{}{}
|
||||
}
|
||||
|
||||
func normalizeOpenAIResponsesRejectedFieldRetryBody(statusCode int, body, responseBody []byte) ([]byte, string, bool, error) {
|
||||
if statusCode != http.StatusBadRequest || len(body) == 0 || len(responseBody) == 0 {
|
||||
return nil, "", false, nil
|
||||
}
|
||||
|
||||
code := strings.ToLower(strings.TrimSpace(extractUpstreamErrorCode(responseBody)))
|
||||
message := strings.ToLower(strings.TrimSpace(extractUpstreamErrorMessage(responseBody)))
|
||||
if !isExplicitOpenAIResponsesFieldRejection(code, message) {
|
||||
return nil, "", false, nil
|
||||
}
|
||||
|
||||
param := strings.ToLower(strings.TrimSpace(gjson.GetBytes(responseBody, "error.param").String()))
|
||||
if param == "" {
|
||||
param = openAIResponsesRejectedParamFromMessage(message)
|
||||
}
|
||||
if index, ok := openAIResponsesRejectedNamespaceIndex(param); ok {
|
||||
return removeOpenAIResponsesRejectedNamespaceAtIndex(body, index)
|
||||
}
|
||||
if param == "max_output_tokens" && gjson.GetBytes(body, "max_output_tokens").Exists() {
|
||||
retryBody, err := sjson.DeleteBytes(body, "max_output_tokens")
|
||||
if err != nil {
|
||||
return nil, "", false, fmt.Errorf("delete rejected max_output_tokens: %w", err)
|
||||
}
|
||||
return retryBody, "max_output_tokens parameter rejection", true, nil
|
||||
}
|
||||
return nil, "", false, nil
|
||||
}
|
||||
|
||||
func isExplicitOpenAIResponsesFieldRejection(code, message string) bool {
|
||||
switch strings.TrimSpace(code) {
|
||||
case "unknown_parameter", "unsupported_parameter":
|
||||
return true
|
||||
}
|
||||
return strings.Contains(message, "unknown parameter") ||
|
||||
strings.Contains(message, "unsupported parameter")
|
||||
}
|
||||
|
||||
func openAIResponsesRejectedParamFromMessage(message string) string {
|
||||
match := openAIResponsesRejectedMessageParamPattern.FindStringSubmatch(strings.TrimSpace(message))
|
||||
if len(match) != 2 {
|
||||
return ""
|
||||
}
|
||||
return strings.ToLower(strings.TrimSpace(match[1]))
|
||||
}
|
||||
|
||||
func openAIResponsesRejectedNamespaceIndex(param string) (int, bool) {
|
||||
match := openAIResponsesRejectedNamespaceParamPattern.FindStringSubmatch(strings.TrimSpace(param))
|
||||
if len(match) != 2 {
|
||||
return 0, false
|
||||
}
|
||||
index, err := strconv.Atoi(match[1])
|
||||
if err == nil && index >= 0 {
|
||||
return index, true
|
||||
}
|
||||
return 0, false
|
||||
}
|
||||
|
||||
func removeOpenAIResponsesRejectedNamespaceAtIndex(body []byte, index int) ([]byte, string, bool, error) {
|
||||
itemPath := fmt.Sprintf("input.%d", index)
|
||||
itemType := strings.ToLower(strings.TrimSpace(gjson.GetBytes(body, itemPath+".type").String()))
|
||||
switch itemType {
|
||||
case "function_call", "tool_call", "custom_tool_call", "mcp_tool_call":
|
||||
default:
|
||||
return nil, "", false, nil
|
||||
}
|
||||
|
||||
namespacePath := itemPath + ".namespace"
|
||||
if !gjson.GetBytes(body, namespacePath).Exists() {
|
||||
return nil, "", false, nil
|
||||
}
|
||||
retryBody, err := sjson.DeleteBytes(body, namespacePath)
|
||||
if err != nil {
|
||||
return nil, "", false, fmt.Errorf("delete rejected namespace at input[%d]: %w", index, err)
|
||||
}
|
||||
return retryBody, "indexed namespace parameter rejection", true, nil
|
||||
}
|
||||
@@ -0,0 +1,230 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"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 TestOpenAIResponsesRejectedFieldRetryStateRejectsDuplicateBodyAndCap(t *testing.T) {
|
||||
initialBody := []byte(`{"model":"gpt-5.5"}`)
|
||||
state := newOpenAIResponsesRejectedFieldRetryState(initialBody)
|
||||
|
||||
require.False(t, state.Allow(initialBody))
|
||||
for attempt := 0; attempt < maxOpenAIResponsesRejectedFieldRetries; attempt++ {
|
||||
nextBody := []byte(fmt.Sprintf(`{"model":"gpt-5.5","variant":%d}`, attempt))
|
||||
require.True(t, state.Allow(nextBody))
|
||||
require.False(t, state.Allow(nextBody))
|
||||
}
|
||||
require.False(t, state.Allow([]byte(`{"model":"gpt-5.5","variant":"overflow"}`)))
|
||||
}
|
||||
|
||||
func TestNormalizeOpenAIResponsesRejectedFieldRetryBodyRejectsAmbiguousErrors(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
body []byte
|
||||
responseBody []byte
|
||||
}{
|
||||
{
|
||||
name: "namespace belongs to message",
|
||||
body: []byte(`{"input":[{"type":"message","namespace":"keep"}]}`),
|
||||
responseBody: []byte(`{"error":{"code":"unknown_parameter","message":"Unknown parameter: 'input[0].namespace'.","param":"input[0].namespace"}}`),
|
||||
},
|
||||
{
|
||||
name: "max output tokens only mentioned",
|
||||
body: []byte(`{"max_output_tokens":4096}`),
|
||||
responseBody: []byte(`{"error":{"code":"invalid_request_error","message":"max_output_tokens must be positive","param":"max_output_tokens"}}`),
|
||||
},
|
||||
{
|
||||
name: "structured param overrides namespace mention",
|
||||
body: []byte(`{"input":[{"type":"function_call","namespace":"keep","arguments":"{}"}]}`),
|
||||
responseBody: []byte(`{"error":{"code":"unknown_parameter","message":"Unknown parameter: 'input[0].namespace'.","param":"tools"}}`),
|
||||
},
|
||||
{
|
||||
name: "nested max output tokens param is not top level",
|
||||
body: []byte(`{"max_output_tokens":4096,"input":[{"type":"message","content":{"max_output_tokens":"keep"}}]}`),
|
||||
responseBody: []byte(`{"error":{"code":"unknown_parameter","message":"Unknown parameter: input[0].content.max_output_tokens","param":"input[0].content.max_output_tokens"}}`),
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
retryBody, _, changed, err := normalizeOpenAIResponsesRejectedFieldRetryBody(http.StatusBadRequest, tt.body, tt.responseBody)
|
||||
require.NoError(t, err)
|
||||
require.False(t, changed)
|
||||
require.Nil(t, retryBody)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeOpenAIResponsesRejectedFieldRetryBodyFindsNamespacePathInMessage(t *testing.T) {
|
||||
body := []byte(`{"input":[{"type":"function_call","namespace":"keep","arguments":"{}"},{"type":"function_call","namespace":"remove","arguments":"{}"}]}`)
|
||||
responseBody := []byte(`{"error":{"code":"unknown_parameter","message":"input[0] was accepted; Unknown parameter: 'input[1].namespace'."}}`)
|
||||
|
||||
retryBody, _, changed, err := normalizeOpenAIResponsesRejectedFieldRetryBody(http.StatusBadRequest, body, responseBody)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.True(t, changed)
|
||||
require.Equal(t, "keep", gjson.GetBytes(retryBody, "input.0.namespace").String())
|
||||
require.False(t, gjson.GetBytes(retryBody, "input.1.namespace").Exists())
|
||||
}
|
||||
|
||||
func TestNormalizeOpenAIResponsesRejectedFieldRetryBodyBindsNamespacePathToRejectionPhrase(t *testing.T) {
|
||||
body := []byte(`{"input":[{"type":"function_call","namespace":"keep","arguments":"{}"},{"type":"function_call","namespace":"remove","arguments":"{}"}]}`)
|
||||
responseBody := []byte(`{"error":{"code":"unknown_parameter","message":"input[0].namespace is supported; Unknown parameter: input[1].namespace."}}`)
|
||||
|
||||
retryBody, _, changed, err := normalizeOpenAIResponsesRejectedFieldRetryBody(http.StatusBadRequest, body, responseBody)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.True(t, changed)
|
||||
require.Equal(t, "keep", gjson.GetBytes(retryBody, "input.0.namespace").String())
|
||||
require.False(t, gjson.GetBytes(retryBody, "input.1.namespace").Exists())
|
||||
}
|
||||
|
||||
func TestNormalizeOpenAIResponsesRejectedFieldRetryBodyDoesNotTreatMaxOutputTokensSuggestionAsRejection(t *testing.T) {
|
||||
body := []byte(`{"max_tokens":4096,"max_output_tokens":2048}`)
|
||||
responseBody := []byte(`{"error":{"code":"unknown_parameter","message":"Unknown parameter: max_tokens. Use max_output_tokens instead."}}`)
|
||||
|
||||
retryBody, _, changed, err := normalizeOpenAIResponsesRejectedFieldRetryBody(http.StatusBadRequest, body, responseBody)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.False(t, changed)
|
||||
require.Nil(t, retryBody)
|
||||
}
|
||||
|
||||
func TestNormalizeOpenAIResponsesRejectedFieldRetryBodyBindsMaxOutputTokensToRejectionPhrase(t *testing.T) {
|
||||
body := []byte(`{"max_output_tokens":2048}`)
|
||||
responseBody := []byte(`{"error":{"code":"unsupported_parameter","message":"Unsupported parameter: max_output_tokens."}}`)
|
||||
|
||||
retryBody, _, changed, err := normalizeOpenAIResponsesRejectedFieldRetryBody(http.StatusBadRequest, body, responseBody)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.True(t, changed)
|
||||
require.False(t, gjson.GetBytes(retryBody, "max_output_tokens").Exists())
|
||||
}
|
||||
|
||||
func TestOpenAIGatewayService_RetriesRejectedIndexedNamespaceField(t *testing.T) {
|
||||
body := []byte(`{"model":"gpt-5.5","stream":false,"input":[{"type":"function_call","name":"first","namespace":"keep","arguments":"{}"},{"type":"custom_tool_call","name":"second","namespace":"remove","input":"{}"}]}`)
|
||||
upstream := &httpUpstreamRecorder{responses: []*http.Response{
|
||||
newOpenAIRejectedFieldTestResponse(http.StatusBadRequest, `{"error":{"code":"unknown_parameter","message":"Unknown parameter: 'input[1].namespace'.","param":"input[1].namespace","type":"invalid_request_error"}}`),
|
||||
newOpenAIRejectedFieldTestResponse(http.StatusOK, `{"output":[],"usage":{"input_tokens":1,"output_tokens":1,"input_tokens_details":{"cached_tokens":0}}}`),
|
||||
}}
|
||||
|
||||
result, err := newOpenAIRejectedFieldTestService(upstream).Forward(
|
||||
context.Background(),
|
||||
newOpenAIRejectedFieldTestContext(body),
|
||||
newOpenAIRejectedFieldTestAccount(),
|
||||
body,
|
||||
)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, result)
|
||||
require.Len(t, upstream.bodies, 2)
|
||||
require.Equal(t, "keep", gjson.GetBytes(upstream.bodies[1], "input.0.namespace").String())
|
||||
require.False(t, gjson.GetBytes(upstream.bodies[1], "input.1.namespace").Exists())
|
||||
}
|
||||
|
||||
func TestOpenAIGatewayService_RetriesExplicitMaxOutputTokensRejection(t *testing.T) {
|
||||
body := []byte(`{"model":"gpt-5.5","stream":false,"max_output_tokens":4096,"input":[{"type":"message","role":"user","content":{"max_output_tokens":"keep"}}]}`)
|
||||
upstream := &httpUpstreamRecorder{responses: []*http.Response{
|
||||
newOpenAIRejectedFieldTestResponse(http.StatusBadRequest, `{"error":{"code":"unsupported_parameter","message":"Unsupported parameter: max_output_tokens","param":"max_output_tokens","type":"invalid_request_error"}}`),
|
||||
newOpenAIRejectedFieldTestResponse(http.StatusOK, `{"output":[],"usage":{"input_tokens":1,"output_tokens":1,"input_tokens_details":{"cached_tokens":0}}}`),
|
||||
}}
|
||||
|
||||
result, err := newOpenAIRejectedFieldTestService(upstream).Forward(
|
||||
context.Background(),
|
||||
newOpenAIRejectedFieldTestContext(body),
|
||||
newOpenAIRejectedFieldTestAccount(),
|
||||
body,
|
||||
)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, result)
|
||||
require.Len(t, upstream.bodies, 2)
|
||||
require.Equal(t, int64(4096), gjson.GetBytes(upstream.bodies[0], "max_output_tokens").Int())
|
||||
require.False(t, gjson.GetBytes(upstream.bodies[1], "max_output_tokens").Exists())
|
||||
require.Equal(t, "keep", gjson.GetBytes(upstream.bodies[1], "input.0.content.max_output_tokens").String())
|
||||
}
|
||||
|
||||
func TestOpenAIGatewayService_ComposesDistinctRejectedFieldRetries(t *testing.T) {
|
||||
body := []byte(`{"model":"gpt-5.5","stream":false,"max_output_tokens":2048,"input":[{"type":"function_call","name":"first","namespace":"keep","arguments":"{}"},{"type":"custom_tool_call","name":"second","namespace":"remove","input":"{}"}]}`)
|
||||
upstream := &httpUpstreamRecorder{responses: []*http.Response{
|
||||
newOpenAIRejectedFieldTestResponse(http.StatusBadRequest, `{"error":{"code":"unknown_parameter","message":"Unknown parameter: 'input[1].namespace'.","param":"input[1].namespace"}}`),
|
||||
newOpenAIRejectedFieldTestResponse(http.StatusBadRequest, `{"error":{"code":"unsupported_parameter","message":"Unsupported parameter: max_output_tokens","param":"max_output_tokens"}}`),
|
||||
newOpenAIRejectedFieldTestResponse(http.StatusOK, `{"output":[],"usage":{"input_tokens":1,"output_tokens":1,"input_tokens_details":{"cached_tokens":0}}}`),
|
||||
}}
|
||||
|
||||
result, err := newOpenAIRejectedFieldTestService(upstream).Forward(
|
||||
context.Background(),
|
||||
newOpenAIRejectedFieldTestContext(body),
|
||||
newOpenAIRejectedFieldTestAccount(),
|
||||
body,
|
||||
)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, result)
|
||||
require.Len(t, upstream.bodies, 3)
|
||||
require.True(t, gjson.GetBytes(upstream.bodies[0], "input.1.namespace").Exists())
|
||||
require.False(t, gjson.GetBytes(upstream.bodies[1], "input.1.namespace").Exists())
|
||||
require.Equal(t, int64(2048), gjson.GetBytes(upstream.bodies[1], "max_output_tokens").Int())
|
||||
require.False(t, gjson.GetBytes(upstream.bodies[2], "input.1.namespace").Exists())
|
||||
require.False(t, gjson.GetBytes(upstream.bodies[2], "max_output_tokens").Exists())
|
||||
}
|
||||
|
||||
func newOpenAIRejectedFieldTestService(upstream *httpUpstreamRecorder) *OpenAIGatewayService {
|
||||
return &OpenAIGatewayService{
|
||||
cfg: &config.Config{Security: config.SecurityConfig{
|
||||
URLAllowlist: config.URLAllowlistConfig{Enabled: false},
|
||||
}},
|
||||
httpUpstream: upstream,
|
||||
}
|
||||
}
|
||||
|
||||
func newOpenAIRejectedFieldTestContext(body []byte) *gin.Context {
|
||||
gin.SetMode(gin.TestMode)
|
||||
recorder := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(recorder)
|
||||
c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", bytes.NewReader(body))
|
||||
c.Request.Header.Set("Content-Type", "application/json")
|
||||
c.Request.Header.Set("User-Agent", "curl/8.0")
|
||||
return c
|
||||
}
|
||||
|
||||
func newOpenAIRejectedFieldTestAccount() *Account {
|
||||
return &Account{
|
||||
ID: 5107,
|
||||
Name: "responses-compatible",
|
||||
Platform: PlatformOpenAI,
|
||||
Type: AccountTypeAPIKey,
|
||||
Concurrency: 1,
|
||||
Credentials: map[string]any{
|
||||
"api_key": "sk-test",
|
||||
"base_url": "https://compat.example",
|
||||
},
|
||||
Extra: map[string]any{
|
||||
openai_compat.ExtraKeyResponsesMode: string(openai_compat.ResponsesSupportModeAuto),
|
||||
openai_compat.ExtraKeyResponsesSupported: true,
|
||||
},
|
||||
Status: StatusActive,
|
||||
Schedulable: true,
|
||||
}
|
||||
}
|
||||
|
||||
func newOpenAIRejectedFieldTestResponse(status int, body string) *http.Response {
|
||||
return &http.Response{
|
||||
StatusCode: status,
|
||||
Header: http.Header{"Content-Type": []string{"application/json"}},
|
||||
Body: io.NopCloser(strings.NewReader(body)),
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user