From c1d69e7b47788619feb79647b43dab26f6bf4d46 Mon Sep 17 00:00:00 2001 From: sususu Date: Thu, 6 Aug 2026 19:57:19 +0800 Subject: [PATCH] fix(auth): avoid penalizing credentials for client faults --- internal/clienterror/client_error.go | 79 ++++++++++++++ internal/clienterror/client_error_test.go | 100 ++++++++++++++++++ .../openai_responses_websocket_forward.go | 73 +------------ sdk/cliproxy/auth/conductor_cooldown.go | 62 ++--------- sdk/cliproxy/auth/conductor_overrides_test.go | 59 ++++++++--- 5 files changed, 235 insertions(+), 138 deletions(-) create mode 100644 internal/clienterror/client_error.go create mode 100644 internal/clienterror/client_error_test.go diff --git a/internal/clienterror/client_error.go b/internal/clienterror/client_error.go new file mode 100644 index 000000000..81ce6b1d0 --- /dev/null +++ b/internal/clienterror/client_error.go @@ -0,0 +1,79 @@ +// Package clienterror classifies upstream failures caused by the client request. +package clienterror + +import ( + "encoding/json" + "errors" + "net/http" + "strings" + + "github.com/tidwall/gjson" +) + +var requestFaultCodes = map[string]struct{}{ + "cyber_policy": {}, + "context_length_exceeded": {}, + "message_too_big": {}, + "string_above_max_length": {}, + "invalid_prompt": {}, + "invalid_value": {}, + "unsupported_value": {}, + "invalid_request_error": {}, + "previous_response_not_found": {}, +} + +var requestFaultTypes = map[string]struct{}{ + "invalid_request": {}, + "invalid_request_error": {}, + "bad_request_error": {}, + "invalid_prompt": {}, +} + +// IsRequestFault reports whether an upstream failure is caused by the request +// and therefore must not rotate or penalize credentials. +func IsRequestFault(status int, err error) bool { + if status <= 0 && err != nil { + type statusCoder interface { + StatusCode() int + } + var statusErr statusCoder + if errors.As(err, &statusErr) && statusErr != nil { + status = statusErr.StatusCode() + } + } + if hasRequestFaultBody(err) { + return true + } + switch status { + case http.StatusBadRequest, + http.StatusConflict, + http.StatusRequestEntityTooLarge, + http.StatusUnprocessableEntity: + return true + default: + return false + } +} + +func hasRequestFaultBody(err error) bool { + if err == nil { + return false + } + body := strings.TrimSpace(err.Error()) + if body == "" || !json.Valid([]byte(body)) { + return false + } + for _, path := range []string{"error.code", "code", "response.error.code", "body.error.code"} { + code := strings.ToLower(strings.TrimSpace(gjson.Get(body, path).String())) + if _, ok := requestFaultCodes[code]; ok { + return true + } + } + for _, path := range []string{"error.type", "type", "response.error.type", "body.error.type"} { + errType := strings.ToLower(strings.TrimSpace(gjson.Get(body, path).String())) + if _, ok := requestFaultTypes[errType]; ok { + return true + } + } + return false +} diff --git a/internal/clienterror/client_error_test.go b/internal/clienterror/client_error_test.go new file mode 100644 index 000000000..085497f7b --- /dev/null +++ b/internal/clienterror/client_error_test.go @@ -0,0 +1,100 @@ +package clienterror + +import ( + "errors" + "net/http" + "testing" +) + +type statusError struct { + status int + body string +} + +func (e statusError) Error() string { return e.body } +func (e statusError) StatusCode() int { return e.status } + +func TestIsRequestFaultStructuredIdentifiers(t *testing.T) { + for _, code := range []string{ + "cyber_policy", + "context_length_exceeded", + "message_too_big", + "string_above_max_length", + "invalid_prompt", + "invalid_value", + "unsupported_value", + "invalid_request_error", + "previous_response_not_found", + } { + t.Run("code/"+code, func(t *testing.T) { + err := errors.New(`{"error":{"code":"` + code + `"}}`) + if !IsRequestFault(http.StatusBadGateway, err) { + t.Fatalf("code %q was not classified as a request fault", code) + } + }) + } + + for _, errType := range []string{ + "invalid_request", + "invalid_request_error", + "bad_request_error", + "invalid_prompt", + } { + t.Run("type/"+errType, func(t *testing.T) { + err := errors.New(`{"error":{"type":"` + errType + `"}}`) + if !IsRequestFault(http.StatusBadGateway, err) { + t.Fatalf("type %q was not classified as a request fault", errType) + } + }) + } +} + +func TestIsRequestFault(t *testing.T) { + tests := []struct { + name string + status int + err error + want bool + }{ + {name: "bad request status", status: http.StatusBadRequest, err: errors.New("bad request"), want: true}, + {name: "conflict status", status: http.StatusConflict, err: errors.New("conflict"), want: true}, + {name: "entity too large status", status: http.StatusRequestEntityTooLarge, err: errors.New("too large"), want: true}, + {name: "unprocessable status", status: http.StatusUnprocessableEntity, err: errors.New("unprocessable"), want: true}, + { + name: "cyber policy behind bad gateway", + status: http.StatusBadGateway, + err: errors.New(`{"error":{"type":"invalid_request","code":"cyber_policy","message":"blocked"}}`), + want: true, + }, + { + name: "context length behind internal error", + status: http.StatusInternalServerError, + err: errors.New(`{"response":{"error":{"type":"server_error","code":"context_length_exceeded"}}}`), + want: true, + }, + { + name: "invalid request type behind bad gateway", + status: http.StatusBadGateway, + err: errors.New(`{"body":{"error":{"type":"invalid_request","message":"invalid"}}}`), + want: true, + }, + { + name: "status from error", + err: statusError{status: http.StatusConflict, body: "conflict"}, + want: true, + }, + {name: "unauthorized", status: http.StatusUnauthorized, err: errors.New("invalid token")}, + {name: "quota", status: http.StatusTooManyRequests, err: errors.New("quota")}, + {name: "transport", status: http.StatusBadGateway, err: errors.New("unexpected EOF")}, + {name: "invalid JSON body", status: http.StatusBadGateway, err: errors.New(`{"error":`)}, + {name: "nil", status: 0}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := IsRequestFault(tc.status, tc.err); got != tc.want { + t.Fatalf("IsRequestFault(%d, %v) = %t, want %t", tc.status, tc.err, got, tc.want) + } + }) + } +} diff --git a/sdk/api/handlers/openai/openai_responses_websocket_forward.go b/sdk/api/handlers/openai/openai_responses_websocket_forward.go index 3d7d31abf..775f57000 100644 --- a/sdk/api/handlers/openai/openai_responses_websocket_forward.go +++ b/sdk/api/handlers/openai/openai_responses_websocket_forward.go @@ -12,6 +12,7 @@ import ( "github.com/gin-gonic/gin" "github.com/gorilla/websocket" + "github.com/router-for-me/CLIProxyAPI/v7/internal/clienterror" "github.com/router-for-me/CLIProxyAPI/v7/internal/interfaces" "github.com/router-for-me/CLIProxyAPI/v7/sdk/api/handlers" log "github.com/sirupsen/logrus" @@ -196,65 +197,6 @@ func responsesWebsocketErrorStatus(errMsg *interfaces.ErrorMessage) int { return status } -// responsesClientFaultErrorCodes lists upstream error codes caused by the request -// payload itself. These must reach the client verbatim regardless of the HTTP -// status the upstream attached, because retrying or rotating credentials cannot -// change the outcome. -var responsesClientFaultErrorCodes = map[string]struct{}{ - "cyber_policy": {}, - "context_length_exceeded": {}, - "message_too_big": {}, - "string_above_max_length": {}, - "invalid_prompt": {}, - "invalid_value": {}, - "unsupported_value": {}, - "invalid_request_error": {}, - "previous_response_not_found": {}, -} - -// responsesClientFaultErrorTypes mirrors responsesClientFaultErrorCodes for -// upstreams that only classify the failure through `error.type`. -var responsesClientFaultErrorTypes = map[string]struct{}{ - "invalid_request": {}, - "invalid_request_error": {}, - "bad_request_error": {}, - "invalid_prompt": {}, -} - -// isResponsesClientFaultError reports whether the upstream error body identifies -// a request-shape failure. Upstreams are inconsistent about the status paired -// with these bodies: Codex reports `cyber_policy` as 400 on the stream error path -// but as 502 when the same rejection arrives through the websocket disconnect -// channel, so the body is authoritative here rather than the status. -func isResponsesClientFaultError(errMsg *interfaces.ErrorMessage) bool { - if errMsg == nil || errMsg.Error == nil { - return false - } - body := strings.TrimSpace(errMsg.Error.Error()) - if body == "" || !json.Valid([]byte(body)) { - return false - } - for _, path := range []string{"error.code", "code", "response.error.code"} { - code := strings.ToLower(strings.TrimSpace(gjson.Get(body, path).String())) - if code == "" { - continue - } - if _, ok := responsesClientFaultErrorCodes[code]; ok { - return true - } - } - for _, path := range []string{"error.type", "type", "response.error.type"} { - errType := strings.ToLower(strings.TrimSpace(gjson.Get(body, path).String())) - if errType == "" { - continue - } - if _, ok := responsesClientFaultErrorTypes[errType]; ok { - return true - } - } - return false -} - // shouldExposeResponsesUpstreamError reports whether a terminal upstream error // must reach the downstream client. // @@ -267,18 +209,7 @@ func shouldExposeResponsesUpstreamError(errMsg *interfaces.ErrorMessage) bool { if errMsg == nil { return false } - if isResponsesClientFaultError(errMsg) { - return true - } - switch responsesWebsocketErrorStatus(errMsg) { - case http.StatusBadRequest, - http.StatusConflict, - http.StatusRequestEntityTooLarge, - http.StatusUnprocessableEntity: - return true - default: - return false - } + return clienterror.IsRequestFault(responsesWebsocketErrorStatus(errMsg), errMsg.Error) } func writeResponsesWebsocketTerminalError( diff --git a/sdk/cliproxy/auth/conductor_cooldown.go b/sdk/cliproxy/auth/conductor_cooldown.go index 2820ef4a0..cb1dbe661 100644 --- a/sdk/cliproxy/auth/conductor_cooldown.go +++ b/sdk/cliproxy/auth/conductor_cooldown.go @@ -11,6 +11,7 @@ import ( "sync/atomic" "time" + "github.com/router-for-me/CLIProxyAPI/v7/internal/clienterror" internalconfig "github.com/router-for-me/CLIProxyAPI/v7/internal/config" "github.com/router-for-me/CLIProxyAPI/v7/internal/registry" "github.com/router-for-me/CLIProxyAPI/v7/internal/thinking" @@ -1532,53 +1533,14 @@ func isMissingModelPhrase(value string) bool { } } -func isCyberPolicyError(err error) bool { - if err == nil { - return false - } - var payload any - if errJSON := json.Unmarshal([]byte(strings.TrimSpace(err.Error())), &payload); errJSON != nil { - return false - } - return containsStructuredErrorCode(payload, "cyber_policy") -} - -func containsStructuredErrorCode(value any, target string) bool { - target = strings.ToLower(strings.TrimSpace(target)) - switch typed := value.(type) { - case map[string]any: - for key, item := range typed { - if strings.EqualFold(strings.TrimSpace(key), "code") { - if code, ok := item.(string); ok && strings.ToLower(strings.TrimSpace(code)) == target { - return true - } - } - if containsStructuredErrorCode(item, target) { - return true - } - } - case []any: - for _, item := range typed { - if containsStructuredErrorCode(item, target) { - return true - } - } - } - return false -} - // isRequestInvalidError returns true if the error represents a client request -// error that should not be retried. Specifically, it treats cyber policy -// rejections, 400 responses with "invalid_request_error", request-scoped 404 -// item misses caused by `store=false`, 413 payload/frame size rejections, and all -// 422 responses as request-shape failures, where switching auths or pooled -// upstream models will not help. Model-support errors are excluded so routing can -// fall through to another auth or upstream. +// error that should neither rotate nor penalize credentials. Model-support +// errors remain eligible for alternate routing and keep their model-level state. func isRequestInvalidError(err error) bool { if err == nil { return false } - if isRequestScopedError(err) || isCyberPolicyError(err) { + if isRequestScopedError(err) { return true } if isCloudflareChallengeError(err) { @@ -1591,22 +1553,12 @@ func isRequestInvalidError(err error) bool { return false } status := statusCodeFromError(err) + if clienterror.IsRequestFault(status, err) { + return true + } switch status { - case http.StatusBadRequest: - msg := err.Error() - return strings.Contains(msg, "invalid_request_error") || - strings.Contains(msg, "bad_request_error") || - strings.Contains(msg, "INVALID_ARGUMENT") || - strings.Contains(msg, "FAILED_PRECONDITION") case http.StatusNotFound: return isRequestScopedNotFoundMessage(err.Error()) - case http.StatusRequestEntityTooLarge: - // The request payload (or websocket frame) is too large for the upstream. - // Every other credential enforces the same limit, so retrying elsewhere only - // burns the pool and marks healthy credentials unavailable. - return true - case http.StatusUnprocessableEntity: - return true case http.StatusInternalServerError: msg := err.Error() return strings.Contains(msg, "\"status\":\"UNKNOWN\"") || diff --git a/sdk/cliproxy/auth/conductor_overrides_test.go b/sdk/cliproxy/auth/conductor_overrides_test.go index 8a0484342..23e17fc61 100644 --- a/sdk/cliproxy/auth/conductor_overrides_test.go +++ b/sdk/cliproxy/auth/conductor_overrides_test.go @@ -179,6 +179,7 @@ type authFallbackExecutor struct { streamCalls []string executeErrors map[string]error streamFirstErrors map[string]error + streamTailErrors map[string]error countTokenErrors map[string]error } @@ -200,16 +201,20 @@ func (e *authFallbackExecutor) Execute(_ context.Context, auth *Auth, _ cliproxy func (e *authFallbackExecutor) ExecuteStream(_ context.Context, auth *Auth, _ cliproxyexecutor.Request, _ cliproxyexecutor.Options) (*cliproxyexecutor.StreamResult, error) { e.mu.Lock() e.streamCalls = append(e.streamCalls, auth.ID) - err := e.streamFirstErrors[auth.ID] + firstErr := e.streamFirstErrors[auth.ID] + tailErr := e.streamTailErrors[auth.ID] e.mu.Unlock() - ch := make(chan cliproxyexecutor.StreamChunk, 1) - if err != nil { - ch <- cliproxyexecutor.StreamChunk{Err: err} + ch := make(chan cliproxyexecutor.StreamChunk, 2) + if firstErr != nil { + ch <- cliproxyexecutor.StreamChunk{Err: firstErr} close(ch) return &cliproxyexecutor.StreamResult{Headers: http.Header{"X-Auth": {auth.ID}}, Chunks: ch}, nil } ch <- cliproxyexecutor.StreamChunk{Payload: []byte(auth.ID)} + if tailErr != nil { + ch <- cliproxyexecutor.StreamChunk{Err: tailErr} + } close(ch) return &cliproxyexecutor.StreamResult{Headers: http.Header{"X-Auth": {auth.ID}}, Chunks: ch}, nil } @@ -1199,12 +1204,29 @@ func TestManager_RequestScopedErrorStopsCredentialFallbackWithoutSuspendingAuth( HTTPStatus: http.StatusRequestEntityTooLarge, Message: `{"error":{"code":"message_too_big","message":"upstream websocket message too big"}}`, } + plainBadRequestErr := &Error{ + HTTPStatus: http.StatusBadRequest, + Message: "bad request", + } + conflictErr := &Error{ + HTTPStatus: http.StatusConflict, + Message: `{"error":{"type":"conflict_error","code":"conflict","message":"request conflict"}}`, + } + contextLengthErr := &Error{ + HTTPStatus: http.StatusBadGateway, + Message: `{"error":{"type":"server_error","code":"context_length_exceeded","message":"input too long"}}`, + } + invalidRequestTypeErr := &Error{ + HTTPStatus: http.StatusBadGateway, + Message: `{"body":{"error":{"type":"invalid_request","message":"invalid input"}}}`, + } tests := []struct { - name string - provider string - stream bool - err error - wantStatus int + name string + provider string + stream bool + streamAfterPayload bool + err error + wantStatus int }{ {name: "non-streaming incomplete", err: incompleteErr, wantStatus: http.StatusRequestTimeout}, {name: "streaming incomplete", stream: true, err: incompleteErr, wantStatus: http.StatusRequestTimeout}, @@ -1217,6 +1239,14 @@ func TestManager_RequestScopedErrorStopsCredentialFallbackWithoutSuspendingAuth( {name: "streaming cyber policy", provider: "codex", stream: true, err: cyberPolicyErr, wantStatus: http.StatusBadGateway}, {name: "non-streaming message too big", provider: "codex", err: tooLargeErr, wantStatus: http.StatusRequestEntityTooLarge}, {name: "streaming message too big", provider: "codex", stream: true, err: tooLargeErr, wantStatus: http.StatusRequestEntityTooLarge}, + {name: "non-streaming plain bad request", err: plainBadRequestErr, wantStatus: http.StatusBadRequest}, + {name: "streaming plain bad request", stream: true, err: plainBadRequestErr, wantStatus: http.StatusBadRequest}, + {name: "non-streaming conflict", err: conflictErr, wantStatus: http.StatusConflict}, + {name: "streaming conflict", stream: true, err: conflictErr, wantStatus: http.StatusConflict}, + {name: "streaming conflict after payload", stream: true, streamAfterPayload: true, err: conflictErr, wantStatus: http.StatusConflict}, + {name: "non-streaming context length behind bad gateway", err: contextLengthErr, wantStatus: http.StatusBadGateway}, + {name: "streaming context length behind bad gateway", stream: true, err: contextLengthErr, wantStatus: http.StatusBadGateway}, + {name: "streaming invalid request type behind bad gateway", stream: true, err: invalidRequestTypeErr, wantStatus: http.StatusBadGateway}, } for _, tc := range tests { @@ -1229,7 +1259,9 @@ func TestManager_RequestScopedErrorStopsCredentialFallbackWithoutSuspendingAuth( m.SetRetryConfig(2, 30*time.Second, 0) executor := &authFallbackExecutor{id: provider} - if tc.stream { + if tc.streamAfterPayload { + executor.streamTailErrors = map[string]error{"aa-bad-auth": tc.err} + } else if tc.stream { executor.streamFirstErrors = map[string]error{"aa-bad-auth": tc.err} } else { executor.executeErrors = map[string]error{"aa-bad-auth": tc.err} @@ -1258,11 +1290,14 @@ func TestManager_RequestScopedErrorStopsCredentialFallbackWithoutSuspendingAuth( var errExecute error if tc.stream { result, errStream := m.ExecuteStream(context.Background(), []string{provider}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{Stream: true}) + errExecute = errStream if result != nil { - for range result.Chunks { + for chunk := range result.Chunks { + if chunk.Err != nil { + errExecute = chunk.Err + } } } - errExecute = errStream } else { _, errExecute = m.Execute(context.Background(), []string{provider}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{}) }