orchestrata_ultimate.refactor.codemod.review_checks

Reference › orchestrata_ultimate › refactor › codemod › review_checks

The implementations behind the review catalogue of SPEC-110 FR-012. review_engine is its importer; the descriptors it realises are in review_catalogue, the per-check input is the ReviewContext, and it reuses the shared banned-import visitor from review. It also imports the check functions of the verification.review packages (correctness, quality, safety, structure) and the ctok tokenizer.

_DescribedCheck is the base: it exposes a descriptor's id, category, phase and substrate. ReusedCheck wraps an existing check_* function from verification/review, running it over the mutated module's AST and over lines rebuilt from its token stream, with a scope (all, source or test), an extension for the part of a requirement the reused function does not cover, and string and comment tokens blanked so a literal never feeds a regular expression. IssueAdapter.to_findings converts the reused functions' Issue objects into Finding objects. The checks written for the engine are CyclomaticComplexityCheck (limit 10), TrailingWhitespaceCheck, TabIndentationCheck, FinalNewlineCheck, NamingConventionCheck, BannedLibraryCheck, BroadRaiseCheck, PrintOutsideMainCheck, MutableCallDefaultCheck, SwallowedExceptionCheck, ModuleScopedFixtureCheck and GlobalMutableStateCheck, most backed by a small ast.NodeVisitor. The reusing and writing of checks is described in the S3 packet of the refactor package narrative.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

IssueAdapter

Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 211

Convert verification/review issue tuples into codemod findings.

_DescribedCheck

Line 262

Shared descriptor plumbing for every implementation.

ReusedCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 332

A catalogue check realised by a verification/review function.

_BranchCounter

Bases: ast.NodeVisitor · Line 406

Count branching nodes of one function body, not descending into nested functions, classes or lambdas (each is scored on its own).

_CyclomaticVisitor

Bases: ast.NodeVisitor · Line 437

Flag every function whose complexity exceeds the limit.

CyclomaticComplexityCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 466

CX_CYCLOMATIC: branching nodes per function, default max 10.

TrailingWhitespaceCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 478

ST_TRAILING_WS: trailing blanks outside multi-line string tokens.

TabIndentationCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 510

ST_TAB_INDENT: INDENT tokens that contain a tab.

FinalNewlineCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 526

ST_FINAL_NEWLINE: a non-empty file must end with a newline.

_NamingVisitor

Bases: ast.NodeVisitor · Line 540

NamingConventionCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 564

ST_NAMING: PascalCase classes, snake_case functions.

_BannedLibraryVisitor

Bases: _BannedImportVisitor · Line 575

The codemod review's banned-import visitor, kept as the one parser.

BannedLibraryCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 580

SF_BANNED_LIBRARY: imports of libraries in the banned registry.

_BroadRaiseVisitor

Bases: ast.NodeVisitor · Line 601

BroadRaiseCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 619

SF_BROAD_RAISE: raise Exception(...) / raise BaseException(...).

_PrintVisitor

Bases: ast.NodeVisitor · Line 644

PrintOutsideMainCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 664

SF_PRINT: print() calls outside the __main__ guard.

_MutableCallDefaultVisitor

Bases: ast.NodeVisitor · Line 675

MutableCallDefaultCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 704

QL_MUTABLE_DEFAULTS extension: set()/list()/dict() defaults.

_SwallowVisitor

Bases: ast.NodeVisitor · Line 719

SwallowedExceptionCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 742

SR_ERROR_HANDLING extension: handlers that only pass/continue/....

_FixtureVisitor

Bases: ast.NodeVisitor · Line 775

ModuleScopedFixtureCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 800

TI_MODULE_FIXTURES: fixtures wider than function scope (test files).

_GlobalStateVisitor

Bases: ast.NodeVisitor · Line 840

Mutations of module-level containers from inside functions, plus global statements (module state rebound by a function).

GlobalMutableStateCheck

Bases: _DescribedCheck · Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-034']) · Line 905

TI_GLOBAL_STATE: module-level mutable state shared across tests.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/codemod/review_checks.py

"""SPEC-110 FR-012 -- the review check implementations.

Thirty distinct implementations realise the 40-check catalogue
(``review_catalogue``). Twenty wrap the mapped ``verification/review/check_*``
functions through ``ReusedCheck`` (REQ-110-034: reuse, never re-implement);
ten are new classes for the bullets no existing function covers.

Every check reads only the MUTATED module's ``ast`` overlay
(``LosslessModule.tree``) or its token stream (``LosslessModule.stream``):
the reused functions receive lines reconstructed from ``stream.source`` and a
forest-relative path, so none of them can reach the disk, spawn a
subprocess or call an external linter.
"""

