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
Dependency diagram
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
"""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
]