scripts.gate

Reference › scripts › gate

scripts/gate.py is the command-line seam for the Prolog gate layer: it extracts a fact base from the AST snapshot, runs the gate rules over it and reports proofs. It is run by a human or by a gate step; no other module imports it. It is a thin front end over orchestrata_ultimate.gates.engine, orchestrata_ultimate.gates.commit_review and orchestrata_ultimate.core.janus_bootstrap, which carry the real work.

build_parser defines seven subcommands, dispatched by main: facts (derive the fact base from the snapshot, with --out, --stats, --snapshot and --repo-root); run (evaluate the gate; --workdir, --max-listed); uncovered (list every policy requirement that no enabled mechanism covers, REQ-GATE-011); explain and why-not (show the proof for a violation at a location, or explain why a rule does not fire, reporting a missing candidate clause instead of printing nothing); diff (compare the Prolog gate with the legacy checker it replaces and account for every disagreement); and review-staged (review the git index).

The exit codes carry three outcomes, not two. run exits 0 on pass, 1 on fail and 2 when the verdict is invalid: a stale fact base, or a run with no rules loaded, is never a pass because the rules did not evaluate the code in front of them (REQ-GATE-003). review-staged follows the same rule: a git that cannot read the index gives exit 2, not a clean verdict. uncovered is a report and exits 0 whenever the query succeeds. Facts come from the snapshot and nothing else (REQ-GATE-001); the only other disk access is the freshness layer, which compares disk with the snapshot. cmd_facts and cmd_run write generated facts under --workdir or a temporary directory.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Constants

Functions

Source

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

#!/usr/bin/env python3
"""CLI seam for the Prolog gate layer.

Contract: ``plans/contracts/prolog-gates.contract.json``.
Plan: ``plans/prolog-gates-implementation-plan.md``.

``facts`` extracts the fact base from the AST snapshot and reports it (P1).
``run`` evaluates the gate and exits **0 pass / 1 fail / 2 invalid** (P2) —
three outcomes, because "the rules did not evaluate the code in front of them"
is not a pass. ``explain`` and ``why-not`` answer, for one location, why a rule
DID or DID NOT fire (P4). ``diff`` runs this layer and the legacy checker it
replaces over one corpus and accounts for every disagreement (P5), because a
replacement that quietly finds LESS than what it replaces is a regression
wearing a cleanup's clothes. ``uncovered`` prints every policy requirement
with no enabled mechanism behind it (REQ-GATE-011) — RUL-014 as a query,
never a gate: it reports, it does not change whether ``run`` passes.
``review-staged`` is the pre-commit review as a derivation: the staged paths
and every reviewer's report become facts, ``commit_review.pl`` decides, and the
exit code is the same 0 pass / 1 fail / 2 invalid — a review that produced no
verdict is invalid, never a pass.

``why-not`` is the question that matters most and the one no checker in this
repo could previously answer. The tree-sitter relative-import checker was dead
for its entire life, twice over, and nothing noticed — because nothing could
ask it why it never fired.

Usage::

    python scripts/gate.py facts --stats
    python scripts/gate.py run
    python scripts/gate.py uncovered
    python scripts/gate.py explain no_relative_import src/pkg/a.py 7
    python scripts/gate.py why-not bare_os_environ src/pkg/a.py 12
    python scripts/gate.py review-staged
"""

from __future__ import annotations

import argparse
import os
import subprocess
import sys
import tempfile
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT / "src") not in sys.path:
    sys.path.insert(0, str(REPO_ROOT / "src"))

from orchestrata_ultimate.gates.commit_review import (  # noqa: E402
    DEFAULT_REVIEW_TIMEOUT_S,
    CommitReview,
    write_walls,
)
from orchestrata_ultimate.gates.engine import (  # noqa: E402
    FRESHNESS,
    GATE_RULES,
    GATE_SEVERITY,
    ONTOLOGY,
    GateEngine,
)
from orchestrata_ultimate.gates.fact_extractor import (  # noqa: E402
    DEFAULT_SNAPSHOT,
    FactExtractor,
    SnapshotMissingError,
)