from __future__ import annotations

import ast
import os
import re
from bisect import bisect_right
from collections.abc import Callable, Iterable, Mapping, Sequence
from typing import TYPE_CHECKING, Any, Literal

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod.report import Finding
from orchestrata_ultimate.refactor.codemod.review import _BannedImportVisitor
from orchestrata_ultimate.refactor.codemod.review_catalogue import (
    _CATALOGUE,
    CheckCategory,
    CheckDescriptor,
    CheckRegistry,
    ReviewCheck,
)
from orchestrata_ultimate.refactor.codemod.review_context import _SIBLING_ROOTS
from orchestrata_ultimate.refactor.ctok.tokenizer import TokenKind
from orchestrata_ultimate.verification.review.checks_correctness import (
    _build_module_candidates,
    check_dependency_declaration,
    check_entry_point,
    check_file_completeness,
    check_imports,
)
from orchestrata_ultimate.verification.review.checks_quality import (
    check_docstrings,
    check_line_length,
    check_mutable_defaults,
    check_type_hints,
)
from orchestrata_ultimate.verification.review.checks_safety import (
    check_env_access,
    check_no_mocks,
    check_security,
    check_test_isolation,
    check_todo_comments,
)
from orchestrata_ultimate.verification.review.checks_structure import (
    check_error_handling,
    check_function_args,
    check_function_length,
    check_god_objects,
    check_nesting_depth,
    check_no_placeholder,
    check_stub_methods,
)
from orchestrata_ultimate.verification.review.types import Issue

if TYPE_CHECKING:
    from orchestrata_ultimate.refactor.codemod.review_context import ReviewContext
    from orchestrata_ultimate.refactor.ctok import LosslessModule

ArgShape = Literal[
    "path_tree",
    "path_tree_lines",
    "tree_lines_path",
    "path_lines",
    "path_tree_modules",
    "path_tree_files_modules",
    "path_tree_deps_modules",
    "path_lines_main",
]
Scope = Literal["all", "source", "test"]

_CYCLOMATIC_MAX = 10
_LINE_IN_PARENS = re.compile(r"\(line (\d+)\)|\bline (\d+)\b")
_LEADING_LINE = re.compile(r"^L(\d+):")
_STRING_KINDS = frozenset({TokenKind.STRING, TokenKind.FSTRING_MIDDLE})
# Literal text of every string flavour (f/t-string middles are text; the
# ``{expr}`` parts between them stay code).
_LITERAL_KINDS = frozenset(
    {TokenKind.STRING, TokenKind.FSTRING_MIDDLE, TokenKind.TSTRING_MIDDLE}
)
_COMMENT_KINDS = frozenset({TokenKind.COMMENT})
_BROAD_EXCEPTIONS = frozenset({"Exception", "BaseException"})
_MUTABLE_CALLS = frozenset({"set", "list", "dict"})
_MUTABLE_CTORS = frozenset(
    {"list", "dict", "set", "defaultdict", "deque", "Counter", "OrderedDict"}
)
_MUTATING_METHODS = frozenset(
    {
        "append",
        "extend",
        "insert",
        "add",
        "update",
        "clear",
        "pop",
        "popitem",
        "remove",
        "discard",
        "setdefault",
        "sort",
        "reverse",
    }
)
_SNAKE_CASE = re.compile(r"^_*[a-z][a-z0-9_]*$|^_+$|^__[a-z0-9_]+__$")
_PASCAL_CASE = re.compile(r"^_*[A-Z][A-Za-z0-9]*$")
_NON_FUNCTION_SCOPES = frozenset({"module", "session", "package", "class"})


def _source_lines(
    module: LosslessModule, blank: frozenset[TokenKind] = frozenset()
) -> list[str]:
    """Lines of the mutated source, newline-terminated like ``readlines``.

    ``blank`` names token kinds whose text is replaced by spaces, one per
    character (newlines kept), so the lexical regex checks cannot match text
    that is a string literal or a comment while line and column geometry
    stay intact in characters, non-ASCII text included.
    """
    source = module.stream.source
    if blank:
        pieces: list[bytes] = []
        cursor = 0
        for token in module.stream.tokens:
            if token.kind not in blank or token.start < cursor:
                continue
            pieces.append(source[cursor : token.start])
            span = source[token.start : token.end].decode("utf-8", errors="replace")
            pieces.append(
                "".join(ch if ch in "\r\n" else " " for ch in span).encode("ascii")
            )
            cursor = token.end
        pieces.append(source[cursor:])
        source = b"".join(pieces)
    text = source.decode("utf-8", errors="replace")
    parts = text.split("\n")
    lines = [part + "\n" for part in parts[:-1]]
    if parts[-1]:
        lines.append(parts[-1])
    return lines


