fix(pipeline): recover truncated codex runs (#126)

* fix(pipeline): recover empty truncated codex responses

* fix(providers): continue fallback after content policy errors

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**

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{

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

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 {

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")

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

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",