From 579f5e30fbd63b36f220dea350d3b0bb1625f38f Mon Sep 17 00:00:00 2001 From: sususu Date: Thu, 6 Aug 2026 20:18:49 +0800 Subject: [PATCH] fix(auth): rotate credentials for unknown upstream failures Stop treating an upstream 500 carrying "status":"UNKNOWN" as a request fault. It is an internal upstream failure, so the request now falls through to the next credential instead of failing immediately, and the resulting cooldown lands on the failing credential and model pair only, leaving sibling models on that credential selectable. Move the store=false item-miss detection into the shared client-error package. The upstream sends that 404 as plain text rather than a JSON error body, so the structured identifiers could never match it and only the conductor recognized it. The proxy now reports it to the client, which is the only party able to rebuild the request without the stale item reference; a reconnect resends the full input and the conversation continues unchanged. --- internal/clienterror/client_error.go | 18 +++ internal/clienterror/client_error_test.go | 16 +++ .../openai/openai_responses_websocket_test.go | 127 +++++++++++++++++- sdk/cliproxy/auth/conductor_cooldown.go | 27 +--- sdk/cliproxy/auth/conductor_overrides_test.go | 81 +++++++++++ 5 files changed, 240 insertions(+), 29 deletions(-) diff --git a/internal/clienterror/client_error.go b/internal/clienterror/client_error.go index 81ce6b1d..ac575e9c 100644 --- a/internal/clienterror/client_error.go +++ b/internal/clienterror/client_error.go @@ -44,6 +44,9 @@ func IsRequestFault(status int, err error) bool { if hasRequestFaultBody(err) { return true } + if err != nil && IsItemNotPersisted(err.Error()) { + return true + } switch status { case http.StatusBadRequest, http.StatusConflict, @@ -55,6 +58,21 @@ func IsRequestFault(status int, err error) bool { } } +// IsItemNotPersisted matches the upstream 404 raised when a request references a +// response item the upstream never stored because `store` was false. The upstream +// sends this as a plain-text message rather than a JSON body, so it cannot be +// recognized through the structured identifiers above. +// +// The request can only succeed once the client rebuilds it without the stale +// reference, so it is a request fault: rotating credentials cannot help, and the +// client must be told rather than left to retry the same broken input. +func IsItemNotPersisted(message string) bool { + lower := strings.ToLower(message) + return strings.Contains(lower, "item with id") && + strings.Contains(lower, "not found") && + strings.Contains(lower, "items are not persisted when `store` is set to false") +} + func hasRequestFaultBody(err error) bool { if err == nil { return false diff --git a/internal/clienterror/client_error_test.go b/internal/clienterror/client_error_test.go index 085497f7..2582b6f9 100644 --- a/internal/clienterror/client_error_test.go +++ b/internal/clienterror/client_error_test.go @@ -83,6 +83,22 @@ func TestIsRequestFault(t *testing.T) { err: statusError{status: http.StatusConflict, body: "conflict"}, want: true, }, + { + // Verbatim upstream text: plain text, not JSON, so it can only be matched + // by message. + name: "item not persisted with store=false", + status: http.StatusNotFound, + err: errors.New("Item with id 'rs_0b5f3eb6f51f175c0169ca74e4a85881998539920821603a74' not found. Items are not persisted when `store` is set to false. Try again with `store` set to true, or remove this item from your input."), + want: true, + }, + { + // An upstream internal error is not a request fault: it must stay eligible + // for credential rotation and (credential, model) cooldown. + name: "upstream unknown internal error", + status: http.StatusInternalServerError, + err: errors.New(`{"error":{"code":500,"message":"Internal error encountered.","status":"UNKNOWN"}}`), + }, + {name: "plain not found", status: http.StatusNotFound, err: errors.New("model not found")}, {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")}, diff --git a/sdk/api/handlers/openai/openai_responses_websocket_test.go b/sdk/api/handlers/openai/openai_responses_websocket_test.go index 9c7fe991..47f2f4a3 100644 --- a/sdk/api/handlers/openai/openai_responses_websocket_test.go +++ b/sdk/api/handlers/openai/openai_responses_websocket_test.go @@ -686,6 +686,8 @@ type websocketCanonicalRollbackExecutor struct { mu sync.Mutex payloads [][]byte calls int + // failErr overrides the default second-call failure when set. + failErr error } type websocketPinnedFailoverStatusError struct { @@ -857,14 +859,18 @@ func (e *websocketCanonicalRollbackExecutor) ExecuteStream(_ context.Context, _ e.calls++ call := e.calls e.payloads = append(e.payloads, bytes.Clone(req.Payload)) + failErr := e.failErr e.mu.Unlock() chunks := make(chan coreexecutor.StreamChunk, 1) if call == 2 { - chunks <- coreexecutor.StreamChunk{Err: websocketPinnedFailoverStatusError{ - status: http.StatusBadRequest, - msg: `{"error":{"message":"bad turn","type":"invalid_request_error","code":"invalid_request"}}`, - }} + if failErr == nil { + failErr = websocketPinnedFailoverStatusError{ + status: http.StatusBadRequest, + msg: `{"error":{"message":"bad turn","type":"invalid_request_error","code":"invalid_request"}}`, + } + } + chunks <- coreexecutor.StreamChunk{Err: failErr} close(chunks) return &coreexecutor.StreamResult{Chunks: chunks}, nil } @@ -3632,6 +3638,119 @@ func TestResponsesWebsocketClosesAfterNonRetryableClientError(t *testing.T) { } } +// itemNotPersistedUpstreamMessage is the verbatim upstream 404 text raised when a +// turn references a response item the upstream never stored because `store` was +// false. It arrives as plain text, not as a JSON error body. +const itemNotPersistedUpstreamMessage = "Item with id 'rs_0b5f3eb6f51f175c0169ca74e4a85881998539920821603a74' not found. Items are not persisted when `store` is set to false. Try again with `store` set to true, or remove this item from your input." + +// TestResponsesWebsocketExposesItemNotPersistedAndRecoversOnReconnect pins the +// store=false item miss end to end. The client must be told (it has to drop the +// stale reference; retrying the same input can never succeed), and the +// conversation must survive: after reconnecting with the full input the turn +// succeeds, and no stale per-socket transcript leaks into the new connection. +func TestResponsesWebsocketExposesItemNotPersistedAndRecoversOnReconnect(t *testing.T) { + gin.SetMode(gin.TestMode) + + modelName := "xai-item-miss-model" + executor := &websocketCanonicalRollbackExecutor{ + failErr: websocketPinnedFailoverStatusError{ + status: http.StatusNotFound, + msg: itemNotPersistedUpstreamMessage, + }, + } + manager := coreauth.NewManager(nil, nil, nil) + manager.RegisterExecutor(executor) + auth := &coreauth.Auth{ID: "auth-xai-item-miss", Provider: "xai", Status: coreauth.StatusActive} + if _, err := manager.Register(context.Background(), auth); err != nil { + t.Fatalf("Register auth: %v", err) + } + registry.GetGlobalRegistry().RegisterClient(auth.ID, auth.Provider, []*registry.ModelInfo{{ID: modelName}}) + t.Cleanup(func() { registry.GetGlobalRegistry().UnregisterClient(auth.ID) }) + + base := handlers.NewBaseAPIHandlers(&sdkconfig.SDKConfig{}, manager) + h := NewOpenAIResponsesAPIHandler(base) + router := gin.New() + router.GET("/v1/responses/ws", h.ResponsesWebsocket) + server := httptest.NewServer(router) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/v1/responses/ws" + sessionHeader := http.Header{"Session-Id": []string{"item-miss-session"}} + + conn, _, err := websocket.DefaultDialer.Dial(wsURL, sessionHeader) + if err != nil { + t.Fatalf("dial websocket: %v", err) + } + defer func() { _ = conn.Close() }() + + firstRequest := fmt.Sprintf(`{"type":"response.create","model":%q,"input":[{"type":"message","id":"msg-1"}]}`, modelName) + if errWrite := conn.WriteMessage(websocket.TextMessage, []byte(firstRequest)); errWrite != nil { + t.Fatalf("write first request: %v", errWrite) + } + if _, firstResponse, errRead := conn.ReadMessage(); errRead != nil || + gjson.GetBytes(firstResponse, "type").String() != wsEventTypeCompleted { + t.Fatalf("first response = %s, err=%v", firstResponse, errRead) + } + + // The turn references a reasoning item the upstream no longer holds. + staleRequest := `{"type":"response.create","previous_response_id":"resp-1","input":[{"type":"reasoning","id":"rs_0b5f3eb6f51f175c0169ca74e4a85881998539920821603a74"},{"type":"message","id":"msg-2"}]}` + if errWrite := conn.WriteMessage(websocket.TextMessage, []byte(staleRequest)); errWrite != nil { + t.Fatalf("write stale request: %v", errWrite) + } + _, errorResponse, errRead := conn.ReadMessage() + if errRead != nil { + t.Fatalf("item miss was hidden from the client: %v", errRead) + } + if got := gjson.GetBytes(errorResponse, "type").String(); got != wsEventTypeError { + t.Fatalf("response type = %q, want %q: %s", got, wsEventTypeError, errorResponse) + } + if got := int(gjson.GetBytes(errorResponse, "status").Int()); got != http.StatusNotFound { + t.Fatalf("status = %d, want %d: %s", got, http.StatusNotFound, errorResponse) + } + if msg := gjson.GetBytes(errorResponse, "error.message").String(); !strings.Contains(msg, "Items are not persisted") { + t.Fatalf("error.message lost the upstream reason: %q", msg) + } + if _, extra, errRead := conn.ReadMessage(); errRead == nil { + t.Fatalf("received frame after terminal error: %s", extra) + } + + // The client rebuilds the conversation on a new socket with the full input. + reconn, _, errDial := websocket.DefaultDialer.Dial(wsURL, sessionHeader) + if errDial != nil { + t.Fatalf("reconnect websocket: %v", errDial) + } + defer func() { _ = reconn.Close() }() + + fullRequest := fmt.Sprintf(`{"type":"response.create","model":%q,"input":[{"type":"message","id":"msg-1"},{"type":"message","id":"msg-2"}]}`, modelName) + if errWrite := reconn.WriteMessage(websocket.TextMessage, []byte(fullRequest)); errWrite != nil { + t.Fatalf("write rebuilt request: %v", errWrite) + } + _, recovered, errRead := reconn.ReadMessage() + if errRead != nil { + t.Fatalf("read rebuilt response: %v", errRead) + } + if got := gjson.GetBytes(recovered, "type").String(); got != wsEventTypeCompleted { + t.Fatalf("rebuilt response type = %q, want %q: %s", got, wsEventTypeCompleted, recovered) + } + + payloads := executor.Payloads() + if len(payloads) != 3 { + t.Fatalf("upstream payload count = %d, want 3", len(payloads)) + } + // The rebuilt turn must carry the full input and none of the failed turn's state. + rebuilt := payloads[2] + if got := gjson.GetBytes(rebuilt, "previous_response_id").String(); got != "" { + t.Fatalf("rebuilt upstream request still pinned previous_response_id=%q: %s", got, rebuilt) + } + inputIDs := gjson.GetBytes(rebuilt, "input.#.id").Array() + if len(inputIDs) != 2 || inputIDs[0].String() != "msg-1" || inputIDs[1].String() != "msg-2" { + t.Fatalf("rebuilt upstream input lost context: %s", rebuilt) + } + if strings.Contains(string(rebuilt), "rs_0b5f3eb6f51f175c0169ca74e4a85881998539920821603a74") { + t.Fatalf("rebuilt upstream request replayed the stale item: %s", rebuilt) + } +} + func TestResponsesWebsocketSwitchesPinnedAuthAcrossProviders(t *testing.T) { for _, testCase := range []struct { name string diff --git a/sdk/cliproxy/auth/conductor_cooldown.go b/sdk/cliproxy/auth/conductor_cooldown.go index cb1dbe66..abd5673d 100644 --- a/sdk/cliproxy/auth/conductor_cooldown.go +++ b/sdk/cliproxy/auth/conductor_cooldown.go @@ -1319,21 +1319,11 @@ func nextCloudflareCooldown(backoffLevel int, disableCooling bool, now time.Time return next, backoffLevel } -func isRequestScopedNotFoundMessage(message string) bool { - if message == "" { - return false - } - lower := strings.ToLower(message) - return strings.Contains(lower, "item with id") && - strings.Contains(lower, "not found") && - strings.Contains(lower, "items are not persisted when `store` is set to false") -} - func isRequestScopedNotFoundResultError(err *Error) bool { if err == nil || statusCodeFromResult(err) != http.StatusNotFound { return false } - return isRequestScopedNotFoundMessage(err.Message) + return clienterror.IsItemNotPersisted(err.Message) } func isRequestScopedResultError(err *Error) bool { @@ -1552,20 +1542,7 @@ func isRequestInvalidError(err error) bool { if isModelSupportError(err) { return false } - status := statusCodeFromError(err) - if clienterror.IsRequestFault(status, err) { - return true - } - switch status { - case http.StatusNotFound: - return isRequestScopedNotFoundMessage(err.Error()) - case http.StatusInternalServerError: - msg := err.Error() - return strings.Contains(msg, "\"status\":\"UNKNOWN\"") || - strings.Contains(msg, "\"status\": \"UNKNOWN\"") - default: - return false - } + return clienterror.IsRequestFault(statusCodeFromError(err), err) } func applyAuthFailureState(auth *Auth, resultErr *Error, retryAfter *time.Duration, now time.Time, disableCooling bool) { diff --git a/sdk/cliproxy/auth/conductor_overrides_test.go b/sdk/cliproxy/auth/conductor_overrides_test.go index 23e17fc6..de35ebc9 100644 --- a/sdk/cliproxy/auth/conductor_overrides_test.go +++ b/sdk/cliproxy/auth/conductor_overrides_test.go @@ -1220,6 +1220,11 @@ func TestManager_RequestScopedErrorStopsCredentialFallbackWithoutSuspendingAuth( HTTPStatus: http.StatusBadGateway, Message: `{"body":{"error":{"type":"invalid_request","message":"invalid input"}}}`, } + // Upstream sends this one as plain text rather than a JSON error body. + itemNotPersistedErr := &Error{ + HTTPStatus: http.StatusNotFound, + Message: requestScopedNotFoundMessage, + } tests := []struct { name string provider string @@ -1247,6 +1252,9 @@ func TestManager_RequestScopedErrorStopsCredentialFallbackWithoutSuspendingAuth( {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}, + {name: "non-streaming item not persisted", err: itemNotPersistedErr, wantStatus: http.StatusNotFound}, + {name: "streaming item not persisted", stream: true, err: itemNotPersistedErr, wantStatus: http.StatusNotFound}, + {name: "streaming item not persisted after payload", stream: true, streamAfterPayload: true, err: itemNotPersistedErr, wantStatus: http.StatusNotFound}, } for _, tc := range tests { @@ -1345,6 +1353,79 @@ func TestManager_RequestScopedErrorStopsCredentialFallbackWithoutSuspendingAuth( } } +// TestManager_UnknownUpstreamErrorRotatesAndPenalizesModelOnly pins the upstream +// 500 "status":"UNKNOWN" contract. It is an upstream internal failure, not a +// request fault, so the request must fall through to the next credential. The +// cooldown that follows must land on the (credential, model) pair only: sibling +// models on the same credential stay selectable. +func TestManager_UnknownUpstreamErrorRotatesAndPenalizesModelOnly(t *testing.T) { + m := NewManager(nil, nil, nil) + m.SetRetryConfig(3, 30*time.Second, 0) + + const provider = "gemini" + const model = "gemini-3.6-pro" + const siblingModel = "gemini-3.6-flash" + + executor := &authFallbackExecutor{id: provider} + executor.executeErrors = map[string]error{ + "aa-bad-auth": &Error{ + HTTPStatus: http.StatusInternalServerError, + Message: `{"error":{"code":500,"message":"Internal error encountered.","status":"UNKNOWN"}}`, + }, + } + m.RegisterExecutor(executor) + + badAuth := &Auth{ID: "aa-bad-auth", Provider: provider} + goodAuth := &Auth{ID: "bb-good-auth", Provider: provider} + + reg := registry.GetGlobalRegistry() + models := []*registry.ModelInfo{{ID: model}, {ID: siblingModel}} + reg.RegisterClient(badAuth.ID, provider, models) + reg.RegisterClient(goodAuth.ID, provider, models) + t.Cleanup(func() { + reg.UnregisterClient(badAuth.ID) + reg.UnregisterClient(goodAuth.ID) + }) + + if _, errRegister := m.Register(context.Background(), badAuth); errRegister != nil { + t.Fatalf("register bad auth: %v", errRegister) + } + if _, errRegister := m.Register(context.Background(), goodAuth); errRegister != nil { + t.Fatalf("register good auth: %v", errRegister) + } + + resp, errExecute := m.Execute(context.Background(), []string{provider}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{}) + if errExecute != nil { + t.Fatalf("expected fallback to the next credential, got error: %v", errExecute) + } + if got := string(resp.Payload); got != goodAuth.ID { + t.Fatalf("served by %q, want %q", got, goodAuth.ID) + } + if calls := executor.ExecuteCalls(); len(calls) != 2 || calls[0] != badAuth.ID || calls[1] != goodAuth.ID { + t.Fatalf("credential calls = %v, want [%s %s]", calls, badAuth.ID, goodAuth.ID) + } + + updatedBad, ok := m.GetByID(badAuth.ID) + if !ok || updatedBad == nil { + t.Fatal("expected bad auth to remain registered") + } + state := updatedBad.ModelStates[model] + if state == nil { + t.Fatal("expected the failing (credential, model) pair to be penalized") + } + if state.NextRetryAfter.IsZero() { + t.Fatal("expected a cooldown on the failing (credential, model) pair") + } + + now := time.Now() + if blocked, _, _ := isAuthBlockedForModel(updatedBad, model, now); !blocked { + t.Fatal("expected the failing model to be blocked on that credential") + } + if blocked, reason, _ := isAuthBlockedForModel(updatedBad, siblingModel, now); blocked { + t.Fatalf("sibling model was blocked on the same credential (reason=%v); the penalty must stay scoped to (credential, model)", reason) + } +} + func TestManager_MarkResult_RequestScopedNotFoundDoesNotCooldownAuth(t *testing.T) { m := NewManager(nil, nil, nil) -- 2.51.2