orchestrata_ultimate.gates.commit_review

Reference › orchestrata_ultimate › gates › commit_review

The pre-commit review verdict: the reviewers report and Prolog decides. It is called by scripts.gate review-staged, which scripts/hooks/pre-commit invokes, and the rules live in the commit_review.pl rulebase. The driver, CommitReview, decides nothing itself. run reads the staged paths (staged_paths, added, copied or modified files from the index), writes them as facts and asks review_required/2 which reviewer must run on which file (required). run_reviewer then runs exactly those reviewers and records a ReviewRun: exit code, the totals the reviewer states about itself, every Finding, a log tail and wall time. parse_code_review reads the JSON of the code reviewer and parse_test_review the text of the test reviewer. render_facts turns the runs into a fact file, and evaluate asks commit_status/1 for a CommitVerdict.

The design point is that scope, blocking policy and every soundness check live in the rulebase, where review_defect/3 names each way a review can fail to be a verdict. A report that does not parse becomes a missing review_report/4 fact, which the rules turn into an invalid verdict, so a silent reviewer cannot read as zero issues. The verdict uses the three values of GateEngine: 0 pass, 1 fail, 2 invalid. Each evaluation consults the rulebase and one generated fact file and unloads both afterwards, because janus shares one process-global SWI database. write_walls writes per-file reviewer wall times as a side channel (ADR-0101 rule 2). See the prolog-gate page for the wider layer.

At a glance

Class diagram

Dependency diagram

API

Classes

Finding (dataclass)

Decorators: dataclass(frozen=True) · Line 89

One issue as a reviewer reported it. at is the file the issue names, which a repo-wide check inside code-review.py can set to a file other than the one under review.

ReviewRun (dataclass)

Decorators: dataclass(frozen=True) · Line 105

What one reviewer reported about one staged file.

ReviewStatus (dataclass)

Decorators: dataclass(frozen=True) · Line 125

kind is pass / fail / invalid; detail is the status term the rules derived, e.g. fail(2) or invalid(count_mismatch(3,2)).

BlockingFinding (dataclass)

Decorators: dataclass(frozen=True) · Line 136

CommitVerdict (dataclass)

Decorators: dataclass · Line 143

What the rules concluded about this commit, and on what evidence.

CommitReview (dataclass)

Decorators: dataclass · Line 413

Run the commit review: staged paths -> required reviews -> reports -> verdict.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gates/commit_review.py

"""Pre-commit review verdict — the reviewers report, Prolog decides.

Rules: ``src/orchestrata_ultimate/prolog/commit_review.pl``.
CLI:   ``scripts/gate.py review-staged`` (called by ``scripts/hooks/pre-commit``).

THE DRIVER DECIDES NOTHING. It does three mechanical things:

1. writes the staged paths as facts and asks ``review_required/2`` which
   reviews to run;
2. runs exactly those reviewers and records what each REPORTED — its exit
   code, the totals it states for itself, every finding — as ground facts;
3. asks ``commit_status/1`` for the verdict.

Scope, blocking policy and every soundness check live in the rulebase, where
``review_defect/3`` names each way a review can fail to be a verdict. The shell
review this replaces read "the reviewer printed nothing" as "0 issues" three
separate ways (measured 2026-10-11; the rulebase header lists them). Here a
report that does not parse is a MISSING ``review_report/4`` fact, which the
rules turn into ``invalid(no_report)`` — it cannot become a pass.

The verdict has the gate layer's three values and exit codes
(``gates.engine``): 0 pass / 1 fail / 2 invalid.

Every run consults the rulebase and one generated fact file, and unloads both
before returning — janus shares ONE process-global SWI database, and
``GateEngine._loaded`` documents what a second run inherits from a first that
did not unload (measured: run 2 still answered run 1's facts).
"""

from __future__ import annotations

import json
import os
import re
import shutil
import subprocess
import tempfile
import time
from collections.abc import Iterator, Sequence
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from orchestrata_ultimate.gates.engine import (
    EXIT_FAIL,
    EXIT_INVALID,
    EXIT_PASS,
    FAIL,
    INVALID,
    PASS,
    PROLOG_DIR,
)
from orchestrata_ultimate.gates.fact_extractor import prolog_atom

