scripts.code-review

Reference › scripts › code-review

The unified code review tool for orchestrata-ultimate, a single script of about 2,000 lines that runs several checkers over Python files and merges their findings into one report. agent_check calls it as its review step, and comprehensive_review runs it as the first section of its report. The file name has a hyphen, so no module imports it.

Invocation: main takes one or more paths (files or directories) and --output/-o, --format (text, json or markdown), --min-severity (ERROR, WARNING or INFO), --fix to apply fixable ruff issues, --format-code to run the ruff formatter first, --max-files, --diagnostics to also run the test-diagnostic tools (the isolation audit and the graph preflight) and --stdout.

Each checker subclasses BaseChecker, which catches errors per file and adds CodeIssue records to a shared ReviewReport, each tagged with a Severity. RuffChecker and RuffFormatChecker run ruff and map its codes to severities. MyPyChecker runs mypy and parses its output. StyleChecker reads the syntax tree for docstrings, type hints, naming, cyclomatic complexity, test isolation, code patterns and imperative accumulators (a collection initialized empty and filled in a loop). ChecklistChecker checks class responsibility, methods that never use self, and abstract classes.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

Severity (enum)

Bases: StrEnum · Line 75

Issue severity levels.

CodeIssue (dataclass)

Decorators: dataclass · Line 84

Represents a code review issue.

ReviewReport (dataclass)

Decorators: dataclass · Line 100

Complete code review report.

BaseChecker

Line 129

Base class for all checkers.

RuffChecker

Bases: BaseChecker · Line 179

Run ruff for formatting and linting.

RuffFormatChecker

Bases: BaseChecker · Line 316

Run ruff format on files.

MyPyChecker

Bases: BaseChecker · Line 381

Run mypy for type checking.

StyleChecker

Bases: BaseChecker · Line 470

AST-based style checks (Google Python Style Guide).

ChecklistChecker

Bases: BaseChecker · Line 930

Checklist-based code review checks.

TraceabilityChecker

Bases: BaseChecker · Line 1282

Check that all classes and methods have @traces decorators.

ConnectionStringChecker

Bases: BaseChecker · Line 1532

Check for hardcoded connection strings and env-var defaults.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/code-review.py

#!/usr/bin/env python3
"""
Unified Code Review Tool for orchestrata-ultimate.

Combines:
- Ruff formatting and linting (replaces black/flake8)
- MyPy type checking
- AST-based style checks (Google Python Style Guide)
- Checklist-based code review checks

Usage:
    python code-review.py src/                         # Review directory
    python code-review.py --format json --output report.json src/  # JSON output
    python code-review.py --min-severity WARNING src/  # Only warnings and errors
"""

import argparse
import ast
import json
import os
import re
import subprocess
import sys
from collections import defaultdict
from dataclasses import asdict, dataclass, field
from datetime import datetime
from enum import StrEnum
from pathlib import Path
from typing import Any

# A tool that DID NOT RUN is not a clean result.
#
# RUFF_NOT_FOUND, RUFF_TIMEOUT, RUFF_FORMAT_ERROR, RUFF_FORMAT_TIMEOUT,
# MYPY_NOT_FOUND and MYPY_TIMEOUT all report that an external checker never
# produced a verdict. They were WARNING severity, and `main()` returns
# `1 if error_count > 0 else 0` — so a run where linting or type checking never
# happened exited 0 and was indistinguishable from a run where it happened and
# found nothing. That is the false-green shape this project has already been
# bitten by twice (a kernel that silently never started; tests passing against
# leaked host services).
#
# Promoted to ERROR on 2026-08-06. Both tools are declared dev dependencies
# (pyproject.toml `ruff>=0.1.0`, `mypy>=1.8.0`) and present in the venv, so
# "not installed" means a broken environment and "timed out" means an
# unfinished check. Neither is a clean review, and neither may report success.
#
# Import conventions are checked from the `ast` tree this file already parses.
# There is no second parser here any more: a tree-sitter copy of the
# relative-import check lived at this call site until 2026-08-06 and was
# deleted, because
#   1. the rule is already owned by the AST toolkit
#      (`serialize_codebase_ast.py standards`, rule `no-relative-imports`) —
#      this module was the THIRD implementation of one rule;
#   2. it re-parsed a file that line ~412 had just parsed, only to answer the
#      same question;
#   3. it was silently dead twice (a NameError on a never-defined `_py_lang()`
#      swallowed by a broad except, then a wrong assumption about the
#      grammar's node shape) — a check nothing depends on;
#   4. the tree_sitter C extension does not declare free-threaded support, so
#      importing it re-enables the GIL for the whole process, poisoning
#      janus-transport (SPEC-023) in anything that loads this script.
# A GIL hazard carried for a triplicated check is not a trade worth keeping.


# Children's output is machine-parsed (regex/substring greps): force color
# OFF for every subprocess this tool spawns. An ambient FORCE_COLOR (e.g.
# exported by the launching terminal/agent harness) made mypy/ruff/pytest
# emit ANSI into captured pipes and silently broke the parsers (gate
# 2026-08-05: mypy "0 issues" on a dirty tree, ruff int(line) crash,
# FAILED-line greps missing their lines).
os.environ["NO_COLOR"] = "1"  # gate-ok: process-bootstrap env routing
os.environ.pop("FORCE_COLOR", None)  # gate-ok: process-bootstrap env routing


class Severity(StrEnum):
    """Issue severity levels."""

    ERROR = "ERROR"
    WARNING = "WARNING"
    INFO = "INFO"


@dataclass
class CodeIssue:
    """Represents a code review issue."""

    file: str
    line: int
    column: int = 0
    code: str = ""
    message: str = ""
    severity: Severity = Severity.WARNING
    suggestion: str | None = None
    category: str = ""
    fixable: bool = False
    context: dict[str, Any] = field(default_factory=dict)


@dataclass
class ReviewReport:
    """Complete code review report."""

    files_analyzed: int = 0
    total_issues: int = 0
    issues: list[CodeIssue] = field(default_factory=list)
    summary_by_category: dict[str, int] = field(
        default_factory=lambda: defaultdict(int)
    )
    summary_by_severity: dict[Severity, int] = field(
        default_factory=lambda: defaultdict(int)
    )
    _seen_issues: set[tuple[str, int, int, str, str]] = field(
        default_factory=set, init=False, repr=False
    )

    def add_issue(self, issue: CodeIssue) -> None:
        """Add an issue to the report."""
        # Create a unique key for deduplication
        key = (issue.file, issue.line, issue.column, issue.code, issue.message)
        if key in self._seen_issues:
            return
        self._seen_issues.add(key)
        self.issues.append(issue)
        self.total_issues += 1
        self.summary_by_category[issue.category] += 1
        self.summary_by_severity[issue.severity] += 1


class BaseChecker:
    """Base class for all checkers."""

    def __init__(self, report: ReviewReport):
        self.report = report
        self.current_file: str = ""
        self.source_lines: list[str] = []
        self.source: str = ""

    def check_file(self, file_path: Path) -> None:
        """Check a single file."""
        self.current_file = str(file_path)
        try:
            self.source = file_path.read_text(encoding="utf-8")
            self.source_lines = self.source.splitlines()
        except Exception as e:
            self.report.add_issue(
                CodeIssue(
                    file=self.current_file,
                    line=1,
                    code="FILE_READ_ERROR",
                    message=f"Cannot read file: {e}",
                    severity=Severity.ERROR,
                    category="system",
                )
            )
            return

        self._check_file_impl(file_path)

    def _check_file_impl(self, file_path: Path) -> None:
        """Implementation of file checking (to be overridden)."""
        pass

    def _report_ruff_not_found(self, message: str) -> None:
        """Report RUFF_NOT_FOUND at its pre-Q2b severity (ERROR): a review
        in which ruff never ran must not be indistinguishable from a clean
        review — the run's exit code comes from the ERROR count."""
        self.report.add_issue(
            CodeIssue(
                file=self.current_file,
                line=1,
                code="RUFF_NOT_FOUND",
                message=message,
                severity=Severity.ERROR,
                category="ruff",
            )
        )


