orchestrata_ultimate.gates.engine

Reference › orchestrata_ultimate › gates › engine

GateEngine is the thin evaluator at the centre of the Prolog gate layer (contract REQ-GATE-003/005); the policy it enforces is described on the prolog-gate page. It is used by commit_review and by scripts.gate. By design it holds no rule logic: a policy decision written in Python would duplicate the rules and would be invisible to gate explain.

GateEngine.run takes the fact base produced by the fact extractor from an AST snapshot, consults the rulebases through the janus bootstrap, asks two questions of the rules (whether the facts are stale, and which violations hold) and maps the answers to a Verdict. The verdict has three values: pass (the rules ran over a faithful fact base and found nothing), fail (they found something) and invalid (they did not evaluate the code in front of them). Invalid has two causes: a stale fact base, inferred by freshness.pl from ordinary facts so every rule inherits the precondition, and no rules loaded, since zero rules finding zero violations would otherwise print a green nobody earned. The exit codes are 0, 1 and 2 respectively.

Supporting types: Staleness (one reason the facts do not describe the tree), Violation (rule id, file, line, detail, severity, file class and whether it blocks), ProofTree and WhyNot (the derivation of one violation, and the subgoal that stopped a candidate clause, returned by explain), and Verdict.report, which renders the result.

At a glance

Source src/orchestrata_ultimate/gates/engine.py
Lines of code 637
Classes 6
Functions 2
Imports (internal) orchestrata_ultimate.core.base.traces, orchestrata_ultimate.core.janus_bootstrap, orchestrata_ultimate.gates.fact_extractor
Imported by orchestrata_ultimate.gates.commit_review, scripts.gate
Tests that import it tests/gates/test_commit_review.py, tests/gates/test_differential_vs_legacy.py, tests/gates/test_freshness_inference.py, tests/gates/test_gate_explain.py, tests/gates/test_gate_rules.py

Class diagram

Class diagram orchestrata_ultimate.gates.engine

Dependency diagram

Dependency diagram orchestrata_ultimate.gates.engine

API

Classes

Staleness (dataclass)

Decorators: dataclass(frozen=True) · Line 69

One reason the fact base does not describe the working tree.

Attribute Type Default
kind str
file str
Method Signature Summary
__str__ __str__() -> str

Violation (dataclass)

Decorators: dataclass(frozen=True) · Line 80

One solution of violation(RuleId, File, Line, Detail), with severity.

Attribute Type Default
rule_id str
file str
line int
detail str
severity str 'error'
file_class str 'unknown'
blocking bool True
Method Signature Summary
__str__ __str__() -> str

ProofTree (dataclass)

Decorators: dataclass(frozen=True) · Line 103

One node of a derivation from explain.pl.

Attribute Type Default
kind str
head str
children tuple[ProofTree, ...] ()
Method Signature Summary
render render(indent: int = 0) -> str An indented tree.

WhyNot (dataclass)

Decorators: dataclass(frozen=True) · Line 127

One candidate clause that COULD have fired, and the subgoal that stopped it.

Attribute Type Default
head str
failed_subgoal str
Method Signature Summary
__str__ __str__() -> str

Verdict (dataclass)

Decorators: dataclass · Line 160

What the gate concluded, and on what evidence.

Attribute Type Default
status str
reasons list[str] field(default_factory=list)
stale list[Staleness] field(default_factory=list)
violations list[Violation] field(default_factory=list)
rules_loaded int 0
facts FactStats | None None
Method Signature Summary
exit_code property exit_code() -> int
report report(*, max_listed: int = 20) -> str Render the verdict.

GateEngine (dataclass)

Decorators: dataclass · Line 201

Run the gate: facts -> precondition -> rules -> verdict.