EXIT_OK = 0
EXIT_ERROR = 2


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="gate",
        description="Prolog gate layer — facts, rules, proofs.",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    facts = sub.add_parser(
        "facts",
        help="Extract the fact base from the AST snapshot.",
        description=(
            "Facts are derived from the AST snapshot and from nothing else "
            "(REQ-GATE-001). The only disk access beyond the snapshot is the "
            "freshness layer, which must compare disk against snapshot."
        ),
    )
    facts.add_argument(
        "--snapshot",
        type=Path,
        default=DEFAULT_SNAPSHOT,
        help=f"AST snapshot to read (default: {DEFAULT_SNAPSHOT}).",
    )
    facts.add_argument(
        "--repo-root",
        type=Path,
        default=REPO_ROOT,
        help="Root the snapshot's relative paths resolve against.",
    )
    facts.add_argument(
        "--out",
        type=Path,
        default=None,
        help="Write the fact base to this path.",
    )
    facts.add_argument(
        "--stats",
        action="store_true",
        help="Print the per-predicate fact-count table.",
    )

    run = sub.add_parser(
        "run",
        help="Evaluate the gate. Exits 0 pass / 1 fail / 2 invalid.",
        description=(
            "Three outcomes, not two. A stale fact base or a run with no rules "
            "loaded is INVALID (exit 2), never a pass — the rules did not "
            "evaluate the code in front of them (REQ-GATE-003)."
        ),
    )
    run.add_argument(
        "--snapshot",
        type=Path,
        default=DEFAULT_SNAPSHOT,
        help=f"AST snapshot to read (default: {DEFAULT_SNAPSHOT}).",
    )
    run.add_argument(
        "--repo-root",
        type=Path,
        default=REPO_ROOT,
        help="Root the snapshot's relative paths resolve against.",
    )
    run.add_argument(
        "--workdir",
        type=Path,
        default=None,
        help="Where to write the generated facts (default: a temp directory).",
    )
    run.add_argument(
        "--max-listed",
        type=int,
        default=20,
        help="How many stale files / violations to list before summarising.",
    )

    uncovered = sub.add_parser(
        "uncovered",
        help="List every policy requirement with no enabled mechanism (REQ-GATE-011).",
        description=(
            "uncovered(Policy): RUL-014 ('every ruling must resolve to a "
            "mechanism') as an inference over the fact base, rather than "
            "scripts/rulings_lint.py's prose scan over ruling pages. A policy "
            "is covered when an enabled ruff rule or a live gate_rules.pl "
            "native rule mechanises it (policy_mechanism/3, covered/1 in "
            "gate_rules.pl). Always exits 0 on a successful query — this is a "
            "report, not a gate; it does not change whether `gate run` passes."
        ),
    )
    uncovered.add_argument(
        "--snapshot",
        type=Path,
        default=DEFAULT_SNAPSHOT,
        help=f"AST snapshot to read (default: {DEFAULT_SNAPSHOT}).",
    )
    uncovered.add_argument(
        "--repo-root",
        type=Path,
        default=REPO_ROOT,
        help="Root the snapshot's relative paths resolve against.",
    )
    uncovered.add_argument(
        "--workdir",
        type=Path,
        default=None,
        help="Where to write the generated facts (default: a temp directory).",
    )

    diff = sub.add_parser(
        "diff",
        help="Compare the gate against the legacy checker it replaces.",
        description=(
            "Runs both implementations over one corpus and accounts for every "
            "disagreement. Exits 0 when nothing is unexplained and nothing is "
            "lost — not when the sets are equal, because a Prolog-only finding "
            "is a coverage gain and is not a reason to refuse the replacement. "
            "A legacy-only finding is coverage the replacement does not have, "
            "and P6 deletes the legacy checker."
        ),
    )
    diff.add_argument(
        "--snapshot",
        type=Path,
        default=DEFAULT_SNAPSHOT,
        help=f"AST snapshot to read (default: {DEFAULT_SNAPSHOT}).",
    )
    diff.add_argument(
        "--repo-root",
        type=Path,
        default=REPO_ROOT,
        help="Root the snapshot's relative paths resolve against.",
    )
    diff.add_argument(
        "--source",
        type=Path,
        default=REPO_ROOT / "src" / "orchestrata_ultimate",
        help="Tree the LEGACY checker walks. Must be the tree the snapshot describes.",
    )
    diff.add_argument(
        "--workdir",
        type=Path,
        default=None,
        help="Where to write the generated facts (default: a temp directory).",
    )
    diff.add_argument(
        "--all-rules",
        action="store_true",
        help=(
            "Include Prolog rules the legacy checker never had. Off by default "
            "so the comparison is over the shared rule set; on, every extra "
            "rule is reported as a named gain."
        ),
    )

    for name, help_text in (
        ("explain", "Print the proof tree for one violation."),
        ("why-not", "Print the failing subgoal for a location that does NOT violate."),
    ):
        parser_ = sub.add_parser(name, help=help_text, description=help_text)
        parser_.add_argument("rule", help="Rule id, e.g. no_relative_import.")
        parser_.add_argument("file", help="Repo-relative path, as it appears in facts.")
        parser_.add_argument("line", type=int, help="Line number.")
        parser_.add_argument(
            "--snapshot",
            type=Path,
            default=DEFAULT_SNAPSHOT,
            help=f"AST snapshot to read (default: {DEFAULT_SNAPSHOT}).",
        )
        parser_.add_argument(
            "--repo-root",
            type=Path,
            default=REPO_ROOT,
            help="Root the snapshot's relative paths resolve against.",
        )

    review = sub.add_parser(
        "review-staged",
        help="Review the staged files. Exits 0 pass / 1 fail / 2 invalid.",
        description=(
            "The pre-commit review, as a derivation: staged paths and every "
            "reviewer's report become facts, and commit_review.pl decides. A "
            "review that did not run, did not finish, or left a report that "
            "does not add up is INVALID (exit 2) — no verdict is not a pass."
        ),
    )
    review.add_argument(
        "--repo-root",
        type=Path,
        default=REPO_ROOT,
        help="The checkout being committed; its index and reviewers are used.",
    )
    review.add_argument(
        "--python",
        type=Path,
        default=Path(sys.executable),
        help="Interpreter that runs the reviewers (default: this one).",
    )
    review.add_argument(
        "--timeout",
        type=float,
        default=DEFAULT_REVIEW_TIMEOUT_S,
        help="Per-reviewer limit in seconds; a reviewer past it is no verdict.",
    )
    return parser