class RuffChecker(BaseChecker):
    """Run ruff for formatting and linting."""

    def __init__(
        self,
        report: ReviewReport,
        fix: bool = False,
        interpreter: str | None = None,
    ) -> None:
        super().__init__(report)
        self.fix = fix
        # ADR-0089: ruff runs as ``<interpreter> -m ruff``, never as a bare
        # name resolved through whatever PATH the session carries. Defaults
        # to the running interpreter, matching lint.LintEvaluator.
        self.interpreter = interpreter if interpreter is not None else sys.executable

    def _check_file_impl(self, file_path: Path) -> None:
        """Run ruff on the file."""
        try:
            cmd = [
                self.interpreter,
                "-m",
                "ruff",
                "check",
                "--exit-zero",
                "--output-format=concise",
            ]
            if self.fix:
                cmd.append("--fix")
            cmd.append(str(file_path))
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=30,
            )

            if result.returncode != 0 and "No module named ruff" in result.stderr:
                # BEFORE the output-parsing branch: a missing ruff MODULE is
                # an infrastructure verdict about the interpreter, not lint
                # findings — mirror lint.LintEvaluator (ADR-0089).
                self._report_ruff_not_found(
                    f"ruff is not installed for {self.interpreter} "
                    "(python -m ruff: No module named ruff)"
                )
            elif result.stdout:
                self._parse_ruff_output(result.stdout)
            elif result.returncode != 0:
                # Non-zero exit with no parseable output (e.g. the child was
                # killed mid-run) is never silence — mirror lint.py, which
                # reports the exit code alone.
                self.report.add_issue(
                    CodeIssue(
                        file=self.current_file,
                        line=1,
                        code="RUFF_CHECK_ERROR",
                        message=f"ruff check failed (exit code {result.returncode})",
                        severity=Severity.ERROR,
                        category="ruff",
                    )
                )

        except subprocess.TimeoutExpired:
            self.report.add_issue(
                CodeIssue(
                    file=self.current_file,
                    line=1,
                    code="RUFF_TIMEOUT",
                    message="Ruff check timed out after 30 seconds",
                    severity=Severity.ERROR,
                    category="ruff",
                )
            )
        except FileNotFoundError:
            self._report_ruff_not_found(
                f"Python interpreter not found: {self.interpreter}"
            )

    def _parse_ruff_output(self, output: str) -> None:
        """Parse ruff output lines."""
        for line in output.strip().split("\n"):
            if not line or ":" not in line:
                continue

            # Format: path:line:col: code message
            parts = line.split(":", 3)
            if len(parts) < 4:
                continue

            file_path, line_str, col_str, message = parts

            # Map ruff codes to severity
            code = message.strip().split()[0] if message.strip() else "RUFF"
            severity = self._map_ruff_severity(code)
            fixable = self._is_ruff_fixable(code)

            self.report.add_issue(
                CodeIssue(
                    file=file_path,
                    line=int(line_str),
                    column=int(col_str),
                    code=code,
                    message=message.strip(),
                    severity=severity,
                    category="ruff",
                    fixable=fixable,
                )
            )

    def _map_ruff_severity(self, code: str) -> Severity:
        """Map ruff error codes to severity levels."""
        # First character indicates category
        if code.startswith("F"):  # PyFlakes
            return Severity.ERROR
        elif code.startswith("E"):  # Pycodestyle errors
            return Severity.ERROR
        elif code.startswith("W"):  # Pycodestyle warnings
            return Severity.WARNING
        elif code.startswith("I"):  # Isort
            return Severity.INFO
        elif code.startswith("UP"):  # Pyupgrade
            return Severity.INFO
        else:
            return Severity.WARNING

    def _is_ruff_fixable(self, code: str) -> bool:
        """Return True if ruff can auto-fix this code."""
        # Codes known to be auto-fixable by ruff
        fixable_prefixes = ("I", "W", "UP")
        fixable_codes = {"F401", "F811", "E501"}
        if code.startswith(fixable_prefixes):
            return True
        if code in fixable_codes:
            return True
        return False


class RuffFormatChecker(BaseChecker):
    """Run ruff format on files."""

    def __init__(self, report: ReviewReport, interpreter: str | None = None) -> None:
        super().__init__(report)
        # ADR-0089: ruff runs as ``<interpreter> -m ruff`` — see RuffChecker.
        self.interpreter = interpreter if interpreter is not None else sys.executable

    def _check_file_impl(self, file_path: Path) -> None:
        """Format the file with ruff."""
        try:
            result = subprocess.run(
                [self.interpreter, "-m", "ruff", "format", str(file_path)],
                capture_output=True,
                text=True,
                timeout=30,
            )
            if result.returncode != 0 and "No module named ruff" in result.stderr:
                # Missing ruff MODULE: infrastructure verdict about the
                # interpreter, reported BEFORE the generic format-error
                # branch — mirror lint.LintEvaluator (ADR-0089).
                self._report_ruff_not_found(
                    f"ruff is not installed for {self.interpreter} "
                    "(python -m ruff: No module named ruff)"
                )
            elif result.returncode != 0 and result.stderr:
                self.report.add_issue(
                    CodeIssue(
                        file=self.current_file,
                        line=1,
                        code="RUFF_FORMAT_ERROR",
                        message=f"Ruff format failed: {result.stderr.strip()}",
                        severity=Severity.ERROR,
                        category="ruff",
                    )
                )
            elif result.returncode != 0:
                # Non-zero exit with no output is never silence.
                self.report.add_issue(
                    CodeIssue(
                        file=self.current_file,
                        line=1,
                        code="RUFF_FORMAT_ERROR",
                        message=f"Ruff format failed (exit code {result.returncode})",
                        severity=Severity.ERROR,
                        category="ruff",
                    )
                )
        except subprocess.TimeoutExpired:
            self.report.add_issue(
                CodeIssue(
                    file=self.current_file,
                    line=1,
                    code="RUFF_FORMAT_TIMEOUT",
                    message="Ruff format timed out after 30 seconds",
                    severity=Severity.ERROR,
                    category="ruff",
                )
            )
        except FileNotFoundError:
            self._report_ruff_not_found(
                f"Python interpreter not found: {self.interpreter}"
            )


class MyPyChecker(BaseChecker):
    """Run mypy for type checking."""

    def __init__(self, report: ReviewReport, mypy_cmd: list[str] | None = None) -> None:
        super().__init__(report)
        # `python -m mypy` (not the bare "mypy" binary) by default: the
        # caller's PATH is not guaranteed to include the venv's bin/, but
        # sys.executable always resolves to this process's own interpreter,
        # which is the venv Python that has mypy installed. ``mypy_cmd`` is
        # a real-subprocess injection seam (not mocking) so tests can point
        # at a genuinely nonexistent binary to exercise the MYPY_NOT_FOUND
        # path without depending on PATH.
        self._mypy_cmd = (
            mypy_cmd if mypy_cmd is not None else [sys.executable, "-m", "mypy"]
        )

    def _check_file_impl(self, file_path: Path) -> None:
        """Run mypy on the file."""
        try:
            result = subprocess.run(
                [*self._mypy_cmd, "--no-error-summary", str(file_path)],
                capture_output=True,
                text=True,
                timeout=30,
            )

            if result.stdout:
                self._parse_mypy_output(result.stdout)

        except subprocess.TimeoutExpired:
            self.report.add_issue(
                CodeIssue(
                    file=self.current_file,
                    line=1,
                    code="MYPY_TIMEOUT",
                    message="MyPy type check timed out after 30 seconds",
                    severity=Severity.ERROR,
                    category="mypy",
                )
            )
        except FileNotFoundError:
            self.report.add_issue(
                CodeIssue(
                    file=self.current_file,
                    line=1,
                    code="MYPY_NOT_FOUND",
                    message="MyPy not installed. Install with: uv add --dev mypy",
                    severity=Severity.ERROR,
                    category="mypy",
                )
            )

    def _parse_mypy_output(self, output: str) -> None:
        """Parse mypy output lines."""
        for line in output.strip().split("\n"):
            if not line or "error:" not in line:
                continue

            # Format: path:line: error: message
            if ":" not in line:
                continue

            # Extract file path, line number, and message
            parts = line.split(":", 3)
            if len(parts) < 4:
                continue

            file_path, line_str, _, message = parts

            # mypy sometimes emits errors without a line number
            # (e.g. "path: error: ..." for cross-file diagnostics).
            # Skip those — they are not tied to the file under review.
            try:
                line_num = int(line_str)
            except ValueError:
                continue

            self.report.add_issue(
                CodeIssue(
                    file=file_path,
                    line=line_num,
                    code="MYPY_ERROR",
                    message=message.strip(),
                    severity=Severity.ERROR,
                    category="mypy",
                )
            )