Attribute Type Default
snapshot_path Path
repo_root Path Path('.')
workdir Path | None None
rulebases tuple[Path, ...] | None None
Method Signature Summary
__post_init__ __post_init__() -> None
run run() -> Verdict Evaluate the gate and return its verdict.
_evaluate _evaluate(janus: Any, stats: FactStats) -> Verdict
explain explain(rule_id: str, file: str, line: int) -> ProofTree | None The derivation of ONE violation, or None if it does not hold.
why_not why_not(rule_id: str, file: str, line: int) -> list[WhyNot] For a NON-violation, the first precondition of each candidate clause that did not hold.
_loaded _loaded(facts_path: Path, *, explain: bool = False) -> Iterator[Any] Consult the rulebases for the duration of one run, then unload them.
_staleness staticmethod _staleness(janus: Any) -> list[Staleness] Collect every staleness finding.
_scope_count staticmethod _scope_count(janus: Any) -> int How many source trees the fact base declares.
_missing_captures staticmethod _missing_captures(janus: Any) -> list[tuple[str, str]] Every (rule, capture) pair whose capture the snapshot lacks.
_rule_count staticmethod _rule_count(janus: Any) -> int How many violation/4 clauses exist.
_declared_rule_count staticmethod _declared_rule_count(path: Path) -> int How many violation/4 clause heads FILE declares, by TEXT.
_expected_rule_count _expected_rule_count() -> tuple[int, list[Path]] The declared violation/4 clause count, summed over every rulebase THIS run actually consults, and which of those files contribute at least one clause.
_clobber_reason staticmethod _clobber_reason(loaded: int, expected: int, contributors: list[Path]) -> str Name WHICH file is implicated, not just that the counts disagree.
_violations staticmethod _violations(janus: Any) -> list[Violation] Every solution of the single entry predicate (REQ-GATE-005), joined to its effective severity.

Functions

Function Signature Summary
_violation_goal _violation_goal(rule_id: str, file: str, line: int) -> str Build the goal both explain surfaces ask about.
_as_tree _as_tree(node: Any) -> ProofTree Rebuild the janus-marshalled dict tree as typed nodes.

Source

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

"""GateEngine — precondition, then rules, then a verdict that can say INVALID.

Contract: ``plans/contracts/prolog-gates.contract.json`` — REQ-GATE-003/005.
Plan: ``plans/prolog-gates-implementation-plan.md`` P2.

THE ENGINE IS DELIBERATELY THIN. It extracts facts, consults the rulebases,
asks two questions, and maps the answers to a verdict. No rule logic lives
here: a policy decision written in Python would be exactly the duplication this
layer exists to remove, and it would be invisible to ``gate explain``.

The verdict has THREE values, and that is the whole point:

    pass     the rules ran over a faithful fact base and found nothing
    fail     the rules ran and found something
    invalid  the rules did not evaluate the code in front of them

Collapsing ``invalid`` into ``pass`` is the false-green class already logged in
this project (2026-08-06, ``62349326`` — six code-review.py codes reporting
success for "the tool never ran"). Two distinct conditions produce it here:

1. THE FACT BASE IS STALE. Inferred by ``freshness.pl`` from ordinary facts, so
   every rule inherits the precondition and a new checker cannot forget it.
2. NO RULES ARE LOADED. A run of zero rules finds zero violations, and
   reporting that as a pass is the same lie in a different costume. Until
   ``gate_rules.pl`` lands (P3), every ``gate run`` on this tree says so out
   loud rather than printing a green nobody earned.
"""

from __future__ import annotations

import tempfile
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.gates.fact_extractor import FactExtractor, FactStats

PROLOG_DIR = Path(__file__).resolve().parents[1] / "prolog"
ONTOLOGY = PROLOG_DIR / "code_facts.pl"
FRESHNESS = PROLOG_DIR / "freshness.pl"

# gate_rules.pl / gate_severity.pl arrive with P3. They are named here, and
# their ABSENCE is handled as a first-class condition rather than silently
# yielding an empty rule set.
GATE_RULES = PROLOG_DIR / "gate_rules.pl"
GATE_SEVERITY = PROLOG_DIR / "gate_severity.pl"

# The proof-tree / why-not meta-interpreter (ADR-0031). Consulted only for
# `explain`/`why-not`: it is a debugging surface, and loading it on every
# `run` would put a meta-interpreter in the path of the hot query.
EXPLAIN = PROLOG_DIR / "explain.pl"

PASS = "pass"
FAIL = "fail"
INVALID = "invalid"

# Exit codes. `invalid` is NOT 0 and NOT 1: a caller that only checks
# "rc == 0" must still be stopped, and a caller that distinguishes them must be
# able to tell "the code is wrong" from "I could not tell".
EXIT_PASS = 0
EXIT_FAIL = 1
EXIT_INVALID = 2


