From fbb99067cc2fcef47b67eedf11edd5d4d53b9bca Mon Sep 17 00:00:00 2001 From: tbphp Date: Sun, 13 Sep 2026 18:06:37 +0800 Subject: [PATCH] =?UTF-8?q?fix(health):=20=E6=94=BE=E5=AE=BD=E4=B8=8A?= =?UTF-8?q?=E6=B8=B8=E9=94=99=E8=AF=AF=E9=87=8D=E8=AF=95=E5=B9=B6=E4=BF=9D?= =?UTF-8?q?=E7=95=99=E5=81=A5=E5=BA=B7=E8=BE=B9=E7=95=8C=20(#643)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(health): 放宽上游错误重试并保留健康边界 * fix(health): 补齐资源操作重放与上下文拒绝规则 * fix(execution): 保留明确请求拒绝的分类优先级 * fix(health): 补齐流内 HTTP 错误重试判断 --- .../bifrost/candidate_rejection_test.go | 10 +- internal/execution/bifrost/mapping.go | 11 +- internal/execution/bifrost/mapping_test.go | 6 +- .../execution/cpa/claude_provider_test.go | 11 + .../execution/cpa/codex_compatibility_test.go | 6 +- internal/execution/cpa/grok_provider.go | 2 - internal/execution/cpa/grok_provider_test.go | 5 +- internal/execution/cpa/provider.go | 9 +- internal/execution/cpa/provider_test.go | 25 +- internal/execution/request_rejection.go | 19 ++ internal/gateway/dialects_integration_test.go | 8 +- internal/gateway/error_retry_policy_test.go | 218 ++++++++++++++++++ internal/gateway/execution_boundary_test.go | 4 +- internal/gateway/handler_test.go | 9 +- internal/health/execution_judge.go | 53 ++++- internal/health/execution_judge_test.go | 12 +- internal/health/execution_judge_v2_test.go | 24 +- internal/health/retry_policy_test.go | 119 ++++++++++ 18 files changed, 489 insertions(+), 62 deletions(-) create mode 100644 internal/execution/request_rejection.go create mode 100644 internal/gateway/error_retry_policy_test.go create mode 100644 internal/health/retry_policy_test.go diff --git a/internal/execution/bifrost/candidate_rejection_test.go b/internal/execution/bifrost/candidate_rejection_test.go index f69ae647..b0eea70c 100644 --- a/internal/execution/bifrost/candidate_rejection_test.go +++ b/internal/execution/bifrost/candidate_rejection_test.go @@ -19,8 +19,14 @@ func TestCandidateCapabilityRejectionsAreNarrow(t *testing.T) { {name: "unsupported model legacy spelling", status: 400, body: `{"error":{"type":"invalid_request_error","code":"unsupported-model"}}`, want: execution.FailureHintModelUnavailable}, {name: "known function calling rejection", status: 400, body: `{"error":{"type":"invalid_request_error","message":"function calling is not supported with this model"}}`, want: execution.FailureHintModelUnavailable}, {name: "context limit", status: 400, body: `{"error":{"type":"invalid_request_error","code":"context_length_exceeded"}}`, want: execution.FailureHintRequestRejected}, - {name: "generic invalid request", status: 400, body: `{"error":{"type":"invalid_request_error","message":"invalid request"}}`, want: execution.FailureHintRequestRejected}, - {name: "quoted message is not rejection evidence", status: 400, body: `{"error":{"type":"invalid_request_error","message":"invalid input: function calling is not supported with this model"}}`, want: execution.FailureHintRequestRejected}, + {name: "Anthropic context limit", status: 400, body: `{"error":{"type":"invalid_request_error","message":"prompt is too long"}}`, want: execution.FailureHintRequestRejected}, + {name: "Anthropic context limit with counts", status: 400, body: `{"error":{"type":"invalid_request_error","message":"prompt is too long: 210000 tokens > 200000 maximum"}}`, want: execution.FailureHintRequestRejected}, + {name: "quoted context limit", status: 400, body: `{"error":{"type":"invalid_request_error","message":"provider diagnostic: prompt is too long"}}`}, + {name: "context near match", status: 400, body: `{"error":{"type":"invalid_request_error","message":"prompt is too long-lived"}}`}, + {name: "generic invalid request", status: 400, body: `{"error":{"type":"invalid_request_error","message":"invalid request"}}`}, + {name: "quoted message is not rejection evidence", status: 400, body: `{"error":{"type":"invalid_request_error","message":"invalid input: function calling is not supported with this model"}}`}, + {name: "quoted error code is not rejection evidence", status: 400, body: `{"error":{"message":"upstream returned invalid_parameter"}}`}, + {name: "content policy rejection", status: 400, body: `{"error":{"code":"content_policy_violation"}}`, want: execution.FailureHintRequestRejected}, {name: "explicit invalid parameter wins", status: 400, body: `{"error":{"type":"invalid_request_error","code":"invalid_parameter","message":"function calling is not supported with this model"}}`, want: execution.FailureHintRequestRejected}, {name: "generic forbidden", status: http.StatusForbidden, body: `{"error":{"code":"permission_denied"}}`}, {name: "resource missing", status: http.StatusNotFound, body: `{"error":{"code":"response_not_found"}}`}, diff --git a/internal/execution/bifrost/mapping.go b/internal/execution/bifrost/mapping.go index e66459ed..ca24076e 100644 --- a/internal/execution/bifrost/mapping.go +++ b/internal/execution/bifrost/mapping.go @@ -298,6 +298,8 @@ func neutralFailureHint(status int, values ...string) execution.FailureHint { switch { case status == http.StatusUnauthorized: return execution.FailureHintInvalidCredential + case len(values) >= 2 && execution.ExplicitRequestRejection(values[0], values[1], values[len(values)-1]): + return execution.FailureHintRequestRejected case candidateCapabilityRejected(status, values): return execution.FailureHintModelUnavailable case containsAnyMarker(markers, @@ -313,9 +315,6 @@ func neutralFailureHint(status int, values ...string) execution.FailureHint { "authentication failed", "invalid credential", "api key not valid"): return execution.FailureHintInvalidCredential - case containsAnyMarker(markers, - "invalid_request_error", "context_length_exceeded"): - return execution.FailureHintRequestRejected case containsAnyMarker(markers, "server_is_overloaded", "service_unavailable", "server overloaded"): return execution.FailureHintHostError @@ -387,12 +386,6 @@ func annotateBifrostErrorEvidence(evidence *execution.ErrorEvidence) { evidence.ScopeHint = execution.ErrorScopeGroup case execution.ErrorKindCanceled, execution.ErrorKindInvalidRequest: evidence.ScopeHint = execution.ErrorScopeRequest - case execution.ErrorKindHTTP: - if evidence.StatusCode >= http.StatusBadRequest && - evidence.StatusCode < http.StatusInternalServerError && - evidence.StatusCode != http.StatusTooManyRequests { - evidence.ScopeHint = execution.ErrorScopeRequest - } } } } diff --git a/internal/execution/bifrost/mapping_test.go b/internal/execution/bifrost/mapping_test.go index 91890d27..c449182d 100644 --- a/internal/execution/bifrost/mapping_test.go +++ b/internal/execution/bifrost/mapping_test.go @@ -67,13 +67,11 @@ func TestPassthroughHTTPErrorProducesNeutralFailureHints(t *testing.T) { }, { name: "generic forbidden permission", status: http.StatusForbidden, - body: `{"error":{"code":"permission_denied"}}`, - scope: execution.ErrorScopeRequest, + body: `{"error":{"code":"permission_denied"}}`, }, { name: "payment required", status: http.StatusPaymentRequired, - body: `{"error":{"message":"billing disabled"}}`, - scope: execution.ErrorScopeRequest, + body: `{"error":{"message":"billing disabled"}}`, }, { name: "explicit invalid key under forbidden", status: http.StatusForbidden, diff --git a/internal/execution/cpa/claude_provider_test.go b/internal/execution/cpa/claude_provider_test.go index 39204e2c..f521eb23 100644 --- a/internal/execution/cpa/claude_provider_test.go +++ b/internal/execution/cpa/claude_provider_test.go @@ -62,6 +62,17 @@ type classifiedClaudeError struct { summary string } +func TestClaudeContextLimitIsAnExplicitRequestRejection(t *testing.T) { + bridge := newClaudeProviderBridge() + _, evidence := bridge.ClassifyError(t.Context(), &classifiedClaudeError{ + status: http.StatusBadRequest, typeValue: "invalid_request_error", + summary: "prompt is too long: 210000 tokens > 200000 maximum", + }, claudeProviderCredential{}) + if evidence == nil || evidence.Hint != execution.FailureHintRequestRejected || evidence.ScopeHint != execution.ErrorScopeRequest { + t.Fatalf("context limit evidence=%#v, want request rejection", evidence) + } +} + func (err *classifiedClaudeError) Error() string { return err.summary } func (err *classifiedClaudeError) StatusCode() int { return err.status } func (err *classifiedClaudeError) ErrorType() string { return err.typeValue } diff --git a/internal/execution/cpa/codex_compatibility_test.go b/internal/execution/cpa/codex_compatibility_test.go index bb3001db..6ea8c29a 100644 --- a/internal/execution/cpa/codex_compatibility_test.go +++ b/internal/execution/cpa/codex_compatibility_test.go @@ -78,7 +78,7 @@ func TestCodexUpgradePreservesQuotaAndRejectionPolicy(t *testing.T) { events := 0 result := adapter.ExecuteStream(ctx, spec, func(execution.StreamEvent) error { events++; return nil }) evidence, status, dispatch = result.Error, result.StatusCode, result.DispatchState - if test.wantReplay && events != 0 { + if events != 0 { t.Fatalf("rejection emitted %d downstream events", events) } } else { @@ -97,8 +97,8 @@ func TestCodexUpgradePreservesQuotaAndRejectionPolicy(t *testing.T) { if decision.Effect != test.wantEffect { t.Fatalf("health decision=%+v, want effect %s", decision, test.wantEffect) } - if got := decision.Retry == health.RetryNextCandidate; got != test.wantReplay { - t.Fatalf("health retry=%s, want retryable=%t", decision.Retry, test.wantReplay) + if decision.Retry != health.RetryNextCandidate { + t.Fatalf("health retry=%s, want next candidate without changing replay evidence", decision.Retry) } }) } diff --git a/internal/execution/cpa/grok_provider.go b/internal/execution/cpa/grok_provider.go index 2a2e1b4c..8cbdfa7f 100644 --- a/internal/execution/cpa/grok_provider.go +++ b/internal/execution/cpa/grok_provider.go @@ -231,8 +231,6 @@ func (*grokProviderBridge) ClassifyError( case status == http.StatusForbidden || status == http.StatusPaymentRequired: evidence.Hint = execution.FailureHintCandidateUnavailable evidence.ReplaySafety = execution.ReplaySafetyRejectedBeforeProcessing - case status == http.StatusBadRequest: - evidence.Hint = execution.FailureHintRequestRejected case status == http.StatusTooManyRequests: evidence.Hint = execution.FailureHintRateLimited case status >= http.StatusInternalServerError: diff --git a/internal/execution/cpa/grok_provider_test.go b/internal/execution/cpa/grok_provider_test.go index 7e812359..91318615 100644 --- a/internal/execution/cpa/grok_provider_test.go +++ b/internal/execution/cpa/grok_provider_test.go @@ -40,7 +40,10 @@ func TestGrokProviderClassifiesOAuthAndQuotaFailures(t *testing.T) { {name: "unauthorized", err: grokProviderTestError{status: http.StatusUnauthorized}, hint: execution.FailureHintRefreshRequired, scope: execution.ErrorScopeCredential, replay: execution.ReplaySafetyRejectedBeforeProcessing}, {name: "forbidden", err: grokProviderTestError{status: http.StatusForbidden}, hint: execution.FailureHintCandidateUnavailable, scope: execution.ErrorScopeModel, replay: execution.ReplaySafetyRejectedBeforeProcessing}, {name: "payment required", err: grokProviderTestError{status: http.StatusPaymentRequired}, hint: execution.FailureHintCandidateUnavailable, scope: execution.ErrorScopeModel, replay: execution.ReplaySafetyRejectedBeforeProcessing}, - {name: "invalid request", err: grokProviderTestError{status: http.StatusBadRequest}, hint: execution.FailureHintRequestRejected, scope: execution.ErrorScopeRequest}, + {name: "generic bad request", err: grokProviderTestError{status: http.StatusBadRequest}}, + {name: "explicit invalid parameter", err: grokProviderTestError{status: http.StatusBadRequest, code: "invalid_parameter"}, hint: execution.FailureHintRequestRejected, scope: execution.ErrorScopeRequest}, + {name: "unsupported model", err: grokProviderTestError{status: http.StatusBadRequest, code: "unsupported_model"}, hint: execution.FailureHintModelUnavailable, scope: execution.ErrorScopeModel}, + {name: "unsupported model alias", err: grokProviderTestError{status: http.StatusBadRequest, code: "unsupported-model"}, hint: execution.FailureHintModelUnavailable, scope: execution.ErrorScopeModel}, {name: "free usage", err: grokProviderTestError{status: http.StatusTooManyRequests, code: "subscription:free-usage-exhausted", retry: 24 * time.Hour}, hint: execution.FailureHintRateLimited, scope: execution.ErrorScopeCredential}, {name: "host", err: grokProviderTestError{status: http.StatusServiceUnavailable}, hint: execution.FailureHintHostError, scope: execution.ErrorScopeGroup}, } { diff --git a/internal/execution/cpa/provider.go b/internal/execution/cpa/provider.go index 67cd32d7..689c8447 100644 --- a/internal/execution/cpa/provider.go +++ b/internal/execution/cpa/provider.go @@ -53,9 +53,14 @@ func annotateProviderErrorEvidence(evidence *execution.ErrorEvidence, err error) if evidence == nil { return } + explicitRequestRejection := execution.ExplicitRequestRejection(evidence.Type, evidence.Code, evidence.Summary) + if evidence.Hint == "" && explicitRequestRejection { + evidence.Hint = execution.FailureHintRequestRejected + } if evidence.Kind == execution.ErrorKindHTTP && evidence.StatusCode == http.StatusBadRequest && - evidence.Hint == execution.FailureHintRequestRejected && !requestScopedFailure(err) { - // 通用 HTTP 400 标签不能盖过明确的模型拒绝；保留 provider 的强请求级判定。 + (evidence.Hint == "" || evidence.Hint == execution.FailureHintRequestRejected) && + !explicitRequestRejection && !requestScopedFailure(err) { + // 仅细化通用拒绝；明确参数错误和 provider 的强请求级判定优先。 switch strings.ToLower(strings.TrimSpace(evidence.Code)) { case "unsupported_model", "unsupported-model": evidence.Hint = execution.FailureHintModelUnavailable diff --git a/internal/execution/cpa/provider_test.go b/internal/execution/cpa/provider_test.go index db2a919e..8d724e6a 100644 --- a/internal/execution/cpa/provider_test.go +++ b/internal/execution/cpa/provider_test.go @@ -157,22 +157,35 @@ var _ providerBridge = (*recordingProviderBridge)(nil) func TestProviderModelRejectionRefinesGenericRequestHint(t *testing.T) { for _, test := range []struct { - name string - err error - want execution.FailureHint + name string + err error + hint execution.FailureHint + typeValue string + code string + want execution.FailureHint }{ - {name: "generic provider rejection", err: errors.New("model unsupported"), want: execution.FailureHintModelUnavailable}, - {name: "explicit request scope preserved", err: requestScopedTestError{}, want: execution.FailureHintRequestRejected}, + {name: "generic provider rejection", err: errors.New("model unsupported"), hint: execution.FailureHintRequestRejected, code: "unsupported_model", want: execution.FailureHintModelUnavailable}, + {name: "model without generic hint", code: "unsupported_model", want: execution.FailureHintModelUnavailable}, + {name: "model alias without generic hint", code: "unsupported-model", want: execution.FailureHintModelUnavailable}, + {name: "explicit parameter error precedes model code", typeValue: "invalid_parameter", code: "unsupported_model", want: execution.FailureHintRequestRejected}, + {name: "explicit request scope preserved", err: requestScopedTestError{}, hint: execution.FailureHintRequestRejected, code: "unsupported_model", want: execution.FailureHintRequestRejected}, } { t.Run(test.name, func(t *testing.T) { evidence := &execution.ErrorEvidence{ Kind: execution.ErrorKindHTTP, StatusCode: http.StatusBadRequest, - Hint: execution.FailureHintRequestRejected, Code: "unsupported_model", + Hint: test.hint, Type: test.typeValue, Code: test.code, } annotateProviderErrorEvidence(evidence, test.err) if evidence.Hint != test.want { t.Fatalf("hint = %s, want %s", evidence.Hint, test.want) } + wantScope := execution.ErrorScopeModel + if test.want == execution.FailureHintRequestRejected { + wantScope = execution.ErrorScopeRequest + } + if evidence.ScopeHint != wantScope { + t.Fatalf("scope = %s, want %s", evidence.ScopeHint, wantScope) + } }) } } diff --git a/internal/execution/request_rejection.go b/internal/execution/request_rejection.go new file mode 100644 index 00000000..025bb50e --- /dev/null +++ b/internal/execution/request_rejection.go @@ -0,0 +1,19 @@ +package execution + +import "strings" + +// ExplicitRequestRejection 识别具体代码及明确的上下文超限，不以通用错误类型或任意消息子串终止重试。 +func ExplicitRequestRejection(typeValue, codeValue, message string) bool { + for _, value := range []string{typeValue, codeValue} { + switch strings.ToLower(strings.TrimSpace(value)) { + case "context_length_exceeded", "invalid_parameter", "content_policy_violation": + return true + } + } + if !strings.EqualFold(strings.TrimSpace(typeValue), "invalid_request_error") || strings.TrimSpace(codeValue) != "" { + return false + } + message = strings.ToLower(strings.TrimSpace(message)) + return strings.TrimSuffix(message, ".") == "prompt is too long" || + strings.HasPrefix(message, "prompt is too long:") +} diff --git a/internal/gateway/dialects_integration_test.go b/internal/gateway/dialects_integration_test.go index ce4ccec5..1ac58267 100644 --- a/internal/gateway/dialects_integration_test.go +++ b/internal/gateway/dialects_integration_test.go @@ -682,7 +682,7 @@ func TestGatewayAppliesEachGroupsParameterOverridesFromOriginal(t *testing.T) { } } -func TestHandlerHostFailureDoesNotReplayGenerationAcrossGroups(t *testing.T) { +func TestHandlerHostFailureRetriesGenerationAcrossGroups(t *testing.T) { primary := fakeupstream.New( fakeupstream.Step{Status: http.StatusInternalServerError, Fixture: "openai/500.json"}, fakeupstream.Step{Status: http.StatusInternalServerError, Fixture: "openai/500.json"}, @@ -709,7 +709,7 @@ func TestHandlerHostFailureDoesNotReplayGenerationAcrossGroups(t *testing.T) { request.Header.Set("Authorization", "Bearer gl-client") recorder := httptest.NewRecorder() engine.ServeHTTP(recorder, request) - if recorder.Code != http.StatusInternalServerError || recorder.Header().Get(debugHeaderAttempts) != "1" { + if recorder.Code != http.StatusOK || recorder.Header().Get(debugHeaderAttempts) != "2" { t.Fatalf("response = %d attempts=%s body=%s", recorder.Code, recorder.Header().Get(debugHeaderAttempts), recorder.Body.String()) } @@ -717,8 +717,8 @@ func TestHandlerHostFailureDoesNotReplayGenerationAcrossGroups(t *testing.T) { if got := len(primary.Requests()); got != 2 { t.Fatalf("primary requests = %d, want one per downstream request", got) } - if got := len(backup.Requests()); got != 0 { - t.Fatalf("backup requests = %d, want none for ambiguous generation failures", got) + if got := len(backup.Requests()); got != 2 { + t.Fatalf("backup requests = %d, want one per downstream request", got) } } diff --git a/internal/gateway/error_retry_policy_test.go b/internal/gateway/error_retry_policy_test.go new file mode 100644 index 00000000..75e425b4 --- /dev/null +++ b/internal/gateway/error_retry_policy_test.go @@ -0,0 +1,218 @@ +package gateway + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "gpt-load/internal/channel" + "gpt-load/internal/dialect" + "gpt-load/internal/protocol" +) + +func TestAnthropicPrechargeFailureRetriesWithoutCredentialPenalty(t *testing.T) { + for _, errorType := range []string{"", "invalid_request_error"} { + t.Run("type="+errorType, func(t *testing.T) { + var mu sync.Mutex + var credentials []string + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + key := r.Header.Get("X-Api-Key") + mu.Lock() + credentials = append(credentials, key) + mu.Unlock() + w.Header().Set("Content-Type", "application/json") + if key == "sk-one" { + w.WriteHeader(http.StatusForbidden) + _, _ = fmt.Fprintf(w, `{"type":"error","error":{"type":%q,"message":"预扣费额度失败，余额 0.10，需要预扣 0.20"}}`, errorType) + return + } + _, _ = w.Write([]byte(`{"id":"msg_test","type":"message","role":"assistant","model":"model-a","content":[{"type":"text","text":"ok"}],"stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}`)) + })) + defer upstream.Close() + engine, registry := newDialectGatewayEngine(t, protocol.Anthropic, "model-a", dialect.NewSet(dialect.NewAnthropic()), + dialectGatewayGroup{id: 1, name: "precharge", upstreamURL: upstream.URL, apiKeys: []string{"sk-one", "sk-two"}}, + ) + request := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(`{"model":"model-a","max_tokens":1,"messages":[{"role":"user","content":"hello"}]}`)) + request.Header.Set("Authorization", "Bearer gl-client") + response := httptest.NewRecorder() + engine.ServeHTTP(response, request) + mu.Lock() + defer mu.Unlock() + if response.Code != http.StatusOK || fmt.Sprint(credentials) != "[sk-one sk-two]" { + t.Fatalf("response=%d attempts=%v body=%s", response.Code, credentials, response.Body.String()) + } + if candidates := registry.CollectCredentialCandidates([]uint{1}, nil, time.Now()); len(candidates) != 2 { + t.Fatalf("available credentials=%d, want both", len(candidates)) + } + entries, err := registry.SnapshotGroupCredentialEntriesExact(1, []uint{1, 2}) + if err != nil { + t.Fatal(err) + } + for _, entry := range entries { + if entry.FailureCount != 0 || entry.Blacklisted || !entry.CooldownUntil.IsZero() || len(entry.ModelCooldowns) != 0 { + t.Fatalf("credential %d unexpectedly penalized", entry.ID) + } + } + }) + } +} + +func TestUnknownResponsesCancellationFailureDoesNotSwitchCredential(t *testing.T) { + var attempts atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + if attempts.Add(1) == 1 { + w.WriteHeader(http.StatusForbidden) + _, _ = fmt.Fprint(w, `{"error":{"message":"resource rejected"}}`) + return + } + _, _ = fmt.Fprint(w, `{"id":"resp_1","status":"cancelled"}`) + })) + defer upstream.Close() + engine, _ := newDialectGatewayEngine(t, protocol.OpenAIResponses, "model-a", dialect.NewSet(dialect.NewOpenAIResponses()), + dialectGatewayGroup{id: 1, name: "resource-retry", channelID: channel.OpenAI, + params: json.RawMessage(fmt.Sprintf(`{"base_url":%q}`, upstream.URL+"/v1")), apiKeys: []string{"sk-one", "sk-two"}}, + ) + request := httptest.NewRequest(http.MethodPost, "/v1/responses/resp_1/cancel", strings.NewReader("{}")) + request.Header.Set("Authorization", "Bearer gl-client") + response := httptest.NewRecorder() + engine.ServeHTTP(response, request) + if response.Code != http.StatusForbidden || attempts.Load() != 1 || !strings.Contains(response.Body.String(), "resource rejected") { + t.Fatalf("status=%d attempts=%d body=%s", response.Code, attempts.Load(), response.Body.String()) + } +} + +func TestUnknownErrorRetryRecordsAttemptsAndHonorsBudget(t *testing.T) { + for _, retryCount := range []int{0, 1} { + t.Run(fmt.Sprintf("retries=%d", retryCount), func(t *testing.T) { + failure := completeScriptedUpstreamResult(UpstreamResult{ + StatusCode: http.StatusForbidden, RequestWritten: true, + Body: []byte(`{"error":{"message":"unclassified upstream rejection"}}`), + }) + forwarder := &scriptedForwarder{results: []UpstreamResult{failure, successfulAffinityResult()}} + handler, manager, _ := newHandlerForTest(t, forwarder, "sk-one", "sk-two") + manager.Current().Settings.RetryCount = retryCount + sink := &recordingRequestLogSink{} + handler.requestLogSink = sink + engine := newAffinityTestEngine(t, handler) + request := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{"model":"gpt-4o","messages":[]}`)) + request.Header.Set("Authorization", "Bearer gl-client") + response := httptest.NewRecorder() + engine.ServeHTTP(response, request) + events := sink.snapshot() + if len(events) != 1 || len(events[0].Attempts) != retryCount+1 || len(forwarder.inputs) != retryCount+1 { + t.Fatalf("attempts=%d events=%#v", len(forwarder.inputs), events) + } + first := events[0].Attempts[0] + if first.RetryDirective != "next_candidate" || first.Effect != "none" || + first.WillRetry != (retryCount > 0) || first.FailureCategory != "ambiguous" { + t.Fatalf("first attempt=%#v", first) + } + if retryCount == 0 && (response.Code != http.StatusForbidden || response.Body.String() != string(failure.Body)) { + t.Fatalf("terminal response=%d %s", response.Code, response.Body.String()) + } + if retryCount > 0 && (response.Code != http.StatusOK || forwarder.inputs[0].APIKey == forwarder.inputs[1].APIKey) { + t.Fatalf("retry did not switch credential: status=%d", response.Code) + } + }) + } +} + +func TestUnknownFirstStreamErrorRetriesBeforeOutput(t *testing.T) { + var mu sync.Mutex + var attempts []string + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + key := r.Header.Get("Authorization") + mu.Lock() + attempts = append(attempts, key) + mu.Unlock() + w.Header().Set("Content-Type", "text/event-stream") + if key == "Bearer sk-one" { + _, _ = fmt.Fprint(w, "data: "+`{"error":{"type":"new_provider_error","code":"unrecognized_condition","message":"first candidate failed"}}`+"\n\n") + return + } + _, _ = fmt.Fprint(w, "data: "+`{"id":"test","object":"chat.completion.chunk","model":"model-a","choices":[{"index":0,"delta":{"content":"recovered"},"finish_reason":"stop"}]}`+"\n\ndata: [DONE]\n\n") + })) + defer upstream.Close() + engine, _ := newDialectGatewayEngine(t, protocol.OpenAICompletions, "model-a", dialect.NewSet(dialect.NewOpenAI()), + dialectGatewayGroup{id: 1, name: "stream-retry", upstreamURL: upstream.URL, apiKeys: []string{"sk-one", "sk-two"}}, + ) + request := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{"model":"model-a","stream":true,"messages":[]}`)) + request.Header.Set("Authorization", "Bearer gl-client") + response := httptest.NewRecorder() + engine.ServeHTTP(response, request) + mu.Lock() + defer mu.Unlock() + if response.Code != http.StatusOK || fmt.Sprint(attempts) != "[Bearer sk-one Bearer sk-two]" || + !strings.Contains(response.Body.String(), "recovered") || strings.Contains(response.Body.String(), "unrecognized_condition") { + t.Fatalf("status=%d attempts=%v body=%s", response.Code, attempts, response.Body.String()) + } +} + +func TestConvertedStreamHTTPErrorRetriesBeforeOutput(t *testing.T) { + for _, test := range []struct { + name string + status int + partial bool + wantAttempts string + }{ + {name: "server error skips failed group", status: 503, wantAttempts: "sk-one,sk-next"}, + {name: "unknown rejection switches credential", status: 403, wantAttempts: "sk-one,sk-two"}, + {name: "server error after output stops", status: 503, partial: true, wantAttempts: "sk-one"}, + } { + t.Run(test.name, func(t *testing.T) { + var mu sync.Mutex + var attempts []string + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1beta/models/model-a:streamGenerateContent" { + t.Errorf("upstream path = %s", r.URL.Path) + } + key := r.Header.Get("X-Goog-Api-Key") + mu.Lock() + attempts = append(attempts, key) + mu.Unlock() + w.Header().Set("Content-Type", "text/event-stream") + if key == "sk-one" { + if test.partial { + _, _ = fmt.Fprint(w, "data: "+`{"candidates":[{"content":{"parts":[{"text":"partial"}],"role":"model"},"index":0}],"modelVersion":"model-a","responseId":"resp_1"}`+"\n\n") + } + _, _ = fmt.Fprintf(w, "data: {\"error\":{\"code\":%d,\"message\":\"first candidate failed\"}}\n\n", test.status) + return + } + _, _ = fmt.Fprint(w, "data: "+`{"candidates":[{"content":{"parts":[{"text":"recovered"}],"role":"model"},"finishReason":"STOP","index":0}],"modelVersion":"model-a","responseId":"resp_2","usageMetadata":{"promptTokenCount":1,"candidatesTokenCount":1,"totalTokenCount":2}}`+"\n\n") + })) + defer upstream.Close() + params, err := json.Marshal(map[string]string{"base_url": upstream.URL + "/v1beta"}) + if err != nil { + t.Fatal(err) + } + engine, _ := newDialectGatewayEngine(t, protocol.Anthropic, "model-a", dialect.NewSet(dialect.NewAnthropic()), + dialectGatewayGroup{id: 1, name: "first", channelID: channel.Gemini, params: params, apiKeys: []string{"sk-one", "sk-two"}}, + dialectGatewayGroup{id: 2, name: "backup", channelID: channel.Gemini, params: params, apiKeys: []string{"sk-next"}}, + ) + request := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(`{"model":"model-a","max_tokens":16,"stream":true,"messages":[{"role":"user","content":"hello"}]}`)) + request.Header.Set("Authorization", "Bearer gl-client") + response := httptest.NewRecorder() + engine.ServeHTTP(response, request) + mu.Lock() + defer mu.Unlock() + if strings.Join(attempts, ",") != test.wantAttempts { + t.Fatalf("attempts=%v, want %s; status=%d body=%s", attempts, test.wantAttempts, response.Code, response.Body.String()) + } + body := response.Body.String() + if test.partial { + if !strings.Contains(body, "partial") || strings.Contains(body, "recovered") { + t.Fatalf("committed stream response=%s", body) + } + } else if response.Code != http.StatusOK || !strings.Contains(body, "recovered") || strings.Contains(body, "first candidate failed") { + t.Fatalf("retry response=%d %s", response.Code, body) + } + }) + } +} diff --git a/internal/gateway/execution_boundary_test.go b/internal/gateway/execution_boundary_test.go index b55829f9..aaebcbf2 100644 --- a/internal/gateway/execution_boundary_test.go +++ b/internal/gateway/execution_boundary_test.go @@ -177,7 +177,7 @@ func TestJudgeUpstreamResultUsesStableStreamTerminalRules(t *testing.T) { wantOrigin execution.ErrorOrigin wantScope execution.ErrorScope }{ - {name: "provider error", endReason: StreamEndSSEError, wantRule: "stream.provider_error", wantOrigin: execution.ErrorOriginUpstream, wantScope: execution.ErrorScopeRequest}, + {name: "provider error", endReason: StreamEndSSEError, wantRule: "safety.committed", wantOrigin: execution.ErrorOriginUpstream, wantScope: execution.ErrorScopeRequest}, {name: "upstream terminated", endReason: StreamEndUpstreamTerminated, wantRule: "stream.upstream_terminated", wantOrigin: execution.ErrorOriginUpstream, wantScope: execution.ErrorScopeRequest}, {name: "protocol error", endReason: StreamEndUpstreamProtocolError, wantRule: "stream.protocol_error", wantOrigin: execution.ErrorOriginUpstream, wantScope: execution.ErrorScopeRequest}, {name: "idle timeout", endReason: StreamEndIdleTimeout, wantRule: "stream.idle_timeout", wantOrigin: execution.ErrorOriginUpstream, wantScope: execution.ErrorScopeRequest}, @@ -192,7 +192,7 @@ func TestJudgeUpstreamResultUsesStableStreamTerminalRules(t *testing.T) { StatusCode: http.StatusOK, Committed: true, Stream: streamTerminalObservation(test.endReason), - }, time.Now(), health.DecisionContext{}) + }, time.Now(), health.DecisionContext{Method: http.MethodPost, Operation: execution.OperationChatCompletion}) if decision.Category != health.FailureCategoryAmbiguous || decision.Origin != test.wantOrigin || decision.Scope != test.wantScope || decision.Retry != health.RetryNone || decision.Effect != health.EffectNone || diff --git a/internal/gateway/handler_test.go b/internal/gateway/handler_test.go index 4fd245bd..6a1de2b3 100644 --- a/internal/gateway/handler_test.go +++ b/internal/gateway/handler_test.go @@ -3303,8 +3303,9 @@ func TestHandlerUsesClassifierForNonStreamingNonSuccess(t *testing.T) { t.Run("client error terminates after one attempt", func(t *testing.T) { forwarder := &scriptedForwarder{results: []UpstreamResult{ {StatusCode: http.StatusBadRequest, Header: make(http.Header), - Body: []byte(`{"error":"invalid input"}`), - ClassificationBody: []byte(`{"error":"invalid input"}`), RequestWritten: true}, + Body: []byte(`{"error":"invalid input"}`), + ExecutionError: &execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, Code: "invalid_parameter"}, + RequestWritten: true}, {StatusCode: http.StatusOK, Header: make(http.Header), Body: []byte(`{"ok":true}`)}, }} engine, _, _ := newHandlerTestRuntime(t, forwarder, "sk-one", "sk-two") @@ -4647,8 +4648,8 @@ func TestHandlerDoesNotExposeAliasedUpstreamModelWhenRetryBudgetIsExhausted(t *t engine.ServeHTTP(recorder, request) - if recorder.Code != http.StatusInternalServerError || attempts.Load() != 1 { - t.Fatalf("response/attempts = %d/%d, want 500/1", recorder.Code, attempts.Load()) + if recorder.Code != http.StatusInternalServerError || attempts.Load() != 3 { + t.Fatalf("response/attempts = %d/%d, want 500/3", recorder.Code, attempts.Load()) } if strings.Contains(recorder.Body.String(), upstreamModel) || !strings.Contains(recorder.Body.String(), externalModel) { diff --git a/internal/health/execution_judge.go b/internal/health/execution_judge.go index 1f485e09..f2a64c00 100644 --- a/internal/health/execution_judge.go +++ b/internal/health/execution_judge.go @@ -297,6 +297,9 @@ func classifyExecutionEvidence(attempt ExecutionAttempt) FailureCategory { case execution.FailureHintHostError: return FailureCategoryUpstreamHostError } + if execution.ExplicitRequestRejection(attempt.Evidence.Type, attempt.Evidence.Code, attempt.Evidence.Summary) { + return FailureCategoryClientError + } markers = strings.ToLower(strings.Join([]string{ attempt.Evidence.Type, attempt.Evidence.Code, @@ -321,8 +324,6 @@ func classifyExecutionEvidence(attempt ExecutionAttempt) FailureCategory { return FailureCategoryInvalidKey case statusCode >= http.StatusInternalServerError && statusCode <= 599: return FailureCategoryUpstreamHostError - case statusCode >= http.StatusBadRequest && statusCode <= 499: - return FailureCategoryClientError case attempt.Evidence != nil && attempt.Evidence.Kind == execution.ErrorKindInvalidRequest: return FailureCategoryClientError default: @@ -409,10 +410,10 @@ func decisionForExecutionCategory( if attempt.Evidence.ReplaySafety == execution.ReplaySafetyRejectedBeforeProcessing { retry = RetryNextCandidate ruleID = "upstream.host_error.rejected_before_processing" - } else if attempt.Evidence.ReplaySafety != execution.ReplaySafetyUnknown && + } else if retryableUpstreamResponse(attempt) && requestMayReplayAfterResponse(decisionContext) { retry = RetryNextCandidate - ruleID = "upstream.host_error.read_only" + ruleID = "upstream.host_error.response_retry" } return decision( category, @@ -443,7 +444,44 @@ func decisionForExecutionCategory( case FailureCategoryOK: return decision(category, origin, scope, RetryNone, EffectNone, "success.upstream_response") default: - return decision(category, origin, scope, RetryNone, EffectNone, ambiguousRuleID(attempt.Evidence)) + ruleID := ambiguousRuleID(attempt.Evidence) + if retryableUpstreamResponse(attempt) { + if !requestMayReplayAfterResponse(decisionContext) && + attempt.Evidence.ReplaySafety != execution.ReplaySafetyRejectedBeforeProcessing { + return decision(category, origin, scope, RetryNone, EffectNone, "safety.operation_replay_unsafe") + } + if ruleID == "fallback.ambiguous" { + ruleID = "fallback.upstream_response" + } + return decision(category, origin, scope, RetryNextCandidate, EffectNone, ruleID) + } + return decision(category, origin, scope, RetryNone, EffectNone, ruleID) + } +} + +// retryableUpstreamResponse 区分已收到的错误响应与超时、断流等执行结果未知的失败。 +func retryableUpstreamResponse(attempt ExecutionAttempt) bool { + evidence := attempt.Evidence + if evidence == nil || originForEvidence(evidence) != execution.ErrorOriginUpstream || + evidence.ReplaySafety == execution.ReplaySafetyUnknown { + return false + } + switch evidence.Kind { + case execution.ErrorKindHTTP: + status := attempt.StatusCode + if status == 0 { + status = evidence.StatusCode + } + // 流建立后仍保留外层 2xx，流内 HTTP 错误由 Kind/Hint 承载。 + return isSuccessStatus(status) || status >= http.StatusBadRequest && status <= 599 + case execution.ErrorKindProvider: + switch evidence.Code { + case "upstream_protocol_error", "upstream_response_incomplete": + return false + } + return attempt.ResponseStarted() + default: + return false } } @@ -695,7 +733,10 @@ func requestMayReplayAfterResponse(value DecisionContext) bool { return true } switch value.Operation { - case execution.OperationResponsesRetrieve, + case execution.OperationChatCompletion, + execution.OperationResponsesCreate, + execution.OperationResponsesCompact, + execution.OperationResponsesRetrieve, execution.OperationResponsesInputItems, execution.OperationResponsesInputTokens, execution.OperationCountTokens, diff --git a/internal/health/execution_judge_test.go b/internal/health/execution_judge_test.go index bdac0ea0..01e2c582 100644 --- a/internal/health/execution_judge_test.go +++ b/internal/health/execution_judge_test.go @@ -216,22 +216,22 @@ func TestJudgeExecutionUsesNeutralEvidenceAndReplayBoundary(t *testing.T) { want: Result{Category: FailureCategoryAmbiguous, Action: ActionTerminate}, }, { - name: "payment required is a client error", + name: "payment required retries without penalty", attempt: ExecutionAttempt{ DispatchState: execution.DispatchMaybeSent, StatusCode: http.StatusPaymentRequired, Evidence: evidence(execution.ErrorKindHTTP, http.StatusPaymentRequired, "billing disabled"), }, - want: Result{Category: FailureCategoryClientError, Action: ActionTerminate}, + want: Result{Category: FailureCategoryAmbiguous, Action: ActionRetry}, }, { - name: "generic forbidden is a client error", + name: "generic forbidden retries without penalty", attempt: ExecutionAttempt{ DispatchState: execution.DispatchMaybeSent, StatusCode: http.StatusForbidden, Evidence: evidence(execution.ErrorKindHTTP, http.StatusForbidden, "permission denied"), }, - want: Result{Category: FailureCategoryClientError, Action: ActionTerminate}, + want: Result{Category: FailureCategoryAmbiguous, Action: ActionRetry}, }, { name: "forbidden model marker retries without penalty", @@ -316,13 +316,13 @@ func TestJudgeExecutionUsesNeutralEvidenceAndReplayBoundary(t *testing.T) { }, }, { - name: "generic not found terminates", + name: "generic not found retries without penalty", attempt: ExecutionAttempt{ DispatchState: execution.DispatchMaybeSent, StatusCode: http.StatusNotFound, Evidence: evidence(execution.ErrorKindHTTP, http.StatusNotFound, "endpoint not found"), }, - want: Result{Category: FailureCategoryClientError, Action: ActionTerminate}, + want: Result{Category: FailureCategoryAmbiguous, Action: ActionRetry}, }, { name: "provider rate limit under success status still cools credential", diff --git a/internal/health/execution_judge_v2_test.go b/internal/health/execution_judge_v2_test.go index 73cbf33e..ea56c122 100644 --- a/internal/health/execution_judge_v2_test.go +++ b/internal/health/execution_judge_v2_test.go @@ -289,7 +289,7 @@ func TestJudgeExecutionPreservesReplayCompatibilityRules(t *testing.T) { wantRule RuleID }{ { - name: "generic payment required terminates", + name: "generic payment required retries", attempt: ExecutionAttempt{ DispatchState: execution.DispatchMaybeSent, StatusCode: http.StatusPaymentRequired, @@ -298,8 +298,9 @@ func TestJudgeExecutionPreservesReplayCompatibilityRules(t *testing.T) { Summary: "billing disabled", }, }, - wantRetry: RetryNone, - wantRule: RuleID("fallback.http_client_error"), + context: DecisionContext{Method: http.MethodPost, Operation: execution.OperationChatCompletion}, + wantRetry: RetryNextCandidate, + wantRule: RuleID("fallback.upstream_response"), }, { name: "candidate payment required retries", @@ -340,10 +341,10 @@ func TestJudgeExecutionPreservesReplayCompatibilityRules(t *testing.T) { }, context: DecisionContext{Method: http.MethodGet, Operation: execution.OperationResponsesRetrieve}, wantRetry: RetryNextCandidate, - wantRule: RuleID("upstream.host_error.read_only"), + wantRule: RuleID("upstream.host_error.response_retry"), }, { - name: "mutating host error terminates", + name: "generation host error retries", attempt: ExecutionAttempt{ DispatchState: execution.DispatchMaybeSent, StatusCode: http.StatusServiceUnavailable, @@ -353,8 +354,8 @@ func TestJudgeExecutionPreservesReplayCompatibilityRules(t *testing.T) { }, }, context: DecisionContext{Method: http.MethodPost, Operation: execution.OperationChatCompletion}, - wantRetry: RetryNone, - wantRule: RuleID("upstream.host_error.replay_unsafe"), + wantRetry: RetryNextCandidate, + wantRule: RuleID("upstream.host_error.response_retry"), }, { name: "explicitly rejected mutating host error retries", @@ -496,13 +497,13 @@ func TestJudgeExecutionReplayUnknownKeepsUnaffectedRuleID(t *testing.T) { wantRule: "rate_limit.scoped", }, { - name: "generic client error", + name: "generic forbidden with unknown replay safety", status: http.StatusForbidden, evidence: execution.ErrorEvidence{ Kind: execution.ErrorKindHTTP, StatusCode: http.StatusForbidden, ReplaySafety: execution.ReplaySafetyUnknown, Summary: "permission denied", }, - wantRule: "fallback.http_client_error", + wantRule: "safety.replay_unknown", }, } @@ -738,8 +739,9 @@ func TestJudgeExecutionCompatibilityRuleMatrix(t *testing.T) { Kind: execution.ErrorKindHTTP, StatusCode: http.StatusForbidden, Summary: "permission denied", }, - category: FailureCategoryClientError, scope: execution.ErrorScopeRequest, - retry: RetryNone, effect: EffectNone, ruleID: "fallback.http_client_error", + context: DecisionContext{Method: http.MethodPost, Operation: execution.OperationChatCompletion}, + category: FailureCategoryAmbiguous, + retry: RetryNextCandidate, effect: EffectNone, ruleID: "fallback.upstream_response", }, { name: "candidate scoped 403", status: http.StatusForbidden, diff --git a/internal/health/retry_policy_test.go b/internal/health/retry_policy_test.go new file mode 100644 index 00000000..c65ed9b0 --- /dev/null +++ b/internal/health/retry_policy_test.go @@ -0,0 +1,119 @@ +package health + +import ( + "net/http" + "testing" + + "gpt-load/internal/execution" +) + +func TestUnknownHTTPRejectionsRetryWithoutCredentialEffects(t *testing.T) { + for _, status := range []int{400, 402, 403, 404, 409, 422} { + for _, code := range []string{"", "unknown_provider_error"} { + result := JudgeExecution(ExecutionAttempt{ + DispatchState: execution.DispatchMaybeSent, StatusCode: status, + Evidence: &execution.ErrorEvidence{ + Kind: execution.ErrorKindHTTP, StatusCode: status, + OriginHint: execution.ErrorOriginUpstream, + Type: "invalid_request_error", Code: code, Summary: "request could not be served", + }, + }, DecisionContext{Method: http.MethodPost, Operation: execution.OperationChatCompletion}) + if result.Category != FailureCategoryAmbiguous || result.Retry != RetryNextCandidate || + result.Effect != EffectNone || !result.CooldownUntil.IsZero() { + t.Errorf("status=%d code=%q decision=%#v", status, code, result) + } + } + } +} + +func TestExplicitRequestRejectionCodesStillTerminate(t *testing.T) { + for _, code := range []string{"context_length_exceeded", "invalid_parameter", "content_policy_violation"} { + result := JudgeExecution(ExecutionAttempt{ + DispatchState: execution.DispatchMaybeSent, StatusCode: http.StatusBadRequest, + Evidence: &execution.ErrorEvidence{ + Kind: execution.ErrorKindHTTP, Code: code, OriginHint: execution.ErrorOriginUpstream, + }, + }, DecisionContext{Method: http.MethodPost, Operation: execution.OperationChatCompletion}) + if result.Category != FailureCategoryClientError || result.Retry != RetryNone || result.Effect != EffectNone { + t.Errorf("code=%s decision=%#v", code, result) + } + } +} + +func TestUpstreamResponseRetryPreservesExecutionBoundaries(t *testing.T) { + for _, test := range []struct { + name string + status int + evidence execution.ErrorEvidence + operation execution.Operation + committed bool + retry RetryDirective + effect Effect + rule RuleID + }{ + {name: "generation server error", status: 503, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP}, retry: RetryNextCandidate, effect: EffectSkipGroup}, + {name: "server error inside successful stream", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, Hint: execution.FailureHintHostError}, retry: RetryNextCandidate, effect: EffectSkipGroup}, + {name: "unknown HTTP error inside successful stream", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP}, retry: RetryNextCandidate, effect: EffectNone}, + {name: "stream HTTP error with unknown outcome", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, Hint: execution.FailureHintHostError, ReplaySafety: execution.ReplaySafetyUnknown}, retry: RetryNone, effect: EffectSkipGroup}, + {name: "committed stream HTTP error", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, Hint: execution.FailureHintHostError}, committed: true, retry: RetryNone, effect: EffectNone}, + {name: "stream HTTP error retains strict replay policy", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, Hint: execution.FailureHintHostError}, operation: execution.OperationImagesGenerate, retry: RetryNone, effect: EffectSkipGroup}, + {name: "stream HTTP error retains resource boundary", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, Hint: execution.FailureHintHostError}, operation: execution.OperationResponsesCancel, retry: RetryNone, effect: EffectSkipGroup}, + {name: "unknown error event", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindProvider, Code: "upstream_sse_error"}, retry: RetryNextCandidate, effect: EffectNone}, + {name: "unknown provider code", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindProvider, Code: "new_provider_error"}, retry: RetryNextCandidate, effect: EffectNone}, + {name: "response status in evidence", evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, StatusCode: 403}, retry: RetryNextCandidate, effect: EffectNone}, + {name: "transport after headers", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindTransport}, retry: RetryNone, effect: EffectNone}, + {name: "timeout after headers", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindTimeout}, retry: RetryNone, effect: EffectNone}, + {name: "protocol failure", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindProvider, Code: "upstream_protocol_error"}, retry: RetryNone, effect: EffectNone}, + {name: "incomplete response", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindProvider, Code: "upstream_response_incomplete"}, retry: RetryNone, effect: EffectNone}, + {name: "internal result with HTTP status", status: 400, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindInternal}, retry: RetryNone, effect: EffectNone}, + {name: "unknown execution result", status: 503, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, ReplaySafety: execution.ReplaySafetyUnknown}, retry: RetryNone, effect: EffectSkipGroup}, + {name: "committed event", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindProvider, Code: "upstream_sse_error"}, committed: true, retry: RetryNone, effect: EffectNone}, + {name: "image needs rejection proof", status: 503, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP}, operation: execution.OperationImagesGenerate, retry: RetryNone, effect: EffectSkipGroup, rule: "upstream.host_error.replay_unsafe"}, + {name: "resource mutation remains restricted", status: 503, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP}, operation: execution.OperationResponsesCancel, retry: RetryNone, effect: EffectSkipGroup}, + } { + t.Run(test.name, func(t *testing.T) { + operation := test.operation + if operation == "" { + operation = execution.OperationChatCompletion + } + result := JudgeExecution(ExecutionAttempt{ + DispatchState: execution.DispatchMaybeSent, StatusCode: test.status, + Evidence: &test.evidence, DownstreamCommitted: test.committed, + }, DecisionContext{Method: http.MethodPost, Operation: operation}) + if result.Retry != test.retry || result.Effect != test.effect { + t.Fatalf("decision=%#v, want retry=%s effect=%s", result, test.retry, test.effect) + } + if test.rule != "" && result.RuleID != test.rule { + t.Fatalf("rule=%s, want %s", result.RuleID, test.rule) + } + }) + } +} + +func TestUnknownErrorRetryRespectsOperationReplayPolicy(t *testing.T) { + for _, test := range []struct { + operation execution.Operation + method string + }{ + {execution.OperationResponsesDelete, http.MethodDelete}, + {execution.OperationResponsesCancel, http.MethodPost}, + {execution.OperationResponsesPassthrough, http.MethodPost}, + {execution.OperationResponsesPassthrough, http.MethodGet}, + {execution.OperationChatCompletion, http.MethodPost}, + } { + for _, proof := range []execution.ReplaySafety{"", execution.ReplaySafetyRejectedBeforeProcessing} { + result := JudgeExecution(ExecutionAttempt{ + DispatchState: execution.DispatchMaybeSent, StatusCode: http.StatusForbidden, + Evidence: &execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, ReplaySafety: proof}, + }, DecisionContext{Method: test.method, Operation: test.operation}) + want := RetryNone + if test.method == http.MethodGet || test.operation == execution.OperationChatCompletion || proof != "" { + want = RetryNextCandidate + } + if result.Retry != want || result.Effect != EffectNone { + t.Errorf("operation=%s method=%s proof=%q decision=%#v, want retry=%s without effects", + test.operation, test.method, proof, result, want) + } + } + } +}