class StyleChecker(BaseChecker):
    """AST-based style checks (Google Python Style Guide)."""

    # Google Style constants
    MAX_LINE_LENGTH = 88  # Black/ruff default
    MAX_FUNCTION_LENGTH = 40
    MAX_CLASS_LENGTH = 200
    MAX_ARGS = 5
    MAX_LOCALS = 15
    MAX_RETURNS = 4
    MAX_BRANCHES = 12

    def _check_file_impl(self, file_path: Path) -> None:
        """Run all style checks."""
        # Basic checks that don't require parsing
        self._check_formatting()
        self._check_test_isolation(file_path)

        # AST-based checks
        try:
            tree = ast.parse(self.source)
            self._check_docstrings(tree)
            self._check_type_hints(tree)
            self._check_naming_conventions(tree)
            self._check_complexity(tree)
            self._check_imperative_accumulator(tree)
            self._check_code_patterns(tree)

        except SyntaxError as e:
            self.report.add_issue(
                CodeIssue(
                    file=self.current_file,
                    line=e.lineno or 1,
                    column=e.offset or 0,
                    code="SYNTAX_ERROR",
                    message=f"Syntax error: {e.msg}",
                    severity=Severity.ERROR,
                    category="syntax",
                )
            )

    def _check_formatting(self) -> None:
        """Check tab indentation.

        TRAILING_WHITESPACE and MISSING_FINAL_NEWLINE were removed on
        2026-08-06: ruff already runs W291 and W292 under the enabled "W"
        selector, and black REWRITES both (measured — trailing whitespace
        removed, final newline added, in one pass). Reporting what an enabled
        linter reports and a formatter silently fixes is duplication, not
        enforcement.

        TAB_INDENTATION stays: it is ERROR severity here, so unlike the other
        two it can actually fail the gate.
        """
        for i, line in enumerate(self.source_lines, 1):
            if "\t" in line:
                self.report.add_issue(
                    CodeIssue(
                        file=self.current_file,
                        line=i,
                        column=line.index("\t"),
                        code="TAB_INDENTATION",
                        message="Use spaces for indentation, not tabs",
                        severity=Severity.ERROR,
                        category="style",
                        fixable=True,
                    )
                )

    def _check_docstrings(self, tree: ast.Module) -> None:
        """Check docstring presence and format."""
        # Check module docstring
        module_doc = ast.get_docstring(tree)
        if not module_doc:
            self.report.add_issue(
                CodeIssue(
                    file=self.current_file,
                    line=1,
                    code="MISSING_MODULE_DOCSTRING",
                    message="Module is missing a docstring",
                    severity=Severity.INFO,
                    suggestion='Add a module docstring at the top: """Module description."""',
                    category="documentation",
                )
            )

        # Check classes and functions
        for node in ast.walk(tree):
            if isinstance(node, ast.ClassDef):
                class_doc = ast.get_docstring(node)
                if not class_doc and not node.name.startswith("_"):
                    self.report.add_issue(
                        CodeIssue(
                            file=self.current_file,
                            line=node.lineno,
                            code="MISSING_CLASS_DOCSTRING",
                            message=f"Class '{node.name}' is missing a docstring",
                            severity=Severity.INFO,
                            category="documentation",
                        )
                    )

            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                func_doc = ast.get_docstring(node)
                if (
                    not func_doc
                    and not node.name.startswith("_")
                    and node.name not in ("__init__", "__call__")
                ):
                    self.report.add_issue(
                        CodeIssue(
                            file=self.current_file,
                            line=node.lineno,
                            code="MISSING_FUNCTION_DOCSTRING",
                            message=f"Function '{node.name}' is missing a docstring",
                            severity=Severity.INFO,
                            category="documentation",
                        )
                    )

    def _check_type_hints(self, tree: ast.Module) -> None:
        """Check type hints on public functions."""
        for node in ast.walk(tree):
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                # Skip private methods
                if node.name.startswith("_") and not node.name.startswith("__"):
                    continue

                # Check return type
                if node.returns is None and node.name != "__init__":
                    self.report.add_issue(
                        CodeIssue(
                            file=self.current_file,
                            line=node.lineno,
                            code="MISSING_RETURN_TYPE",
                            message=f"Function '{node.name}' missing return type annotation",
                            severity=Severity.WARNING,
                            suggestion="Add '-> ReturnType' annotation",
                            category="type-safety",
                        )
                    )

                # Check parameters
                for arg in node.args.args:
                    if arg.annotation is None and arg.arg not in ("self", "cls"):
                        self.report.add_issue(
                            CodeIssue(
                                file=self.current_file,
                                line=node.lineno,
                                code="MISSING_PARAMETER_TYPE",
                                message=f"Parameter '{arg.arg}' in '{node.name}' missing type annotation",
                                severity=Severity.WARNING,
                                category="type-safety",
                            )
                        )

    def _check_naming_conventions(self, tree: ast.Module) -> None:
        """Check naming conventions."""
        for node in ast.walk(tree):
            # Class names: PascalCase
            if isinstance(node, ast.ClassDef):
                if node.name.startswith("_"):
                    continue
                if not re.match(r"^[A-Z][a-zA-Z0-9]*$", node.name):
                    self.report.add_issue(
                        CodeIssue(
                            file=self.current_file,
                            line=node.lineno,
                            code="CLASS_NAMING",
                            message=f"Class name '{node.name}' should use PascalCase",
                            severity=Severity.ERROR,
                            category="naming",
                        )
                    )

            # Function names: snake_case (skip special methods and the
            # ast.NodeVisitor dispatch protocol, whose visit_<NodeName>
            # casing is mandated by the stdlib)
            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                if not (
                    node.name.startswith("__") and node.name.endswith("__")
                ) and not node.name.startswith("visit_"):
                    if not re.match(r"^[a-z_][a-z0-9_]*$", node.name):
                        self.report.add_issue(
                            CodeIssue(
                                file=self.current_file,
                                line=node.lineno,
                                code="FUNCTION_NAMING",
                                message=f"Function name '{node.name}' should use snake_case",
                                severity=Severity.ERROR,
                                category="naming",
                            )
                        )

    # Methods that GROW a collection in place. Each is the imperative form of
    # a declarative construct: append/extend -> comprehension or list(map(..)),
    # add -> set comprehension, update -> dict comprehension or a merge.
    _ACCUMULATOR_METHODS = {"append", "add", "extend", "update"}
    # Empty-collection constructors — the accumulator's initialiser.
    _EMPTY_CTORS = {"list", "set", "dict"}

    @staticmethod
    def _empty_collection(node: ast.AST) -> bool:
        """ATOM `empty_collection(Expr)` — an empty list/set/dict literal or call."""
        if isinstance(node, (ast.List, ast.Dict)) and not getattr(
            node, "elts", getattr(node, "keys", [])
        ):
            return True
        return (
            isinstance(node, ast.Call)
            and isinstance(node.func, ast.Name)
            and node.func.id in StyleChecker._EMPTY_CTORS
            and not node.args
        )

    @classmethod
    def _accumulator_inits(cls, scope: ast.AST) -> dict[str, int]:
        """FACT `accumulator_init(Var, Line)` — name bound to an empty collection."""
        assigns = [
            n for n in ast.walk(scope) if isinstance(n, (ast.Assign, ast.AnnAssign))
        ]
        targets = (
            (t, n)
            for n in assigns
            for t in (n.targets if isinstance(n, ast.Assign) else [n.target])
        )
        return {
            t.id: n.lineno
            for t, n in targets
            if isinstance(t, ast.Name)
            and n.value is not None
            and cls._empty_collection(n.value)
        }

    @classmethod
    def _loop_mutations(cls, loop: ast.AST) -> set[str]:
        """FACT `loop_mutates(Var, Method)` — names grown inside the loop body."""
        calls = [n for n in ast.walk(loop) if isinstance(n, ast.Call)]
        return {
            c.func.value.id
            for c in calls
            if isinstance(c.func, ast.Attribute)
            and c.func.attr in cls._ACCUMULATOR_METHODS
            and isinstance(c.func.value, ast.Name)
        }

    def _check_imperative_accumulator(self, tree: ast.Module) -> None:
        """IMPERATIVE_ACCUMULATOR — a loop that only grows a collection.

        Atoms/facts this rule is composed from, so it lifts into the Prolog
        gate layer unchanged (plans/prolog-gates-implementation-plan.md):

            empty_collection(Expr)          -- atom
            accumulator_init(Var, Line)     -- fact
            loop_mutates(Var, Method)       -- fact
            loop_body_size(Loop, N)         -- fact
            branchless(Loop)                -- fact

            imperative_accumulator(Var, Line) :-
                accumulator_init(Var, _),
                loop_mutates(Var, _),
                loop_body_size(Loop, N), N =< 2,
                branchless(Loop).

        `for x in xs: out.append(f(x))` is `[f(x) for x in xs]`; the
        imperative form hides the shape of the result behind a mutation and
        an initialiser that must be read together. Deliberately narrow: a
        loop with branches or a body doing more than accumulate is left
        alone, because there the loop is carrying real control flow.
        """
        for scope in ast.walk(tree):
            if not isinstance(scope, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            inits = self._accumulator_inits(scope)
            if not inits:
                continue
            loops = [
                n for n in ast.walk(scope) if isinstance(n, (ast.For, ast.AsyncFor))
            ]
            for loop in loops:
                # Every statement must BE an accumulator mutation. Bounding
                # body SIZE instead let `out.append(x); sink.write(x)` through
                # — a loop with a side effect is doing real work and its
                # rewrite is not mechanical.
                if not loop.body or not all(
                    isinstance(stmt, ast.Expr)
                    and isinstance(stmt.value, ast.Call)
                    and isinstance(stmt.value.func, ast.Attribute)
                    and stmt.value.func.attr in self._ACCUMULATOR_METHODS
                    and isinstance(stmt.value.func.value, ast.Name)
                    and stmt.value.func.value.id in inits
                    for stmt in loop.body
                ):
                    continue
                grown = self._loop_mutations(loop) & inits.keys()
                for var in sorted(grown):
                    self.report.add_issue(
                        CodeIssue(
                            file=self.current_file,
                            line=loop.lineno,
                            code="IMPERATIVE_ACCUMULATOR",
                            message=(
                                f"loop only grows '{var}' (initialised empty at "
                                f"line {inits[var]}) — express the result "
                                f"directly"
                            ),
                            severity=Severity.WARNING,
                            suggestion=(
                                "use a comprehension, map/filter, or "
                                "functools.reduce"
                            ),
                            category="quality",
                        )
                    )

    def _check_complexity(self, tree: ast.Module) -> None:
        """Check code complexity metrics."""
        for node in ast.walk(tree):
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                # Function length
                if node.end_lineno and node.lineno:
                    func_lines = node.end_lineno - node.lineno
                    if func_lines > self.MAX_FUNCTION_LENGTH:
                        self.report.add_issue(
                            CodeIssue(
                                file=self.current_file,
                                line=node.lineno,
                                code="FUNCTION_TOO_LONG",
                                message=f"Function '{node.name}' is too long ({func_lines} lines > {self.MAX_FUNCTION_LENGTH})",
                                severity=Severity.WARNING,
                                suggestion="Split into smaller functions",
                                category="complexity",
                            )
                        )

                # Argument count
                args_count = len(node.args.args) + len(node.args.kwonlyargs)
                if node.args.vararg:
                    args_count += 1
                if node.args.kwarg:
                    args_count += 1

                if args_count > self.MAX_ARGS:
                    self.report.add_issue(
                        CodeIssue(
                            file=self.current_file,
                            line=node.lineno,
                            code="TOO_MANY_ARGUMENTS",
                            message=f"Function '{node.name}' has too many arguments ({args_count} > {self.MAX_ARGS})",
                            severity=Severity.WARNING,
                            suggestion="Consider using a dataclass or **kwargs",
                            category="complexity",
                        )
                    )

            elif isinstance(node, ast.ClassDef):
                # Class length
                if node.end_lineno and node.lineno:
                    class_lines = node.end_lineno - node.lineno
                    if class_lines > self.MAX_CLASS_LENGTH:
                        self.report.add_issue(
                            CodeIssue(
                                file=self.current_file,
                                line=node.lineno,
                                code="CLASS_TOO_LONG",
                                message=f"Class '{node.name}' is too long ({class_lines} lines > {self.MAX_CLASS_LENGTH})",
                                severity=Severity.WARNING,
                                suggestion="Consider breaking into smaller classes",
                                category="complexity",
                            )
                        )

    def _check_test_isolation(self, file_path: Path) -> None:
        """Flag global graph/Redis teardowns in tests (standards rule 34).

        Concurrent tests share ONE Neo4j DB and Redis instance, so a
        global ``DETACH DELETE`` / ``flushdb()`` wipes data a peer test
        is mid-assertion on. Mirrors check_test_isolation in the review
        engine and scripts/gate_parallel_safety.py.
        """
        name = file_path.name
        is_test = name.startswith("test_") or "tests" in file_path.parts
        if not is_test:
            return
        delete_re = re.compile(r"DETACH\s+DELETE|\)\s*DELETE\s+\w", re.IGNORECASE)
        flush_re = re.compile(r"\.flush(?:db|all)\s*\(")
        scope_re = re.compile(
            r"\$\w+|test_id|_TEST_|__test\b|project_id|session_id|tmp_|uuid",
            re.IGNORECASE,
        )
        lines = self.source_lines
        for i, raw in enumerate(lines, 1):
            line = raw.split("#", 1)[0]
            if not line.strip():
                continue
            if flush_re.search(line):
                self.report.add_issue(
                    CodeIssue(
                        file=self.current_file,
                        line=i,
                        code="TEST_FLUSH_DB",
                        message="flushdb/flushall wipes Redis shared by concurrent workers",
                        severity=Severity.ERROR,
                        suggestion="Delete only this test's keys (scope by test_id)",
                        category="test-isolation",
                    )
                )
                continue
            if delete_re.search(line):
                window = "".join(lines[max(0, i - 7) : i + 4])
                if not scope_re.search(window):
                    self.report.add_issue(
                        CodeIssue(
                            file=self.current_file,
                            line=i,
                            code="TEST_GLOBAL_DELETE",
                            message="Unscoped Neo4j DELETE can wipe a concurrent test's data",
                            severity=Severity.ERROR,
                            suggestion="Add a $param/test_id/project_id filter to the MATCH",
                            category="test-isolation",
                        )
                    )

    def _check_code_patterns(self, tree: ast.Module) -> None:
        """Check for discouraged code patterns."""
        for node in ast.walk(tree):
            # Check for bare except
            if isinstance(node, ast.ExceptHandler):
                if node.type is None:  # Bare except
                    self.report.add_issue(
                        CodeIssue(
                            file=self.current_file,
                            line=node.lineno,
                            code="BARE_EXCEPT",
                            message="Bare 'except:' clause should not be used",
                            severity=Severity.ERROR,
                            suggestion="Use 'except Exception:' or specific exception types",
                            category="error-handling",
                        )
                    )

            # Check for mutable default arguments
            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                for default in node.args.defaults:
                    if isinstance(default, (ast.List, ast.Dict, ast.Set)):
                        self.report.add_issue(
                            CodeIssue(
                                file=self.current_file,
                                line=node.lineno,
                                code="MUTABLE_DEFAULT_ARGUMENT",
                                message=f"Function '{node.name}' has mutable default argument",
                                severity=Severity.ERROR,
                                suggestion="Use None as default and initialize mutable value inside function",
                                category="best-practices",
                            )
                        )

            # Check for bare os.environ["X"] reads (standards rule 30c):


class ChecklistChecker(BaseChecker):
    """Checklist-based code review checks."""

    def _check_file_impl(self, file_path: Path) -> None:
        """Run checklist-based checks."""
        try:
            tree = ast.parse(self.source)
            self._check_class_responsibility(tree)
            self._check_method_self_access(tree)
            self._check_abstract_classes(tree)
            self._check_test_files(tree, file_path)

        except SyntaxError:
            pass

    def _check_class_responsibility(self, tree: ast.Module) -> None:
        """Check class single responsibility (method count)."""
        for node in ast.walk(tree):
            if isinstance(node, ast.ClassDef):
                methods = [
                    n
                    for n in node.body
                    if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))
                ]
                if len(methods) > 10:  # Arbitrary threshold
                    self.report.add_issue(
                        CodeIssue(
                            file=self.current_file,
                            line=node.lineno,
                            code="CLASS_TOO_MANY_METHODS",
                            message=f"Class '{node.name}' has many methods ({len(methods)}), consider single responsibility",
                            severity=Severity.WARNING,
                            category="design",
                        )
                    )

    def _check_method_self_access(self, tree: ast.Module) -> None:
        """Check that methods access self or cls."""
        for node in ast.walk(tree):
            if isinstance(node, ast.ClassDef):
                for method in node.body:
                    if isinstance(method, (ast.FunctionDef, ast.AsyncFunctionDef)):
                        # Skip special methods
                        if method.name.startswith("__") and method.name.endswith("__"):
                            continue

                        # Check if method uses self or cls
                        uses_self = any(
                            isinstance(n, ast.Name) and n.id == "self"
                            for n in ast.walk(method)
                        )
                        uses_cls = any(
                            isinstance(n, ast.Name) and n.id == "cls"
                            for n in ast.walk(method)
                        )

                        if not (uses_self or uses_cls):
                            # Check if it's marked as staticmethod
                            is_static = any(
                                isinstance(d, ast.Name) and d.id == "staticmethod"
                                for d in method.decorator_list
                            )

                            if not is_static:
                                self.report.add_issue(
                                    CodeIssue(
                                        file=self.current_file,
                                        line=method.lineno,
                                        code="METHOD_NO_SELF_ACCESS",
                                        message=f"Method '{method.name}' does not access self or cls; consider making it a static method or module-level function",
                                        severity=Severity.INFO,
                                        category="design",
                                    )
                                )

    def _check_abstract_classes(self, tree: ast.Module) -> None:
        """Check abstract class usage."""
        for node in ast.walk(tree):
            if isinstance(node, ast.ClassDef):
                # Check for abstract methods
                abstract_methods = []
                for method in node.body:
                    if isinstance(method, (ast.FunctionDef, ast.AsyncFunctionDef)):
                        for decorator in method.decorator_list:
                            if (
                                isinstance(decorator, ast.Name)
                                and decorator.id == "abstractmethod"
                            ):
                                abstract_methods.append(method.name)

                # If has abstract methods, check if inherits from ABC
                if abstract_methods:
                    base_names = []
                    for base in node.bases:
                        if isinstance(base, ast.Name):
                            base_names.append(base.id)
                        elif isinstance(base, ast.Attribute):
                            base_names.append(base.attr)

                    if "ABC" not in base_names and "abc.ABC" not in str(node.bases):
                        self.report.add_issue(
                            CodeIssue(
                                file=self.current_file,
                                line=node.lineno,
                                code="ABSTRACT_CLASS_NO_ABC",
                                message=f"Class '{node.name}' has abstract methods but does not inherit from ABC",
                                severity=Severity.ERROR,
                                category="design",
                            )
                        )

    def _check_test_files(self, tree: ast.Module, file_path: Path) -> None:
        """Special checks for test files."""
        filename = file_path.name
        if not (filename.startswith("test_") or filename.endswith("_test.py")):
            return
        # Name alone is not enough: production modules may legitimately be
        # called test_*.py because they are ABOUT tests, not written in pytest.
        # `src/orchestrata_ultimate/gateway/test_runner.py` is the live
        # test-execution service, and its `except ProcessLookupError: pass`
        # was reported as "Pytest file contains pass statement" — a finding
        # that can never be actioned, because there is no assertion to add.
        if "src" in file_path.parts:
            return

        # Check for pass statements in tests
        class PassVisitor(ast.NodeVisitor):
            # visit_* casing is mandated by the ast.NodeVisitor dispatch
            # protocol; FUNCTION_NAMING findings on these are false positives.
            def __init__(self, checker: BaseChecker) -> None:
                self.checker = checker
                self.current_class: str | None = None
                self.current_function: str | None = None

            def visit_ClassDef(self, node: ast.ClassDef) -> None:  # noqa: N802
                old_class = self.current_class
                self.current_class = node.name
                self.generic_visit(node)
                self.current_class = old_class

            def visit_FunctionDef(self, node: ast.FunctionDef) -> None:  # noqa: N802
                old_function = self.current_function
                self.current_function = node.name
                self.generic_visit(node)
                self.current_function = old_function

            def visit_Pass(self, node: ast.Pass) -> None:  # noqa: N802
                self.checker.report.add_issue(
                    CodeIssue(
                        file=self.checker.current_file,
                        line=node.lineno,
                        code="TEST_PASS_STATEMENT",
                        message="Pytest file contains pass statement - tests should have assertions",
                        severity=Severity.ERROR,
                        category="testing",
                    )
                )

        visitor = PassVisitor(self)
        visitor.visit(tree)

        # Check for mock usage in test files