@dataclass(frozen=True)
class Staleness:
    """One reason the fact base does not describe the working tree."""

    kind: str
    file: str

    def __str__(self) -> str:
        return f"{self.file} ({self.kind})"


@dataclass(frozen=True)
class Violation:
    """One solution of ``violation(RuleId, File, Line, Detail)``, with severity.

    ``severity`` is NOT part of the rule's answer — the rule says what is true
    of the code, ``gate_severity.pl`` says whether that stops the build, and
    the engine joins them (REQ-GATE-006). ``file_class`` is carried because it
    is the context the severity was resolved against, so a downgraded finding
    can be read without re-deriving why.
    """

    rule_id: str
    file: str
    line: int
    detail: str
    severity: str = "error"
    file_class: str = "unknown"
    blocking: bool = True

    def __str__(self) -> str:
        return f"{self.file}:{self.line} [{self.rule_id}/{self.severity}] {self.detail}"


@dataclass(frozen=True)
class ProofTree:
    """One node of a derivation from ``explain.pl``.

    ``kind`` is ``fact`` (a ground fact was consumed), ``proved`` (a rule
    fired and its body is below), or ``builtin`` (a built-in was called and
    not descended into).
    """

    kind: str
    head: str
    children: tuple[ProofTree, ...] = ()

    def render(self, indent: int = 0) -> str:
        """An indented tree. The MARKER matters more than the shape: reading a
        proof, the question is always which steps were facts from the snapshot
        and which were inference, and a uniform tree hides that."""
        marker = {"fact": "·", "proved": "+", "builtin": "="}.get(self.kind, "?")
        lines = [f"{'  ' * indent}{marker} {self.head}"]
        for child in self.children:
            lines.append(child.render(indent + 1))
        return "\n".join(lines)


@dataclass(frozen=True)
class WhyNot:
    """One candidate clause that COULD have fired, and the subgoal that stopped it."""

    head: str
    failed_subgoal: str

    def __str__(self) -> str:
        return f"{self.head}\n    blocked at: {self.failed_subgoal}"


def _violation_goal(rule_id: str, file: str, line: int) -> str:
    """Build the goal both explain surfaces ask about.

    Detail is left FREE. Binding it would require the caller to reproduce the
    rule's own message verbatim before it could ask why the rule fired — which
    inverts the point of asking.
    """
    escaped_rule = rule_id.replace("'", "''")
    escaped_file = file.replace("'", "''")
    return f"violation('{escaped_rule}', '{escaped_file}', {int(line)}, _Detail)"


def _as_tree(node: Any) -> ProofTree:
    """Rebuild the janus-marshalled dict tree as typed nodes."""
    children = node.get("children") or ()
    return ProofTree(
        kind=str(node.get("kind", "")),
        head=str(node.get("head", "")),
        children=tuple(_as_tree(child) for child in children),
    )


@dataclass
class Verdict:
    """What the gate concluded, and on what evidence."""

    status: str
    reasons: list[str] = field(default_factory=list)
    stale: list[Staleness] = field(default_factory=list)
    violations: list[Violation] = field(default_factory=list)
    rules_loaded: int = 0
    facts: FactStats | None = None

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

    def report(self, *, max_listed: int = 20) -> str:
        """Render the verdict.

        Truncation is ANNOUNCED. A list silently cut at 20 reads as "there were
        20", which is how a 400-file staleness problem gets filed as a small
        one.
        """
        lines = [f"gate: {self.status.upper()}"]
        for reason in self.reasons:
            lines.append(f"  - {reason}")
        for label, items in (("stale", self.stale), ("violations", self.violations)):
            if not items:
                continue
            lines.append(f"  {label} ({len(items)}):")
            for item in items[:max_listed]:
                lines.append(f"    {item}")
            if len(items) > max_listed:
                lines.append(f"    ... {len(items) - max_listed} more not shown")
        if self.facts is not None:
            lines.append(
                f"  fact base: {self.facts.total:,} facts across "
                f"{self.facts.distinct_predicates} predicates"
            )
        return "\n".join(lines)