_IMPORT_IN_MESSAGE = re.compile(r"Import '([^']+)'")


def _outside_include_roots(message: str, roots: tuple[str, ...]) -> bool:
    """Whether the import a completeness finding names lies outside ``roots``.

    ``check_file_completeness`` claims "module file not found in workspace"
    for any project import with no file in the set. Under a narrow
    ``include`` a module outside every stored include root is outside the
    workspace, not missing from it; the roots are the ones the workspace was
    loaded from, never inferred from the path set. No roots: no restriction.
    """
    found = _IMPORT_IN_MESSAGE.search(message)
    if not roots or found is None:
        return False
    full = found.group(1)
    candidates = _build_module_candidates(full.split(".")[0], full)
    return not any(
        candidate.replace(os.sep, "/").startswith(root + "/")
        for candidate in candidates
        for root in roots
    )


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033"])
def _resolves_to_sibling_module(
    message: str, workspace_files: frozenset[str], importer_path: str
) -> bool:
    """Whether the import a completeness finding names is a sibling-root module.

    ``_build_module_candidates`` proposes ``src/<top>/...`` shapes only, so a
    bare-name import of ``scripts/<top>.py`` or ``scripts/<top>/__init__.py``
    (the import rule of the sibling roots) is reported "module file not found
    in workspace" although the file is in the set. Such a finding is a false
    positive and is dropped when the importer itself lies under a sibling
    root (``scripts/`` is not on ``sys.path`` for ``src/``, so there the
    import stays reported). A submodule of a plain module (``b.nope`` beside
    ``scripts/b.py``) cannot exist and stays reported; a name that is only a
    nested stem below a package never matches.
    """
    found = _IMPORT_IN_MESSAGE.search(message)
    if found is None or importer_path.split("/")[0] not in _SIBLING_ROOTS:
        return False
    full = found.group(1)
    top = full.split(".")[0]
    if any(f"{root}/{top}/__init__.py" in workspace_files for root in _SIBLING_ROOTS):
        return True
    return "." not in full and any(
        f"{root}/{top}.py" in workspace_files for root in _SIBLING_ROOTS
    )


def _line_at(source: bytes, offset: int) -> int:
    """1-based line number of byte ``offset`` in ``source``."""
    return source.count(b"\n", 0, offset) + 1


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class IssueAdapter:
    """Convert ``verification/review`` issue tuples into codemod findings."""

    @staticmethod
    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033"])
    def to_findings(
        issues: Sequence[Issue],
        path: str,
        severity: Literal["error", "warning"],
        rule: str | None = None,
    ) -> list[Finding]:
        """Map ``(check_id, filepath, description)`` tuples to findings.

        Args:
            issues: Tuples returned by a ``check_*`` function.
            path: Forest-relative path the findings are reported against
                (the tuple's own path is only a basename).
            severity: Severity of every resulting finding.
            rule: Rule id to report under; the tuple's check id when ``None``.

        Returns:
            One finding per issue, in order.
        """
        return [
            Finding(
                severity=severity,
                rule=rule or check_id,
                path=path,
                line=IssueAdapter._line_of(description),
                message=description,
            )
            for check_id, _filepath, description in issues
        ]

    @staticmethod
    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033"])
    def _line_of(description: str) -> int | None:
        """The line a ``check_*`` description names, if it names one.

        Most checks lead with ``L<n>:``; the structural ones embed
        ``(line <n>)`` or ``line <n>`` instead.
        """
        match = _LEADING_LINE.match(description)
        if match:
            return int(match.group(1))
        match = _LINE_IN_PARENS.search(description)
        if match:
            return int(match.group(1) or match.group(2))
        return None


class _DescribedCheck:
    """Shared descriptor plumbing for every implementation."""

    def __init__(self, descriptor: CheckDescriptor) -> None:
        self.descriptor = descriptor

    @property
    def id(self) -> str:
        """Id findings are reported under (the owner's id)."""
        return self.descriptor.id

    @property
    def category(self) -> CheckCategory:
        """Catalogue category of the owner descriptor."""
        return self.descriptor.category

    @property
    def phase(self) -> Literal[1, 2, 3]:
        """Phase the implementation runs in (the owner's)."""
        return self.descriptor.phase

    @property
    def substrate(self) -> Literal["ast-visitor", "token-scanner"]:
        """Lossless layer the implementation reads."""
        return self.descriptor.substrate

    def _findings(
        self,
        ctx: ReviewContext,
        hits: Iterable[tuple[int | None, str]],
    ) -> list[Finding]:
        return [
            Finding(
                severity=self.descriptor.severity,
                rule=self.descriptor.id,
                path=ctx.path,
                line=line,
                message=message,
            )
            for line, message in hits
        ]