def _explain_engine(args: argparse.Namespace) -> GateEngine:
    return GateEngine(snapshot_path=args.snapshot, repo_root=args.repo_root)


def cmd_explain(args: argparse.Namespace) -> int:
    """A location that does NOT violate exits non-zero and says so.

    Exiting 0 with an empty proof would read as "explained", when what
    happened is that there was nothing to explain — and the remedy is to ask
    ``why-not`` instead, so the message says that.
    """
    try:
        tree = _explain_engine(args).explain(args.rule, args.file, args.line)
    except SnapshotMissingError as exc:
        print(f"gate explain: {exc}", file=sys.stderr)
        return EXIT_ERROR

    if tree is None:
        print(
            f"gate explain: {args.rule} does not hold at {args.file}:{args.line} — "
            f"there is no proof to print. Ask why-not instead:\n"
            f"  python scripts/gate.py why-not {args.rule} {args.file} {args.line}",
            file=sys.stderr,
        )
        return EXIT_ERROR

    print(f"{args.rule} at {args.file}:{args.line}")
    print("  · fact   + rule fired   = built-in")
    print(tree.render(indent=1))
    return EXIT_OK


def cmd_why_not(args: argparse.Namespace) -> int:
    """No candidate clause at all is REPORTED, not silently empty.

    An empty answer has two very different causes — the rule id is wrong, or
    no clause head even unifies with this location — and both look identical
    if the command just prints nothing and exits 0.
    """
    try:
        failures = _explain_engine(args).why_not(args.rule, args.file, args.line)
    except SnapshotMissingError as exc:
        print(f"gate why-not: {exc}", file=sys.stderr)
        return EXIT_ERROR

    if not failures:
        print(
            f"gate why-not: no clause of {args.rule} even unifies with "
            f"{args.file}:{args.line}. Either the rule id is wrong, or the rule "
            f"HOLDS here — try explain.",
            file=sys.stderr,
        )
        return EXIT_ERROR

    print(f"{args.rule} does NOT hold at {args.file}:{args.line}")
    for failure in failures:
        print(f"  {failure}")
    return EXIT_OK


