diff --git a/docs/project-changelog.md b/docs/project-changelog.md index 55e1b23e..616a38f8 100644 --- a/docs/project-changelog.md +++ b/docs/project-changelog.md @@ -6,6 +6,33 @@ Significant changes, features, and fixes in reverse chronological order. ## 2026-05-31 +### Provider fallback content-policy recovery + +**Fixes** + +- Classifies provider content-policy rejections such as DashScope `data_inspection_failed` so model fallback can continue to the next configured candidate instead of stopping on `unknown`. +- Prevents Telegram runs from aborting silently when one fallback provider rejects a long/private history but another configured fallback remains available. + +**Tests** + +- Added provider classifier and model fallback coverage for continuing past a content-policy fallback failure. + +--- + +### Empty Codex OAuth truncation recovery + +**Fixes** + +- Detects successful LLM calls that still return `finish_reason=length` with no text/tool/image output, the observed failure mode when long Codex OAuth runs spend the remaining budget on reasoning. +- Runs emergency history compaction and retries instead of finalizing empty assistant content into repeated `"..."` replies. +- Preserves configured/model context windows so future provider context upgrades do not require code changes. + +**Tests** + +- Added ThinkStage coverage for empty length responses, retry compaction, usage accounting, and repeated failure handling. + +--- + ### Agent Access git credential follow-up (issue #117) **Fixes** diff --git a/internal/pipeline/stages_test.go b/internal/pipeline/stages_test.go index a868c476..39185e10 100644 --- a/internal/pipeline/stages_test.go +++ b/internal/pipeline/stages_test.go @@ -500,6 +500,83 @@ func TestThinkStage_ContextOverflow_TriggersCompaction(t *testing.T) { } } +func TestThinkStage_EmptyLengthResponse_TriggersCompaction(t *testing.T) { + t.Parallel() + compacted := false + + deps := &PipelineDeps{ + Config: PipelineConfig{MaxIterations: 10, MaxTokens: 1000}, + CallLLM: func(_ context.Context, _ *RunState, _ providers.ChatRequest) (*providers.ChatResponse, error) { + return &providers.ChatResponse{ + FinishReason: "length", + Usage: &providers.Usage{ + PromptTokens: 271_343, + CompletionTokens: 203, + TotalTokens: 271_546, + ThinkingTokens: 203, + }, + }, nil + }, + CompactMessages: func(_ context.Context, _ []providers.Message, _ string) ([]providers.Message, error) { + compacted = true + return []providers.Message{{Role: "user", Content: "[Summary]"}}, nil + }, + } + + stage := NewThinkStage(deps) + state := defaultState() + state.Messages.SetHistory([]providers.Message{{Role: "user", Content: "long history"}}) + + err := stage.Execute(context.Background(), state) + if err != nil { + t.Fatalf("Execute() error: %v", err) + } + if !compacted { + t.Fatal("expected empty length response to trigger compaction") + } + if state.Think.LastResponse != nil { + t.Fatal("LastResponse should stay nil so ObserveStage does not finalize empty content") + } + if state.Think.OverflowRetries != 1 { + t.Errorf("OverflowRetries = %d, want 1", state.Think.OverflowRetries) + } + if state.Think.TotalUsage.TotalTokens != 271_546 { + t.Errorf("TotalTokens = %d, want failed-call usage recorded", state.Think.TotalUsage.TotalTokens) + } + if got := state.Messages.History(); len(got) != 1 || got[0].Content != "[Summary]" { + t.Fatalf("history after compaction = %#v", got) + } + if stage.Result() != Continue { + t.Errorf("Result() = %v, want Continue", stage.Result()) + } +} + +func TestThinkStage_EmptyLengthResponse_FailsAfterOneRetry(t *testing.T) { + t.Parallel() + + deps := &PipelineDeps{ + Config: PipelineConfig{MaxIterations: 10, MaxTokens: 1000}, + CallLLM: func(_ context.Context, _ *RunState, _ providers.ChatRequest) (*providers.ChatResponse, error) { + return &providers.ChatResponse{FinishReason: "length"}, nil + }, + CompactMessages: func(_ context.Context, _ []providers.Message, _ string) ([]providers.Message, error) { + return []providers.Message{{Role: "user", Content: "[Summary]"}}, nil + }, + } + + stage := NewThinkStage(deps) + state := defaultState() + state.Think.OverflowRetries = 1 + + err := stage.Execute(context.Background(), state) + if err == nil { + t.Fatal("expected error after repeated empty length response") + } + if !strings.Contains(err.Error(), "truncated before content after compaction") { + t.Fatalf("unexpected error: %v", err) + } +} + func TestThinkStage_ContextOverflow_FailsAfterOneRetry(t *testing.T) { t.Parallel() deps := &PipelineDeps{ diff --git a/internal/pipeline/think_stage.go b/internal/pipeline/think_stage.go index 0dd48135..d6714790 100644 --- a/internal/pipeline/think_stage.go +++ b/internal/pipeline/think_stage.go @@ -67,26 +67,12 @@ func (s *ThinkStage) Execute(ctx context.Context, state *RunState) error { if state.Think.OverflowRetries > 0 { return fmt.Errorf("context overflow after compaction: %w", err) } - state.Think.OverflowRetries++ - // Attempt emergency compaction - if s.deps.CompactMessages != nil { - originalLen := len(state.Messages.History()) - compacted, compactErr := s.deps.CompactMessages(ctx, state.Messages.History(), state.Model) - if compactErr == nil { - state.Messages.ReplaceHistory(compacted) - slog.Info("emergency_compaction_triggered", - "run_id", state.RunID, - "original_msgs", originalLen, - "compacted_msgs", len(compacted), - ) - return nil // Retry this iteration (Continue result) - } - slog.Warn("emergency_compaction_failed", "error", compactErr) + if s.tryEmergencyCompaction(ctx, state, "context_overflow_error") { + return nil // Retry this iteration (Continue result) } } return fmt.Errorf("llm call: %w", err) } - state.Think.LastResponse = resp // 5. Accumulate usage (including ThinkingTokens for reasoning models) if resp.Usage != nil { @@ -96,6 +82,18 @@ func (s *ThinkStage) Execute(ctx context.Context, state *RunState) error { state.Think.TotalUsage.ThinkingTokens += resp.Usage.ThinkingTokens } + if isEmptyLengthResponse(resp) { + if state.Think.OverflowRetries > 0 { + return fmt.Errorf("llm response truncated before content after compaction") + } + if s.tryEmergencyCompaction(ctx, state, "empty_length_response") { + return nil // Retry next iteration with compacted history. + } + return fmt.Errorf("llm response truncated before content") + } + + state.Think.LastResponse = resp + // 6. Handle truncation: retry when tool call args are truncated or malformed. // Gemini returns finish_reason="tool_calls" (not "length") even when the thinking // budget exhausted max_tokens before args could be emitted — detect via empty @@ -153,6 +151,36 @@ func (s *ThinkStage) Execute(ctx context.Context, state *RunState) error { return nil } +func (s *ThinkStage) tryEmergencyCompaction(ctx context.Context, state *RunState, reason string) bool { + state.Think.OverflowRetries++ + if s.deps.CompactMessages == nil { + return false + } + + originalLen := len(state.Messages.History()) + compacted, compactErr := s.deps.CompactMessages(ctx, state.Messages.History(), state.Model) + if compactErr != nil { + slog.Warn("emergency_compaction_failed", "reason", reason, "error", compactErr) + return false + } + state.Messages.ReplaceHistory(compacted) + slog.Info("emergency_compaction_triggered", + "run_id", state.RunID, + "reason", reason, + "original_msgs", originalLen, + "compacted_msgs", len(compacted), + ) + return true +} + +func isEmptyLengthResponse(resp *providers.ChatResponse) bool { + return resp != nil && + resp.FinishReason == "length" && + strings.TrimSpace(resp.Content) == "" && + len(resp.ToolCalls) == 0 && + len(resp.Images) == 0 +} + func (s *ThinkStage) emitToolIterationBlockReply(ctx context.Context, resp *providers.ChatResponse) { content := resp.Content source := protocol.BlockReplySourceLLMProgress diff --git a/internal/providers/error_classify.go b/internal/providers/error_classify.go index ec73e9ed..8e5838ad 100644 --- a/internal/providers/error_classify.go +++ b/internal/providers/error_classify.go @@ -18,6 +18,7 @@ const ( FailoverBilling FailoverReason = "billing" FailoverTimeout FailoverReason = "timeout" FailoverModelNotFound FailoverReason = "model_not_found" + FailoverContentPolicy FailoverReason = "content_policy" FailoverUnknown FailoverReason = "unknown" ) @@ -109,6 +110,9 @@ 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 { + return classifyReason(FailoverContentPolicy) + } // Provider-specific patterns for _, patterns := range c.providerPatterns { diff --git a/internal/providers/error_classify_test.go b/internal/providers/error_classify_test.go index 7ec52120..fdd3b2ed 100644 --- a/internal/providers/error_classify_test.go +++ b/internal/providers/error_classify_test.go @@ -285,6 +285,14 @@ func TestClassifyFormatInvalidRequest(t *testing.T) { } } +func TestClassifyContentPolicyDataInspectionFailed(t *testing.T) { + classifier := NewDefaultClassifier() + result := classifier.Classify(nil, 400, `{"error":{"code":"data_inspection_failed","message":"Input text data may contain inappropriate content."}}`) + if result.Reason != FailoverContentPolicy { + t.Errorf("expected FailoverContentPolicy, got %s", result.Reason) + } +} + func TestClassifyHTTP401WithExpired(t *testing.T) { classifier := NewDefaultClassifier() result := classifier.Classify(nil, 401, "Token has expired") diff --git a/internal/providers/failover.go b/internal/providers/failover.go index d2c18a74..c5ec2e81 100644 --- a/internal/providers/failover.go +++ b/internal/providers/failover.go @@ -55,7 +55,7 @@ func isProfileRotatable(reason FailoverReason) bool { // isModelFallbackRequired returns true for permanent errors requiring a different model. func isModelFallbackRequired(reason FailoverReason) bool { switch reason { - case FailoverAuthPermanent, FailoverBilling, FailoverFormat, FailoverModelNotFound: + case FailoverAuthPermanent, FailoverBilling, FailoverFormat, FailoverModelNotFound, FailoverContentPolicy: return true } return false diff --git a/internal/providers/model_fallback_test.go b/internal/providers/model_fallback_test.go index a891374d..d550e49e 100644 --- a/internal/providers/model_fallback_test.go +++ b/internal/providers/model_fallback_test.go @@ -173,6 +173,39 @@ func TestModelFallbackProviderDoesNotFallbackOnUnknownError(t *testing.T) { } } +func TestModelFallbackProviderContinuesAfterContentPolicyFallback(t *testing.T) { + primary := &testFallbackProvider{ + name: "primary", + model: "primary-model", + err: &HTTPError{Status: 429, Body: "rate limited"}, + } + blocked := &testFallbackProvider{ + name: "blocked", + model: "blocked-model", + err: &HTTPError{Status: 400, Body: `{"error":{"code":"data_inspection_failed","message":"Input text data may contain inappropriate content."}}`}, + } + backup := &testFallbackProvider{name: "backup", model: "backup-model"} + provider := NewModelFallbackProvider(FallbackCandidate{ + ProviderName: "primary", + Provider: primary, + Model: "primary-model", + }, []FallbackCandidate{ + {ProviderName: "blocked", Provider: blocked, Model: "blocked-model"}, + {ProviderName: "backup", Provider: backup, Model: "backup-model"}, + }, 0, false) + + resp, err := provider.Chat(context.Background(), ChatRequest{}) + if err != nil { + t.Fatalf("Chat() error = %v", err) + } + if resp.Content != "backup-model" { + t.Fatalf("Chat() content = %q, want backup model", resp.Content) + } + if primary.calls != 1 || blocked.calls != 1 || backup.calls != 1 { + t.Fatalf("calls primary=%d blocked=%d backup=%d, want 1/1/1", primary.calls, blocked.calls, backup.calls) + } +} + func TestModelFallbackProviderMaxAttemptsCapsTotalAttempts(t *testing.T) { primary := &testFallbackProvider{ name: "primary",