COMMIT_REVIEW_RULES = PROLOG_DIR / "commit_review.pl"

CODE_REVIEW = "code_review"
TEST_REVIEW = "test_review"

# The reviewers, relative to the checkout being committed. A checkout reviews
# its own commit with its own reviewers (RUL-021: every worktree owns its tree).
REVIEWER_SCRIPTS = {
    CODE_REVIEW: Path("scripts") / "code-review.py",
    TEST_REVIEW: Path("scripts") / "test-review.py",
}

# A reviewer that has not answered in this long is not going to. 124 is
# timeout(1)'s exit code; it is not a verdict exit, so the rules read it as
# ``invalid(exit(124))``.
DEFAULT_REVIEW_TIMEOUT_S = 600.0
EXIT_TIMED_OUT = 124
# The interpreter itself could not be started (missing or not executable).
EXIT_NOT_STARTED = 127

LOG_TAIL_LINES = 15

# test-review.py's text report (``output_text``): it has no machine format
# since e1e1a4f77 removed --format, so these four shapes ARE its contract.
# A line that stops matching makes the parsed count disagree with the
# reviewer's own "Total Issues" line, and the rules report count_mismatch.
_TR_ANALYZED = re.compile(r"^Files Analyzed: (\d+)$", re.MULTILINE)
_TR_TOTAL = re.compile(r"^Total Issues: (\d+)$", re.MULTILINE)
_TR_SECTION = re.compile(r"^(ERROR|WARNING|INFO) \(\d+\):$")
_TR_ITEM = re.compile(r"^  (.+):(\d+) \[([^\]]+)\] (.*)$")


@dataclass(frozen=True)
class Finding:
    """One issue as a reviewer reported it. ``at`` is the file the issue
    names, which a repo-wide check inside code-review.py can set to a file
    other than the one under review."""

    severity: str
    code: str
    at: str
    line: int
    message: str

    def __str__(self) -> str:
        return f"[{self.severity}] {self.at}:{self.line} {self.code} {self.message}"


@dataclass(frozen=True)
class ReviewRun:
    """What one reviewer reported about one staged file.

    ``analyzed``/``total`` are ``None`` when the report did not parse — the
    fact is then simply not emitted, and the rules decide what that means.
    """

    file: str
    reviewer: str
    exit_code: int
    analyzed: int | None = None
    total: int | None = None
    findings: tuple[Finding, ...] = ()
    log_tail: str = ""
    #: Wall-clock seconds the reviewer took (ADR-0101 rule 2 side channel).
    #: ``None`` only on runs constructed by hand in tests.
    wall_s: float | None = None


@dataclass(frozen=True)
class ReviewStatus:
    """``kind`` is pass / fail / invalid; ``detail`` is the status term the
    rules derived, e.g. ``fail(2)`` or ``invalid(count_mismatch(3,2))``."""

    file: str
    reviewer: str
    kind: str
    detail: str


@dataclass(frozen=True)
class BlockingFinding:
    file: str
    reviewer: str
    finding: Finding


@dataclass
class CommitVerdict:
    """What the rules concluded about this commit, and on what evidence."""

    status: str
    reviews: list[ReviewStatus] = field(default_factory=list)
    defects: list[str] = field(default_factory=list)
    blocking: list[BlockingFinding] = field(default_factory=list)
    runs: list[ReviewRun] = field(default_factory=list)

    @property
    def exit_code(self) -> int:
        return {PASS: EXIT_PASS, FAIL: EXIT_FAIL, INVALID: EXIT_INVALID}[self.status]

    def report(self) -> str:
        """Render the verdict: every review that did not pass, why, and the
        evidence — the findings that block, or the reviewer's own output when
        it produced no verdict."""
        lines = [f"commit review: {self.status.upper()}"]
        if self.status == PASS:
            lines.append(
                f"  {len(self.reviews)} review(s), no blocking findings"
                if self.reviews
                else "  no staged file requires a review"
            )
        for defect in self.defects:
            lines.append(f"  run is not a verdict: {defect}")
        runs = {(run.file, run.reviewer): run for run in self.runs}
        for review in self.reviews:
            if review.kind == PASS:
                continue
            lines.append(f"  {review.file}  {review.reviewer}  {review.detail}")
            if review.kind == FAIL:
                lines.extend(
                    f"    {item.finding}"
                    for item in self.blocking
                    if (item.file, item.reviewer) == (review.file, review.reviewer)
                )
                continue
            run = runs.get((review.file, review.reviewer))
            if run is not None and run.log_tail:
                lines.append("    reviewer output (tail):")
                lines.extend(f"      {line}" for line in run.log_tail.splitlines())
        return "\n".join(lines)