def find_python_files(paths: list[str]) -> list[Path]:
    """Find all Python files in given paths."""
    files = []
    for path_str in paths:
        path = Path(path_str)
        if path.is_file() and path.suffix == ".py":
            files.append(path)
        elif path.is_dir():
            files.extend(path.rglob("*.py"))
    return files


def filter_issues_by_severity(
    issues: list[CodeIssue], min_severity: Severity
) -> list[CodeIssue]:
    """Filter issues by minimum severity."""
    severity_order = {Severity.ERROR: 0, Severity.WARNING: 1, Severity.INFO: 2}
    min_level = severity_order[min_severity]
    return [issue for issue in issues if severity_order[issue.severity] <= min_level]


def output_text(
    report: ReviewReport,
    filtered_issues: list[CodeIssue],
    output_file: str | None = None,
) -> None:
    """Output report in text format."""
    output = sys.stdout if output_file is None else open(output_file, "w")

    print(f"\n{'=' * 80}", file=output)
    print("UNIFIED CODE REVIEW REPORT", file=output)
    print(f"{'=' * 80}", file=output)
    print(f"\nFiles Analyzed: {report.files_analyzed}", file=output)
    print(f"Total Issues: {len(filtered_issues)}", file=output)

    # Quick fixes section
    quick_fixes = [i for i in filtered_issues if i.fixable]
    if quick_fixes:
        print("\n" + "-" * 40, file=output)
        print(
            f"QUICK FIXES ({len(quick_fixes)} issues — run with --fix and --format-code)",
            file=output,
        )
        print("-" * 40, file=output)
        for issue in sorted(quick_fixes, key=lambda x: (x.file, x.line, x.column)):
            print(
                f"  [FIXABLE] Line {issue.line}:{issue.column} [{issue.code}] {issue.message}",
                file=output,
            )
            if issue.suggestion:
                print(f"    Suggestion: {issue.suggestion}", file=output)

    # Summary by category
    print("\n" + "-" * 40, file=output)
    print("SUMMARY BY CATEGORY", file=output)
    print("-" * 40, file=output)
    for category, count in sorted(report.summary_by_category.items()):
        print(f"  {category}: {count}", file=output)

    # Summary by severity
    print("\n" + "-" * 40, file=output)
    print("SUMMARY BY SEVERITY", file=output)
    print("-" * 40, file=output)
    for severity in Severity:
        count = report.summary_by_severity.get(severity, 0)
        print(f"  {severity.value}: {count}", file=output)

    # All issues
    print("\n" + "-" * 40, file=output)
    print("ALL ISSUES", file=output)
    print("-" * 40, file=output)

    current_file = None
    for issue in sorted(filtered_issues, key=lambda x: (x.file, x.line, x.column)):
        if issue.file != current_file:
            current_file = issue.file
            print(f"\n{current_file}:", file=output)

        fix_marker = "[FIXABLE] " if issue.fixable else ""
        severity_marker = {
            "ERROR": "[ERROR]",
            "WARNING": "[WARN]",
            "INFO": "[INFO]",
        }.get(issue.severity.value, "•")
        print(
            f"  {fix_marker}{severity_marker} Line {issue.line}:{issue.column} [{issue.code}] {issue.message}",
            file=output,
        )
        if issue.suggestion:
            print(f"    Suggestion: {issue.suggestion}", file=output)

    print(f"\n{'=' * 80}", file=output)

    if output_file:
        output.close()


