orchestrata_ultimate.refactor.codemod.review

Reference › orchestrata_ultimate › refactor › codemod › review

The review passes that run over a ChangeSet before a commit, and the Prolog safety pass that completes the in-memory review engine. The facade, review_checks, the engine and workspace import it. It builds on the Finding type in report, core.banned_libraries and core.prolog_adapter.

ReviewPass is a protocol with a single review(ws, changes) method that returns findings. The passes are:

• ParseIntegrity: re-parses each mutated module's bytes with ast.parse and reports an error finding for any that fail. Byte-level fidelity is guaranteed by the edit buffer, so only syntax is checked here.
• BannedConstructs: reports bare os.environ["X"] subscript reads (found by an AST walk), mock, stub and fake patterns (regular expressions over the text) and imports of banned libraries (from the registry file .claude/banned-libraries.txt).
• DiffBudget: a guard against runaway passes that fails the review when a change set touches more than max_files files (1000 by default) or more than max_lines changed lines (50,000).
• PrologSemanticReview: asserts the mutations as Prolog facts and queries safety predicates defined in codemod_review.pl, using the explain module for proof trees on failure. When Prolog is unavailable it logs a warning and returns no findings (ERR-110-015), and it removes its facts afterwards.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

_OsEnvironVisitor

Bases: ast.NodeVisitor · Line 76

Walk the ast tree to find bare os.environ["X"] subscript reads.

_BannedImportVisitor

Bases: ast.NodeVisitor · Line 103

Walk the ast tree to find imports of banned libraries.

CodemodWorkspace

Line 135

Forward reference for type checking.