# -- reviewer reports --------------------------------------------------------


def parse_code_review(document: str) -> tuple[int, int, tuple[Finding, ...]] | None:
    """``(files_analyzed, total_issues, findings)`` from code-review.py's JSON
    report, or ``None`` when it is not that document."""
    try:
        data = json.loads(document)
        analyzed = int(data["files_analyzed"])
        total = int(data["total_issues"])
        findings = tuple(
            Finding(
                severity=str(issue["severity"]).lower(),
                code=str(issue["code"]),
                at=str(issue["file"]),
                line=int(issue["line"]),
                message=str(issue["message"]),
            )
            for issue in data["issues"]
        )
    except ValueError, KeyError, TypeError:
        return None
    return analyzed, total, findings


def parse_test_review(text: str) -> tuple[int, int, tuple[Finding, ...]] | None:
    """``(files_analyzed, total_issues, findings)`` from test-review.py's text
    report, or ``None`` when the text carries no report at all."""
    analyzed = _TR_ANALYZED.search(text)
    total = _TR_TOTAL.search(text)
    if analyzed is None or total is None:
        return None
    findings: list[Finding] = []
    severity: str | None = None
    for line in text.splitlines():
        section = _TR_SECTION.match(line)
        if section is not None:
            severity = section.group(1).lower()
            continue
        item = _TR_ITEM.match(line)
        if item is not None and severity is not None:
            findings.append(
                Finding(
                    severity=severity,
                    code=item.group(3),
                    at=item.group(1),
                    line=int(item.group(2)),
                    message=item.group(4),
                )
            )
    return int(analyzed.group(1)), int(total.group(1)), tuple(findings)


def _tail(text: str, lines: int = LOG_TAIL_LINES) -> str:
    return "\n".join(text.strip().splitlines()[-lines:])


def run_reviewer(
    repo_root: Path,
    python: Path,
    reviewer: str,
    file: str,
    *,
    timeout: float = DEFAULT_REVIEW_TIMEOUT_S,
) -> ReviewRun:
    """Run one reviewer on one file and record what it reported.

    Both reviewers run at ``--min-severity INFO``: the reviewer reports
    everything, and ``blocking_severity/2`` — not a command-line flag chosen
    here — decides what stops the commit.

    The interpreter's directory leads PATH because code-review.py shells out
    to ruff and mypy by name; without it, a PATH lacking the venv made
    RUFF_NOT_FOUND the review's only ERROR (measured 2026-10-11).
    """
    script = repo_root / REVIEWER_SCRIPTS[reviewer]
    env = dict(os.environ)
    env["PATH"] = f"{python.parent}{os.pathsep}{env.get('PATH', '')}"
    started = time.monotonic()
    with tempfile.TemporaryDirectory(prefix="commit_review_run_") as tmp:
        report_path = Path(tmp) / "report.json"
        if reviewer == CODE_REVIEW:
            argv = [
                str(python), str(script), "--format", "json",
                "--min-severity", "INFO", "--output", str(report_path), file,
            ]  # fmt: skip
        else:
            argv = [
                str(python),
                str(script),
                "--min-severity",
                "INFO",
                "--stdout",
            ]
            baseline = repo_root / "reports" / "t21-worklist-baseline.json"
            if baseline.is_file():
                # T21-ROUTING: downgrade named-debt exact-key WARNINGs to INFO so
                # the dispatch-time baseline does not block existing files.
                argv += ["--worklist-baseline", str(baseline)]
            argv.append(file)
        try:
            proc = subprocess.run(
                argv,
                cwd=repo_root,
                env=env,
                capture_output=True,
                text=True,
                timeout=timeout,
                check=False,
            )
        except subprocess.TimeoutExpired:
            return ReviewRun(
                file, reviewer, EXIT_TIMED_OUT,
                log_tail=f"{reviewer} did not finish within {timeout:.0f}s",
                wall_s=time.monotonic() - started,
            )  # fmt: skip
        except OSError as exc:
            return ReviewRun(
                file, reviewer, EXIT_NOT_STARTED,
                log_tail=f"could not start {python}: {exc}",
                wall_s=time.monotonic() - started,
            )  # fmt: skip

        if reviewer == CODE_REVIEW:
            document = (
                report_path.read_text(encoding="utf-8") if report_path.is_file() else ""
            )
            parsed = parse_code_review(document)
        else:
            parsed = parse_test_review(proc.stdout)

    log_tail = _tail(f"{proc.stdout}\n{proc.stderr}")
    wall_s = time.monotonic() - started
    if parsed is None:
        return ReviewRun(
            file, reviewer, proc.returncode, log_tail=log_tail, wall_s=wall_s
        )
    analyzed, total, findings = parsed
    return ReviewRun(
        file, reviewer, proc.returncode,
        analyzed=analyzed, total=total, findings=findings, log_tail=log_tail,
        wall_s=wall_s,
    )  # fmt: skip