def output_json(
    report: ReviewReport,
    filtered_issues: list[CodeIssue],
    output_file: str | None = None,
) -> None:
    """Output report in JSON format."""
    output = {
        "timestamp": datetime.now().isoformat(),
        "files_analyzed": report.files_analyzed,
        "total_issues": len(filtered_issues),
        "summary_by_category": dict(report.summary_by_category),
        "summary_by_severity": {
            k.value: v for k, v in report.summary_by_severity.items()
        },
        "quick_fixes": [asdict(issue) for issue in filtered_issues if issue.fixable],
        "issues": [asdict(issue) for issue in filtered_issues],
    }

    json_str = json.dumps(output, indent=2)
    if output_file:
        Path(output_file).write_text(json_str)
        print(f"JSON report written to {output_file}")
    else:
        print(json_str)


def output_markdown(
    report: ReviewReport,
    filtered_issues: list[CodeIssue],
    output_file: str | None = None,
) -> None:
    """Output report in markdown format."""
    lines = [
        "# Unified Code Review Report\n",
        f"**Generated:** {datetime.now().isoformat()}\n",
        f"**Files Analyzed:** {report.files_analyzed}\n",
        f"**Total Issues:** {len(filtered_issues)}\n",
        "\n## Summary by Category\n",
        "| Category | Count |\n|----------|-------|",
    ]

    for category, count in sorted(report.summary_by_category.items()):
        lines.append(f"| {category} | {count} |")

    lines.extend(
        ["\n## Summary by Severity\n", "| Severity | Count |\n|----------|-------|"]
    )

    for severity in Severity:
        count = report.summary_by_severity.get(severity, 0)
        lines.append(f"| {severity.value} | {count} |")

    # Quick fixes section
    quick_fixes = [i for i in filtered_issues if i.fixable]
    if quick_fixes:
        lines.extend(
            [
                "\n## Quick Fixes\n",
                "| File | Line | Code | Message | Suggestion |\n",
                "|------|------|------|---------|------------|",
            ]
        )
        for issue in sorted(quick_fixes, key=lambda x: (x.file, x.line, x.column)):
            suggestion = issue.suggestion or ""
            lines.append(
                f"| {issue.file} | {issue.line} | {issue.code} | {issue.message} | {suggestion} |"
            )

    lines.extend(
        [
            "\n## All Issues\n",
            "| File | Line | Code | Severity | Message | Suggestion |\n",
            "|------|------|------|----------|---------|------------|",
        ]
    )

    for issue in sorted(filtered_issues, key=lambda x: (x.file, x.line, x.column)):
        suggestion = issue.suggestion or ""
        fix_badge = "[FIXABLE] " if issue.fixable else ""
        lines.append(
            f"| {issue.file} | {issue.line} | {issue.code} | {issue.severity.value} | {fix_badge}{issue.message} | {suggestion} |"
        )

    output = "\n".join(lines)
    if output_file:
        Path(output_file).write_text(output)
        print(f"Markdown report written to {output_file}")
    else:
        print(output)