_ARG_BUILDERS: Mapping[str, Callable[[str, ast.Module, list[str], Any], tuple]] = {
    "path_tree": lambda path, tree, lines, ctx: (path, tree),
    "path_tree_lines": lambda path, tree, lines, ctx: (path, tree, lines),
    "tree_lines_path": lambda path, tree, lines, ctx: (tree, lines, path),
    "path_lines": lambda path, tree, lines, ctx: (path, lines),
    "path_tree_modules": lambda path, tree, lines, ctx: (
        path,
        tree,
        set(ctx.project_modules),
    ),
    "path_tree_files_modules": lambda path, tree, lines, ctx: (
        path,
        tree,
        set(ctx.workspace_files),
        set(ctx.project_modules),
    ),
    "path_tree_deps_modules": lambda path, tree, lines, ctx: (
        path,
        tree,
        set(ctx.declared_deps),
        set(ctx.project_modules),
    ),
    "path_lines_main": lambda path, tree, lines, ctx: (path, lines, ctx.is_main_file),
}


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class ReusedCheck(_DescribedCheck):
    """A catalogue check realised by a ``verification/review`` function.

    The function runs over the mutated module's ``ast`` tree and lines
    reconstructed from its token stream; ``extension`` adds the part of the
    spec bullet the function does not cover (REQ-110-034: reuse where one
    exists).
    """

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033"])
    def __init__(
        self,
        descriptor: CheckDescriptor,
        fn: Callable[..., Sequence[Issue]],
        arg_shape: ArgShape,
        extension: ReviewCheck | None = None,
        *,
        scope: Scope = "all",
        blank: frozenset[TokenKind] = frozenset(),
    ) -> None:
        """Bind ``fn`` to ``descriptor``.

        Args:
            descriptor: The owner descriptor findings are reported under.
            fn: The ``check_*`` function to reuse.
            arg_shape: How ``fn`` takes its arguments.
            extension: Optional check run in addition, same descriptor.
            scope: ``source`` runs on non-test files only, ``test`` on test
                files only, ``all`` on every file (the engine's own split).
            blank: Token kinds blanked out of the lines handed to ``fn``
                (the lexical regex checks must not read string literals or
                comments as code); empty hands ``fn`` the raw lines.
        """
        super().__init__(descriptor)
        self._fn = fn
        self._arg_shape: ArgShape = arg_shape
        self._extension = extension
        self._scope: Scope = scope
        self._blank = blank

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Run the reused function (and extension) over the mutated module."""
        if self._scope == "source" and ctx.is_test:
            return []
        if self._scope == "test" and not ctx.is_test:
            return []
        findings = IssueAdapter.to_findings(
            self._fn(*self._args(module, ctx)),
            ctx.path,
            self.descriptor.severity,
            self.descriptor.id,
        )
        if self._arg_shape == "path_tree_files_modules":
            findings = [
                finding
                for finding in findings
                if not _outside_include_roots(finding.message, ctx.include_roots)
                and not _resolves_to_sibling_module(
                    finding.message, ctx.workspace_files, ctx.path
                )
            ]
        if self._extension is not None:
            findings.extend(self._extension.run(module, ctx))
        return findings

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-034"])
    def _args(self, module: LosslessModule, ctx: ReviewContext) -> tuple[Any, ...]:
        """Positional arguments for ``fn`` from the mutated module."""
        return _ARG_BUILDERS[self._arg_shape](
            ctx.path, module.tree, _source_lines(module, self._blank), ctx
        )


class _BranchCounter(ast.NodeVisitor):
    """Count branching nodes of one function body, not descending into
    nested functions, classes or lambdas (each is scored on its own)."""

    _BRANCHES = (
        ast.If,
        ast.For,
        ast.AsyncFor,
        ast.While,
        ast.With,
        ast.AsyncWith,
        ast.Try,
        ast.TryStar,
        ast.ExceptHandler,
        ast.BoolOp,
        ast.IfExp,
        ast.comprehension,
    )
    _SCOPES = (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)

    def __init__(self) -> None:
        self.branches = 0

    def visit(self, node: ast.AST) -> None:
        if isinstance(node, self._BRANCHES):
            self.branches += 1
        if isinstance(node, self._SCOPES):
            return
        self.generic_visit(node)


class _CyclomaticVisitor(ast.NodeVisitor):
    """Flag every function whose complexity exceeds the limit."""

    def __init__(self) -> None:
        self.hits: list[tuple[int | None, str]] = []

    def _function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
        counter = _BranchCounter()
        for statement in node.body:
            counter.visit(statement)
        complexity = 1 + counter.branches
        if complexity > _CYCLOMATIC_MAX:
            self.hits.append(
                (
                    node.lineno,
                    f"`{node.name}()` has cyclomatic complexity {complexity} "
                    f"(max {_CYCLOMATIC_MAX}).",
                )
            )
        self.generic_visit(node)

    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        self._function(node)

    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        self._function(node)


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class CyclomaticComplexityCheck(_DescribedCheck):
    """CX_CYCLOMATIC: branching nodes per function, default max 10."""

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Flag functions above the cyclomatic limit."""
        visitor = _CyclomaticVisitor()
        visitor.visit(module.tree)
        return self._findings(ctx, visitor.hits)


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class TrailingWhitespaceCheck(_DescribedCheck):
    """ST_TRAILING_WS: trailing blanks outside multi-line string tokens."""

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Scan the source for trailing whitespace the tokens do not own."""
        source = module.stream.source
        spans = sorted(
            (token.start, token.end)
            for token in module.stream.tokens
            if token.kind in _STRING_KINDS
        )
        starts = [start for start, _ in spans]
        hits: list[tuple[int | None, str]] = []
        offset = 0
        for lineno, raw in enumerate(source.split(b"\n"), 1):
            body = raw[:-1] if raw.endswith(b"\r") else raw
            kept = len(body.rstrip(b" \t\f"))
            if kept != len(body) and not self._inside(starts, spans, offset + kept):
                hits.append((lineno, "Trailing whitespace."))
            offset += len(raw) + 1
        return self._findings(ctx, hits)

    @staticmethod
    @traces(spec="110", story="US-110-004", reqs=["REQ-110-034"])
    def _inside(starts: list[int], spans: list[tuple[int, int]], position: int) -> bool:
        """Whether ``position`` falls inside a string token's span."""
        index = bisect_right(starts, position) - 1
        return index >= 0 and spans[index][0] <= position < spans[index][1]


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class TabIndentationCheck(_DescribedCheck):
    """ST_TAB_INDENT: INDENT tokens that contain a tab."""

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Flag every INDENT token whose bytes include a tab."""
        source = module.stream.source
        hits = [
            (_line_at(source, token.start), "Tab indentation (use spaces).")
            for token in module.stream.tokens
            if token.kind is TokenKind.INDENT and b"\t" in token.text(source)
        ]
        return self._findings(ctx, hits)


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class FinalNewlineCheck(_DescribedCheck):
    """ST_FINAL_NEWLINE: a non-empty file must end with a newline."""

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Flag a non-empty source that does not end with a newline."""
        source = module.stream.source
        if not source or source.endswith(b"\n"):
            return []
        return self._findings(
            ctx, [(_line_at(source, len(source)), "File does not end with a newline.")]
        )