def write_walls(runs: Sequence[ReviewRun], path: Path) -> None:
    """The ADR-0101 rule 2 side channel: one JSON array of per-file reviewer
    walls for the pre-commit hook's ledger (``reports/precommit-walls.ndjson``).

    The hook passes ``path`` in ``PRECOMMIT_WALLS_FILE``; when the variable is
    absent nothing calls this and ``review-staged`` is byte-identical to before
    the channel existed. ``reviewer`` is hyphenated (``code-review``) to match
    the ledger schema; the run's ``wall_s`` and ``exit_code`` are the only
    numbers that exist here, so they are the numbers reported.
    """
    payload = [
        {
            "file": run.file,
            "reviewer": run.reviewer.replace("_", "-"),
            "wall_s": round(run.wall_s, 3) if run.wall_s is not None else None,
            "rc": run.exit_code,
        }
        for run in runs
    ]
    path.write_text(json.dumps(payload), encoding="utf-8")


def staged_paths(repo_root: Path) -> list[str]:
    """Every path staged as added, copied or modified, repo-relative.

    A failing ``git`` raises: the caller reports that as no verdict rather
    than reviewing an empty list.
    """
    proc = subprocess.run(
        ["git", "diff", "--cached", "--name-only", "--diff-filter=ACM", "-z"],
        cwd=repo_root,
        capture_output=True,
        check=True,
    )
    raw = proc.stdout.decode("utf-8", errors="surrogateescape")
    return [path for path in raw.split("\0") if path]


# -- facts -------------------------------------------------------------------


def render_facts(staged: Sequence[str], runs: Sequence[ReviewRun]) -> str:
    """The fact file for one evaluation, grouped by predicate so SWI never
    warns that a predicate's clauses are not together.

    A report that did not parse emits ``review_exit/3`` and nothing else —
    the absence is the evidence.
    """
    staged_lines = [f"staged_file({prolog_atom(path)})." for path in staged]
    exit_lines: list[str] = []
    report_lines: list[str] = []
    finding_lines: list[str] = []
    for run in runs:
        file, reviewer = prolog_atom(run.file), prolog_atom(run.reviewer)
        exit_lines.append(f"review_exit({file}, {reviewer}, {int(run.exit_code)}).")
        if run.analyzed is None or run.total is None:
            continue
        report_lines.append(
            f"review_report({file}, {reviewer}, {int(run.analyzed)}, {int(run.total)})."
        )
        finding_lines.extend(
            f"review_finding({file}, {reviewer}, {prolog_atom(f.severity)}, "
            f"{prolog_atom(f.code)}, {prolog_atom(f.at)}, {int(f.line)}, "
            f"{prolog_atom(f.message)})."
            for f in run.findings
        )
    lines = [
        f"staged_count({len(staged)}).",
        *staged_lines,
        *exit_lines,
        *report_lines,
        *finding_lines,
    ]
    return "\n".join(lines) + "\n"


# -- evaluation --------------------------------------------------------------


