diff --git a/apps/backend/agents/tools_pkg/models.py b/apps/backend/agents/tools_pkg/models.py index 46671dfa..069eb322 100644 --- a/apps/backend/agents/tools_pkg/models.py +++ b/apps/backend/agents/tools_pkg/models.py @@ -292,6 +292,14 @@ AGENT_CONFIGS = { "auto_claude_tools": [], "thinking_default": "high", }, + "pr_followup_extraction": { + # Lightweight extraction call for recovering data when structured output fails + # Pure structured output extraction, no tools needed + "tools": [], + "mcp_servers": [], + "auto_claude_tools": [], + "thinking_default": "low", + }, "pr_finding_validator": { # Standalone validator for re-checking findings against actual code # Called separately from orchestrator to validate findings with fresh context diff --git a/apps/backend/runners/github/services/parallel_followup_reviewer.py b/apps/backend/runners/github/services/parallel_followup_reviewer.py index b5e11652..b49a76f3 100644 --- a/apps/backend/runners/github/services/parallel_followup_reviewer.py +++ b/apps/backend/runners/github/services/parallel_followup_reviewer.py @@ -51,7 +51,7 @@ try: from .category_utils import map_category from .io_utils import safe_print from .pr_worktree_manager import PRWorktreeManager - from .pydantic_models import ParallelFollowupResponse + from .pydantic_models import FollowupExtractionResponse, ParallelFollowupResponse from .sdk_utils import process_sdk_stream except (ImportError, ValueError, SystemError): from context_gatherer import _validate_git_ref @@ -75,7 +75,10 @@ except (ImportError, ValueError, SystemError): from services.category_utils import map_category from services.io_utils import safe_print from services.pr_worktree_manager import PRWorktreeManager - from services.pydantic_models import ParallelFollowupResponse + from services.pydantic_models import ( + FollowupExtractionResponse, + ParallelFollowupResponse, + ) from services.sdk_utils import process_sdk_stream @@ -576,16 +579,36 @@ The SDK will run invoked agents in parallel automatically. ) # Check for stream processing errors - if stream_result.get("error"): - logger.error( - f"[ParallelFollowup] SDK stream failed: {stream_result['error']}" - ) - raise RuntimeError( - f"SDK stream processing failed: {stream_result['error']}" - ) + stream_error = stream_result.get("error") + if stream_error: + if stream_result.get("error_recoverable"): + # Recoverable error — attempt extraction call fallback + logger.warning( + f"[ParallelFollowup] Recoverable error: {stream_error}. " + f"Attempting extraction call fallback." + ) + safe_print( + f"[ParallelFollowup] WARNING: {stream_error} — " + f"attempting recovery with minimal extraction...", + flush=True, + ) + else: + # Fatal error — raise as before + logger.error( + f"[ParallelFollowup] SDK stream failed: {stream_error}" + ) + raise RuntimeError( + f"SDK stream processing failed: {stream_error}" + ) result_text = stream_result["result_text"] - structured_output = stream_result["structured_output"] + last_assistant_text = stream_result.get("last_assistant_text", "") + # Nullify structured output on recoverable errors to force Tier 2 fallback + structured_output = ( + None + if (stream_error and stream_result.get("error_recoverable")) + else stream_result["structured_output"] + ) agents_invoked = stream_result["agents_invoked"] msg_count = stream_result["msg_count"] @@ -596,22 +619,28 @@ The SDK will run invoked agents in parallel automatically. pr_number=context.pr_number, ) - # Parse findings from output + # Parse findings from output (three-tier recovery cascade) if structured_output: result_data = self._parse_structured_output(structured_output, context) else: - # Log when structured output is missing - this shouldn't happen normally - # when output_format is configured, so it indicates a problem + # Structured output missing or validation failed. + # Tier 2: Attempt extraction call with minimal schema logger.warning( - "[ParallelFollowup] No structured output received from SDK - " - "falling back to text parsing. Resolution data may be incomplete." + "[ParallelFollowup] No structured output — attempting extraction call" ) - safe_print( - "[ParallelFollowup] WARNING: Structured output not captured, " - "using text fallback (resolution tracking may be incomplete)", - flush=True, + # Use last_assistant_text (cleaner) if available, fall back to full transcript + fallback_text = last_assistant_text or result_text + result_data = await self._attempt_extraction_call( + fallback_text, context ) - result_data = self._parse_text_output(result_text, context) + if result_data is None: + # Tier 3: Fall back to basic text parsing + safe_print( + "[ParallelFollowup] WARNING: Extraction call failed, " + "using text fallback (resolution tracking may be incomplete)", + flush=True, + ) + result_data = self._parse_text_output(result_text, context) # Extract data findings = result_data.get("findings", []) @@ -730,7 +759,9 @@ The SDK will run invoked agents in parallel automatically. blockers.append(f"{finding.category.value}: {finding.title}") # Extract validation counts - dismissed_count = len(result_data.get("dismissed_false_positive_ids", [])) + dismissed_count = len( + result_data.get("dismissed_false_positive_ids", []) + ) or result_data.get("dismissed_finding_count", 0) confirmed_count = result_data.get("confirmed_valid_count", 0) needs_human_count = result_data.get("needs_human_review_count", 0) @@ -1074,17 +1105,129 @@ The SDK will run invoked agents in parallel automatically. elif "needs revision" in text_lower or "request changes" in text_lower: verdict = MergeVerdict.NEEDS_REVISION else: - verdict = MergeVerdict.MERGE_WITH_CHANGES + verdict = MergeVerdict.NEEDS_REVISION return { "findings": findings, "resolved_ids": [], "unresolved_ids": [], "new_finding_ids": [], + "dismissed_false_positive_ids": [], + "confirmed_valid_count": 0, + "dismissed_finding_count": 0, + "needs_human_review_count": 0, "verdict": verdict, "verdict_reasoning": text[:500] if text else "Unable to parse response", + "agents_invoked": [], } + async def _attempt_extraction_call( + self, text: str, context: FollowupReviewContext + ) -> dict | None: + """Attempt a short SDK call with a minimal schema to recover review data. + + This is the Tier 2 recovery step when full structured output validation fails. + Uses FollowupExtractionResponse (~6 flat fields) which has near-100% success rate. + + Returns parsed result dict on success, None on failure. + """ + if not text or not text.strip(): + logger.warning("[ParallelFollowup] No text available for extraction call") + return None + + try: + safe_print( + "[ParallelFollowup] Attempting recovery with minimal extraction schema...", + flush=True, + ) + + extraction_prompt = ( + "Extract the key review data from the following AI analysis output. " + "Return the verdict, reasoning, resolved finding IDs, unresolved finding IDs, " + "one-line summaries of any new findings, and counts of confirmed/dismissed findings.\n\n" + f"--- AI ANALYSIS OUTPUT ---\n{text[:8000]}\n--- END ---" + ) + + model_shorthand = self.config.model or "sonnet" + model = resolve_model_id(model_shorthand) + + extraction_client = create_client( + project_dir=self.project_dir, + spec_dir=self.github_dir, + model=model, + agent_type="pr_followup_extraction", + fast_mode=self.config.fast_mode, + output_format={ + "type": "json_schema", + "schema": FollowupExtractionResponse.model_json_schema(), + }, + ) + + async with extraction_client: + await extraction_client.query(extraction_prompt) + + stream_result = await process_sdk_stream( + client=extraction_client, + context_name="FollowupExtraction", + model=model, + system_prompt=extraction_prompt, + max_messages=20, + ) + + if stream_result.get("error"): + logger.warning( + f"[ParallelFollowup] Extraction call also failed: {stream_result['error']}" + ) + return None + + extraction_output = stream_result.get("structured_output") + if not extraction_output: + logger.warning( + "[ParallelFollowup] Extraction call returned no structured output" + ) + return None + + # Parse the minimal extraction response + extracted = FollowupExtractionResponse.model_validate(extraction_output) + + # Map verdict string to MergeVerdict enum + verdict_map = { + "READY_TO_MERGE": MergeVerdict.READY_TO_MERGE, + "MERGE_WITH_CHANGES": MergeVerdict.MERGE_WITH_CHANGES, + "NEEDS_REVISION": MergeVerdict.NEEDS_REVISION, + "BLOCKED": MergeVerdict.BLOCKED, + } + verdict = verdict_map.get(extracted.verdict, MergeVerdict.NEEDS_REVISION) + + safe_print( + f"[ParallelFollowup] Extraction recovered: verdict={extracted.verdict}, " + f"{len(extracted.resolved_finding_ids)} resolved, " + f"{len(extracted.new_finding_summaries)} new findings", + flush=True, + ) + + return { + "findings": [], # Full findings not recoverable via extraction + "resolved_ids": extracted.resolved_finding_ids, + "unresolved_ids": extracted.unresolved_finding_ids, + "new_finding_ids": [], + "dismissed_false_positive_ids": [], + "confirmed_valid_count": extracted.confirmed_finding_count, + "dismissed_finding_count": extracted.dismissed_finding_count, + "needs_human_review_count": 0, + "verdict": verdict, + "verdict_reasoning": f"[Recovered via extraction] {extracted.verdict_reasoning}", + "agents_invoked": [], + } + + except Exception as e: + logger.warning(f"[ParallelFollowup] Extraction call failed: {e}") + safe_print( + f"[ParallelFollowup] Extraction call failed: {e}", + flush=True, + ) + return None + def _create_empty_result(self) -> dict: """Create empty result structure.""" return { @@ -1092,8 +1235,13 @@ The SDK will run invoked agents in parallel automatically. "resolved_ids": [], "unresolved_ids": [], "new_finding_ids": [], + "dismissed_false_positive_ids": [], + "confirmed_valid_count": 0, + "dismissed_finding_count": 0, + "needs_human_review_count": 0, "verdict": MergeVerdict.NEEDS_REVISION, "verdict_reasoning": "Unable to parse review results", + "agents_invoked": [], } def _extract_partial_data(self, data: dict) -> dict | None: diff --git a/apps/backend/runners/github/services/parallel_orchestrator_reviewer.py b/apps/backend/runners/github/services/parallel_orchestrator_reviewer.py index 0c76cf56..81609e6d 100644 --- a/apps/backend/runners/github/services/parallel_orchestrator_reviewer.py +++ b/apps/backend/runners/github/services/parallel_orchestrator_reviewer.py @@ -1785,6 +1785,7 @@ For EACH finding above: or "concurrency" in error_str or "circuit breaker" in error_str or "tool_use" in error_str + or "structured_output" in error_str ) if is_retryable and attempt < MAX_VALIDATION_RETRIES: diff --git a/apps/backend/runners/github/services/pydantic_models.py b/apps/backend/runners/github/services/pydantic_models.py index 10054230..68b403f2 100644 --- a/apps/backend/runners/github/services/pydantic_models.py +++ b/apps/backend/runners/github/services/pydantic_models.py @@ -710,3 +710,39 @@ class FindingValidationResponse(BaseModel): "how many dismissed, how many need human review" ) ) + + +# ============================================================================= +# Minimal Extraction Schema (Fallback for structured output validation failure) +# ============================================================================= + + +class FollowupExtractionResponse(BaseModel): + """Minimal extraction schema for recovering data when full structured output fails. + + Deliberately kept small (~6 fields, no nesting) for near-100% validation success. + Used as an intermediate recovery step before falling back to raw text parsing. + """ + + verdict: Literal[ + "READY_TO_MERGE", "MERGE_WITH_CHANGES", "NEEDS_REVISION", "BLOCKED" + ] = Field(description="Overall merge verdict") + verdict_reasoning: str = Field(description="Explanation for the verdict") + resolved_finding_ids: list[str] = Field( + default_factory=list, + description="IDs of previous findings that are now resolved", + ) + unresolved_finding_ids: list[str] = Field( + default_factory=list, + description="IDs of previous findings that remain unresolved", + ) + new_finding_summaries: list[str] = Field( + default_factory=list, + description="One-line summary of each new finding (e.g. 'HIGH: cleanup deletes QA-rejected specs in batch_commands.py')", + ) + confirmed_finding_count: int = Field( + 0, description="Number of findings confirmed as valid" + ) + dismissed_finding_count: int = Field( + 0, description="Number of findings dismissed as false positives" + ) diff --git a/apps/backend/runners/github/services/sdk_utils.py b/apps/backend/runners/github/services/sdk_utils.py index 1c277f76..23fe632c 100644 --- a/apps/backend/runners/github/services/sdk_utils.py +++ b/apps/backend/runners/github/services/sdk_utils.py @@ -133,6 +133,13 @@ def _get_tool_detail(tool_name: str, tool_input: dict[str, Any]) -> str: # Prevents runaway retry loops from consuming unbounded resources MAX_MESSAGE_COUNT = 500 +# Errors that are recoverable (callers can fall back to text parsing or retry) +# vs fatal errors (auth failures, circuit breaker) that should propagate +RECOVERABLE_ERRORS = { + "structured_output_validation_failed", + "tool_use_concurrency_error", +} + # Abort after 1 consecutive repeat (2 total identical responses). # Low threshold catches error loops quickly (e.g., auth errors returned as AI text). # Normal AI responses never produce the exact same text block twice in a row. @@ -261,8 +268,11 @@ async def process_sdk_stream( - msg_count: Total message count - subagent_tool_ids: Mapping of tool_id -> agent_name - error: Error message if stream processing failed (None on success) + - error_recoverable: Boolean indicating if the error is recoverable (fallback possible) vs fatal + - last_assistant_text: Last non-empty assistant text block (for cleaner fallback parsing) """ result_text = "" + last_assistant_text = "" # Last assistant text block (for cleaner fallback parsing) structured_output = None agents_invoked = [] msg_count = 0 @@ -481,6 +491,9 @@ async def process_sdk_stream( block_type = type(block).__name__ if block_type == "TextBlock" and hasattr(block, "text"): result_text += block.text + # Track last non-empty text for fallback parsing + if block.text.strip(): + last_assistant_text = block.text # Check for auth/access error returned as AI response text. # Note: break exits this inner for-loop over msg.content; # the outer message loop exits via `if stream_error: break`. @@ -647,11 +660,16 @@ async def process_sdk_stream( f"[{context_name}] Tool use concurrency error detected - caller should retry" ) + # Categorize error as recoverable (fallback possible) vs fatal + error_recoverable = stream_error in RECOVERABLE_ERRORS if stream_error else False + return { "result_text": result_text, + "last_assistant_text": last_assistant_text, "structured_output": structured_output, "agents_invoked": agents_invoked, "msg_count": msg_count, "subagent_tool_ids": subagent_tool_ids, "error": stream_error, + "error_recoverable": error_recoverable, } diff --git a/tests/test_structured_output_recovery.py b/tests/test_structured_output_recovery.py new file mode 100644 index 00000000..bc02e32c --- /dev/null +++ b/tests/test_structured_output_recovery.py @@ -0,0 +1,145 @@ +""" +Tests for Structured Output Recovery +====================================== + +Tests the three-tier recovery cascade when structured output validation fails: +1. FollowupExtractionResponse model validation +2. Error categorization imported from sdk_utils +3. Agent config registration for pr_followup_extraction +""" + +import json +import sys +from pathlib import Path + +import pytest + +# Add paths for imports — conftest.py adds apps/backend, but there's a +# services/ package at both apps/backend/services/ and runners/github/services/. +# To avoid collision, add the github services dir directly and import bare module names. +_backend_dir = Path(__file__).parent.parent / "apps" / "backend" +_github_services_dir = _backend_dir / "runners" / "github" / "services" +if str(_backend_dir) not in sys.path: + sys.path.insert(0, str(_backend_dir)) +if str(_github_services_dir) not in sys.path: + sys.path.insert(0, str(_github_services_dir)) + +from agents.tools_pkg.models import AGENT_CONFIGS +from pydantic_models import ( + FollowupExtractionResponse, + ParallelFollowupResponse, +) +from sdk_utils import RECOVERABLE_ERRORS + + +# ============================================================================ +# Test FollowupExtractionResponse model +# ============================================================================ + + +class TestFollowupExtractionResponse: + """Tests for the minimal extraction schema.""" + + def test_minimal_valid_response(self): + """Accepts minimal response with just verdict and reasoning.""" + resp = FollowupExtractionResponse( + verdict="NEEDS_REVISION", + verdict_reasoning="Found issues that need fixing", + ) + assert resp.verdict == "NEEDS_REVISION" + assert resp.resolved_finding_ids == [] + assert resp.new_finding_summaries == [] + assert resp.confirmed_finding_count == 0 + assert resp.dismissed_finding_count == 0 + + def test_full_valid_response(self): + """Accepts fully populated response.""" + resp = FollowupExtractionResponse( + verdict="READY_TO_MERGE", + verdict_reasoning="All findings resolved", + resolved_finding_ids=["NCR-001", "NCR-002"], + unresolved_finding_ids=[], + new_finding_summaries=["HIGH: potential cleanup issue in batch_commands.py"], + confirmed_finding_count=1, + dismissed_finding_count=1, + ) + assert len(resp.resolved_finding_ids) == 2 + assert len(resp.new_finding_summaries) == 1 + assert resp.confirmed_finding_count == 1 + + def test_schema_is_small(self): + """Schema should be significantly smaller than ParallelFollowupResponse.""" + extraction_schema = json.dumps( + FollowupExtractionResponse.model_json_schema() + ) + followup_schema = json.dumps( + ParallelFollowupResponse.model_json_schema() + ) + # Extraction schema should be less than half the size of the full schema + assert len(extraction_schema) < len(followup_schema) / 2, ( + f"Extraction schema ({len(extraction_schema)} chars) should be " + f"less than half of full schema ({len(followup_schema)} chars)" + ) + + def test_all_verdict_values_accepted(self): + """All four verdict values should be accepted.""" + for verdict in ["READY_TO_MERGE", "MERGE_WITH_CHANGES", "NEEDS_REVISION", "BLOCKED"]: + resp = FollowupExtractionResponse( + verdict=verdict, + verdict_reasoning=f"Test {verdict}", + ) + assert resp.verdict == verdict + + +# ============================================================================ +# Test error categorization using the actual RECOVERABLE_ERRORS from sdk_utils +# ============================================================================ + + +class TestErrorCategorization: + """Tests that sdk_utils RECOVERABLE_ERRORS constant classifies errors correctly.""" + + def test_structured_output_error_is_recoverable(self): + """structured_output_validation_failed should be in RECOVERABLE_ERRORS.""" + assert "structured_output_validation_failed" in RECOVERABLE_ERRORS + + def test_concurrency_error_is_recoverable(self): + """tool_use_concurrency_error should be in RECOVERABLE_ERRORS.""" + assert "tool_use_concurrency_error" in RECOVERABLE_ERRORS + + def test_auth_error_is_fatal(self): + """Auth errors should NOT be in RECOVERABLE_ERRORS.""" + assert "Authentication error detected in AI response: please login again" not in RECOVERABLE_ERRORS + + def test_circuit_breaker_is_fatal(self): + """Circuit breaker errors should NOT be in RECOVERABLE_ERRORS.""" + for error in RECOVERABLE_ERRORS: + assert "circuit breaker" not in error.lower() + + def test_none_is_not_recoverable(self): + """None should not be in RECOVERABLE_ERRORS.""" + assert None not in RECOVERABLE_ERRORS + + +# ============================================================================ +# Test agent config registration +# ============================================================================ + + +class TestAgentConfigRegistration: + """Tests that pr_followup_extraction agent type is registered.""" + + def test_extraction_agent_type_registered(self): + """pr_followup_extraction must exist in AGENT_CONFIGS.""" + assert "pr_followup_extraction" in AGENT_CONFIGS + + def test_extraction_agent_needs_no_tools(self): + """Extraction agent should have no tools (pure structured output).""" + config = AGENT_CONFIGS["pr_followup_extraction"] + assert config["tools"] == [] + assert config["mcp_servers"] == [] + + def test_extraction_agent_low_thinking(self): + """Extraction agent should use low thinking (lightweight call).""" + config = AGENT_CONFIGS["pr_followup_extraction"] + assert config["thinking_default"] == "low"