def cmd_run(args: argparse.Namespace) -> int:
    engine = GateEngine(
        snapshot_path=args.snapshot,
        repo_root=args.repo_root,
        workdir=args.workdir,
    )
    try:
        verdict = engine.run()
    except SnapshotMissingError as exc:
        print(f"gate run: {exc}", file=sys.stderr)
        return EXIT_ERROR

    stream = sys.stdout if verdict.exit_code == EXIT_OK else sys.stderr
    print(verdict.report(max_listed=args.max_listed), file=stream)
    return verdict.exit_code


def cmd_review_staged(args: argparse.Namespace) -> int:
    """A ``git`` that cannot read the index is no verdict (exit 2), never an
    empty list of files to review."""
    review = CommitReview(
        repo_root=args.repo_root, python=args.python, timeout=args.timeout
    )
    try:
        verdict = review.run()
    except subprocess.CalledProcessError as exc:
        print(f"gate review-staged: cannot read the index: {exc}", file=sys.stderr)
        return EXIT_ERROR

    # ADR-0101 rule 2: the pre-commit hook passes PRECOMMIT_WALLS_FILE and
    # reads the per-file reviewer walls for its append-only ledger
    # (reports/precommit-walls.ndjson). Absent, nothing is written and the
    # command is byte-identical to before the side channel existed.
    walls = os.environ.get("PRECOMMIT_WALLS_FILE")
    if walls:
        write_walls(verdict.runs, Path(walls))

    stream = sys.stdout if verdict.exit_code == EXIT_OK else sys.stderr
    print(verdict.report(), file=stream)
    return verdict.exit_code


def cmd_uncovered(args: argparse.Namespace) -> int:
    """REQ-GATE-011: print every policy requirement with no enabled mechanism.

    Deliberately NOT a GateEngine method — GateEngine's surface is `run` /
    `explain` / `why_not`, and this is a fourth, independent query (`gate
    facts --stats` is the same shape: a report, not a verdict). It follows
    `GateEngine._loaded`'s own pattern exactly — consult ontology, generated
    facts, freshness, the native rulebase, then unload everything in a
    `finally` so a second `gate` invocation in the same process never inherits
    this run's clauses (measured consequence if it did: gates/engine.py's own
    docstring on `_loaded`).
    """
    from orchestrata_ultimate.core.janus_bootstrap import janus

    workdir = (
        Path(tempfile.mkdtemp(prefix="orchestrata_gate_"))
        if args.workdir is None
        else Path(args.workdir)
    )
    workdir.mkdir(parents=True, exist_ok=True)
    facts_path = workdir / "facts.pl"
    try:
        FactExtractor(snapshot_path=args.snapshot, repo_root=args.repo_root).write(
            facts_path
        )
    except SnapshotMissingError as exc:
        print(f"gate uncovered: {exc}", file=sys.stderr)
        return EXIT_ERROR

    files = [
        path
        for path in (ONTOLOGY, facts_path, FRESHNESS, GATE_RULES, GATE_SEVERITY)
        if path.is_file()
    ]
    loaded: list[Path] = []
    try:
        for path in files:
            janus.consult(str(path))
            loaded.append(path)
        result = janus.query_once(
            "( current_predicate(uncovered/1)"
            "-> ( findall(_P, uncovered(_P), _Rows), sort(_Rows, Policies) )"
            " ; Policies = [] )"
        )
    finally:
        for path in reversed(loaded):
            escaped = str(path).replace("'", "''")
            janus.query_once(f"unload_file('{escaped}')")

    policies = [str(p) for p in (result.get("Policies", []) if result else [])]
    # ADVISORY, and the word is IN the line: this always returns EXIT_OK on
    # a successful query regardless of what it finds (below), by design — a
    # policy with no mechanism is DRIFT (RUL-014), not a violation. There is
    # no file:line and nothing is broken, so it must never read as a failure
    # a reader learns to scroll past. gates.py wires this gate on the
    # ordinary ok/fail mechanism (`returncode == 0`) rather than a second
    # severity channel, per the "severity tier in ONE channel" ruling
    # (policy_requirement/1, gate_rules.pl) — this print is what makes that
    # channel legible without the reader needing to know the exit code means
    # something different here.
    noun = "policy" if len(policies) == 1 else "policies"
    if not policies:
        print(f"gate uncovered [ADVISORY, drift not failure]: 0 unmechanized {noun}")
        return EXIT_OK
    print(
        f"gate uncovered [ADVISORY, drift not failure]: "
        f"{len(policies)} unmechanized {noun}"
    )
    for policy in policies:
        print(f"  {policy}")
    return EXIT_OK