@dataclass
class CommitReview:
    """Run the commit review: staged paths -> required reviews -> reports ->
    verdict.

    Args:
        repo_root: the checkout being committed; reviewers run from here.
        python: interpreter that runs the reviewers — the checkout's own.
        timeout: per-reviewer wall-clock limit, in seconds.
    """

    repo_root: Path
    python: Path
    timeout: float = DEFAULT_REVIEW_TIMEOUT_S

    def run(self, staged: Sequence[str] | None = None) -> CommitVerdict:
        """Review the staged paths (read from the index when not given)."""
        paths = list(staged) if staged is not None else staged_paths(self.repo_root)
        runs = [
            run_reviewer(
                self.repo_root, self.python, reviewer, file, timeout=self.timeout
            )
            for file, reviewer in self.required(paths)
        ]
        return self.evaluate(paths, runs)

    def required(self, staged: Sequence[str]) -> list[tuple[str, str]]:
        """``review_required/2`` over these staged paths, as (file, reviewer)."""
        with _consulted(render_facts(staged, ())) as janus:
            result = janus.query_once(
                "findall([_F, _R], commit_review:review_required(_F, _R), Rows)"
            )
        rows = result.get("Rows", []) if result else []
        return [(str(row[0]), str(row[1])) for row in rows]

    @staticmethod
    def evaluate(staged: Sequence[str], runs: Sequence[ReviewRun]) -> CommitVerdict:
        """The rules' verdict over these staged paths and these reports."""
        with _consulted(render_facts(staged, runs)) as janus:
            statuses = janus.query_once(
                "findall([_F, _R, _K, _T],"
                " ( commit_review:review_status(_F, _R, _S),"
                "   commit_review:status_kind(_S, _K),"
                "   format(atom(_T), '~q', [_S]) ),"
                " Rows)"
            )
            defects = janus.query_once(
                "findall(_T,"
                " ( commit_review:commit_defect(_D), format(atom(_T), '~q', [_D]) ),"
                " Rows)"
            )
            blocking = janus.query_once(
                "findall([_F, _R, _Sev, _C, _At, _L, _M],"
                " commit_review:blocking_finding(_F, _R, _Sev, _C, _At, _L, _M),"
                " Rows)"
            )
            outcome = janus.query_once(
                "findall(_S, commit_review:commit_status(_S), Rows)"
            )

        defect_texts = [str(text) for text in _rows(defects)]
        verdicts = [str(status) for status in _rows(outcome)]
        if len(verdicts) != 1 or verdicts[0] not in (PASS, FAIL, INVALID):
            # commit_status/1 is documented det. Anything else is a broken
            # rulebase, which is no verdict — never a pass.
            defect_texts.append(
                f"commit_status/1 answered {verdicts!r}, not one status"
            )
            status = INVALID
        else:
            status = verdicts[0]
        return CommitVerdict(
            status=status,
            reviews=[
                ReviewStatus(str(r[0]), str(r[1]), str(r[2]), str(r[3]))
                for r in _rows(statuses)
            ],
            defects=defect_texts,
            blocking=[
                BlockingFinding(
                    str(r[0]),
                    str(r[1]),
                    Finding(str(r[2]), str(r[3]), str(r[4]), int(r[5]), str(r[6])),
                )
                for r in _rows(blocking)
            ],
            runs=list(runs),
        )


def _rows(result: Any) -> list[Any]:
    return list(result.get("Rows", [])) if result else []


@contextmanager
def _consulted(facts_text: str) -> Iterator[Any]:
    """Consult the rulebase, then this run's facts; unload both on the way out.

    Rules first: their ``:- dynamic user:...`` declarations must exist before
    the facts load (``GateEngine._loaded`` documents the same ordering for
    ``code_facts.pl``). The ``finally`` is load-bearing for the reason given in
    the module docstring.
    """
    from orchestrata_ultimate.core.janus_bootstrap import janus

    workdir = Path(tempfile.mkdtemp(prefix="commit_review_"))
    facts_path = workdir / "facts.pl"
    facts_path.write_text(facts_text, encoding="utf-8")
    loaded: list[Path] = []
    try:
        for path in (COMMIT_REVIEW_RULES, facts_path):
            janus.consult(str(path))
            loaded.append(path)
        yield janus
    finally:
        for path in reversed(loaded):
            escaped = str(path).replace("'", "''")
            janus.query_once(f"unload_file('{escaped}')")
        shutil.rmtree(workdir, ignore_errors=True)