fix(auth): avoid penalizing credentials for client faults

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
+}

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)
+			}
+		})
+	}
+}

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(

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\"") ||

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{})
 			}