orchestrata_ultimate.refactor.codemod.review_engine

Reference › orchestrata_ultimate › refactor › codemod › review_engine

The in-memory review engine that workspace runs over a mutated forest before it writes anything. The daemon server and scripts.codemod also use it. It combines the catalogue, the check implementations, the context and the Prolog pass from review.

ReviewPhaseRunner.run_phase runs one phase's checks over a set of modules: phase 1 runs one task per file on a thread pool, phase 2 walks the files in order after it so graph reads are not interleaved. A failing check raises, so a crashed review is never treated as a clean one. InMemoryReviewEngine is configured with a registry (the default catalogue is imported lazily, because the verification checks it pulls in are numerous), an optional Prolog pass, a baseline and include_phase3. review_detailed runs the checks on the mutated modules and returns a ReviewOutcome of introduced and legacy findings, then appends the Prolog findings; review returns the introduced findings only. The default baseline is delta (ADR-0073): the same checks run over the original trees, and a finding counts as introduced only when it is absent there, matched on path, rule, a message with line numbers removed (_line_free) and the stripped line text. Findings already present are returned as legacy. The absolute baseline reports every finding of the mutated modules, and any other value raises ValueError.

_shared and _context build the ReviewContext for each file, including the dependency names declared in pyproject.toml (_declared_dependencies) and the banned registry. _top_level, _sibling_name and _is_test_path decide whether an import is local to the importer's root. without_phase3 returns a copy that skips the Prolog pass.

At a glance

Class diagram

Dependency diagram

API

Classes

ReviewOutcome (dataclass)

Decorators: traces(spec='110', story='US-110-004', reqs=['REQ-110-033', 'REQ-110-035']), dataclass(frozen=True, slots=True) · Line 60

The two finding sets of one review (ADR-0073).

ReviewPhaseRunner

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

Run one phase's checks over a set of modules.

InMemoryReviewEngine

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

The 40-check review over the mutated forest (a ReviewPass).

_LineTexts

Line 450

Stripped source line of a finding, read from its own tree.

Functions

Source

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

"""SPEC-110 FR-012 -- the in-memory review engine.

``InMemoryReviewEngine`` runs the 40-check catalogue over the MUTATED
forest in three phases (REQ-110-035): phase 1 is the file-local structural
checks, run in parallel per file; phase 2 is the correctness and
test-isolation checks, run after phase 1 with the workspace's
``CodeIntelligenceGraph`` on the context; phase 3 is the existing
``PrologSemanticReview``. Findings come only from the mutated modules'
``ast`` overlay and token streams (REQ-110-034): the engine never re-reads
source from disk.

``baseline='delta'`` (the default, ADR-0073) reports as findings only those
the pass introduced, so an error finding aborts the commit (ERR-110-002)
only when the change introduced it; the findings the original trees already
carried come back separately as legacy findings and never veto.
``baseline='absolute'`` (the opt-in clean-slate gate) reports every finding
of the mutated modules.
"""

from __future__ import annotations

import os
import re
import tomllib
from collections import Counter
from collections.abc import Callable, Mapping
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from importlib.metadata import packages_distributions
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, cast

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod.report import ChangeSet, Finding
from orchestrata_ultimate.refactor.codemod.review import (
    PrologSemanticReview,
    ReviewPass,
    _banned_libraries,
)
from orchestrata_ultimate.refactor.codemod.review_catalogue import (
    CheckRegistry,
    ReviewCheck,
)
from orchestrata_ultimate.refactor.codemod.review_context import (
    _SIBLING_ROOTS,
    ReviewableWorkspace,
    ReviewContext,
)

if TYPE_CHECKING:
    from orchestrata_ultimate.refactor.ctok import LosslessModule

_MAIN_FILES = frozenset({"cli.py", "__main__.py", "main.py"})
_DIST_NAME = re.compile(r"\s*([A-Za-z0-9][A-Za-z0-9._-]*)")
_Baseline = Literal["absolute", "delta"]


@traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-035"])
@dataclass(frozen=True, slots=True)
class ReviewOutcome:
    """The two finding sets of one review (ADR-0073).

    Attributes:
        introduced: Findings the change introduced (every finding under the
            ``absolute`` baseline); the only set that can veto a commit.
        legacy: Findings the original trees already carried, reported but
            never vetoing; always empty under the ``absolute`` baseline.
    """

    introduced: tuple[Finding, ...]
    legacy: tuple[Finding, ...] = ()


@traces(spec="110", story="US-110-004", reqs=["REQ-110-035"])
class ReviewPhaseRunner:
    """Run one phase's checks over a set of modules.

    Phase 1 fans out one task per file; phase 2 walks the files in order so
    its graph reads are not interleaved.
    """

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-035"])
    def __init__(self, registry: CheckRegistry, max_workers: int | None = None) -> None:
        """Bind the registry and the phase-1 worker count.

        Args:
            registry: Supplies each phase's checks.
            max_workers: Phase-1 pool size; the CPU count when ``None``.
        """
        self._registry = registry
        self._max_workers = max_workers

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-035"])
    def run_phase(
        self,
        phase: Literal[1, 2],
        modules: Mapping[str, LosslessModule],
        ctx_factory: Callable[[str], ReviewContext],
    ) -> list[Finding]:
        """Run every check of ``phase`` over every module.

        Args:
            phase: 1 (parallel per file) or 2 (after phase 1).
            modules: Forest-relative path -> module, in report order.
            ctx_factory: Builds the context for one path.

        Returns:
            Findings in module order, then check order; a failing check
            raises (a crashed review is not a clean one).
        """
        checks = self._registry.checks_for_phase(phase)
        if not checks or not modules:
            return []
        paths = list(modules)
        workers = min(len(paths), self._max_workers or os.cpu_count() or 1)
        if phase == 1 and workers > 1:
            with ThreadPoolExecutor(max_workers=workers) as pool:
                batches = list(
                    pool.map(
                        lambda path: self._run_file(
                            checks, path, modules[path], ctx_factory(path)
                        ),
                        paths,
                    )
                )
        else:
            batches = [
                self._run_file(checks, path, modules[path], ctx_factory(path))
                for path in paths
            ]
        return [finding for batch in batches for finding in batch]

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-035"])
    def _run_file(
        self,
        checks: tuple[ReviewCheck, ...],
        path: str,
        module: LosslessModule,
        ctx: ReviewContext,
    ) -> list[Finding]:
        """Every check of the phase over one module."""
        findings: list[Finding] = []
        for check in checks:
            findings.extend(check.run(module, ctx))
        return findings