ReviewPass (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 143

ReviewPass protocol — structural interface for review passes.

ParseIntegrity

Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-012', 'REQ-110-124']) · Line 152

Verify mutated LosslessModule trees still parse.

BannedConstructs

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

Detect banned patterns: bare os.environ, mocks, banned libraries.

DiffBudget

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

Enforce max_files and max_lines limits on a ChangeSet.

PrologSemanticReview

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

Prolog-based semantic safety checks via codemod_review.pl.

Functions

Source

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

"""SPEC-110 codemod engine — review passes.

Four built-in review passes that operate on the mutated LosslessModule
forest (ast overlay over a byte-exact token stream, see
``orchestrata_ultimate.refactor.ctok``) before any bytes are written
to disk:

- ParseIntegrity: verify mutated trees still parse
- BannedConstructs: detect bare os.environ, mocks, banned libraries
- DiffBudget: enforce max_files and max_lines limits
- PrologSemanticReview: Prolog-based semantic safety checks
"""

from __future__ import annotations

import ast
import logging
import re
from pathlib import Path
from typing import Any, Protocol, runtime_checkable

from orchestrata_ultimate.core.banned_libraries import (
    BANNED_LIBRARIES,
    banned_names,
    is_banned,
)
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.paths import resolve_project_root
from orchestrata_ultimate.refactor.codemod.report import ChangeSet, Finding

logger = logging.getLogger(__name__)

_MOCK_PATTERNS: list[tuple[str, str]] = [
    (r"from\s+unittest\.mock\s+import", "unittest.mock import"),
    (r"import\s+unittest\.mock", "unittest.mock import"),
    (r"from\s+unittest\s+import\s+mock", "unittest import mock"),
    (r"Mock\s*\(", "Mock() instantiation"),
    (r"MagicMock\s*\(", "MagicMock() instantiation"),
    (r"AsyncMock\s*\(", "AsyncMock() instantiation"),
    (r"@\s*(mock\.)?patch\s*\(", "@patch decorator"),
    (r"mocker\.(patch|spy|stub|mock)\s*\(", "pytest-mock usage"),
    (r"create_autospec\s*\(", "create_autospec()"),
    (r"patch\.object\s*\(", "patch.object()"),
    (r"patch\.dict\s*\(", "patch.dict()"),
    (r"patch\.multiple\s*\(", "patch.multiple()"),
    (r"class\s+(Mock|Fake|Stub|Dummy)\w*\s*[\(:]", "custom mock/stub/fake class"),
    (r"monkeypatch\.\w+\s*\(", "monkeypatch usage"),
]


# READ from .claude/banned-libraries.txt, never restated here. This was a
# hardcoded ("pytest_xdist", "xdist") tuple until 2026-08-08 — while the finding
# message below told the reader the list came from the registry. It was missing
# docker, grpcio, grpcio-tools, grpc, unittest.mock, mock and pyswip, so an
# agent's pending edit could introduce any of them and this review would pass.
#
# Read at call time rather than at import: this process can outlive an edit to
# the registry (the codemod daemon is long-lived), and a ban that only takes
# effect after a restart is a ban nobody can rely on.
def _banned_libraries() -> frozenset[str]:
    return frozenset(banned_names(_repo_root() / BANNED_LIBRARIES))


def _repo_root() -> Path:
    """The checkout this module was loaded from.

    Resolved through ``resolve_project_root``, never a fixed ``__file__`` parent
    count. Since the proper install (dd919d55c) this module sits in
    site-packages, where four parents up is ``.venv/lib/python3.14t``; the
    banned-libraries registry read from there was empty, so BannedConstructs
    flagged no banned import (test_review_engine, 2026-10-22).
    """
    return resolve_project_root()


class _OsEnvironVisitor(ast.NodeVisitor):
    """Walk the ast tree to find bare os.environ["X"] subscript reads.

    ``ast.Subscript.lineno`` is always populated by CPython's parser
    (unlike a bare ``cst.parse_module`` tree without a MetadataWrapper +
    PositionProvider), so every finding carries a real line number.
    """

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

    def visit_Subscript(self, node: ast.Subscript) -> None:
        val = node.value
        if (
            isinstance(val, ast.Attribute)
            and val.attr == "environ"
            and isinstance(val.value, ast.Name)
            and val.value.id == "os"
        ):
            key = ""
            sl = node.slice
            if isinstance(sl, ast.Constant) and isinstance(sl.value, str):
                key = sl.value
            self.violations.append((node.lineno, key))
        self.generic_visit(node)


class _BannedImportVisitor(ast.NodeVisitor):
    """Walk the ast tree to find imports of banned libraries.

    ``ast.Import.lineno`` / ``ast.ImportFrom.lineno`` are always
    populated, so every finding carries a real line number (see
    ``_OsEnvironVisitor``'s docstring for the bug this fixes). Dotted
    module names are already plain strings on the ast nodes
    (``alias.name`` / ``ImportFrom.module``), so no Name/Attribute
    tree-walk is needed to reconstruct them.

    Matching goes through ``core.banned_libraries.is_banned``, which tries the
    full dotted path AND the top-level package. This visitor folded to the root
    only, so a dotted registry entry like ``unittest.mock`` could never match
    anything — the same gap ``ban_target/3`` closed on the Prolog side.
    """

    def __init__(self, banned: tuple[str, ...] | frozenset[str]) -> None:
        self._banned = {lib.replace("-", "_").lower() for lib in banned}
        self.violations: list[tuple[int, str]] = []

    def visit_Import(self, node: ast.Import) -> None:
        for alias in node.names:
            if is_banned(alias.name, self._banned):
                self.violations.append((node.lineno, alias.name))
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
        if node.module is not None and is_banned(node.module, self._banned):
            self.violations.append((node.lineno, node.module))
        self.generic_visit(node)


class CodemodWorkspace:
    """Forward reference for type checking."""

    _cig: Any
    _root: Path


@runtime_checkable
class ReviewPass(Protocol):
    """ReviewPass protocol — structural interface for review passes."""

    name: str

    def review(self, ws: CodemodWorkspace, changes: ChangeSet) -> list[Finding]: ...


@traces(spec="110", story="US-110-004", reqs=["REQ-110-012", "REQ-110-124"])
class ParseIntegrity:
    """Verify mutated LosslessModule trees still parse.

    Byte-for-byte round-trip fidelity is no longer a reparse-diff check
    here — it is a structural guarantee of ``EditBuffer`` (REQ-110-124):
    every byte outside an applied edit span is byte-identical to the
    original by construction, so ``LosslessModule.code`` cannot silently
    drift the way a re-serialized CST tree could. What remains to
    verify is syntactic validity of the mutated bytes.
    """

    name: str = "ParseIntegrity"

    def review(self, ws: CodemodWorkspace, changes: ChangeSet) -> list[Finding]:
        """Re-parse each mutated module's bytes and verify they parse.

        For each ChangeEntry: run ``ast.parse`` over
        ``mutated_tree.code`` (bytes) and return an error finding for
        any module that fails to parse.
        """
        findings: list[Finding] = []
        for entry in changes.entries:
            code = entry.mutated_tree.code
            try:
                ast.parse(code)
            except Exception as exc:
                findings.append(
                    Finding(
                        severity="error",
                        rule="PARSE_INTEGRITY",
                        path=entry.path,
                        line=None,
                        message=f"Mutated tree fails to re-parse: {exc}",
                    )
                )
        return findings


@traces(spec="110", story="US-110-004", reqs=["REQ-110-012"])
class BannedConstructs:
    """Detect banned patterns: bare os.environ, mocks, banned libraries.

    Re-hosts checks from verification/review/checks_safety.py, adapted
    to operate on the ast overlay of a LosslessModule.
    """

    name: str = "BannedConstructs"

    def review(self, ws: CodemodWorkspace, changes: ChangeSet) -> list[Finding]:
        """Check all mutated trees for banned constructs.

        Checks:
        1. Bare os.environ["X"] subscript reads (must use .get())
        2. 14 mock/stub/fake patterns (regex on code text)
        3. Banned library imports (from .claude/banned-libraries.txt)
        """
        return [
            finding
            for entry in changes.entries
            for finding in (
                self._check_os_environ(entry)
                + self._check_mocks(entry)
                + self._check_banned_imports(entry)
            )
        ]

    def _check_os_environ(self, entry: Any) -> list[Finding]:
        """Detect bare os.environ["X"] subscript reads via an ast walk."""
        findings: list[Finding] = []
        visitor = _OsEnvironVisitor()
        visitor.visit(entry.mutated_tree.tree)

        for line, key in visitor.violations:
            target = f'"{key}"' if key else "..."
            findings.append(
                Finding(
                    severity="error",
                    rule="BARE_OS_ENVIRON",
                    path=entry.path,
                    line=line,
                    message=(
                        f"Bare os.environ[{target}] raises KeyError when "
                        f"unset — use os.environ.get({target}, default)"
                    ),
                )
            )
        return findings

    def _check_mocks(self, entry: Any) -> list[Finding]:
        """Detect 14 mock/stub/fake patterns via regex on code text."""
        findings: list[Finding] = []
        code = entry.mutated_tree.code.decode("utf-8")
        for line_num, line in enumerate(code.split("\n"), 1):
            stripped = line.strip()
            if not stripped or stripped.startswith("#"):
                continue
            for pattern, desc in _MOCK_PATTERNS:
                if re.search(pattern, stripped):
                    findings.append(
                        Finding(
                            severity="error",
                            rule="MOCK_PATTERN",
                            path=entry.path,
                            line=line_num,
                            message=f"{desc} — `{stripped[:60]}`",
                        )
                    )
                    break
        return findings

    def _check_banned_imports(self, entry: Any) -> list[Finding]:
        """Detect banned library imports via an ast walk."""
        visitor = _BannedImportVisitor(_banned_libraries())
        visitor.visit(entry.mutated_tree.tree)

        return [
            Finding(
                severity="error",
                rule="BANNED_LIBRARY_IMPORT",
                path=entry.path,
                line=line,
                message=(
                    f"Banned library import: {mod_name} "
                    f"(listed in .claude/banned-libraries.txt)"
                ),
            )
            for line, mod_name in visitor.violations
        ]


@traces(spec="110", story="US-110-004", reqs=["REQ-110-012"])
class DiffBudget:
    """Enforce max_files and max_lines limits on a ChangeSet.

    A fat-finger guard: if a pass mutates more files or lines than
    its declared budget, the review aborts the commit.
    """

    name: str = "DiffBudget"

    def __init__(self, max_files: int = 1000, max_lines: int = 50000) -> None:
        """Configure diff budget limits.

        Args:
            max_files: Maximum number of files a ChangeSet may touch.
            max_lines: Maximum total changed lines across all files.
        """
        self.max_files = max_files
        self.max_lines = max_lines

    def review(self, ws: CodemodWorkspace, changes: ChangeSet) -> list[Finding]:
        """Count files and lines in ChangeSet, enforce limits.

        File count: len(changes.entries)
        Line count: sum of line counts of all mutated trees' code.
        """
        findings: list[Finding] = []

        file_count = changes.file_count
        if file_count > self.max_files:
            findings.append(
                Finding(
                    severity="error",
                    rule="DIFF_BUDGET_FILES",
                    path="<changeset>",
                    line=None,
                    message=(
                        f"ChangeSet touches {file_count} files, "
                        f"exceeding max_files budget of {self.max_files}"
                    ),
                )
            )

        total_lines = sum(
            entry.mutated_tree.code.count(b"\n") + 1 for entry in changes.entries
        )
        if total_lines > self.max_lines:
            findings.append(
                Finding(
                    severity="error",
                    rule="DIFF_BUDGET_LINES",
                    path="<changeset>",
                    line=None,
                    message=(
                        f"ChangeSet contains {total_lines} lines, "
                        f"exceeding max_lines budget of {self.max_lines}"
                    ),
                )
            )

        return findings


@traces(spec="110", story="US-110-004", reqs=["REQ-110-012"])
class PrologSemanticReview:
    """Prolog-based semantic safety checks via codemod_review.pl.

    Asserts ChangeSet mutations as Prolog facts, queries safety
    predicates (pass_safe/3, blast_radius_ok/3, import_cycle_free/2,
    rename_safe/3, string_replace_safe/3), and uses
    explain:explain_whynot/2 for proof trees on failure.

    Gracefully degrades when Prolog is unavailable: logs a warning
    and returns an empty findings list (ERR-110-015).
    """

    name: str = "PrologSemanticReview"

    def __init__(self) -> None:
        """Initialize PrologSemanticReview."""
        self._engine: Any = None
        self._facts_asserted: bool = False

    def review(self, ws: CodemodWorkspace, changes: ChangeSet) -> list[Finding]:
        """Assert mutations as Prolog facts, query safety predicates.

        Returns warning findings for any failed safety check.
        Returns empty list if Prolog is unavailable (ERR-110-015).
        """
        try:
            from orchestrata_ultimate.core.prolog_adapter import (
                PROLOG_AVAILABLE,
                PrologEngine,
            )
        except ImportError:
            logger.warning("PrologSemanticReview: prolog_adapter not importable")
            return []

        if not PROLOG_AVAILABLE:
            logger.warning("PrologSemanticReview: janus-swi not available, skipping")
            return []

        try:
            self._engine = PrologEngine(persistent=False)
            self._assert_facts(changes)
            return self._query_safety()
        except Exception as exc:
            logger.warning("PrologSemanticReview failed: %s", exc)
            return []
        finally:
            self._cleanup_facts()
            if self._engine is not None:
                try:
                    self._engine.close()
                except Exception:
                    pass
                self._engine = None

    def _assert_facts(self, changes: ChangeSet) -> None:
        """Assert ChangeSet mutations as Prolog ground facts.

        Asserts:
        - codemod_mutated(ModulePath, PassName) for each entry
        - codemod_affected(PassName, ModulePath) for each entry
        """
        if self._engine is None:
            return

        self._engine.retractall("codemod_mutated(_, _)")
        self._engine.retractall("codemod_affected(_, _)")

        for entry in changes.entries:
            safe_path = entry.path.replace("'", "\\'")
            safe_pass = entry.pass_name.replace("'", "\\'")
            self._engine.assertz(f"codemod_mutated('{safe_path}', '{safe_pass}')")
            self._engine.assertz(f"codemod_affected('{safe_pass}', '{safe_path}')")

        self._facts_asserted = True

    def _query_safety(self) -> list[Finding]:
        """Query codemod_review.pl safety predicates.

        Queries pass_safe/3 for each mutated module. On failure,
        attempts explain:explain_whynot/2 for a proof tree.
        """
        if self._engine is None:
            return []

        findings: list[Finding] = []

        codemod_review_path = (
            Path(__file__).resolve().parent.parent.parent
            / "prolog"
            / "codemod_review.pl"
        )
        if codemod_review_path.exists():
            try:
                self._engine.consult(str(codemod_review_path))
            except Exception as exc:
                logger.warning("Failed to consult codemod_review.pl: %s", exc)
                return []
        else:
            logger.debug(
                "codemod_review.pl not found at %s, skipping Prolog queries",
                codemod_review_path,
            )
            return []

        results = self._engine.query("codemod_mutated(Module, PassName)")
        seen: set[tuple[str, str]] = set()
        for binding in results:
            module = binding.get("Module", "")
            pass_name = binding.get("PassName", "")
            key = (module, pass_name)
            if key in seen:
                continue
            seen.add(key)

            result = self._engine.query_once(
                f"codemod_review:pass_safe('{pass_name}', '{module}', Safe)"
            )
            # janus marshals the Prolog atoms true/false as the strings
            # 'true'/'false', not Python bools (measured against the
            # process-global interpreter 2026-08-24) -- `is True` never
            # matched, so every safe (pass, module) pair produced a
            # false-positive PROLOG_SEMANTIC_REVIEW warning.
            if result is not None and result.get("Safe") in (True, "true"):
                continue

            proof_tree = self._explain_failure(pass_name, module)
            findings.append(
                Finding(
                    severity="warning",
                    rule="PROLOG_SEMANTIC_REVIEW",
                    path=module,
                    line=None,
                    message=(
                        f"Prolog safety check failed for pass "
                        f"'{pass_name}' on module '{module}'"
                    ),
                    proof_tree=proof_tree,
                )
            )

        return findings

    def _explain_failure(self, pass_name: str, module: str) -> dict[str, Any] | None:
        """Get proof tree for a failed safety check via explain_whynot."""
        if self._engine is None:
            return None

        try:
            proof = self._engine.query_explained(
                f"codemod_review:pass_safe('{pass_name}', '{module}', true)"
            )
            if not proof.succeeded and proof.whynot:
                return {
                    "goal": proof.goal,
                    "whynot": proof.whynot,
                }
        except Exception as exc:
            logger.debug("explain_whynot failed: %s", exc)

        return None

    def _cleanup_facts(self) -> None:
        """Retract asserted Prolog facts to avoid leaking state."""
        if not self._facts_asserted or self._engine is None:
            return
        try:
            self._engine.retractall("codemod_mutated(_, _)")
            self._engine.retractall("codemod_affected(_, _)")
        except Exception:
            pass
        self._facts_asserted = False


DEFAULT_REVIEWS: tuple[ReviewPass, ...] = (
    ParseIntegrity(),
    BannedConstructs(),
    DiffBudget(),
    PrologSemanticReview(),
)