AI Code Review Agent

_DIFF_HEADER_RE = re.compile(
    r"^diff --git a/(.+?) b/(.+)$", re.MULTILINE
)

@observe(name="parse_diff")
def parse_diff(state: ReviewState) -> ReviewState:
    raw_diff = state["raw_diff"]
    file_diffs: list[dict[str, str]] = []

    splits = _DIFF_HEADER_RE.split(raw_diff)
    i = 1
    while i + 2 < len(splits):
        b_path = splits[i + 1]
        chunk = splits[i + 2]
        file_diffs.append({"path": b_path, "diff": chunk.strip()})
        i += 3

    return {"file_diffs": file_diffs}

SKIP_EXTENSIONS = frozenset({
    ".lock", ".md", ".json", ".yaml", ".yml", ".toml",
    ".png", ".jpg", ".jpeg", ".gif", ".svg", ".ico",
    ".woff", ".woff2", ".ttf", ".eot",
    ".pyc", ".pyo", ".so", ".dll",
})

@observe(name="filter_files")
def filter_files(state: ReviewState) -> ReviewState:
    filtered = []
    for file_diff in state["file_diffs"]:
        ext = PurePosixPath(file_diff["path"]).suffix.lower()
        if ext not in SKIP_EXTENSIONS:
            filtered.append(file_diff)
    return {"filtered_files": filtered}

def _extract_json_array(text: str) -> list[dict[str, Any]]:
    """Extract a JSON array from LLM output with fallback strategies."""
    stripped = text.strip()

    # Strategy 1: direct parse
    if stripped.startswith("["):
        try:
            return json.loads(stripped)
        except json.JSONDecodeError:
            pass

    # Strategy 2: extract from markdown code fences
    fence_match = re.search(
        r"```(?:json)?\s*\n?(.*?)```", stripped, re.DOTALL
    )
    if fence_match:
        try:
            return json.loads(fence_match.group(1).strip())
        except json.JSONDecodeError:
            pass

    # Strategy 3: find any JSON array in the text
    array_match = re.search(r"\[.*]", stripped, re.DOTALL)
    if array_match:
        try:
            return json.loads(array_match.group(0))
        except json.JSONDecodeError:
            pass

    raise json.JSONDecodeError("No JSON array found", text, 0)

@observe(name="aggregate")
def aggregate(state: ReviewState) -> ReviewState:
    raw_findings = state["findings"]

    all_findings: list[ReviewFinding] = []
    for item in raw_findings:
        all_findings.append(ReviewFinding.model_validate(item))

    stats: dict[str, int] = {}
    for finding in all_findings:
        sev = finding.severity.value
        cat = finding.category.value
        stats[sev] = stats.get(sev, 0) + 1
        stats[cat] = stats.get(cat, 0) + 1
    stats["total"] = len(all_findings)

    summary = ReviewSummary(
        findings=all_findings, stats=stats,
        model_used=_MODEL_NAME,
        tokens_used=partial.get("tokens_used", 0),
        # ...
    )
    return {"summary": summary.model_dump()}

severity_order = [Severity.CRITICAL, Severity.WARNING, Severity.SUGGESTION]
severity_icons = {
    Severity.CRITICAL: "🔴",
    Severity.WARNING:  "🟡",
    Severity.SUGGESTION: "🔵",
}

for severity in severity_order:
    matching = [f for f in summary.findings if f.severity == severity]
    if not matching:
        continue

    lines.append(f"### {severity_icons[severity]} {severity.value.title()}")
    for finding in matching:
        lines.append(
            f"- **{finding.file_path}:{finding.line_number}** "
            f"[{finding.category.value}] "
            f"(confidence: {finding.confidence:.0%})"
        )
        lines.append(f"  {finding.message}")
        if finding.suggested_fix:
            lines.append(f"  > **Fix:** {finding.suggested_fix}")