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)