class TraceabilityChecker(BaseChecker):
    """Check that all classes and methods have @traces decorators."""

    ENUM_BASES = {"Enum", "StrEnum", "IntEnum", "Flag", "IntFlag"}
    PYDANTIC_BASES = {"BaseModel", "BaseSettings", "CXPMixin", "BaseAgent", "BaseVolk"}
    PYDANTIC_DECORATORS = {
        "field_validator",
        "model_validator",
        "validator",
        "computed_field",
        "field_serializer",
        "model_serializer",
    }

    def _check_file_impl(self, file_path: Path) -> None:
        """Check traceability of a single file."""
        rel = str(file_path)
        # @traces applies to the traced product package only — the tools
        # package (src/orchestrata_tools) and scripts/ are untraced by
        # convention (two-lane rule, AGENTS.md "@traces Conventions";
        # trace-lint scopes to src/orchestrata_ultimate for the same reason).
        parts = Path(rel).parts
        if "orchestrata_tools" in parts or "scripts" in parts:
            return
        # Skip domain/, __init__.py, traceability internals, tests
        if "/domain/" in rel or file_path.name == "__init__.py":
            return
        if "traceability_utils.py" in rel or rel.endswith(
            "core/tracing/traceability.py"
        ):
            return
        if "tests" in Path(rel).parts:
            return
        if "proto" in Path(rel).parts:
            return
        if "in_memory" in rel:
            return

        try:
            tree = ast.parse(self.source, filename=str(file_path))
        except SyntaxError:
            return

        for node in tree.body:
            if isinstance(node, ast.ClassDef):
                self._check_class(node, file_path)

    def _check_class(self, node: ast.ClassDef, file_path: Path) -> None:
        """Check a single class and its methods."""
        if self._is_enum(node):
            return
        if self._is_exception(node):
            return
        if self._is_protocol(node):
            return
        if self._is_dataclass(node):
            return
        if self._is_typeddict(node):
            return
        if self._is_pydantic(node):
            return
        if self._is_abc(node):
            return
        if node.name.startswith("_"):
            return

        if not self._has_traces(node):
            # WARNING, not ERROR: per AGENTS.md "@traces Conventions"
            # (sparse-and-accurate beats padded), a class with no
            # genuinely-matching requirement STAYS untraced — missing
            # @traces is a finding to review, not a failure to block on.
            self.report.add_issue(
                CodeIssue(
                    file=str(file_path),
                    line=node.lineno,
                    column=node.col_offset,
                    code="TRACE_MISSING_CLASS",
                    message=f"Class '{node.name}' has no @traces decorator",
                    severity=Severity.WARNING,
                    category="traceability",
                )
            )

        is_pydantic = self._is_pydantic(node)
        for item in node.body:
            if not isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            # Skip property, pydantic validators, and dunders on pydantic models
            if self._has_property(item):
                continue
            if self._has_pydantic_decorator(item):
                continue
            if is_pydantic and item.name.startswith("__") and item.name.endswith("__"):
                continue

            if not self._has_traces(item):
                self.report.add_issue(
                    CodeIssue(
                        file=str(file_path),
                        line=item.lineno,
                        column=item.col_offset,
                        code="TRACE_MISSING_METHOD",
                        message=f"Method '{node.name}.{item.name}' has no @traces decorator",
                        severity=Severity.INFO,
                        category="traceability",
                    )
                )

    @staticmethod
    def _has_traces(
        node: ast.ClassDef | ast.FunctionDef | ast.AsyncFunctionDef,
    ) -> bool:
        for dec in node.decorator_list:
            if isinstance(dec, ast.Name) and dec.id == "traces":
                return True
            if (
                isinstance(dec, ast.Call)
                and isinstance(dec.func, ast.Name)
                and dec.func.id == "traces"
            ):
                return True
        return False

    def _is_enum(self, node: ast.ClassDef) -> bool:
        return any(
            (
                b.id
                if isinstance(b, ast.Name)
                else (b.attr if isinstance(b, ast.Attribute) else "")
            )
            in self.ENUM_BASES
            for b in node.bases
        )

    def _is_pydantic(self, node: ast.ClassDef) -> bool:
        return any(
            (
                b.id
                if isinstance(b, ast.Name)
                else (b.attr if isinstance(b, ast.Attribute) else "")
            )
            in self.PYDANTIC_BASES
            for b in node.bases
        )

    @staticmethod
    def _is_exception(node: ast.ClassDef) -> bool:
        base_names = [
            (
                b.id
                if isinstance(b, ast.Name)
                else (b.attr if isinstance(b, ast.Attribute) else "")
            )
            for b in node.bases
        ]
        return any(
            name
            in ("Exception", "BaseException", "ValueError", "RuntimeError", "TypeError")
            or name.endswith("Exception")
            or name.endswith("Error")
            for name in base_names
        )

    @staticmethod
    def _is_protocol(node: ast.ClassDef) -> bool:
        return any(
            (
                b.id
                if isinstance(b, ast.Name)
                else (b.attr if isinstance(b, ast.Attribute) else "")
            )
            == "Protocol"
            for b in node.bases
        )

    @staticmethod
    def _is_dataclass(node: ast.ClassDef) -> bool:
        for dec in node.decorator_list:
            if isinstance(dec, ast.Name) and dec.id == "dataclass":
                return True
            if (
                isinstance(dec, ast.Call)
                and isinstance(dec.func, ast.Name)
                and dec.func.id == "dataclass"
            ):
                return True
        return False

    @staticmethod
    def _is_typeddict(node: ast.ClassDef) -> bool:
        return any(
            (
                b.id
                if isinstance(b, ast.Name)
                else (b.attr if isinstance(b, ast.Attribute) else "")
            )
            == "TypedDict"
            for b in node.bases
        )

    @staticmethod
    def _is_abc(node: ast.ClassDef) -> bool:
        return any(
            (
                b.id
                if isinstance(b, ast.Name)
                else (b.attr if isinstance(b, ast.Attribute) else "")
            )
            in ("ABC",)
            for b in node.bases
        )

    @staticmethod
    def _has_property(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
        for dec in node.decorator_list:
            if isinstance(dec, ast.Name) and dec.id == "property":
                return True
            if isinstance(dec, ast.Attribute) and dec.attr in (
                "setter",
                "deleter",
                "getter",
            ):
                return True
        return False

    def _has_pydantic_decorator(
        self, node: ast.FunctionDef | ast.AsyncFunctionDef
    ) -> bool:
        for dec in node.decorator_list:
            if isinstance(dec, ast.Name) and dec.id in self.PYDANTIC_DECORATORS:
                return True
            if (
                isinstance(dec, ast.Call)
                and isinstance(dec.func, ast.Name)
                and dec.func.id in self.PYDANTIC_DECORATORS
            ):
                return True
        return False


# Standards-body namespace IRIs: identifiers in RDF/OWL/XSD/XML documents.
_NAMESPACE_IRI_RE = re.compile(
    r"https?://(?:www\.)?(?:w3\.org|purl\.org|xmlns\.com|schema\.org"
    r"|openrdf\.org|rdfs?\.org)/",
    re.I,
)
# %s / %d / {name} / $VAR placeholders mark a template, not a literal address.
_URI_PLACEHOLDER_RE = re.compile(r"%[sd]|\{[^}]*\}|\$\{?\w+")


class ConnectionStringChecker(BaseChecker):
    """Check for hardcoded connection strings and env-var defaults."""

    # Regex patterns
    IPV4_RE = re.compile(r"(?:\d{1,3}\.){3}\d{1,3}")
    URI_RE = re.compile(r"(bolt|redis|mongodb|postgres|mysql|https?)://[^\s\"']+")
    LOCALHOST_PORT_RE = re.compile(r"localhost:\d{2,5}")
    # TEST-NET and documentation IPs (RFC 5737)
    TEST_NET_PREFIXES = ("192.0.2.", "198.51.100.", "203.0.113.")
    # Loopback / multicast / reserved
    SKIP_IP_PREFIXES = (
        "127.",
        "0.0.0.0",
        "255.",
        "240.",
        "192.0.2.",
        "198.51.100.",
        "203.0.113.",
        # Link-local (RFC 3927) — includes the 169.254.169.254 cloud-metadata
        # endpoint, which appears in SSRF *blocklists* as a security constant,
        # never as reachable deployment infrastructure.
        "169.254.",
    )

    def _check_file_impl(self, file_path: Path) -> None:
        """Scan file for hardcoded connection values."""
        if "tests" in Path(str(file_path)).parts:
            return
        try:
            tree = ast.parse(self.source)
        except SyntaxError:
            return

        # Collect docstring line numbers to skip
        docstring_lines = set()
        for node in ast.walk(tree):
            if isinstance(
                node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)
            ):
                if (
                    node.body
                    and isinstance(node.body[0], ast.Expr)
                    and isinstance(node.body[0].value, ast.Constant)
                    and isinstance(node.body[0].value.value, str)
                ):
                    docstring_lines.add(node.body[0].lineno)

        # Collect string constants that are default args to os.environ.get / os.getenv
        env_get_default_lines = self._collect_env_get_defaults(tree)

        for node in ast.walk(tree):
            if isinstance(node, ast.Constant) and isinstance(node.value, str):
                if node.lineno in docstring_lines:
                    continue
                if node.lineno in env_get_default_lines:
                    continue
                self._check_string(node.value, node.lineno)

    @staticmethod
    def _is_env_get_call(func: ast.expr) -> bool:
        """True for the call target of os.environ.get(...), bare
        environ.get(...), os.getenv(...), or bare getenv(...).

        Matches both the ``os.``-qualified forms (the standard convention
        used throughout this codebase — see AGENTS.md "no bare
        os.environ['X']") and the bare forms (``from os import environ,
        getenv``), so callers don't silently stop recognizing the common
        case.
        """
        if isinstance(func, ast.Attribute) and func.attr == "get":
            if isinstance(func.value, ast.Name) and func.value.id == "environ":
                return True
            if (
                isinstance(func.value, ast.Attribute)
                and func.value.attr == "environ"
                and isinstance(func.value.value, ast.Name)
                and func.value.value.id == "os"
            ):
                return True
            return False
        if isinstance(func, ast.Attribute) and func.attr == "getenv":
            return isinstance(func.value, ast.Name) and func.value.id == "os"
        if isinstance(func, ast.Name) and func.id == "getenv":
            return True
        return False

    def _collect_env_get_defaults(self, tree: ast.AST) -> set[int]:
        """Return line numbers of string constants used as defaults in os.environ.get()."""
        lines: set[int] = set()
        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            func = node.func
            if not self._is_env_get_call(func):
                continue
            # Second positional arg or "default" keyword arg is the fallback
            if len(node.args) >= 2:
                default_arg = node.args[1]
                if isinstance(default_arg, ast.Constant) and isinstance(
                    default_arg.value, str
                ):
                    lines.add(default_arg.lineno)
            for kw in node.keywords:
                if (
                    kw.arg == "default"
                    and isinstance(kw.value, ast.Constant)
                    and isinstance(kw.value.value, str)
                ):
                    lines.add(kw.value.lineno)
        return lines

    def _check_string(self, value: str, lineno: int) -> None:
        """Flag strings that look like connection endpoints."""
        if len(value) < 4:
            return

        # Skip third-party API URLs (not infrastructure)
        if (
            value.startswith("https://api.")
            or value.startswith("https://dashscope.")
            or value.startswith("https://chat.")
            or value.startswith("https://kimi.")
        ):
            return
        if "openrouter.ai" in value or "orchestrata.dev" in value:
            return
        # Public package index (pinned-sdist vendor builds), not deployment infra
        if "pypi.org" in value or "pythonhosted.org" in value:
            return
        # Browser User-Agent strings: version tokens like Chrome/122.0.0.0
        # false-positive the IPv4 regex; a UA string is not infrastructure.
        if value.startswith("Mozilla/"):
            return
        # Public web endpoints used as env-overridable defaults, not infra
        if "duckduckgo.com" in value:
            return

        # IPv4 address anywhere in the string
        ip_match = self.IPV4_RE.search(value)
        if ip_match:
            ip = ip_match.group(0)
            if any(ip.startswith(p) for p in self.SKIP_IP_PREFIXES):
                return
            self.report.add_issue(
                CodeIssue(
                    file=self.current_file,
                    line=lineno,
                    code="HARDCODED_IP",
                    message=f"Hardcoded IP address found: {value[:60]}",
                    severity=Severity.ERROR,
                    category="config",
                )
            )
            return

        # Connection URI (bolt://, redis://, etc.)
        uri_match = self.URI_RE.search(value)
        if uri_match:
            uri = uri_match.group(0)
            # Skip nonexistent/test hostnames
            if "nonexistent" in uri or "custom" in uri:
                return
            # A namespace IRI is an IDENTIFIER, not an endpoint: nothing
            # connects to http://www.w3.org/2002/07/owl#. Firing on the
            # @prefix lines of an RDF/TTL template reported a hardcoded
            # connection where no connection exists.
            if _NAMESPACE_IRI_RE.search(uri):
                return
            # A format template is not an address: "http://%s:%d" and
            # "http://{host}:{port}" are how the endpoint is BUILT from
            # resolved values, which is the pattern RUL-003 asks for.
            if _URI_PLACEHOLDER_RE.search(uri):
                return
            self.report.add_issue(
                CodeIssue(
                    file=self.current_file,
                    line=lineno,
                    code="HARDCODED_URI",
                    message=f"Hardcoded connection URI found: {value[:60]}",
                    severity=Severity.ERROR,
                    category="config",
                )
            )
            return

        # localhost with port
        if self.LOCALHOST_PORT_RE.search(value):
            self.report.add_issue(
                CodeIssue(
                    file=self.current_file,
                    line=lineno,
                    code="HARDCODED_LOCALHOST",
                    message=f"Hardcoded localhost endpoint found: {value[:60]}",
                    severity=Severity.ERROR,
                    category="config",
                )
            )


