electron-rare/Aperant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(pr-review): reduce structured output failures and preserve findings in recovery (#1806)

Browse Source 
* fix(pr-review): reduce structured output failures and preserve findings in recovery

Simplify Pydantic schemas to prevent validation failures: make VerificationEvidence
optional, relax severity/category from Literal enums to str with field_validators,
remove deprecated evidence field, and clean up 15 unused legacy schemas.

Fix all recovery tiers to reconstruct findings instead of returning empty arrays:
Tier 2 now converts extraction summaries to PRReviewFinding objects and looks up
unresolved findings from previous review context. Tier 1.5 defensively extracts
individual findings from raw dicts. Added extraction recovery to followup_reviewer
and specialist sessions which previously had none.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(pr-review): address PR review findings - deduplicate, use create_client, add consistency

Extract duplicated severity-from-summary parsing into shared recovery_utils.py with
consistent prefixed ID generation (FR-/FU-). Use create_client() + process_sdk_stream()
instead of raw SDK query in followup_reviewer extraction. Add unresolved finding
reconstruction from previous review context. Add missing dismissed_finding_count key
to _extract_partial_data return dict.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(pr-review): remove duplicate unresolved finding reconstruction in extraction recovery

Unresolved findings were being added twice: once by reconstructing PRReviewFinding
objects directly, and again via finding_resolutions + _apply_ai_resolutions. Remove
the direct reconstruction so unresolved IDs are only handled through the resolution
pipeline.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Andy
2026-02-13 12:36:25 +01:00
committed by GitHub
parent 4d4234378f
commit f1b8cd3a7a
6 changed files with 703 additions and 600 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/runners/github/services/followup_reviewer.py
+150 -3
View File
@@ -25,6 +25,8 @@ if TYPE_CHECKING:
from ..models import FollowupReviewContext, GitHubRunnerConfig
try:
from ...core.client import create_client
from ...phase_config import resolve_model_id
from ..gh_client import GHClient
from ..models import (
MergeVerdict,
@@ -37,8 +39,11 @@ try:
from .category_utils import map_category
from .io_utils import safe_print
from .prompt_manager import PromptManager
from .pydantic_models import FollowupReviewResponse
from .pydantic_models import FollowupExtractionResponse, FollowupReviewResponse
from .recovery_utils import create_finding_from_summary
from .sdk_utils import process_sdk_stream
except (ImportError, ValueError, SystemError):
from core.client import create_client
from gh_client import GHClient
from models import (
MergeVerdict,
@@ -48,10 +53,16 @@ except (ImportError, ValueError, SystemError):
ReviewSeverity,
_utc_now_iso,
)
from phase_config import resolve_model_id
from services.category_utils import map_category
from services.io_utils import safe_print
from services.prompt_manager import PromptManager
from services.pydantic_models import FollowupReviewResponse
from services.pydantic_models import (
FollowupExtractionResponse,
FollowupReviewResponse,
)
from services.recovery_utils import create_finding_from_summary
from services.sdk_utils import process_sdk_stream
logger = logging.getLogger(__name__)
@@ -698,6 +709,9 @@ Analyze this follow-up review context and provide your structured response.
)
safe_print(f"[Followup] SDK query with output_format, model={model}")
# Capture assistant text for extraction fallback
captured_text = ""
# Iterate through messages from the query
# Note: max_turns=2 because structured output uses a tool call + response
async for message in query(
@@ -722,7 +736,9 @@ Analyze this follow-up review context and provide your structured response.
content = getattr(message, "content", [])
for block in content:
block_type = type(block).__name__
if block_type == "ToolUseBlock":
if block_type == "TextBlock":
captured_text += getattr(block, "text", "")
elif block_type == "ToolUseBlock":
tool_name = getattr(block, "name", "")
if tool_name == "StructuredOutput":
# Extract structured data from tool input
@@ -765,9 +781,31 @@ Analyze this follow-up review context and provide your structured response.
logger.warning(
"Claude could not produce valid structured output after retries"
)
# Attempt extraction call recovery before giving up
if captured_text:
safe_print(
"[Followup] Attempting extraction call recovery...",
flush=True,
)
extraction_result = await self._attempt_extraction_call(
captured_text, context
)
if extraction_result is not None:
return extraction_result
return None
logger.warning("No structured output received from AI")
# Attempt extraction call recovery before giving up
if captured_text:
safe_print(
"[Followup] No structured output — attempting extraction call recovery...",
flush=True,
)
extraction_result = await self._attempt_extraction_call(
captured_text, context
)
if extraction_result is not None:
return extraction_result
return None
except ValueError as e:
@@ -840,6 +878,115 @@ Analyze this follow-up review context and provide your structured response.
"verdict_reasoning": result.verdict_reasoning,
}
async def _attempt_extraction_call(
self,
text: str,
context: FollowupReviewContext,
) -> dict[str, Any] | None:
"""Attempt a short SDK call with minimal schema to recover review data.
This is the extraction recovery step when full structured output validation fails.
Uses FollowupExtractionResponse (~6 flat fields) which has near-100% success rate.
Uses create_client() + process_sdk_stream() for proper OAuth handling,
matching the pattern in parallel_followup_reviewer.py.
Returns parsed result dict on success, None on failure.
"""
if not text or not text.strip():
return None
try:
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",
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"[Followup] Extraction call also failed: {stream_result['error']}"
)
return None
extraction_output = stream_result.get("structured_output")
if not extraction_output:
logger.warning(
"[Followup] Extraction call returned no structured output"
)
return None
extracted = FollowupExtractionResponse.model_validate(extraction_output)
# Convert extraction to internal format with reconstructed findings
new_findings = []
for i, summary in enumerate(extracted.new_finding_summaries):
new_findings.append(
create_finding_from_summary(summary, i, id_prefix="FR")
)
# Build finding_resolutions from extraction data for _apply_ai_resolutions
# (unresolved findings are handled via finding_resolutions + _apply_ai_resolutions)
finding_resolutions = []
for fid in extracted.resolved_finding_ids:
finding_resolutions.append(
{"finding_id": fid, "status": "resolved", "resolution_notes": None}
)
for fid in extracted.unresolved_finding_ids:
finding_resolutions.append(
{
"finding_id": fid,
"status": "unresolved",
"resolution_notes": None,
}
)
safe_print(
f"[Followup] Extraction recovered: verdict={extracted.verdict}, "
f"{len(extracted.resolved_finding_ids)} resolved, "
f"{len(extracted.unresolved_finding_ids)} unresolved, "
f"{len(new_findings)} new findings",
flush=True,
)
return {
"finding_resolutions": finding_resolutions,
"new_findings": new_findings,
"comment_findings": [],
"verdict": extracted.verdict,
"verdict_reasoning": f"[Recovered via extraction] {extracted.verdict_reasoning}",
}
except Exception as e:
logger.warning(f"[Followup] Extraction call failed: {e}")
return None
def _apply_ai_resolutions(
self,
previous_findings: list[PRReviewFinding],
apps/backend/runners/github/services/parallel_followup_reviewer.py
+106 -13
View File
@@ -52,6 +52,7 @@ try:
from .io_utils import safe_print
from .pr_worktree_manager import PRWorktreeManager
from .pydantic_models import FollowupExtractionResponse, ParallelFollowupResponse
from .recovery_utils import create_finding_from_summary
from .sdk_utils import process_sdk_stream
except (ImportError, ValueError, SystemError):
from context_gatherer import _validate_git_ref
@@ -79,6 +80,7 @@ except (ImportError, ValueError, SystemError):
FollowupExtractionResponse,
ParallelFollowupResponse,
)
from services.recovery_utils import create_finding_from_summary
from services.sdk_utils import process_sdk_stream
@@ -1199,18 +1201,52 @@ The SDK will run invoked agents in parallel automatically.
}
verdict = verdict_map.get(extracted.verdict, MergeVerdict.NEEDS_REVISION)
# Reconstruct findings from extraction data
findings = []
new_finding_ids = []
# 1. Convert new_finding_summaries to minimal PRReviewFinding objects
# Uses shared helper for "SEVERITY: description" parsing and ID generation
for i, summary in enumerate(extracted.new_finding_summaries):
finding = create_finding_from_summary(summary, i, id_prefix="FU")
new_finding_ids.append(finding.id)
findings.append(finding)
# 2. Reconstruct unresolved findings from previous review context
if extracted.unresolved_finding_ids and context.previous_review.findings:
previous_map = {f.id: f for f in context.previous_review.findings}
for uid in extracted.unresolved_finding_ids:
original = previous_map.get(uid)
if original:
findings.append(
PRReviewFinding(
id=original.id,
severity=original.severity,
category=original.category,
title=f"[UNRESOLVED] {original.title}",
description=original.description,
file=original.file,
line=original.line,
suggested_fix=original.suggested_fix,
fixable=original.fixable,
is_impact_finding=original.is_impact_finding,
)
)
safe_print(
f"[ParallelFollowup] Extraction recovered: verdict={extracted.verdict}, "
f"{len(extracted.resolved_finding_ids)} resolved, "
f"{len(extracted.new_finding_summaries)} new findings",
f"{len(extracted.unresolved_finding_ids)} unresolved, "
f"{len(new_finding_ids)} new findings, "
f"{len(findings)} total findings reconstructed",
flush=True,
)
return {
"findings": [], # Full findings not recoverable via extraction
"findings": findings,
"resolved_ids": extracted.resolved_finding_ids,
"unresolved_ids": extracted.unresolved_finding_ids,
"new_finding_ids": [],
"new_finding_ids": new_finding_ids,
"dismissed_false_positive_ids": [],
"confirmed_valid_count": extracted.confirmed_finding_count,
"dismissed_finding_count": extracted.dismissed_finding_count,
@@ -1250,6 +1286,7 @@ The SDK will run invoked agents in parallel automatically.
This handles cases where the AI produced valid data but it doesn't exactly
match the expected schema (missing optional fields, type mismatches, etc.).
Defensively extracts findings from the raw dict so partial results are preserved.
"""
if not isinstance(data, dict):
return None
@@ -1257,6 +1294,7 @@ The SDK will run invoked agents in parallel automatically.
resolved_ids = []
unresolved_ids = []
new_finding_ids = []
findings = []
# Try to extract resolution verifications
resolution_verifications = data.get("resolution_verifications", [])
@@ -1275,14 +1313,68 @@ The SDK will run invoked agents in parallel automatically.
):
unresolved_ids.append(finding_id)
# Try to extract new findings
new_findings = data.get("new_findings", [])
if isinstance(new_findings, list):
for nf in new_findings:
if isinstance(nf, dict):
finding_id = nf.get("id", "")
if finding_id:
new_finding_ids.append(finding_id)
# Try to extract new findings as PRReviewFinding objects
new_findings_raw = data.get("new_findings", [])
if isinstance(new_findings_raw, list):
for nf in new_findings_raw:
if not isinstance(nf, dict):
continue
try:
finding_id = nf.get("id", "") or self._generate_finding_id(
nf.get("file", "unknown"),
nf.get("line", 0),
nf.get("title", "unknown"),
)
new_finding_ids.append(finding_id)
findings.append(
PRReviewFinding(
id=finding_id,
severity=_map_severity(nf.get("severity", "medium")),
category=map_category(nf.get("category", "quality")),
title=nf.get("title", "Unknown issue"),
description=nf.get("description", ""),
file=nf.get("file", "unknown"),
line=nf.get("line", 0) or 0,
suggested_fix=nf.get("suggested_fix"),
fixable=bool(nf.get("fixable", False)),
is_impact_finding=bool(nf.get("is_impact_finding", False)),
)
)
except Exception as e:
logger.debug(
f"[ParallelFollowup] Skipping malformed new finding: {e}"
)
# Try to extract comment findings as PRReviewFinding objects
comment_findings_raw = data.get("comment_findings", [])
if isinstance(comment_findings_raw, list):
for cf in comment_findings_raw:
if not isinstance(cf, dict):
continue
try:
finding_id = cf.get("id", "") or self._generate_finding_id(
cf.get("file", "unknown"),
cf.get("line", 0),
cf.get("title", "unknown"),
)
new_finding_ids.append(finding_id)
findings.append(
PRReviewFinding(
id=finding_id,
severity=_map_severity(cf.get("severity", "medium")),
category=map_category(cf.get("category", "quality")),
title=f"[FROM COMMENTS] {cf.get('title', 'Unknown issue')}",
description=cf.get("description", ""),
file=cf.get("file", "unknown"),
line=cf.get("line", 0) or 0,
suggested_fix=cf.get("suggested_fix"),
fixable=bool(cf.get("fixable", False)),
)
)
except Exception as e:
logger.debug(
f"[ParallelFollowup] Skipping malformed comment finding: {e}"
)
# Try to extract verdict
verdict_str = data.get("verdict", "NEEDS_REVISION")
@@ -1297,14 +1389,15 @@ The SDK will run invoked agents in parallel automatically.
verdict_reasoning = data.get("verdict_reasoning", "Extracted from partial data")
# Only return if we got any useful data
if resolved_ids or unresolved_ids or new_finding_ids:
if resolved_ids or unresolved_ids or new_finding_ids or findings:
return {
"findings": [], # Can't reliably extract full findings without validation
"findings": findings,
"resolved_ids": resolved_ids,
"unresolved_ids": unresolved_ids,
"new_finding_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": f"[Partial extraction] {verdict_reasoning}",
apps/backend/runners/github/services/parallel_orchestrator_reviewer.py
+70 -4
View File
@@ -633,7 +633,14 @@ Report findings with specific file paths, line numbers, and code evidence.
logger.error(
f"[Specialist:{specialist_name}] Failed to parse structured output: {e}"
)
# Fall through to text parsing
# Attempt to extract findings from raw dict before falling to text parsing
findings = self._extract_specialist_partial_data(
specialist_name, structured_output
)
if findings:
logger.info(
f"[Specialist:{specialist_name}] Recovered {len(findings)} findings from partial extraction"
)
if not findings and result_text:
# Fallback to text parsing
@@ -643,6 +650,63 @@ Report findings with specific file paths, line numbers, and code evidence.
return findings
def _extract_specialist_partial_data(
self,
specialist_name: str,
data: dict[str, Any],
) -> list[PRReviewFinding]:
"""Extract findings from raw specialist dict when Pydantic validation fails.
Defensively extracts each finding individually so partial results are preserved
even if some findings have validation issues.
"""
findings = []
raw_findings = data.get("findings", [])
if not isinstance(raw_findings, list):
return findings
for f in raw_findings:
if not isinstance(f, dict):
continue
try:
file_path = f.get("file", "unknown")
line = f.get("line", 0) or 0
title = f.get("title", "Unknown issue")
finding_id = hashlib.md5(
f"{file_path}:{line}:{title}".encode(),
usedforsecurity=False,
).hexdigest()[:12]
category = map_category(f.get("category", "quality"))
try:
severity = ReviewSeverity(str(f.get("severity", "medium")).lower())
except ValueError:
severity = ReviewSeverity.MEDIUM
finding = PRReviewFinding(
id=finding_id,
file=file_path,
line=line,
end_line=f.get("end_line"),
title=title,
description=f.get("description", ""),
category=category,
severity=severity,
suggested_fix=f.get("suggested_fix", ""),
evidence=f.get("evidence"),
source_agents=[specialist_name],
is_impact_finding=bool(f.get("is_impact_finding", False)),
)
findings.append(finding)
except Exception as e:
logger.debug(
f"[Specialist:{specialist_name}] Skipping malformed finding: {e}"
)
return findings
async def _run_parallel_specialists(
self,
context: PRContext,
@@ -910,13 +974,15 @@ The SDK will run invoked agents in parallel automatically.
except ValueError:
severity = ReviewSeverity.MEDIUM
# Extract evidence: prefer verification.code_examined, fallback to evidence field
evidence = finding_data.evidence
# Extract evidence from verification.code_examined if available
evidence = None
if hasattr(finding_data, "verification") and finding_data.verification:
# Structured verification has more detailed evidence
verification = finding_data.verification
if hasattr(verification, "code_examined") and verification.code_examined:
evidence = verification.code_examined
# Fallback to evidence field if present (e.g. from dict-based parsing)
if not evidence:
evidence = getattr(finding_data, "evidence", None)
# Extract end_line if present
end_line = getattr(finding_data, "end_line", None)
apps/backend/runners/github/services/pydantic_models.py
+110 -294
View File
@@ -26,10 +26,10 @@ from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, field_validator
# =============================================================================
# Verification Evidence (Required for All Findings)
# Verification Evidence (Optional for findings — only code_examined is consumed)
# =============================================================================
@@ -50,102 +50,28 @@ class VerificationEvidence(BaseModel):
# =============================================================================
# Common Finding Types
# Severity / Category Validators
# =============================================================================
class BaseFinding(BaseModel):
"""Base class for all finding types."""
id: str = Field(description="Unique identifier for this finding")
severity: Literal["critical", "high", "medium", "low"] = Field(
description="Issue severity level"
)
title: str = Field(description="Brief issue title (max 80 chars)")
description: str = Field(description="Detailed explanation of the issue")
file: str = Field(description="File path where issue was found")
line: int = Field(0, description="Line number of the issue")
suggested_fix: str | None = Field(None, description="How to fix this issue")
fixable: bool = Field(False, description="Whether this can be auto-fixed")
evidence: str | None = Field(
None,
description="DEPRECATED: Use verification.code_examined instead. Will be removed in Phase 5.",
)
verification: VerificationEvidence = Field(
description="Evidence that this finding was verified against actual code"
)
_VALID_SEVERITIES = {"critical", "high", "medium", "low"}
class SecurityFinding(BaseFinding):
"""A security vulnerability finding."""
category: Literal["security"] = Field(
default="security", description="Always 'security' for security findings"
)
def _normalize_severity(v: str) -> str:
"""Normalize severity to a valid value, defaulting to 'medium'."""
if isinstance(v, str):
v = v.lower().strip()
if v not in _VALID_SEVERITIES:
return "medium"
return v
class QualityFinding(BaseFinding):
"""A code quality or redundancy finding."""
category: Literal[
"redundancy", "quality", "test", "performance", "pattern", "docs"
] = Field(description="Issue category")
redundant_with: str | None = Field(
None, description="Reference to duplicate code (file:line) if redundant"
)
class DeepAnalysisFinding(BaseFinding):
"""A finding from deep analysis with verification info."""
category: Literal[
"verification_failed",
"redundancy",
"quality",
"pattern",
"performance",
"logic",
] = Field(description="Issue category")
verification_note: str | None = Field(
None, description="What evidence is missing or couldn't be verified"
)
class StructuralIssue(BaseModel):
"""A structural issue with the PR."""
id: str = Field(description="Unique identifier")
issue_type: Literal[
"feature_creep", "scope_creep", "architecture_violation", "poor_structure"
] = Field(description="Type of structural issue")
severity: Literal["critical", "high", "medium", "low"] = Field(
description="Issue severity"
)
title: str = Field(description="Brief issue title")
description: str = Field(description="Detailed explanation")
impact: str = Field(description="Why this matters")
suggestion: str = Field(description="How to fix")
class AICommentTriage(BaseModel):
"""Triage result for an AI tool comment."""
comment_id: int = Field(description="GitHub comment ID")
tool_name: str = Field(
description="AI tool name (CodeRabbit, Cursor, Greptile, etc.)"
)
verdict: Literal[
"critical",
"important",
"nice_to_have",
"trivial",
"addressed",
"false_positive",
] = Field(description="Verdict on the comment")
reasoning: str = Field(description="Why this verdict was chosen")
response_comment: str | None = Field(
None, description="Optional comment to post in reply"
)
def _normalize_category(v: str, valid_set: set[str], default: str = "quality") -> str:
"""Normalize category to a valid value, defaulting to given default."""
if isinstance(v, str):
v = v.lower().strip().replace("-", "_")
if v not in valid_set:
return default
return v
# =============================================================================
@@ -163,25 +89,34 @@ class FindingResolution(BaseModel):
)
_FOLLOWUP_CATEGORIES = {"security", "quality", "logic", "test", "docs"}
class FollowupFinding(BaseModel):
"""A new finding from follow-up review (simpler than initial review)."""
"""A new finding from follow-up review (simpler than initial review).
verification is intentionally omitted — not consumed by followup_reviewer.py.
"""
id: str = Field(description="Unique identifier for this finding")
severity: Literal["critical", "high", "medium", "low"] = Field(
description="Issue severity level"
)
category: Literal["security", "quality", "logic", "test", "docs"] = Field(
description="Issue category"
)
severity: str = Field(description="Issue severity level")
category: str = Field(description="Issue category")
title: str = Field(description="Brief issue title")
description: str = Field(description="Detailed explanation of the issue")
file: str = Field(description="File path where issue was found")
line: int = Field(0, description="Line number of the issue")
suggested_fix: str | None = Field(None, description="How to fix this issue")
fixable: bool = Field(False, description="Whether this can be auto-fixed")
verification: VerificationEvidence = Field(
description="Evidence that this finding was verified against actual code"
)
@field_validator("severity", mode="before")
@classmethod
def _normalize_severity(cls, v: str) -> str:
return _normalize_severity(v)
@field_validator("category", mode="before")
@classmethod
def _normalize_category(cls, v: str) -> str:
return _normalize_category(v, _FOLLOWUP_CATEGORIES)
class FollowupReviewResponse(BaseModel):
@@ -203,81 +138,6 @@ class FollowupReviewResponse(BaseModel):
verdict_reasoning: str = Field(description="Explanation for the verdict")
# =============================================================================
# Initial Review Responses (Multi-Pass)
# =============================================================================
class QuickScanResult(BaseModel):
"""Result from the quick scan pass."""
purpose: str = Field(description="Brief description of what the PR claims to do")
actual_changes: str = Field(
description="Brief description of what the code actually does"
)
purpose_match: bool = Field(
description="Whether actual changes match the claimed purpose"
)
purpose_match_note: str | None = Field(
None, description="Explanation if purpose doesn't match actual changes"
)
risk_areas: list[str] = Field(
default_factory=list, description="Areas needing careful review"
)
red_flags: list[str] = Field(
default_factory=list, description="Obvious issues or concerns"
)
requires_deep_verification: bool = Field(
description="Whether deep verification is needed"
)
complexity: Literal["low", "medium", "high"] = Field(description="PR complexity")
class SecurityPassResult(BaseModel):
"""Result from the security pass - array of security findings."""
findings: list[SecurityFinding] = Field(
default_factory=list, description="Security vulnerabilities found"
)
class QualityPassResult(BaseModel):
"""Result from the quality pass - array of quality findings."""
findings: list[QualityFinding] = Field(
default_factory=list, description="Quality and redundancy issues found"
)
class DeepAnalysisResult(BaseModel):
"""Result from the deep analysis pass."""
findings: list[DeepAnalysisFinding] = Field(
default_factory=list,
description="Deep analysis findings with verification info",
)
class StructuralPassResult(BaseModel):
"""Result from the structural pass."""
issues: list[StructuralIssue] = Field(
default_factory=list, description="Structural issues found"
)
verdict: Literal[
"READY_TO_MERGE", "MERGE_WITH_CHANGES", "NEEDS_REVISION", "BLOCKED"
] = Field(description="Structural verdict")
verdict_reasoning: str = Field(description="Explanation for the verdict")
class AICommentTriageResult(BaseModel):
"""Result from AI comment triage pass."""
triages: list[AICommentTriage] = Field(
default_factory=list, description="Triage results for each AI comment"
)
# =============================================================================
# Issue Triage Response
# =============================================================================
@@ -320,88 +180,21 @@ class IssueTriageResponse(BaseModel):
comment: str | None = Field(None, description="Optional bot comment to post")
# =============================================================================
# Orchestrator Review Response
# =============================================================================
class OrchestratorFinding(BaseModel):
"""A finding from the orchestrator review."""
file: str = Field(description="File path where issue was found")
line: int = Field(0, description="Line number of the issue")
title: str = Field(description="Brief issue title")
description: str = Field(description="Detailed explanation of the issue")
category: Literal[
"security",
"quality",
"style",
"docs",
"redundancy",
"verification_failed",
"pattern",
"performance",
"logic",
"test",
] = Field(description="Issue category")
severity: Literal["critical", "high", "medium", "low"] = Field(
description="Issue severity level"
)
suggestion: str | None = Field(None, description="How to fix this issue")
evidence: str | None = Field(
None,
description="DEPRECATED: Use verification.code_examined instead. Will be removed in Phase 5.",
)
verification: VerificationEvidence = Field(
description="Evidence that this finding was verified against actual code"
)
class OrchestratorReviewResponse(BaseModel):
"""Complete response schema for orchestrator PR review."""
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")
findings: list[OrchestratorFinding] = Field(
default_factory=list, description="Issues found during review"
)
summary: str = Field(description="Brief summary of the review")
# =============================================================================
# Parallel Orchestrator Review Response (SDK Subagents)
# =============================================================================
class LogicFinding(BaseFinding):
"""A logic/correctness finding from the logic review agent."""
category: Literal["logic"] = Field(
default="logic", description="Always 'logic' for logic findings"
)
example_input: str | None = Field(
None, description="Concrete input that triggers the bug"
)
actual_output: str | None = Field(None, description="What the buggy code produces")
expected_output: str | None = Field(
None, description="What the code should produce"
)
class CodebaseFitFinding(BaseFinding):
"""A codebase fit finding from the codebase fit review agent."""
category: Literal["codebase_fit"] = Field(
default="codebase_fit", description="Always 'codebase_fit' for fit findings"
)
existing_code: str | None = Field(
None, description="Reference to existing code that should be used instead"
)
codebase_pattern: str | None = Field(
None, description="Description of the established pattern being violated"
)
_ORCHESTRATOR_CATEGORIES = {
"security",
"quality",
"logic",
"codebase_fit",
"test",
"docs",
"redundancy",
"pattern",
"performance",
}
class ParallelOrchestratorFinding(BaseModel):
@@ -413,26 +206,11 @@ class ParallelOrchestratorFinding(BaseModel):
end_line: int | None = Field(None, description="End line for multi-line issues")
title: str = Field(description="Brief issue title (max 80 chars)")
description: str = Field(description="Detailed explanation of the issue")
category: Literal[
"security",
"quality",
"logic",
"codebase_fit",
"test",
"docs",
"redundancy",
"pattern",
"performance",
] = Field(description="Issue category")
severity: Literal["critical", "high", "medium", "low"] = Field(
description="Issue severity level"
)
evidence: str | None = Field(
category: str = Field(description="Issue category")
severity: str = Field(description="Issue severity level")
verification: VerificationEvidence | None = Field(
None,
description="DEPRECATED: Use verification.code_examined instead. Will be removed in Phase 5.",
)
verification: VerificationEvidence = Field(
description="Evidence that this finding was verified against actual code"
description="Evidence that this finding was verified against actual code",
)
is_impact_finding: bool = Field(
False,
@@ -459,6 +237,16 @@ class ParallelOrchestratorFinding(BaseModel):
False, description="Whether multiple agents agreed on this finding"
)
@field_validator("severity", mode="before")
@classmethod
def _normalize_severity(cls, v: str) -> str:
return _normalize_severity(v)
@field_validator("category", mode="before")
@classmethod
def _normalize_category(cls, v: str) -> str:
return _normalize_category(v, _ORCHESTRATOR_CATEGORIES)
class AgentAgreement(BaseModel):
"""Tracks agreement between agents on findings."""
@@ -514,15 +302,22 @@ class ValidationSummary(BaseModel):
)
_SPECIALIST_CATEGORIES = {
"security",
"quality",
"logic",
"performance",
"pattern",
"test",
"docs",
}
class SpecialistFinding(BaseModel):
"""A finding from a specialist agent (used in parallel SDK sessions)."""
severity: Literal["critical", "high", "medium", "low"] = Field(
description="Issue severity level"
)
category: Literal[
"security", "quality", "logic", "performance", "pattern", "test", "docs"
] = Field(description="Issue category")
severity: str = Field(description="Issue severity level")
category: str = Field(description="Issue category")
title: str = Field(description="Brief issue title (max 80 chars)")
description: str = Field(description="Detailed explanation of the issue")
file: str = Field(description="File path where issue was found")
@@ -530,14 +325,24 @@ class SpecialistFinding(BaseModel):
end_line: int | None = Field(None, description="End line number if multi-line")
suggested_fix: str | None = Field(None, description="How to fix this issue")
evidence: str = Field(
min_length=1,
description="Actual code snippet examined that shows the issue. Required.",
default="",
description="Actual code snippet examined that shows the issue.",
)
is_impact_finding: bool = Field(
False,
description="True if this is about affected code outside the PR (callers, dependencies)",
)
@field_validator("severity", mode="before")
@classmethod
def _normalize_severity(cls, v: str) -> str:
return _normalize_severity(v)
@field_validator("category", mode="before")
@classmethod
def _normalize_category(cls, v: str) -> str:
return _normalize_category(v, _SPECIALIST_CATEGORIES)
class SpecialistResponse(BaseModel):
"""Response schema for individual specialist agent (parallel SDK sessions).
@@ -611,6 +416,17 @@ class ResolutionVerification(BaseModel):
)
_PARALLEL_FOLLOWUP_CATEGORIES = {
"security",
"quality",
"logic",
"test",
"docs",
"regression",
"incomplete_fix",
}
class ParallelFollowupFinding(BaseModel):
"""A finding from parallel follow-up review."""
@@ -619,18 +435,8 @@ class ParallelFollowupFinding(BaseModel):
line: int = Field(0, description="Line number of the issue")
title: str = Field(description="Brief issue title")
description: str = Field(description="Detailed explanation of the issue")
category: Literal[
"security",
"quality",
"logic",
"test",
"docs",
"regression",
"incomplete_fix",
] = Field(description="Issue category")
severity: Literal["critical", "high", "medium", "low"] = Field(
description="Issue severity level"
)
category: str = Field(description="Issue category")
severity: str = Field(description="Issue severity level")
suggested_fix: str | None = Field(None, description="How to fix this issue")
fixable: bool = Field(False, description="Whether this can be auto-fixed")
is_impact_finding: bool = Field(
@@ -638,6 +444,16 @@ class ParallelFollowupFinding(BaseModel):
description="True if this finding is about impact on OTHER files outside the PR diff",
)
@field_validator("severity", mode="before")
@classmethod
def _normalize_severity(cls, v: str) -> str:
return _normalize_severity(v)
@field_validator("category", mode="before")
@classmethod
def _normalize_category(cls, v: str) -> str:
return _normalize_category(v, _PARALLEL_FOLLOWUP_CATEGORIES)
class ParallelFollowupResponse(BaseModel):
"""Complete response schema for parallel follow-up PR review.
apps/backend/runners/github/services/recovery_utils.py
+108
View File
@@ -0,0 +1,108 @@
"""
Recovery Utilities for PR Review
=================================
Shared helpers for extraction recovery in followup and parallel followup reviewers.
These utilities consolidate duplicated logic for:
- Parsing "SEVERITY: description" patterns from extraction summaries
- Generating consistent, traceable finding IDs with prefixes
- Creating PRReviewFinding objects from extraction data
"""
from __future__ import annotations
import hashlib
try:
from ..models import (
PRReviewFinding,
ReviewCategory,
ReviewSeverity,
)
except (ImportError, ValueError, SystemError):
from models import (
PRReviewFinding,
ReviewCategory,
ReviewSeverity,
)
# Severity mapping for parsing "SEVERITY: description" patterns
_EXTRACTION_SEVERITY_MAP: list[tuple[str, ReviewSeverity]] = [
("CRITICAL:", ReviewSeverity.CRITICAL),
("HIGH:", ReviewSeverity.HIGH),
("MEDIUM:", ReviewSeverity.MEDIUM),
("LOW:", ReviewSeverity.LOW),
]
def parse_severity_from_summary(
summary: str,
) -> tuple[ReviewSeverity, str]:
"""Parse a "SEVERITY: description" pattern from an extraction summary.
Args:
summary: Raw summary string, e.g. "HIGH: Missing null check in parser.py"
Returns:
Tuple of (severity, cleaned_description).
Defaults to MEDIUM severity if no prefix is found.
"""
upper_summary = summary.upper()
for sev_name, sev_val in _EXTRACTION_SEVERITY_MAP:
if upper_summary.startswith(sev_name):
return sev_val, summary[len(sev_name) :].strip()
return ReviewSeverity.MEDIUM, summary
def generate_recovery_finding_id(
index: int, description: str, prefix: str = "FR"
) -> str:
"""Generate a consistent, traceable finding ID for recovery findings.
Args:
index: The index of the finding in the extraction list.
description: The finding description (used for hash uniqueness).
prefix: ID prefix for traceability. Default "FR" (Followup Recovery).
Use "FU" for parallel followup findings.
Returns:
A prefixed finding ID like "FR-A1B2C3D4" or "FU-A1B2C3D4".
"""
content = f"extraction-{index}-{description}"
hex_hash = (
hashlib.md5(content.encode(), usedforsecurity=False).hexdigest()[:8].upper()
)
return f"{prefix}-{hex_hash}"
def create_finding_from_summary(
summary: str,
index: int,
id_prefix: str = "FR",
) -> PRReviewFinding:
"""Create a PRReviewFinding from an extraction summary string.
Parses "SEVERITY: description" patterns, generates a traceable finding ID,
and returns a fully constructed PRReviewFinding.
Args:
summary: Raw summary string, e.g. "HIGH: Missing null check in parser.py"
index: The index of the finding in the extraction list.
id_prefix: ID prefix for traceability. Default "FR" (Followup Recovery).
Returns:
A PRReviewFinding with parsed severity, generated ID, and description.
"""
severity, description = parse_severity_from_summary(summary)
finding_id = generate_recovery_finding_id(index, description, prefix=id_prefix)
return PRReviewFinding(
id=finding_id,
severity=severity,
category=ReviewCategory.QUALITY,
title=description[:80],
description=f"[Recovered via extraction] {description}",
file="unknown",
line=0,
)
tests/test_structured_outputs.py
+159 -286
View File
@@ -29,20 +29,13 @@ from pydantic_models import (
FindingResolution,
FollowupFinding,
FollowupReviewResponse,
# Orchestrator review models
OrchestratorFinding,
OrchestratorReviewResponse,
# Initial review models
QuickScanResult,
SecurityFinding,
QualityFinding,
DeepAnalysisFinding,
StructuralIssue,
AICommentTriage,
# Verification evidence models (Phase 2)
# Verification evidence models
VerificationEvidence,
ParallelOrchestratorFinding,
BaseFinding,
# Specialist models
SpecialistFinding,
# Parallel follow-up models
ParallelFollowupFinding,
)
@@ -86,7 +79,7 @@ class TestFollowupFinding:
"""Tests for FollowupFinding model."""
def test_valid_finding(self):
"""Test valid follow-up finding."""
"""Test valid follow-up finding (no verification required)."""
data = {
"id": "new-1",
"severity": "high",
@@ -97,11 +90,6 @@ class TestFollowupFinding:
"line": 42,
"suggested_fix": "Use parameterized queries",
"fixable": True,
"verification": {
"code_examined": "query = 'SELECT * FROM users WHERE id=' + user_input",
"line_range_examined": [42, 42],
"verification_method": "direct_code_inspection",
},
}
result = FollowupFinding.model_validate(data)
assert result.id == "new-1"
@@ -119,44 +107,52 @@ class TestFollowupFinding:
"title": "Missing docstring",
"description": "Function lacks documentation",
"file": "utils.py",
"verification": {
"code_examined": "def process_data(data):\n return data",
"line_range_examined": [1, 2],
"verification_method": "direct_code_inspection",
},
}
result = FollowupFinding.model_validate(data)
assert result.line == 0 # Default
assert result.suggested_fix is None
assert result.fixable is False
def test_invalid_severity_rejected(self):
"""Test that invalid severity is rejected."""
def test_invalid_severity_normalized(self):
"""Test that invalid severity is normalized to 'medium'."""
data = {
"id": "new-1",
"severity": "extreme", # Invalid
"severity": "extreme", # Invalid — normalized to medium
"category": "security",
"title": "Test",
"description": "Test",
"file": "test.py",
}
with pytest.raises(ValidationError) as exc_info:
FollowupFinding.model_validate(data)
assert "severity" in str(exc_info.value)
result = FollowupFinding.model_validate(data)
assert result.severity == "medium"
def test_invalid_category_rejected(self):
"""Test that invalid category is rejected."""
def test_invalid_category_normalized(self):
"""Test that invalid category is normalized to 'quality'."""
data = {
"id": "new-1",
"severity": "high",
"category": "unknown_category", # Invalid
"category": "unknown_category", # Invalid — normalized to quality
"title": "Test",
"description": "Test",
"file": "test.py",
}
with pytest.raises(ValidationError) as exc_info:
FollowupFinding.model_validate(data)
assert "category" in str(exc_info.value)
result = FollowupFinding.model_validate(data)
assert result.category == "quality"
def test_verification_not_required(self):
"""Test that verification field is not required on FollowupFinding."""
data = {
"id": "new-1",
"severity": "medium",
"category": "quality",
"title": "Test",
"description": "Test",
"file": "test.py",
}
result = FollowupFinding.model_validate(data)
assert not hasattr(result, "verification") or not hasattr(
result.__class__.model_fields, "verification"
)
class TestFollowupReviewResponse:
@@ -177,11 +173,6 @@ class TestFollowupReviewResponse:
"description": "Complex method",
"file": "service.py",
"line": 100,
"verification": {
"code_examined": "def process(self, data):\n # 50 lines of nested if statements",
"line_range_examined": [100, 150],
"verification_method": "direct_code_inspection",
},
}
],
"comment_findings": [],
@@ -238,125 +229,6 @@ class TestFollowupReviewResponse:
assert result.verdict == verdict
class TestOrchestratorFinding:
"""Tests for OrchestratorFinding model."""
def test_valid_finding(self):
"""Test valid orchestrator finding with evidence field."""
data = {
"file": "src/api.py",
"line": 25,
"title": "Missing error handling",
"description": "API endpoint lacks try-catch block",
"category": "quality",
"severity": "medium",
"suggestion": "Add error handling with proper logging",
"evidence": "def handle_request(req):\n result = db.query(req.id) # no try-catch",
"verification": {
"code_examined": "def handle_request(req):\n result = db.query(req.id) # no try-catch",
"line_range_examined": [25, 26],
"verification_method": "direct_code_inspection",
},
}
result = OrchestratorFinding.model_validate(data)
assert result.file == "src/api.py"
assert result.evidence is not None
assert "no try-catch" in result.evidence
def test_evidence_optional(self):
"""Test that evidence field is optional."""
data = {
"file": "test.py",
"title": "Test",
"description": "Test finding",
"category": "quality",
"severity": "low",
"verification": {
"code_examined": "def test():\n pass",
"line_range_examined": [1, 2],
"verification_method": "direct_code_inspection",
},
}
result = OrchestratorFinding.model_validate(data)
assert result.evidence is None
class TestOrchestratorReviewResponse:
"""Tests for OrchestratorReviewResponse model."""
def test_valid_response(self):
"""Test valid orchestrator review response."""
data = {
"verdict": "NEEDS_REVISION",
"verdict_reasoning": "Critical security issue found",
"findings": [
{
"file": "auth.py",
"line": 10,
"title": "Hardcoded secret",
"description": "API key exposed in source",
"category": "security",
"severity": "critical",
"evidence": "API_KEY = 'sk-prod-12345abcdef'",
"verification": {
"code_examined": "API_KEY = 'sk-prod-12345abcdef'",
"line_range_examined": [10, 10],
"verification_method": "direct_code_inspection",
},
}
],
"summary": "Found 1 critical security issue",
}
result = OrchestratorReviewResponse.model_validate(data)
assert result.verdict == "NEEDS_REVISION"
assert len(result.findings) == 1
assert result.findings[0].severity == "critical"
def test_empty_findings(self):
"""Test response with no findings."""
data = {
"verdict": "READY_TO_MERGE",
"verdict_reasoning": "All checks passed",
"findings": [],
"summary": "Clean PR, ready for merge",
}
result = OrchestratorReviewResponse.model_validate(data)
assert len(result.findings) == 0
class TestQuickScanResult:
"""Tests for QuickScanResult model."""
def test_valid_quick_scan(self):
"""Test valid quick scan result."""
data = {
"purpose": "Add user authentication",
"actual_changes": "Implements OAuth login flow",
"purpose_match": True,
"risk_areas": ["Security", "Session management"],
"red_flags": [],
"requires_deep_verification": True,
"complexity": "medium",
}
result = QuickScanResult.model_validate(data)
assert result.purpose_match is True
assert result.complexity == "medium"
assert len(result.risk_areas) == 2
def test_complexity_values(self):
"""Test all valid complexity values."""
for complexity in ["low", "medium", "high"]:
data = {
"purpose": "Test",
"actual_changes": "Test",
"purpose_match": True,
"requires_deep_verification": False,
"complexity": complexity,
}
result = QuickScanResult.model_validate(data)
assert result.complexity == complexity
class TestSchemaGeneration:
"""Tests for JSON schema generation."""
@@ -374,15 +246,6 @@ class TestSchemaGeneration:
verdict_schema = schema["properties"]["verdict"]
assert "enum" in verdict_schema or "$ref" in str(schema)
def test_orchestrator_schema_generation(self):
"""Test that OrchestratorReviewResponse generates valid JSON schema."""
schema = OrchestratorReviewResponse.model_json_schema()
assert "properties" in schema
assert "verdict" in schema["properties"]
assert "findings" in schema["properties"]
assert "summary" in schema["properties"]
def test_schema_has_descriptions(self):
"""Test that schema includes field descriptions for AI guidance."""
schema = FollowupReviewResponse.model_json_schema()
@@ -392,108 +255,8 @@ class TestSchemaGeneration:
assert "properties" in schema or "$defs" in schema
class TestSecurityFinding:
"""Tests for SecurityFinding model."""
def test_security_category_default(self):
"""Test that SecurityFinding has security category by default."""
data = {
"id": "sec-1",
"severity": "high",
"title": "XSS vulnerability",
"description": "Unescaped user input",
"file": "template.html",
"line": 50,
"verification": {
"code_examined": "<div>{{ user_input }}</div>",
"line_range_examined": [50, 50],
"verification_method": "direct_code_inspection",
},
}
result = SecurityFinding.model_validate(data)
assert result.category == "security"
class TestDeepAnalysisFinding:
"""Tests for DeepAnalysisFinding model."""
def test_evidence_field(self):
"""Test evidence field for proof of issue."""
data = {
"id": "deep-1",
"severity": "medium",
"title": "Potential race condition",
"description": "Concurrent access without lock",
"file": "worker.py",
"line": 100,
"category": "logic",
"evidence": "shared_state += 1 # no lock protection",
"verification": {
"code_examined": "shared_state += 1 # no lock protection",
"line_range_examined": [100, 100],
"verification_method": "direct_code_inspection",
},
}
result = DeepAnalysisFinding.model_validate(data)
assert result.evidence == "shared_state += 1 # no lock protection"
def test_verification_note(self):
"""Test verification note field."""
data = {
"id": "deep-2",
"severity": "low",
"title": "Unverified assumption",
"description": "Could not verify behavior",
"file": "lib.py",
"category": "verification_failed",
"verification_note": "Unable to find test coverage",
"verification": {
"code_examined": "def some_function():\n return process_data()",
"line_range_examined": [1, 2],
"verification_method": "cross_file_trace",
},
}
result = DeepAnalysisFinding.model_validate(data)
assert result.verification_note == "Unable to find test coverage"
class TestAICommentTriage:
"""Tests for AICommentTriage model."""
def test_valid_triage(self):
"""Test valid AI comment triage."""
data = {
"comment_id": 12345,
"tool_name": "CodeRabbit",
"verdict": "important",
"reasoning": "Valid security concern raised",
"response_comment": "Thank you, we will address this.",
}
result = AICommentTriage.model_validate(data)
assert result.comment_id == 12345
assert result.verdict == "important"
def test_all_verdict_values(self):
"""Test all valid triage verdict values."""
for verdict in [
"critical",
"important",
"nice_to_have",
"trivial",
"false_positive",
]:
data = {
"comment_id": 1,
"tool_name": "Test",
"verdict": verdict,
"reasoning": f"Testing {verdict}",
}
result = AICommentTriage.model_validate(data)
assert result.verdict == verdict
# =============================================================================
# Phase 2: Schema Enforcement Tests
# Verification Evidence Tests
# =============================================================================
@@ -570,10 +333,10 @@ class TestVerificationEvidence:
class TestParallelOrchestratorFindingVerification:
"""Tests for verification field requirement on ParallelOrchestratorFinding."""
"""Tests for verification field on ParallelOrchestratorFinding."""
def test_missing_verification_rejected(self):
"""Test that findings without verification are rejected."""
def test_missing_verification_accepted(self):
"""Test that findings without verification are accepted (now optional)."""
data = {
"id": "test-1",
"file": "test.py",
@@ -582,11 +345,10 @@ class TestParallelOrchestratorFindingVerification:
"description": "A test finding without verification",
"category": "quality",
"severity": "medium",
# No verification field - should fail
# No verification field — should succeed (now optional)
}
with pytest.raises(ValidationError) as exc_info:
ParallelOrchestratorFinding.model_validate(data)
assert "verification" in str(exc_info.value)
result = ParallelOrchestratorFinding.model_validate(data)
assert result.verification is None
def test_valid_finding_with_verification(self):
"""Test valid finding with verification evidence."""
@@ -618,11 +380,6 @@ class TestParallelOrchestratorFindingVerification:
"description": "Test",
"category": "quality",
"severity": "medium",
"verification": {
"code_examined": "code",
"line_range_examined": [10, 10],
"verification_method": "direct_code_inspection",
},
}
result = ParallelOrchestratorFinding.model_validate(data)
assert result.is_impact_finding is False
@@ -657,11 +414,6 @@ class TestParallelOrchestratorFindingVerification:
"description": "No try-catch",
"category": "quality",
"severity": "medium",
"verification": {
"code_examined": "code",
"line_range_examined": [10, 10],
"verification_method": "direct_code_inspection",
},
}
result = ParallelOrchestratorFinding.model_validate(data)
assert result.checked_for_handling_elsewhere is False
@@ -686,6 +438,34 @@ class TestParallelOrchestratorFindingVerification:
result = ParallelOrchestratorFinding.model_validate(data)
assert result.checked_for_handling_elsewhere is True
def test_invalid_severity_normalized(self):
"""Test invalid severity is normalized to 'medium'."""
data = {
"id": "test-1",
"file": "test.py",
"line": 10,
"title": "Test",
"description": "Test",
"category": "quality",
"severity": "super_critical",
}
result = ParallelOrchestratorFinding.model_validate(data)
assert result.severity == "medium"
def test_invalid_category_normalized(self):
"""Test invalid category is normalized to 'quality'."""
data = {
"id": "test-1",
"file": "test.py",
"line": 10,
"title": "Test",
"description": "Test",
"category": "unknown_thing",
"severity": "medium",
}
result = ParallelOrchestratorFinding.model_validate(data)
assert result.category == "quality"
class TestVerificationSchemaGeneration:
"""Tests for JSON schema generation with VerificationEvidence."""
@@ -713,3 +493,96 @@ class TestVerificationSchemaGeneration:
assert "is_impact_finding" in schema["properties"]
assert "checked_for_handling_elsewhere" in schema["properties"]
# =============================================================================
# Specialist Finding Tests
# =============================================================================
class TestSpecialistFinding:
"""Tests for SpecialistFinding model."""
def test_empty_evidence_accepted(self):
"""Test that empty evidence is accepted (no min_length)."""
data = {
"severity": "medium",
"category": "quality",
"title": "Test finding",
"description": "A test",
"file": "test.py",
"evidence": "",
}
result = SpecialistFinding.model_validate(data)
assert result.evidence == ""
def test_evidence_defaults_to_empty(self):
"""Test that evidence defaults to empty string."""
data = {
"severity": "medium",
"category": "quality",
"title": "Test finding",
"description": "A test",
"file": "test.py",
}
result = SpecialistFinding.model_validate(data)
assert result.evidence == ""
def test_invalid_severity_normalized(self):
"""Test invalid severity is normalized."""
data = {
"severity": "urgent",
"category": "security",
"title": "Test",
"description": "Test",
"file": "test.py",
}
result = SpecialistFinding.model_validate(data)
assert result.severity == "medium"
def test_invalid_category_normalized(self):
"""Test invalid category is normalized."""
data = {
"severity": "high",
"category": "style",
"title": "Test",
"description": "Test",
"file": "test.py",
}
result = SpecialistFinding.model_validate(data)
assert result.category == "quality"
# =============================================================================
# Parallel Follow-up Finding Tests
# =============================================================================
class TestParallelFollowupFinding:
"""Tests for ParallelFollowupFinding model."""
def test_invalid_severity_normalized(self):
"""Test invalid severity is normalized."""
data = {
"id": "pf-1",
"file": "test.py",
"title": "Test",
"description": "Test",
"category": "quality",
"severity": "extreme",
}
result = ParallelFollowupFinding.model_validate(data)
assert result.severity == "medium"
def test_invalid_category_normalized(self):
"""Test invalid category is normalized."""
data = {
"id": "pf-1",
"file": "test.py",
"title": "Test",
"description": "Test",
"category": "unknown",
"severity": "medium",
}
result = ParallelFollowupFinding.model_validate(data)
assert result.category == "quality"