class _NamingVisitor(ast.NodeVisitor):
    def __init__(self) -> None:
        self.hits: list[tuple[int | None, str]] = []

    def visit_ClassDef(self, node: ast.ClassDef) -> None:
        if not _PASCAL_CASE.match(node.name):
            self.hits.append((node.lineno, f"Class `{node.name}` is not PascalCase."))
        self.generic_visit(node)

    def _function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
        if not _SNAKE_CASE.match(node.name):
            self.hits.append(
                (node.lineno, f"Function `{node.name}` is not snake_case.")
            )
        self.generic_visit(node)

    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        self._function(node)

    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        self._function(node)


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class NamingConventionCheck(_DescribedCheck):
    """ST_NAMING: PascalCase classes, snake_case functions."""

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Flag classes and functions that break the naming convention."""
        visitor = _NamingVisitor()
        visitor.visit(module.tree)
        return self._findings(ctx, visitor.hits)


class _BannedLibraryVisitor(_BannedImportVisitor):
    """The codemod review's banned-import visitor, kept as the one parser."""


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class BannedLibraryCheck(_DescribedCheck):
    """SF_BANNED_LIBRARY: imports of libraries in the banned registry."""

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Flag imports listed in ``ctx.banned_libraries``."""
        visitor = _BannedLibraryVisitor(ctx.banned_libraries)
        visitor.visit(module.tree)
        return self._findings(
            ctx,
            [
                (
                    line,
                    f"Banned library import: {name} "
                    f"(listed in .claude/banned-libraries.txt)",
                )
                for line, name in visitor.violations
            ],
        )


class _BroadRaiseVisitor(ast.NodeVisitor):
    def __init__(self) -> None:
        self.hits: list[tuple[int | None, str]] = []

    def visit_Raise(self, node: ast.Raise) -> None:
        exc = node.exc
        target = exc.func if isinstance(exc, ast.Call) else exc
        if isinstance(target, ast.Name) and target.id in _BROAD_EXCEPTIONS:
            self.hits.append(
                (
                    node.lineno,
                    f"`raise {target.id}` is too broad -- raise a specific type.",
                )
            )
        self.generic_visit(node)


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class BroadRaiseCheck(_DescribedCheck):
    """SF_BROAD_RAISE: ``raise Exception(...)`` / ``raise BaseException(...)``."""

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Flag raises of the two broad exception types."""
        visitor = _BroadRaiseVisitor()
        visitor.visit(module.tree)
        return self._findings(ctx, visitor.hits)