def run_diagnostics(report: ReviewReport) -> None:
    """Run test diagnostic tools and add findings to the report."""
    import subprocess

    # Test isolation audit
    try:
        result = subprocess.run(
            [sys.executable, "scripts/test_isolation_audit.py", "tests/"],
            capture_output=True,
            text=True,
            timeout=60,
        )
        if result.returncode != 0:
            # Parse violations from output
            for line in result.stdout.split("\n"):
                if "[CREATE_VS_MERGE]" in line or "[HARDCODED_NAME]" in line:
                    report.add_issue(
                        CodeIssue(
                            file="tests/",
                            line=0,
                            code="TEST-ISOLATION",
                            message=line.strip(),
                            severity=Severity.WARNING,
                            category="test-isolation",
                        )
                    )
                elif "[PYTEST_SKIP]" in line:
                    report.add_issue(
                        CodeIssue(
                            file="tests/",
                            line=0,
                            code="TEST-SKIP",
                            message=line.strip(),
                            severity=Severity.INFO,
                            category="test-isolation",
                        )
                    )
    except Exception as e:
        print(f"Diagnostic audit failed: {e}", file=sys.stderr)

    # graph preflight
    try:
        result = subprocess.run(
            [sys.executable, "scripts/graph_test_preflight.py", "tests/", "--offline"],
            capture_output=True,
            text=True,
            timeout=60,
        )
        if result.returncode != 0:
            for line in result.stdout.split("\n"):
                if "[HIGH]" in line:
                    report.add_issue(
                        CodeIssue(
                            file="tests/",
                            line=0,
                            code="GRAPH-CONSTRAINT",
                            message=line.strip(),
                            severity=Severity.WARNING,
                            category="neo4j-constraints",
                        )
                    )
    except Exception as e:
        print(f"graph preflight failed: {e}", file=sys.stderr)