@dataclass
class GateEngine:
    """Run the gate: facts -> precondition -> rules -> verdict.

    Args:
        snapshot_path: the AST snapshot the facts derive from.
        repo_root: root the snapshot's relative paths resolve against.
        workdir: where the generated facts file is written. Defaults to a
            fresh temp directory, following ``CodebaseTools``' precedent.
        rulebases: extra ``.pl`` files to consult after the ontology and
            freshness layer. Defaults to whichever of ``gate_rules.pl`` /
            ``gate_severity.pl`` exist — so P3 landing them changes behaviour
            with no edit here, and their absence is REPORTED rather than
            silently producing an empty rule set.
    """

    snapshot_path: Path
    repo_root: Path = Path(".")
    workdir: Path | None = None
    rulebases: tuple[Path, ...] | None = None

    def __post_init__(self) -> None:
        self.snapshot_path = Path(self.snapshot_path)
        self.repo_root = Path(self.repo_root)
        workdir = (
            Path(tempfile.mkdtemp(prefix="orchestrata_gate_"))
            if self.workdir is None
            else Path(self.workdir)
        )
        workdir.mkdir(parents=True, exist_ok=True)
        self.workdir = workdir
        self._workdir = workdir
        if self.rulebases is None:
            self.rulebases = tuple(
                path for path in (GATE_RULES, GATE_SEVERITY) if path.is_file()
            )

    # -- public surface ----------------------------------------------------

    def run(self) -> Verdict:
        """Evaluate the gate and return its verdict."""
        facts_path = self._workdir / "facts.pl"
        stats = FactExtractor(
            snapshot_path=self.snapshot_path, repo_root=self.repo_root
        ).write(facts_path)

        with self._loaded(facts_path) as janus:
            return self._evaluate(janus, stats)

    @traces(spec="114", story="FR2", reqs=["REQ-114-FR2-004"])
    def _evaluate(self, janus: Any, stats: FactStats) -> Verdict:
        stale = self._staleness(janus)
        if stale:
            files = {reason.file for reason in stale}
            return Verdict(
                status=INVALID,
                reasons=[
                    f"fact base does not describe the working tree: "
                    f"{len(files)} file(s) across {len(stale)} finding(s). "
                    "Re-run: uv run ast-sync (then "
                    "scripts/build_runtime_callgraph.py reaugment)"
                ],
                stale=stale,
                facts=stats,
            )

        if self._scope_count(janus) == 0:
            return Verdict(
                status=INVALID,
                reasons=[
                    "no source_root/1 fact declares what this fact base covers. "
                    "in_project/1 joins on it, so EVERY rule would silently "
                    "match nothing and the run would report a clean tree it "
                    "never looked at (REQ-GATE-003) — the same false-PASS the "
                    "'.'-root hole produced on 2026-08-08."
                ],
                facts=stats,
            )

        missing_captures = self._missing_captures(janus)
        if missing_captures:
            named = ", ".join(
                f"{rule_id} requires {capture}" for rule_id, capture in missing_captures
            )
            return Verdict(
                status=INVALID,
                reasons=[
                    f"the snapshot never captured a rule's input: {named}. A "
                    "rule over a missing capture matches nothing and reports "
                    "PASS — the same false-PASS class as the '.'-root defect "
                    "(REQ-GATE-003). Fix: uv run ast-sync --snapshot-only"
                ],
                facts=stats,
            )

        rules_loaded = self._rule_count(janus)
        expected, contributors = self._expected_rule_count()
        if rules_loaded != expected:
            return Verdict(
                status=INVALID,
                reasons=[self._clobber_reason(rules_loaded, expected, contributors)],
                rules_loaded=rules_loaded,
                facts=stats,
            )

        if rules_loaded == 0:
            return Verdict(
                status=INVALID,
                reasons=[
                    "no rules are loaded — violation/4 has no clauses, so this "
                    "run evaluated nothing. Zero violations found by zero rules "
                    "is not a pass (REQ-GATE-003). Rules land with P3 "
                    "(gate_rules.pl)."
                ],
                facts=stats,
            )

        violations = self._violations(janus)
        return Verdict(
            status=FAIL if violations else PASS,
            reasons=[
                (
                    f"{len(violations)} violation(s) from {rules_loaded} rule clause(s)"
                    if violations
                    else f"{rules_loaded} rule clause(s), no violations"
                )
            ],
            violations=violations,
            rules_loaded=rules_loaded,
            facts=stats,
        )

    def explain(self, rule_id: str, file: str, line: int) -> ProofTree | None:
        """The derivation of ONE violation, or None if it does not hold.

        REQ-GATE-008. A gate that can only say "this is a violation" leaves the
        reader to re-derive WHY from the rule source; a gate that can print the
        clause that fired and every fact it consumed makes the finding
        auditable. It is also the only way to distinguish "the rule is right"
        from "the facts are wrong".
        """
        facts_path = self._workdir / "facts.pl"
        FactExtractor(snapshot_path=self.snapshot_path, repo_root=self.repo_root).write(
            facts_path
        )
        goal = _violation_goal(rule_id, file, line)
        with self._loaded(facts_path, explain=True) as janus:
            result = janus.query_once(f"explain_call({goal}, Tree)")
            if not result or not result.get("truth"):
                return None
            return _as_tree(result["Tree"])

    def why_not(self, rule_id: str, file: str, line: int) -> list[WhyNot]:
        """For a NON-violation, the first precondition of each candidate clause
        that did not hold.

        This is the question no checker in this repo could previously answer,
        and the tree-sitter relative-import checker died of exactly that: dead
        for its whole life, two separate ways, and nothing noticed because
        nothing could ask why it never fired.
        """
        facts_path = self._workdir / "facts.pl"
        FactExtractor(snapshot_path=self.snapshot_path, repo_root=self.repo_root).write(
            facts_path
        )
        goal = _violation_goal(rule_id, file, line)
        with self._loaded(facts_path, explain=True) as janus:
            result = janus.query_once(f"explain_whynot({goal}, Failures)")
            rows = result.get("Failures", []) if result else []
            return [
                WhyNot(
                    head=str(row.get("head", "")),
                    failed_subgoal=str(row.get("failed_subgoal", "")),
                )
                for row in rows
            ]

    # -- prolog ------------------------------------------------------------

    @contextmanager
    def _loaded(self, facts_path: Path, *, explain: bool = False) -> Iterator[Any]:
        """Consult the rulebases for the duration of one run, then unload them.

        Order matters on the way in: ``code_facts.pl``'s dynamic declarations
        must exist before the facts load, or SWI infers each predicate's
        properties from its first clause instead.

        UNLOADING MATTERS MORE. janus shares ONE process-global SWI database,
        so a second run in the same process inherits the first run's clauses.
        SWI replaces a predicate when a later file redefines it — but a
        predicate the second run emits NO clauses for is never redefined, and
        the first run's facts survive. Measured: two runs where run 2's
        snapshot had no ``h/1`` facts still answered ``h(9)`` from run 1. In a
        freshness layer that is fatal — stale disk hashes from a previous run
        are exactly the evidence this gate reasons over.

        ``unload_file/1`` retracts precisely what each file loaded, leaving the
        predicates DECLARED (so ``\\+`` still works) with zero clauses. The
        ``finally`` is load-bearing: a run that raises mid-query must not leave
        90k facts behind for the next one.
        """
        from orchestrata_ultimate.core.janus_bootstrap import janus

        files = [ONTOLOGY, facts_path, FRESHNESS, *(self.rulebases or ())]
        if explain:
            files.append(EXPLAIN)
        loaded: list[Path] = []
        try:
            for path in files:
                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}')")

    @staticmethod
    def _staleness(janus: Any) -> list[Staleness]:
        """Collect every staleness finding.

        The goal's variables are ``_``-prefixed deliberately: janus marshals
        every NAMED variable in a goal back to Python, and one still unbound
        when the goal succeeds raises "Arguments are not sufficiently
        instantiated" — a failure that reads like a fact-base defect and is
        not one. Measured while building P1.
        """
        result = janus.query_once(
            "findall([_Kind, _File], staleness(_Kind, _File), Rows)"
        )
        rows = result.get("Rows", []) if result else []
        return [Staleness(kind=str(row[0]), file=str(row[1])) for row in rows]

    @staticmethod
    def _scope_count(janus: Any) -> int:
        """How many source trees the fact base declares.

        Zero is INVALID, not empty: ``in_project/1`` is a join over
        ``source_root/1``, so a fact base without one turns every rule off at
        once and reports a pass. ``current_predicate/1`` guards the query for
        the same reason it does in :meth:`_rule_count` — an undefined predicate
        raises, and a crash here would be indistinguishable from the condition
        being detected.
        """
        result = janus.query_once(
            "( current_predicate(source_root/1)"
            "-> aggregate_all(count, source_root(_), N)"
            " ; N = 0 )"
        )
        return int(result["N"]) if result else 0

    @staticmethod
    def _missing_captures(janus: Any) -> list[tuple[str, str]]:
        """Every ``(rule, capture)`` pair whose capture the snapshot lacks.

        A rule whose input was never captured matches nothing and reports
        PASS — the same false-PASS class as the ``'.'``-root defect, and the
        reason ``rule_requires_capture/2`` (gate_rules.pl) exists: the
        snapshot's top-level ``captures`` list becomes ``snapshot_capture/1``
        facts, and a required capture with no fact is an INVALID verdict, not
        a clean one. An empty result means every rule's declared input was
        attested (or no loaded rulebase declares any).

        ``current_predicate/1`` guards BOTH predicates, as ``_rule_count`` and
        ``_violations`` guard theirs: an undefined predicate RAISES in SWI
        rather than failing, and a crash here would be indistinguishable from
        the condition being detected. A rulebase that declares no
        ``rule_requires_capture/2`` fact (the freshness module's own fixtures,
        a pre-Q2b gate_rules.pl) legitimately requires nothing.
        """
        result = janus.query_once(
            "( ( current_predicate(rule_requires_capture/2),"
            "    current_predicate(snapshot_capture/1) )"
            "-> findall([_R, _C],"
            "     ( rule_requires_capture(_R, _C),"
            "       \\+ snapshot_capture(_C) ), Rows)"
            " ; Rows = [] )"
        )
        rows = result.get("Rows", []) if result else []
        return [(str(row[0]), str(row[1])) for row in rows]

    @staticmethod
    @traces(spec="114", story="FR2", reqs=["REQ-114-FR2-004"])
    def _rule_count(janus: Any) -> int:
        """How many ``violation/4`` clauses exist.

        ``current_predicate/1`` guards the query because ``clause/2`` over an
        undefined predicate RAISES, and an engine that crashed here would be
        indistinguishable from one with no rules — the distinction this method
        exists to make.
        """
        result = janus.query_once(
            "( current_predicate(violation/4)"
            "-> aggregate_all(count, clause(violation(_, _, _, _), _), N)"
            " ; N = 0 )"
        )
        return int(result["N"]) if result else 0

    @staticmethod
    @traces(spec="114", story="FR2", reqs=["REQ-114-FR2-004"])
    def _declared_rule_count(path: Path) -> int:
        """How many ``violation/4`` clause heads FILE declares, by TEXT.

        Parsed the identical way ``FactExtractor._native_mechanism_facts`` and
        ``tests/gates/test_gate_rules.py::test_every_declared_rule_has_a_corpus``
        already parse ``gate_rules.pl`` — a line starting with ``violation(``
        at column 0 is one clause head. This is independent of what SWI
        actually has loaded, which is the whole point (REQ-114-FR2-004 clause (a)): it
        is the GROUND TRUTH the runtime count below is checked against, not a
        second read of the same live state.

        A missing file counts as zero rather than raising — the caller
        already treats "no rulebases" (``rulebases=()``) as a legitimate
        configuration, handled by the ``rules_loaded == 0`` branch that runs
        after this invariant.
        """
        if not path.is_file():
            return 0
        text = path.read_text(encoding="utf-8")
        return sum(1 for line in text.splitlines() if line.startswith("violation("))

    @traces(spec="114", story="FR2", reqs=["REQ-114-FR2-004"])
    def _expected_rule_count(self) -> tuple[int, list[Path]]:
        """The declared ``violation/4`` clause count, summed over every
        rulebase THIS run actually consults, and which of those files
        contribute at least one clause.

        DESIGN DECISION (REQ-114-FR2-004 clause (a)): the expected count is DERIVED by
        parsing ``self.rulebases`` at load time rather than a hardcoded
        integer in this module. A hardcoded ``5`` would mean every legitimate
        new rule in ``gate_rules.pl`` requires an engine.py edit too — friction
        that invites someone to loosen or delete the check the next time it is
        merely inconvenient. Deriving it from the consulted files' own text is
        self-maintaining: a new rule clause changes the expected count on the
        same commit that adds it, no second place to update.

        The accepted tradeoff: this cannot detect ``gate_rules.pl`` ITSELF
        being the file that is replaced (a tampered ``gate_rules.pl`` with
        fewer clauses would report its own smaller count as "expected" and
        agree with itself). That is a different, already-covered threat —
        ``gate_rules.pl`` is a reviewed, version-controlled file; freshness
        content-hash coverage for consulted ``.pl`` files is the known gap
        this spec (REQ-114-FR3-005) tracks separately. What this invariant
        closes is the LIVE hole: a second, module-less rulebase silently
        replacing the enforcing rulebase's clauses in the shared SWI process,
        which needs no `gate_rules.pl` edit at all.
        """
        contributors = [
            path
            for path in (self.rulebases or ())
            if self._declared_rule_count(path) > 0
        ]
        total = sum(self._declared_rule_count(path) for path in contributors)
        return total, contributors

    @staticmethod
    @traces(spec="114", story="FR2", reqs=["REQ-114-FR2-004"])
    def _clobber_reason(loaded: int, expected: int, contributors: list[Path]) -> str:
        """Name WHICH file is implicated, not just that the counts disagree.

        Consult order is clobber order: SWI keeps only the clauses of the
        MOST RECENTLY consulted file that defines a given predicate (the
        ``Warning: Redefined static procedure violation/4`` case), so the
        last contributor in consult order is the one whose clauses SURVIVED,
        and any earlier contributors are the ones a mismatch means got wiped.
        With only one contributor (or zero), no redefinition among the
        consulted rulebases can explain the mismatch, and the reason says so
        rather than naming an innocent file.
        """
        declared = ", ".join(str(path) for path in contributors) or "(none)"
        base = (
            f"violation/4 clause-count invariant violated (REQ-114-FR2-004): "
            f"{len(contributors)} consulted rulebase(s) declare {expected} "
            f"violation/4 clause(s) in total ({declared}), but {loaded} "
            f"clause(s) are actually loaded right now. Zero violations found "
            f"by a rule count that does not match what was consulted is not a "
            f"pass."
        )
        if len(contributors) > 1:
            survivor = contributors[-1]
            clobbered = ", ".join(str(path) for path in contributors[:-1])
            return (
                f"{base} SWI replaces a module-less predicate's clauses when a "
                f"later file redefines it rather than adding to them: the last "
                f"rulebase consulted that declares violation/4 is {survivor} — "
                f"it most likely clobbered the earlier declaration(s) in "
                f"{clobbered}."
            )
        return (
            f"{base} No single consulted rulebase's redefinition explains "
            f"this on its own (only {len(contributors)} of them declare "
            f"violation/4) — check for a stale process-global SWI state from "
            f"a previous run, or a rulebase whose clause heads this text scan "
            f"does not recognise."
        )

    @staticmethod
    def _violations(janus: Any) -> list[Violation]:
        """Every solution of the single entry predicate (REQ-GATE-005), joined
        to its effective severity.

        The guards are load-bearing, and for two different reasons.

        ``-> ;`` covers a file with no ``file_class`` fact or a rule with no
        severity clause: without it the conjunction fails and the violation
        VANISHES, and a rule silently producing nothing is exactly how the
        tree-sitter checker died.

        ``current_predicate/1`` covers ``gate_severity.pl`` not being loaded at
        all. An undefined predicate RAISES in SWI rather than failing, so the
        whole findall would throw and the engine would report zero violations
        for a fact base full of them — a false green produced by the very code
        meant to prevent one. Measured: it took out four tests that supply
        their own rulebase without a severity layer.
        """
        result = janus.query_once(
            "findall([_R, _F, _L, _D, _C, _S, _B],"
            " ( violation(_R, _F, _L, _D),"
            "   ( file_class(_F, _C) -> true ; _C = unknown ),"
            "   ( ( current_predicate(severity/3), severity(_R, _C, _S) )"
            "     -> true ; _S = error ),"
            "   ( ( current_predicate(blocking/1), blocking(_S) )"
            "     -> _B = true ; _B = false ) ),"
            " Rows)"
        )
        rows = result.get("Rows", []) if result else []
        return [
            Violation(
                rule_id=str(row[0]),
                file=str(row[1]),
                line=int(row[2]),
                detail=str(row[3]),
                file_class=str(row[4]),
                severity=str(row[5]),
                blocking=str(row[6]) == "true",
            )
            for row in rows
        ]