def _is_main_guard(test: ast.expr) -> bool:
    """Whether ``test`` is ``__name__ == "__main__"`` (either order)."""
    if not (isinstance(test, ast.Compare) and len(test.ops) == 1):
        return False
    operands = [test.left, *test.comparators]
    names = [o.id for o in operands if isinstance(o, ast.Name)]
    values = [o.value for o in operands if isinstance(o, ast.Constant)]
    return (
        isinstance(test.ops[0], ast.Eq)
        and names == ["__name__"]
        and values == ["__main__"]
    )


class _PrintVisitor(ast.NodeVisitor):
    def __init__(self) -> None:
        self.hits: list[tuple[int | None, str]] = []

    def visit_If(self, node: ast.If) -> None:
        if _is_main_guard(node.test):
            for statement in node.orelse:
                self.visit(statement)
            return
        self.generic_visit(node)

    def visit_Call(self, node: ast.Call) -> None:
        if isinstance(node.func, ast.Name) and node.func.id == "print":
            self.hits.append(
                (node.lineno, "`print()` outside an `if __name__ == '__main__'` guard.")
            )
        self.generic_visit(node)


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class PrintOutsideMainCheck(_DescribedCheck):
    """SF_PRINT: ``print()`` calls outside the ``__main__`` guard."""

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Flag ``print()`` calls that are not under the main guard."""
        visitor = _PrintVisitor()
        visitor.visit(module.tree)
        return self._findings(ctx, visitor.hits)


class _MutableCallDefaultVisitor(ast.NodeVisitor):
    def __init__(self) -> None:
        self.hits: list[tuple[int | None, str]] = []

    def _function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
        defaults = [*node.args.defaults, *node.args.kw_defaults]
        for default in defaults:
            if (
                isinstance(default, ast.Call)
                and isinstance(default.func, ast.Name)
                and default.func.id in _MUTABLE_CALLS
            ):
                self.hits.append(
                    (
                        node.lineno,
                        f"`{node.name}()` has mutable default "
                        f"`{default.func.id}()`. Use None instead.",
                    )
                )
        self.generic_visit(node)

    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        self._function(node)

    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        self._function(node)


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class MutableCallDefaultCheck(_DescribedCheck):
    """QL_MUTABLE_DEFAULTS extension: ``set()``/``list()``/``dict()`` defaults.

    ``check_mutable_defaults`` matches the literal forms only; the spec
    bullet also names the constructor-call form.
    """

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Flag defaults written as ``set()``, ``list()`` or ``dict()``."""
        visitor = _MutableCallDefaultVisitor()
        visitor.visit(module.tree)
        return self._findings(ctx, visitor.hits)