def load_banned_libraries(repo_root: Path) -> list[str]:
    """Read .claude/banned-libraries.txt — distribution names that must not be
    declared as dependencies. Comments (#) and blanks are ignored. Returns []
    if the file is absent."""
    banned_file = repo_root / ".claude" / "banned-libraries.txt"
    if not banned_file.exists():
        return []
    names: list[str] = []
    for raw in banned_file.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if line and not line.startswith("#"):
            names.append(line)
    return names


def check_banned_dependencies(report: ReviewReport, repo_root: Path) -> None:
    """SHOWSTOPPER gate: fail if pyproject.toml declares a banned dependency.

    Matches a banned name only as a quoted TOML requirement string
    (e.g. ``"pytest-xdist>=3.5.0"``) so a name inside a comment (``# pytest-xdist
    is BANNED``) is allowed. Single source of truth: .claude/banned-libraries.txt.
    """
    pyproject = repo_root / "pyproject.toml"
    if not pyproject.exists():
        return
    banned = load_banned_libraries(repo_root)
    if not banned:
        return
    # Boundary after the name: version op / extras '[' / quote / comma / space /
    # end. Includes '[' so "pytest-xdist[psutil]>=3" is still caught.
    patterns = {
        name: re.compile(
            rf"""["']{re.escape(name)}(["'>=~<!;,\s\[]|$)""", re.IGNORECASE
        )
        for name in banned
    }
    for lineno, raw in enumerate(
        pyproject.read_text(encoding="utf-8").splitlines(), start=1
    ):
        code = raw.split("#", 1)[0]  # ignore trailing comments
        for name, pat in patterns.items():
            if pat.search(code):
                report.add_issue(
                    CodeIssue(
                        file=str(pyproject),
                        line=lineno,
                        code="BANNED-DEP",
                        message=(
                            f"Banned dependency '{name}' declared in pyproject.toml "
                            "— see .claude/banned-libraries.txt. Parallelism comes "
                            "from scripts/run_tests.py, never pytest-xdist."
                        ),
                        severity=Severity.ERROR,
                        category="dependencies",
                    )
                )


def check_prolog_boundary(report: ReviewReport, repo_root: Path) -> None:
    """BOUNDARY gate: fail if any Python Prolog query site targets a
    predicate/arity that does not exist in the consulted rule base
    (reports/proposal-wire-prolog-governance.md §7 Phase 1). Same
    subprocess + repo-root-independent pattern as ``check_banned_dependencies``
    — a src-only review must still catch a boundary mismatch anywhere in
    ``src/orchestrata_ultimate``.
    """
    script = repo_root / "scripts" / "check_prolog_boundary.py"
    if not script.exists():
        return
    try:
        result = subprocess.run(
            [sys.executable, str(script), "--json"],
            cwd=str(repo_root),
            capture_output=True,
            text=True,
            timeout=90,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        print(f"Prolog boundary gate failed to run: {exc}", file=sys.stderr)
        return
    if result.returncode == 0:
        return
    try:
        payload = json.loads(result.stdout)
    except json.JSONDecodeError:
        report.add_issue(
            CodeIssue(
                file=str(script),
                line=0,
                code="BOUNDARY",
                message=f"check_prolog_boundary.py failed: {result.stderr.strip()}",
                severity=Severity.ERROR,
                category="prolog-boundary",
            )
        )
        return
    for failure in payload.get("failures", []):
        report.add_issue(
            CodeIssue(
                file=failure.get("file", str(script)),
                line=failure.get("line", 0),
                code="BOUNDARY",
                message=(
                    f"{failure.get('functor')}/{failure.get('arity')} → "
                    f"{failure.get('candidate')}"
                ),
                severity=Severity.ERROR,
                category="prolog-boundary",
            )
        )


def main() -> int:
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description="Unified Code Review Tool for orchestrata-ultimate"
    )
    parser.add_argument(
        "paths", nargs="+", help="Python files or directories to analyze"
    )
    parser.add_argument("--output", "-o", help="Output file for report")
    parser.add_argument(
        "--format",
        choices=["text", "json", "markdown"],
        default="text",
        help="Output format",
    )
    parser.add_argument(
        "--min-severity",
        choices=["ERROR", "WARNING", "INFO"],
        default="INFO",
        help="Minimum severity to report",
    )
    parser.add_argument(
        "--fix", action="store_true", help="Auto-fix fixable ruff issues"
    )
    parser.add_argument(
        "--format-code",
        action="store_true",
        help="Format code with ruff before checking",
    )
    parser.add_argument(
        "--max-files",
        type=int,
        default=None,
        help="Maximum number of files to process (for testing)",
    )
    parser.add_argument(
        "--diagnostics",
        action="store_true",
        help="Run test diagnostic tools (isolation audit, graph preflight)",
    )
    parser.add_argument(
        "--stdout",
        action="store_true",
        help="Also print the formatted report to stdout",
    )

    args = parser.parse_args()

    # Find files
    files = find_python_files(args.paths)

    # Filter out virtual environment files
    files = [f for f in files if ".venv" not in str(f) and "__pycache__" not in str(f)]

    if args.max_files:
        files = files[: args.max_files]
        print(f"Limiting to {args.max_files} files", file=sys.stderr)

    if not files:
        print("No Python files found to analyze", file=sys.stderr)
        return 0

    # Create report and checkers — all run every time for historical record
    report = ReviewReport()
    checkers: list[BaseChecker] = []

    if args.format_code:
        checkers.append(RuffFormatChecker(report))

    checkers.append(RuffChecker(report, fix=args.fix))
    checkers.append(MyPyChecker(report))
    checkers.append(StyleChecker(report))
    checkers.append(ChecklistChecker(report))
    checkers.append(ConnectionStringChecker(report))
    checkers.append(TraceabilityChecker(report))

    # Run diagnostic tools if requested
    if args.diagnostics:
        print("Running diagnostic tools...", file=sys.stderr)
        run_diagnostics(report)

    # Repo root for the gates below — overridable at call time (tests point
    # this at an isolated tmp_path so the gates take their no-op early-return
    # paths instead of spawning against the real repo). Production callers
    # never set the env var, so this resolves identically to before.
    gate_root = Path(
        os.environ.get(
            "CODE_REVIEW_REPO_ROOT", str(Path(__file__).resolve().parent.parent)
        )
    )

    # Banned-dependency gate — one-shot scan of the repo's pyproject.toml,
    # independent of which files were passed for review (a src-only review must
    # still catch a banned dep). Repo root = parent of this script's dir.
    check_banned_dependencies(report, gate_root)

    # Prolog boundary gate — one-shot AST walk of the whole src tree,
    # independent of which files were passed for review (same rationale as
    # the banned-dependency gate above).
    check_prolog_boundary(report, gate_root)

    # Run checkers on each file
    for i, file_path in enumerate(files):
        if i % 10 == 0:
            print(
                f"Processing file {i + 1}/{len(files)}: {file_path.name}",
                file=sys.stderr,
            )

        report.files_analyzed += 1
        for checker in checkers:
            checker.check_file(file_path)

    # Filter issues by severity
    min_severity = Severity(args.min_severity)
    filtered_issues = filter_issues_by_severity(report.issues, min_severity)

    # Always write a dated report to reviews/ for historical record
    reviews_dir = Path("reviews")
    reviews_dir.mkdir(exist_ok=True)
    timestamp = datetime.now().strftime("%Y-%m-%d-%H%M%S")
    ext = {"json": "json", "markdown": "md", "text": "txt"}[args.format]
    dated_path = reviews_dir / f"code-review-{timestamp}.{ext}"

    if args.format == "json":
        output_json(report, filtered_issues, str(dated_path))
        if args.output:
            output_json(report, filtered_issues, args.output)
        if args.stdout:
            output_json(report, filtered_issues, None)
    elif args.format == "markdown":
        output_markdown(report, filtered_issues, str(dated_path))
        if args.output:
            output_markdown(report, filtered_issues, args.output)
        if args.stdout:
            output_markdown(report, filtered_issues, None)
    else:  # text format
        output_text(report, filtered_issues, str(dated_path))
        if args.output:
            output_text(report, filtered_issues, args.output)
        if args.stdout:
            output_text(report, filtered_issues, None)

    print(f"Report written to {dated_path}", file=sys.stderr)

    # Return non-zero exit code if there are errors
    error_count = report.summary_by_severity.get(Severity.ERROR, 0)
    return 1 if error_count > 0 else 0


if __name__ == "__main__":
    sys.exit(main())