#: The replace-then-delete ledger, made executable — every legacy rule this
#: layer replaces, mapped onto the Prolog rule that replaced it.
#:
#: EMPTY as of P6: `serialize_codebase_ast.py standards` and all four
#: `code-review.py` checks are deleted, so there is no second implementation
#: left to compare against. It is kept rather than removed because P7 repeats
#: P3->P6 for the next slice and refills it with that slice's checkers; the
#: guard in `cmd_diff` is what makes an empty map refuse instead of reporting a
#: vacuous "no differences".
LEGACY_RULE_MAP: dict[str, str] = {}


def cmd_diff(args: argparse.Namespace) -> int:
    """Compare the Prolog gate against the checkers it replaces (P5).

    Exits 0 when nothing is unexplained and nothing is lost — NOT when the two
    sets are equal. A gain is not a reason to refuse a replacement; a silent
    loss is the only thing this command exists to catch.

    Every row of the ledger is deleted as of P6, so there is nothing left to
    compare against and the command refuses. The comparator it drives
    (``gates/differential.py``) is intact and tested; P7 repeats P3->P6 for the
    next slice and refills ``LEGACY_RULE_MAP`` with that slice's checkers.
    """
    if not LEGACY_RULE_MAP:
        # An empty map means every count is zero and the report reads "no
        # differences" — a vacuous pass, and the same false-green shape as an
        # INVALID run reporting clean. Checked FIRST, before the engine runs:
        # doing the work and then refusing would burn a fact-base build to
        # reach a conclusion that was decided by the map alone.
        print(
            "gate diff: no legacy checker is left to compare against — every "
            "row of the replace-then-delete ledger is deleted. A differential "
            "with no counterpart reports zero differences, which is not a "
            "result. Point it at the next slice's checkers (P7).",
            file=sys.stderr,
        )
        return EXIT_ERROR

    # Unreachable until P7 refills the map, and deliberately not left as a
    # half-written comparison against checkers that no longer exist.
    raise NotImplementedError(
        "LEGACY_RULE_MAP is non-empty but no collector is wired for it — "
        "P7 must restore _legacy_findings for the slice it names."
    )


def cmd_facts(args: argparse.Namespace) -> int:
    extractor = FactExtractor(
        snapshot_path=args.snapshot,
        repo_root=args.repo_root,
    )
    try:
        if args.out is not None:
            stats = extractor.write(args.out)
        else:
            _, stats = extractor.extract()
    except SnapshotMissingError as exc:
        print(f"gate facts: {exc}", file=sys.stderr)
        return EXIT_ERROR

    if args.out is not None:
        print(f"wrote {stats.total:,} facts to {args.out}")
    if args.stats or args.out is None:
        print(stats.table())
    return EXIT_OK


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    if args.command == "facts":
        return cmd_facts(args)
    if args.command == "run":
        return cmd_run(args)
    if args.command == "uncovered":
        return cmd_uncovered(args)
    if args.command == "explain":
        return cmd_explain(args)
    if args.command == "why-not":
        return cmd_why_not(args)
    if args.command == "diff":
        return cmd_diff(args)
    if args.command == "review-staged":
        return cmd_review_staged(args)
    # argparse's required=True makes this unreachable; returning an error
    # rather than falling through keeps an added subcommand from silently
    # exiting 0 before it is wired.
    print(f"gate: unhandled command {args.command!r}", file=sys.stderr)
    return EXIT_ERROR


if __name__ == "__main__":
    raise SystemExit(main())