class _SwallowVisitor(ast.NodeVisitor):
    def __init__(self) -> None:
        self.hits: list[tuple[int | None, str]] = []

    @staticmethod
    def _is_noop(statement: ast.stmt) -> bool:
        if isinstance(statement, (ast.Pass, ast.Continue)):
            return True
        return (
            isinstance(statement, ast.Expr)
            and isinstance(statement.value, ast.Constant)
            and statement.value.value is Ellipsis
        )

    def visit_ExceptHandler(self, node: ast.ExceptHandler) -> None:
        if all(self._is_noop(statement) for statement in node.body):
            self.hits.append(
                (node.lineno, "Exception swallowed -- handler neither logs nor raises.")
            )
        self.generic_visit(node)


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class SwallowedExceptionCheck(_DescribedCheck):
    """SR_ERROR_HANDLING extension: handlers that only pass/continue/``...``.

    ``check_error_handling`` flags the bare ``except:`` only; the spec bullet
    also forbids swallowed exceptions.
    """

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Flag handlers whose whole body is a no-op."""
        visitor = _SwallowVisitor()
        visitor.visit(module.tree)
        return self._findings(ctx, visitor.hits)


def _fixture_scope(decorator: ast.expr) -> str | None:
    """The ``scope=`` string of a ``fixture`` decorator call, if any."""
    if not isinstance(decorator, ast.Call):
        return None
    func = decorator.func
    name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "")
    if name != "fixture":
        return None
    for keyword in decorator.keywords:
        if (
            keyword.arg == "scope"
            and isinstance(keyword.value, ast.Constant)
            and isinstance(keyword.value.value, str)
        ):
            return keyword.value.value
    return None


class _FixtureVisitor(ast.NodeVisitor):
    def __init__(self) -> None:
        self.hits: list[tuple[int | None, str]] = []

    def _function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
        for decorator in node.decorator_list:
            scope = _fixture_scope(decorator)
            if scope in _NON_FUNCTION_SCOPES:
                self.hits.append(
                    (
                        node.lineno,
                        f"Fixture `{node.name}` has scope='{scope}' -- shared "
                        f"mutable state; use function scope.",
                    )
                )
        self.generic_visit(node)

    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        self._function(node)

    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        self._function(node)


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class ModuleScopedFixtureCheck(_DescribedCheck):
    """TI_MODULE_FIXTURES: fixtures wider than function scope (test files)."""

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Flag fixtures declared with module/session/package/class scope."""
        if not ctx.is_test:
            return []
        visitor = _FixtureVisitor()
        visitor.visit(module.tree)
        return self._findings(ctx, visitor.hits)


def _is_mutable_value(value: ast.expr | None) -> bool:
    if isinstance(
        value, (ast.List, ast.Dict, ast.Set, ast.ListComp, ast.DictComp, ast.SetComp)
    ):
        return True
    if not isinstance(value, ast.Call):
        return False
    func = value.func
    name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "")
    return name in _MUTABLE_CTORS


def _module_mutables(tree: ast.Module) -> set[str]:
    """Module-level names bound to a mutable container."""
    names: set[str] = set()
    for statement in tree.body:
        if isinstance(statement, ast.Assign) and _is_mutable_value(statement.value):
            names.update(t.id for t in statement.targets if isinstance(t, ast.Name))
        elif (
            isinstance(statement, ast.AnnAssign)
            and isinstance(statement.target, ast.Name)
            and _is_mutable_value(statement.value)
        ):
            names.add(statement.target.id)
    return names


class _GlobalStateVisitor(ast.NodeVisitor):
    """Mutations of module-level containers from inside functions, plus
    ``global`` statements (module state rebound by a function)."""

    def __init__(self, mutables: set[str]) -> None:
        self._mutables = mutables
        self._depth = 0
        self._reported: set[str] = set()
        self.hits: list[tuple[int | None, str]] = []

    def _function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
        self._depth += 1
        self.generic_visit(node)
        self._depth -= 1

    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        self._function(node)

    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        self._function(node)

    def _report(self, name: str, line: int, how: str) -> None:
        if name in self._reported:
            return
        self._reported.add(name)
        self.hits.append(
            (line, f"Module-level `{name}` {how} -- global mutable state in a test.")
        )

    def visit_Global(self, node: ast.Global) -> None:
        for name in node.names:
            self._report(name, node.lineno, "is rebound via `global`")

    def visit_Call(self, node: ast.Call) -> None:
        func = node.func
        if (
            self._depth
            and isinstance(func, ast.Attribute)
            and func.attr in _MUTATING_METHODS
            and isinstance(func.value, ast.Name)
            and func.value.id in self._mutables
        ):
            self._report(func.value.id, node.lineno, f"is mutated via .{func.attr}()")
        self.generic_visit(node)

    def _subscript_target(self, target: ast.expr, line: int) -> None:
        if (
            self._depth
            and isinstance(target, ast.Subscript)
            and isinstance(target.value, ast.Name)
            and target.value.id in self._mutables
        ):
            self._report(target.value.id, line, "is mutated via item assignment")

    def visit_Assign(self, node: ast.Assign) -> None:
        for target in node.targets:
            self._subscript_target(target, node.lineno)
        self.generic_visit(node)

    def visit_AugAssign(self, node: ast.AugAssign) -> None:
        self._subscript_target(node.target, node.lineno)
        self.generic_visit(node)


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
class GlobalMutableStateCheck(_DescribedCheck):
    """TI_GLOBAL_STATE: module-level mutable state shared across tests.

    The spec states no rule, so the check flags the two forms that make a
    test's outcome depend on another's: a module-level container mutated
    from inside a function, and any ``global`` statement. A constant list a
    test only reads is not state.
    """

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
    def run(self, module: LosslessModule, ctx: ReviewContext) -> list[Finding]:
        """Flag module-level containers mutated by functions, and ``global``."""
        if not ctx.is_test:
            return []
        visitor = _GlobalStateVisitor(_module_mutables(module.tree))
        visitor.visit(module.tree)
        return self._findings(ctx, visitor.hits)


