From 95fd1f8f2dfb83a42704fb937c28fb6725a0e929 Mon Sep 17 00:00:00 2001 From: Duy /zuey/ Date: Fri, 12 Jun 2026 00:43:09 +0700 Subject: [PATCH] fix(providers): fallback on codex safety errors (#157) --- internal/agent/loop_pipeline_callbacks.go | 20 +++++++ internal/agent/loop_tracing.go | 24 +++++++- internal/providers/error_classify.go | 18 +++++- internal/providers/error_classify_test.go | 17 ++++-- internal/providers/failover.go | 2 +- .../providers/model_fallback_observation.go | 57 +++++++++++++++++++ .../model_fallback_observation_test.go | 36 ++++++++++++ internal/providers/model_fallback_test.go | 27 +++++++++ 8 files changed, 192 insertions(+), 9 deletions(-) create mode 100644 internal/providers/model_fallback_observation.go create mode 100644 internal/providers/model_fallback_observation_test.go diff --git a/internal/agent/loop_pipeline_callbacks.go b/internal/agent/loop_pipeline_callbacks.go index 5aba6f89..1a0c991e 100644 --- a/internal/agent/loop_pipeline_callbacks.go +++ b/internal/agent/loop_pipeline_callbacks.go @@ -361,6 +361,7 @@ func (l *Loop) makeCallLLM(req *RunRequest, emitRun func(AgentEvent)) func(ctx c }) } } + fallbackTraceClassifier := providers.NewDefaultClassifier() callProvider := func(attempt string, request providers.ChatRequest) (*providers.ChatResponse, error) { if fallbackProvider, ok := provider.(*providers.ModelFallbackProvider); ok { before := func(callCtx context.Context, entry providers.FallbackCandidate, actualReq providers.ChatRequest) (providers.FallbackAfterCall, error) { @@ -371,6 +372,25 @@ func (l *Loop) makeCallLLM(req *RunRequest, emitRun func(AgentEvent)) func(ctx c return nil, reserveErr } return func(callResp *providers.ChatResponse, callErr error, info providers.FallbackCallInfo) { + fallbackMeta := providers.ModelFallbackAttemptMetadata{ + ProviderName: entry.ProviderName, + Model: actualReq.Model, + Status: "success", + Streamed: info.Streamed, + } + if callErr != nil { + classification := providers.ClassifyHTTPError(fallbackTraceClassifier, callErr) + reason := string(classification.Reason) + if classification.Kind == "context_overflow" { + reason = "context_overflow" + } + fallbackMeta.Status = "error" + fallbackMeta.Reason = reason + fallbackMeta.Error = callErr.Error() + } else { + opts = append(opts, withProvider(entry.ProviderName), withModel(actualReq.Model)) + } + opts = append(opts, withModelFallbackAttempt(fallbackMeta)) if reservation != nil { if info.Streamed { reservation.ReconcileStream(callCtx, callResp, callErr, true) diff --git a/internal/agent/loop_tracing.go b/internal/agent/loop_tracing.go index 469362a7..604150bb 100644 --- a/internal/agent/loop_tracing.go +++ b/internal/agent/loop_tracing.go @@ -52,9 +52,10 @@ func (l *Loop) IsRunning() bool { return l.activeRuns.Load() > 0 } type spanOption func(*spanOverrides) type spanOverrides struct { - model string - provider string - usageCapAttempts []usagecaps.TraceMetadata + model string + provider string + usageCapAttempts []usagecaps.TraceMetadata + modelFallbackAttempts []providers.ModelFallbackAttemptMetadata } func withModel(m string) spanOption { return func(o *spanOverrides) { o.model = m } } @@ -67,6 +68,14 @@ func withUsageCapMetadata(metadata usagecaps.TraceMetadata) spanOption { } } +func withModelFallbackAttempt(metadata providers.ModelFallbackAttemptMetadata) spanOption { + return func(o *spanOverrides) { + if !metadata.Empty() { + o.modelFallbackAttempts = append(o.modelFallbackAttempts, metadata) + } + } +} + // resolveSpan returns (model, provider) applying any overrides on top of agent defaults. func (l *Loop) resolveSpan(opts []spanOption) (string, string) { o := l.resolveSpanOverrides(opts) @@ -165,6 +174,12 @@ func (l *Loop) emitLLMSpanEnd(ctx context.Context, spanID uuid.UUID, start time. } var spanMetadata json.RawMessage spanOpts := l.resolveSpanOverrides(opts) + if spanOpts.provider != "" { + updates["provider"] = spanOpts.provider + } + if spanOpts.model != "" { + updates["model"] = spanOpts.model + } if callErr != nil { updates["status"] = store.SpanStatusError @@ -220,6 +235,9 @@ func (l *Loop) emitLLMSpanEnd(ctx context.Context, spanID uuid.UUID, start time. if len(spanOpts.usageCapAttempts) > 0 { spanMetadata = usagecaps.MergeTraceMetadata(spanMetadata, spanOpts.usageCapAttempts) } + if len(spanOpts.modelFallbackAttempts) > 0 { + spanMetadata = providers.MergeModelFallbackMetadata(spanMetadata, spanOpts.modelFallbackAttempts) + } if len(spanMetadata) > 0 { updates["metadata"] = spanMetadata } diff --git a/internal/providers/error_classify.go b/internal/providers/error_classify.go index 8e5838ad..44ff5bd3 100644 --- a/internal/providers/error_classify.go +++ b/internal/providers/error_classify.go @@ -15,6 +15,7 @@ const ( FailoverFormat FailoverReason = "format" FailoverRateLimit FailoverReason = "rate_limit" FailoverOverloaded FailoverReason = "overloaded" + FailoverServerError FailoverReason = "server_error" FailoverBilling FailoverReason = "billing" FailoverTimeout FailoverReason = "timeout" FailoverModelNotFound FailoverReason = "model_not_found" @@ -101,6 +102,7 @@ func (c *DefaultClassifier) Classify(err error, statusCode int, body string) Fai if containsAny(lower, "overload", "capacity", "too many") { return classifyReason(FailoverOverloaded) } + return classifyReason(FailoverServerError) } // Body pattern matching for specific error types @@ -110,7 +112,7 @@ func (c *DefaultClassifier) Classify(err error, statusCode int, body string) Fai if containsAny(lower, "tool_call", "function_call", "invalid_request") && statusCode == 400 { return classifyReason(FailoverFormat) } - if containsAny(lower, "data_inspection_failed", "inappropriate content", "content_policy_violation") && statusCode != 0 { + if isContentPolicyRefusal(lower, statusCode) { return classifyReason(FailoverContentPolicy) } @@ -140,6 +142,10 @@ func ClassifyHTTPError(classifier ErrorClassifier, err error) FailoverClassifica if errors.As(err, &httpErr) { return classifier.Classify(err, httpErr.Status, httpErr.Body) } + classification := classifier.Classify(err, 0, err.Error()) + if classification.Kind == "context_overflow" || classification.Reason != FailoverUnknown { + return classification + } // Non-HTTP error — check for network errors if isNetworkError(err) { return classifyReason(FailoverTimeout) @@ -185,6 +191,16 @@ func isNetworkError(err error) bool { return containsAny(s, "connection reset", "broken pipe", "EOF", "timeout", "ECONNREFUSED") } +func isContentPolicyRefusal(lower string, statusCode int) bool { + if statusCode != 0 && containsAny(lower, "data_inspection_failed", "inappropriate content", "content_policy_violation") { + return true + } + if containsAny(lower, "limited access to this content for safety reasons", "content for safety reasons") { + return true + } + return strings.Contains(lower, "invalid prompt") && strings.Contains(lower, "safety") +} + // containsAny returns true if s contains any of the substrings. func containsAny(s string, subs ...string) bool { for _, sub := range subs { diff --git a/internal/providers/error_classify_test.go b/internal/providers/error_classify_test.go index fdd3b2ed..bb9c3ab7 100644 --- a/internal/providers/error_classify_test.go +++ b/internal/providers/error_classify_test.go @@ -98,8 +98,8 @@ func TestClassifyHTTP500WithCapacity(t *testing.T) { func TestClassifyHTTP502BadGateway(t *testing.T) { classifier := NewDefaultClassifier() result := classifier.Classify(nil, 502, "Bad gateway") - if result.Reason != FailoverUnknown { - t.Errorf("expected FailoverUnknown for 502, got %s", result.Reason) + if result.Reason != FailoverServerError { + t.Errorf("expected FailoverServerError for 502, got %s", result.Reason) } } @@ -245,6 +245,15 @@ func TestClassifyHTTPErrorNonHTTPError(t *testing.T) { } } +func TestClassifyHTTPErrorCodexSafetyRefusalString(t *testing.T) { + classifier := NewDefaultClassifier() + err := errors.New("codex: response failed: Invalid prompt: we've limited access to this content for safety reasons") + result := ClassifyHTTPError(classifier, err) + if result.Reason != FailoverContentPolicy { + t.Errorf("expected FailoverContentPolicy, got %s", result.Reason) + } +} + func TestClassifyHTTPErrorNil(t *testing.T) { classifier := NewDefaultClassifier() result := ClassifyHTTPError(classifier, nil) @@ -328,8 +337,8 @@ func TestClassifyLowercaseInsensitive(t *testing.T) { func TestClassifyUnknownError(t *testing.T) { classifier := NewDefaultClassifier() result := classifier.Classify(nil, 500, "Unknown server error") - if result.Reason != FailoverUnknown { - t.Errorf("expected FailoverUnknown, got %s", result.Reason) + if result.Reason != FailoverServerError { + t.Errorf("expected FailoverServerError, got %s", result.Reason) } } diff --git a/internal/providers/failover.go b/internal/providers/failover.go index c5ec2e81..c923bbfa 100644 --- a/internal/providers/failover.go +++ b/internal/providers/failover.go @@ -46,7 +46,7 @@ func (e *FailoverSummaryError) Error() string { // isProfileRotatable returns true for transient errors where rotating API key/profile may help. func isProfileRotatable(reason FailoverReason) bool { switch reason { - case FailoverRateLimit, FailoverOverloaded, FailoverTimeout, FailoverAuth: + case FailoverRateLimit, FailoverOverloaded, FailoverServerError, FailoverTimeout, FailoverAuth: return true } return false diff --git a/internal/providers/model_fallback_observation.go b/internal/providers/model_fallback_observation.go new file mode 100644 index 00000000..decfd37b --- /dev/null +++ b/internal/providers/model_fallback_observation.go @@ -0,0 +1,57 @@ +package providers + +import ( + "encoding/json" +) + +const ModelFallbackMetadataKey = "model_fallback" + +type ModelFallbackAttemptMetadata struct { + ProviderName string `json:"provider_name,omitempty"` + Model string `json:"model,omitempty"` + Status string `json:"status,omitempty"` + Reason string `json:"reason,omitempty"` + Error string `json:"error,omitempty"` + Streamed bool `json:"streamed,omitempty"` +} + +func (m ModelFallbackAttemptMetadata) Empty() bool { + return m.ProviderName == "" && m.Model == "" && m.Status == "" && + m.Reason == "" && m.Error == "" && !m.Streamed +} + +func MergeModelFallbackMetadata(existing json.RawMessage, entries []ModelFallbackAttemptMetadata) json.RawMessage { + clean := make([]ModelFallbackAttemptMetadata, 0, len(entries)) + selectedProvider := "" + selectedModel := "" + for _, entry := range entries { + if entry.Empty() { + continue + } + clean = append(clean, entry) + if entry.Status == "success" { + selectedProvider = entry.ProviderName + selectedModel = entry.Model + } + } + if len(clean) == 0 { + return existing + } + payload := map[string]any{} + if len(existing) > 0 { + _ = json.Unmarshal(existing, &payload) + } + section := map[string]any{"attempts": clean} + if selectedProvider != "" { + section["selected_provider_name"] = selectedProvider + } + if selectedModel != "" { + section["selected_model"] = selectedModel + } + payload[ModelFallbackMetadataKey] = section + data, err := json.Marshal(payload) + if err != nil { + return existing + } + return json.RawMessage(data) +} diff --git a/internal/providers/model_fallback_observation_test.go b/internal/providers/model_fallback_observation_test.go new file mode 100644 index 00000000..916dacc7 --- /dev/null +++ b/internal/providers/model_fallback_observation_test.go @@ -0,0 +1,36 @@ +package providers + +import ( + "encoding/json" + "testing" +) + +func TestMergeModelFallbackMetadataPreservesExistingSections(t *testing.T) { + existing := json.RawMessage(`{"reasoning":{"effort":"high"}}`) + merged := MergeModelFallbackMetadata(existing, []ModelFallbackAttemptMetadata{ + {ProviderName: "primary", Model: "primary-model", Status: "error", Reason: string(FailoverContentPolicy), Error: "refused"}, + {ProviderName: "backup", Model: "backup-model", Status: "success"}, + }) + + var payload map[string]json.RawMessage + if err := json.Unmarshal(merged, &payload); err != nil { + t.Fatalf("metadata JSON invalid: %v", err) + } + if len(payload["reasoning"]) == 0 { + t.Fatal("existing reasoning section missing") + } + var fallback struct { + Attempts []ModelFallbackAttemptMetadata `json:"attempts"` + SelectedProviderName string `json:"selected_provider_name"` + SelectedModel string `json:"selected_model"` + } + if err := json.Unmarshal(payload[ModelFallbackMetadataKey], &fallback); err != nil { + t.Fatalf("model fallback section invalid: %v", err) + } + if len(fallback.Attempts) != 2 { + t.Fatalf("attempts = %d, want 2", len(fallback.Attempts)) + } + if fallback.SelectedProviderName != "backup" || fallback.SelectedModel != "backup-model" { + t.Fatalf("selected = %s/%s, want backup/backup-model", fallback.SelectedProviderName, fallback.SelectedModel) + } +} diff --git a/internal/providers/model_fallback_test.go b/internal/providers/model_fallback_test.go index d550e49e..d8262e80 100644 --- a/internal/providers/model_fallback_test.go +++ b/internal/providers/model_fallback_test.go @@ -206,6 +206,33 @@ func TestModelFallbackProviderContinuesAfterContentPolicyFallback(t *testing.T) } } +func TestModelFallbackProviderFallsBackOnCodexSafetyRefusalString(t *testing.T) { + primary := &testFallbackProvider{ + name: "codex-digitop", + model: "gpt-5.5", + err: errors.New("codex: response failed: Invalid prompt: we've limited access to this content for safety reasons"), + } + backup := &testFallbackProvider{name: "anthropic", model: "claude-sonnet-4-5"} + provider := NewModelFallbackProvider(FallbackCandidate{ + ProviderName: "codex-digitop", + Provider: primary, + Model: "gpt-5.5", + }, []FallbackCandidate{ + {ProviderName: "anthropic", Provider: backup, Model: "claude-sonnet-4-5"}, + }, 2, false) + + resp, err := provider.Chat(context.Background(), ChatRequest{}) + if err != nil { + t.Fatalf("Chat() error = %v", err) + } + if resp.Content != "claude-sonnet-4-5" { + t.Fatalf("Chat() content = %q, want fallback model response", resp.Content) + } + if primary.calls != 1 || backup.calls != 1 { + t.Fatalf("calls primary=%d backup=%d, want 1/1", primary.calls, backup.calls) + } +} + func TestModelFallbackProviderMaxAttemptsCapsTotalAttempts(t *testing.T) { primary := &testFallbackProvider{ name: "primary",