@traces(
    spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-034", "REQ-110-035"]
)
class InMemoryReviewEngine:
    """The 40-check review over the mutated forest (a ``ReviewPass``)."""

    name: str = "InMemoryReviewEngine"

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-035"])
    def __init__(
        self,
        registry: CheckRegistry | None = None,
        prolog: ReviewPass | None = None,
        baseline: _Baseline = "delta",
        include_phase3: bool = True,
        max_workers: int | None = None,
    ) -> None:
        """Configure the engine.

        Args:
            registry: The catalogue; ``default_registry()`` when ``None``.
            prolog: The phase-3 pass; a fresh ``PrologSemanticReview`` per
                review when ``None``.
            baseline: ``delta`` (default, ADR-0073) reports only the
                findings the change introduced, matched on path, rule and
                line-free message; ``absolute`` reports every finding of the
                mutated modules (the clean-slate gate).
            include_phase3: Run the Prolog pass after phases 1 and 2.
            max_workers: Phase-1 pool size.

        Raises:
            ValueError: ``baseline`` is neither ``absolute`` nor ``delta``.
        """
        if baseline not in ("absolute", "delta"):
            raise ValueError(f"baseline must be 'absolute' or 'delta': {baseline!r}")
        if registry is None:
            # Deferred on purpose: review_checks pulls the verification review
            # checks (103 eager modules over six packages, measured
            # 2026-10-08). An eager import here broke the deployed
            # codemod-daemon's import smoke on 2026-10-08 (ship of 3bb5319df)
            # while the tools deploy vendored a hand-written subtree list; the
            # list is computed and gate-pinned since WI-DEPLOY
            # (deploy/tools-vendor-manifest.txt, tests/deploy/
            # test_tools_vendor_manifest.py), so the deployed daemon resolves
            # this import. It stays lazy so the daemon's eager closure remains
            # one module (scripts/tools_import_closure.py --eager) and the
            # package imports wherever a catalogue is injected.
            from orchestrata_ultimate.refactor.codemod.review_checks import (
                default_registry,
            )

            registry = default_registry()
        self._registry = registry
        self._prolog = prolog
        self._baseline: _Baseline = baseline
        self._include_phase3 = include_phase3
        self._max_workers = max_workers
        self._runner = ReviewPhaseRunner(self._registry, max_workers)

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033", "REQ-110-035"])
    def without_phase3(self) -> InMemoryReviewEngine:
        """A copy that skips the Prolog pass (the caller already runs one)."""
        return InMemoryReviewEngine(
            registry=self._registry,
            prolog=self._prolog,
            baseline=self._baseline,
            include_phase3=False,
            max_workers=self._max_workers,
        )

    @traces(
        spec="110",
        story="US-110-004",
        reqs=["REQ-110-033", "REQ-110-034", "REQ-110-035"],
    )
    def review(self, ws: ReviewableWorkspace, changes: ChangeSet) -> list[Finding]:
        """Review the mutated modules of ``changes``.

        Args:
            ws: The workspace the changes were computed against.
            changes: The pass output; each entry's ``mutated_tree`` is read.

        Returns:
            The introduced findings of :meth:`review_detailed`: phase-1
            findings, phase-2 findings, then phase-3 findings.
        """
        return list(self.review_detailed(ws, changes).introduced)

    @traces(
        spec="110",
        story="US-110-004",
        reqs=["REQ-110-033", "REQ-110-034", "REQ-110-035"],
    )
    def review_detailed(
        self, ws: ReviewableWorkspace, changes: ChangeSet
    ) -> ReviewOutcome:
        """Review the mutated modules and keep the legacy findings (ADR-0073).

        Args:
            ws: The workspace the changes were computed against.
            changes: The pass output; each entry's ``mutated_tree`` is read.

        Returns:
            The introduced findings (phase 1, phase 2, then phase 3) and, under
            the ``delta`` baseline, the legacy findings the original trees
            already carried; ``legacy`` is empty under ``absolute``.
        """
        mutated = {entry.path: entry.mutated_tree for entry in changes.entries}
        findings = self._run_checks(ws, mutated, {**ws.forest, **mutated})
        legacy: list[Finding] = []
        if self._baseline == "delta":
            original = {entry.path: entry.original_tree for entry in changes.entries}
            findings, legacy = self._delta(
                findings,
                self._run_checks(ws, original, {**ws.forest, **original}),
                mutated,
                original,
            )
        if self._include_phase3:
            prolog = (
                self._prolog if self._prolog is not None else PrologSemanticReview()
            )
            # review.py's ReviewPass names a forward-reference stand-in for
            # the workspace; the real one is passed through unchanged.
            findings.extend(prolog.review(cast("Any", ws), changes))
        return ReviewOutcome(introduced=tuple(findings), legacy=tuple(legacy))

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-035"])
    def _run_checks(
        self,
        ws: ReviewableWorkspace,
        modules: Mapping[str, LosslessModule],
        forest: Mapping[str, LosslessModule],
    ) -> list[Finding]:
        """Phases 1 and 2 over ``modules``, with ``forest`` on the context."""
        shared = self._shared(ws, forest)

        def ctx_factory(path: str) -> ReviewContext:
            return self._context(ws, path, forest, shared)

        findings = self._runner.run_phase(1, modules, ctx_factory)
        findings.extend(self._runner.run_phase(2, modules, ctx_factory))
        return findings

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033"])
    def _shared(
        self, ws: ReviewableWorkspace, forest: Mapping[str, LosslessModule]
    ) -> dict[str, frozenset[str]]:
        """The per-review facts every file's context shares."""
        paths = set(forest) | {path for path, _ in ws.unparsable}
        return {
            "project_modules": frozenset(_top_level(path) for path in paths),
            "sibling_modules": frozenset(
                name for path in paths if (name := _sibling_name(path)) is not None
            ),
            "workspace_files": frozenset(paths),
            "declared_deps": self._declared_dependencies(ws.root),
            "banned_libraries": _banned_libraries(),
        }

    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033"])
    def _context(
        self,
        ws: ReviewableWorkspace,
        path: str,
        forest: Mapping[str, LosslessModule],
        shared: Mapping[str, frozenset[str]],
    ) -> ReviewContext:
        """The context for one file of the review.

        Only a module under a sibling-import root may import a sibling by bare
        name (``scripts/`` is not on ``sys.path`` for ``src/``), so only such a
        module's ``project_modules`` carry the sibling names.
        """
        name = os.path.basename(path)
        project_modules = shared["project_modules"]
        if path.split("/")[0] in _SIBLING_ROOTS:
            project_modules = project_modules | shared["sibling_modules"]
        return ReviewContext(
            path=path,
            forest=forest,
            cig=ws.cig,
            project_modules=project_modules,
            workspace_files=shared["workspace_files"],
            declared_deps=shared["declared_deps"],
            root=ws.root,
            is_test=_is_test_path(path),
            is_main_file=name in _MAIN_FILES,
            banned_libraries=shared["banned_libraries"],
            include_roots=ws.include_roots,
        )

    @staticmethod
    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033"])
    def _declared_dependencies(root: Path) -> frozenset[str]:
        """Dependency names from the workspace's ``pyproject.toml``.

        Reads every place a dependency is declared (see ``_declared_specs``)
        and adds the import names the declared distributions provide, so an
        import of ``google`` is declared by ``protobuf``. This is
        configuration and installed-package metadata, not source: it is read
        once per review by the engine, never by a check.
        """
        try:
            data = tomllib.loads((root / "pyproject.toml").read_text(encoding="utf-8"))
        except OSError, tomllib.TOMLDecodeError:
            return frozenset()
        declared = {
            name for name in map(_requirement_name, _declared_specs(data)) if name
        }
        return frozenset(declared | _import_names_of(declared))

    @staticmethod
    @traces(spec="110", story="US-110-004", reqs=["REQ-110-033"])
    def _delta(
        mutated: list[Finding],
        original: list[Finding],
        mutated_trees: Mapping[str, LosslessModule],
        original_trees: Mapping[str, LosslessModule],
    ) -> tuple[list[Finding], list[Finding]]:
        """Split ``mutated`` into (introduced, legacy) against ``original``.

        Matched on (path, rule, line-free message, stripped text of the
        finding's line) with multiplicity, never the line number: an edit
        above a legacy finding shifts it (and the line number the reused
        checks embed in the message) without introducing it. Each side reads
        the line text from its own tree, so an in-place edit of a line that
        already carries a content-free finding (the secret check's message
        names no content) is an introduced finding, not a legacy one.
        """
        mutated_lines = _LineTexts(mutated_trees)
        original_lines = _LineTexts(original_trees)
        known = Counter(
            (f.path, f.rule, _line_free(f.message), original_lines.text(f))
            for f in original
        )
        introduced: list[Finding] = []
        legacy: list[Finding] = []
        for finding in mutated:
            key = (
                finding.path,
                finding.rule,
                _line_free(finding.message),
                mutated_lines.text(finding),
            )
            if known[key] > 0:
                known[key] -= 1
                legacy.append(finding)
            else:
                introduced.append(finding)
        return introduced, legacy