_CheckFactory = Callable[[CheckDescriptor], ReviewCheck]

# impl_key -> (function, arg shape, scope, extension class or None)
_REUSED: Mapping[
    str,
    tuple[Callable[..., Sequence[Issue]], ArgShape, Scope, _CheckFactory | None],
] = {
    "nesting": (check_nesting_depth, "tree_lines_path", "all", None),
    "func_length": (check_function_length, "path_tree", "all", None),
    "func_args": (check_function_args, "path_tree", "all", None),
    "line_length": (check_line_length, "path_lines", "all", None),
    "imports": (check_imports, "path_tree_modules", "all", None),
    "type_hints": (check_type_hints, "path_tree", "source", None),
    "docstrings": (check_docstrings, "path_tree", "source", None),
    "mutable_defaults": (
        check_mutable_defaults,
        "path_tree",
        "all",
        MutableCallDefaultCheck,
    ),
    "god_objects": (check_god_objects, "tree_lines_path", "all", None),
    "stub_methods": (check_stub_methods, "tree_lines_path", "source", None),
    "placeholders": (check_no_placeholder, "path_tree", "source", None),
    "error_handling": (
        check_error_handling,
        "path_tree",
        "all",
        SwallowedExceptionCheck,
    ),
    "env_access": (check_env_access, "path_tree", "all", None),
    "no_mocks": (check_no_mocks, "path_lines", "all", None),
    "security": (check_security, "path_tree_lines", "all", None),
    "isolation": (check_test_isolation, "path_lines", "test", None),
    "todo": (check_todo_comments, "path_lines", "source", None),
    "file_completeness": (
        check_file_completeness,
        "path_tree_files_modules",
        "source",
        None,
    ),
    "entry_point": (check_entry_point, "path_lines_main", "source", None),
    "dependency_declaration": (
        check_dependency_declaration,
        "path_tree_deps_modules",
        "source",
        None,
    ),
}

# impl_key -> token kinds blanked from the lines the reused function reads.
# ``no_mocks`` hunts code, so literals and comments are blanked. ``security``
# hunts a hardcoded secret, whose subject IS the string literal, so only
# comments are blanked. The other lexical checks (isolation: Cypher inside
# strings; todo: comments; line length; entry-point guard) read raw lines.
_BLANKED: Mapping[str, frozenset[TokenKind]] = {
    "no_mocks": _LITERAL_KINDS | _COMMENT_KINDS,
    "security": _COMMENT_KINDS,
}

_NEW: Mapping[str, _CheckFactory] = {
    "cyclomatic": CyclomaticComplexityCheck,
    "trailing_ws": TrailingWhitespaceCheck,
    "tab_indent": TabIndentationCheck,
    "final_newline": FinalNewlineCheck,
    "naming": NamingConventionCheck,
    "banned_library": BannedLibraryCheck,
    "broad_raise": BroadRaiseCheck,
    "print_call": PrintOutsideMainCheck,
    "module_fixtures": ModuleScopedFixtureCheck,
    "global_state": GlobalMutableStateCheck,
}


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034"])
def build_default_implementations(
    descriptors: Sequence[CheckDescriptor],
) -> dict[str, ReviewCheck]:
    """Build the 30 implementations the 40 descriptors share.

    Each ``impl_key`` is built once, against its owner descriptor (the row
    with ``alias_of is None``), so aliases report through the owner's id.

    Args:
        descriptors: The catalogue rows.

    Returns:
        ``impl_key`` -> implementation.

    Raises:
        KeyError: A descriptor names an ``impl_key`` with no implementation.
    """
    built: dict[str, ReviewCheck] = {}
    for descriptor in descriptors:
        if descriptor.alias_of is not None or descriptor.impl_key in built:
            continue
        key = descriptor.impl_key
        if key in _REUSED:
            fn, shape, scope, extension_cls = _REUSED[key]
            extension = extension_cls(descriptor) if extension_cls else None
            built[key] = ReusedCheck(
                descriptor,
                fn,
                shape,
                extension,
                scope=scope,
                blank=_BLANKED.get(key, frozenset()),
            )
        else:
            built[key] = _NEW[key](descriptor)
    return built


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033"])
def default_registry() -> CheckRegistry:
    """Build the 40-descriptor catalogue with its 30 implementations."""
    return CheckRegistry(_CATALOGUE, build_default_implementations(_CATALOGUE))