def _top_level(path: str) -> str:
    """The top-level name a forest path contributes to every importer.

    ``src/<pkg>/...`` contributes ``<pkg>``; any other path contributes its
    first component (the root's own name). The bare names a sibling-import
    root contributes to its own members are ``_sibling_name``'s.
    """
    parts = path.removesuffix(".py").split("/")
    if parts[0] == "src" and len(parts) > 1:
        return parts[1]
    return parts[0]


def _sibling_name(path: str) -> str | None:
    """The bare import name a file directly under a sibling root contributes.

    Under a sibling-import root (``scripts``, ``tests``, ``claude``) a file
    directly under the root contributes its stem and a package directory its
    own name, because those trees are put on ``sys.path`` and imported by bare
    name (the rule ``check-deps`` applies). The root's own ``__init__.py``,
    nested stems below a package and every other path contribute nothing.
    """
    parts = path.removesuffix(".py").split("/")
    if parts[0] not in _SIBLING_ROOTS or len(parts) < 2 or parts[1:] == ["__init__"]:
        return None
    return parts[1]


def _is_test_path(path: str) -> bool:
    """Whether ``path`` names a test file.

    A test file has a ``tests``/``test`` directory component in its
    repo-relative path, or is a ``conftest.py``. A ``test_*.py`` basename
    under ``src/`` or ``scripts/`` is production (``core/threading/
    test_scheduler.py`` is a scheduler, not a test).
    """
    parts = path.split("/")
    return parts[-1] == "conftest.py" or any(
        part in ("tests", "test") for part in parts[:-1]
    )


# ``L12: msg`` / ``L12 msg`` leads, ``(line 7)`` tokens and bare ``line 9``
# references: the three shapes the reused ``check_*`` functions embed.
_LEADING_LINE = re.compile(r"^L\d+(?::|\s)\s*")
_PAREN_LINE = re.compile(r"\s*\(line \d+\)")
_BARE_LINE = re.compile(r"\bline \d+\b")


class _LineTexts:
    """Stripped source line of a finding, read from its own tree.

    The lines of each module are split once, on first use; a finding with no
    line, a line out of range or a path outside the trees reads ``""``.
    """

    def __init__(self, trees: Mapping[str, LosslessModule]) -> None:
        self._trees = trees
        self._lines: dict[str, list[str]] = {}

    def text(self, finding: Finding) -> str:
        """The stripped text of ``finding.line`` in ``finding.path``."""
        module = self._trees.get(finding.path)
        if module is None or finding.line is None:
            return ""
        lines = self._lines.get(finding.path)
        if lines is None:
            lines = module.stream.source.decode("utf-8", errors="replace").split("\n")
            self._lines[finding.path] = lines
        if not 1 <= finding.line <= len(lines):
            return ""
        return lines[finding.line - 1].strip()


def _line_free(message: str) -> str:
    """``message`` without the line numbers it embeds.

    Delta matching compares messages across the original and mutated trees;
    an edit above a legacy finding moves its line number into the message,
    and the moved finding is still the same finding.
    """
    text = _LEADING_LINE.sub("", message)
    text = _PAREN_LINE.sub("", text)
    text = _BARE_LINE.sub("line", text)
    return " ".join(text.split())


def _requirement_name(spec: str) -> str:
    """Normalised distribution name of one dependency specifier."""
    match = _DIST_NAME.match(spec)
    return re.sub(r"[-_.]+", "_", match.group(1)).lower() if match else ""


def _declared_specs(data: Mapping[str, Any]) -> list[str]:
    """Every dependency specifier a parsed ``pyproject.toml`` declares.

    ``[project] dependencies``, every ``[project.optional-dependencies]``
    group, ``[dependency-groups]`` (string entries; include tables are not
    specifiers) and ``[tool.uv] dev-dependencies``.
    """
    project = data.get("project", {})
    specs: list[str] = list(project.get("dependencies", []))
    for group in (project.get("optional-dependencies", {}) or {}).values():
        specs.extend(group)
    for group in (data.get("dependency-groups", {}) or {}).values():
        specs.extend(spec for spec in group if isinstance(spec, str))
    specs.extend(data.get("tool", {}).get("uv", {}).get("dev-dependencies", []))
    return specs


def _import_names_of(declared: set[str]) -> set[str]:
    """Import names the declared distributions provide (``protobuf`` -> ``google``).

    The distribution -> import-name table is ``importlib.metadata``'s, the
    same one ``scripts/serialize_codebase_ast.py check-deps`` consults.
    """
    return {
        module.lower()
        for module, dists in packages_distributions().items()
        if any(re.sub(r"[-_.]+", "_", d).lower() in declared for d in dists)
    }