orchestrata_ultimate.gates.fact_extractor

Reference › orchestrata_ultimate › gates › fact_extractor

The fact base builder of the Prolog gate layer (contract REQ-GATE-001/002/004), and the largest module in the package. FactExtractor turns the AST snapshot plans/ast-ultimate.json into a ground Prolog fact base laid out by the ontology in code_facts.pl. It is used by engine, commit_review and scripts.gate, and re-exported by the package __init__.

The governing rule is that structural facts (what a module defines, imports, calls or reads from the environment) come from the snapshot and nowhere else; the extractor never opens a .py file to learn them. The one sanctioned disk access is the freshness layer, _freshness_facts and _disk_side, which hashes bytes to emit content_hash(disk, ...) so that a stale snapshot can be detected; it never parses. extract returns the fact lines with FactStats (per-predicate counts, with a table rendering). Fact groups include project and scope, modules, classes and callables, policy and # gate-ok: waivers, and tooling facts that record the ruff rule catalogue (obtained by running ruff rule --all, with each failure mode named) against the project's select and ignore lists, with rule_status and classify_file as pure helpers. banned_registry, env_index and import_index give the indexes the differential needs. prolog_atom and prolog_list quote values safely. A missing snapshot raises SnapshotMissingError. Policy it feeds is described on the prolog-gate and ruff pages.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

SnapshotMissingError

Bases: RuntimeError · Line 167

The snapshot the extractor was pointed at does not exist.

FactStats (dataclass)

Decorators: dataclass · Line 177

Per-predicate fact counts, for gate facts --stats.

FactExtractor (dataclass)

Decorators: dataclass · Line 358

Turn plans/ast-ultimate.json into a ground Prolog fact base.

Functions

Source

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

"""Emit the Prolog gate fact base FROM the AST snapshot.

Contract: ``plans/contracts/prolog-gates.contract.json`` — REQ-GATE-001/002/004.
Ontology: ``src/orchestrata_ultimate/prolog/code_facts.pl``.

REQ-GATE-001 IS THE POINT OF THIS MODULE. Structural facts come from
``plans/ast-ultimate.json`` and from nowhere else — this module never opens a
``.py`` file to learn what a module defines, imports, calls or reads from the
environment. Every existing checker in this repo re-reads and re-parses files
the snapshot already holds, which is both wasted work and a second source of
truth that can disagree with the first.

The ONE sanctioned disk access is the freshness layer. REQ-GATE-002 requires
the fact base to carry ``content_hash(disk, ...)``, and staleness cannot be
detected without hashing both sides of the comparison.
That access is confined to :meth:`FactExtractor._freshness_facts`, it reads
bytes only to hash them, and it never parses. Any structural fact traceable to
a disk read would be a REQ-GATE-001 violation; ``tests/gates/
test_fact_extractor.py`` proves the separation by extracting against a snapshot
whose on-disk files deliberately differ.
"""

from __future__ import annotations

import hashlib
import json
import subprocess
import time
import tomllib
from collections import Counter
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from orchestrata_ast.analyzer import CodebaseAnalyzer
from orchestrata_ultimate.core.banned_libraries import (
    BANNED_LIBRARIES,
    load_banned_libraries,
    normalise_package,
)
from orchestrata_ultimate.core.gate_waivers import waiver_reasons
from orchestrata_ultimate.core.infra_env_rule import INFRA_ENDPOINT_KEYS

# Where the snapshot lives, relative to the repo root. The extractor takes an
# explicit path; this is only the CLI's default.
DEFAULT_SNAPSHOT = Path("plans/ast-ultimate.json")

# BANNED_LIBRARIES and the parser both live in core.banned_libraries now: the
# registry being shared DATA is only half of what matters, and a second parser
# is a second chance to disagree about what an entry means. Re-exported here
# because this module's public surface already named it.

# The project's own config, which decides which ruff rules are ENFORCED. Its
# select/ignore lists are the only place that answers "is this policy
# mechanised", which is REQ-GATE-011 / RUL-014 as a query.
PYPROJECT = Path("pyproject.toml")

# The NATIVE rulebase. Read (never re-derived by hand) to discover which
# policies gate_rules.pl itself mechanises — REQ-GATE-011's "no enabled tool
# rule OR native rule". Parsed the same way tests/gates/test_gate_rules.py::
# TestCorpusDiscipline already parses it, so a rule renamed or added there
# changes this fact base on the next extraction with no second place to
# update.
GATE_RULES_PATH = Path("src/orchestrata_ultimate/prolog/gate_rules.pl")

# Ruff codes a configured BLACK demonstrably, UNCONDITIONALLY rewrites on
# EVERY run, regardless of content — the third mechanism kind covered/1
# (gate_rules.pl) needs, and REQ-GATE-011's own reasoning for why a per-line
# lexical layer was rejected: "black already rewrites three of the four".
# Verified reproducibly, not asserted from a config comment's claim:
#   1. `uv run black` on a file with trailing whitespace / tab indentation /
#      no final newline rewrites all three, unconditionally.
#   2. `uv run black` on a file with a long STRING LITERAL or a long
#      single-token comment (a URL) leaves the line untouched — verified
#      2026-08-19 with a 209-char string assignment: black left it at 209
#      chars, and ruff's own E501 (run separately, `--select E501`) flagged
#      exactly that line. Black only reflows constructs it has freedom to
#      reflow (call arguments, etc.); it does not enforce a hard width.
# E501/line-too-long is DELIBERATELY absent from this set for that reason —
# asserting black as its mechanism would be a false "covered", the exact
# false-green class RUL-014/REQ-GATE-003 exist to catch, not produce.
BLACK_UNCONDITIONAL_RUFF_CODES: frozenset[str] = frozenset(
    {
        "W191",  # tab-indentation
        "W291",  # trailing-whitespace
        "W292",  # missing-newline-at-end-of-file
        "W293",  # blank-line-with-whitespace
    }
)

# Ruff's own catalogue, asked of ruff. REQ-GATE-010 forbids re-deriving what a
# linter computes; the catalogue is the linter TELLING US what it computes.
RUFF_CATALOGUE_COMMAND: tuple[str, ...] = (
    "ruff",
    "rule",
    "--all",
    "--output-format",
    "json",
)
RUFF_CATALOGUE_TIMEOUT_S = 30

# tool_outcome/3 statuses.
OUTCOME_OK = "ok"
OUTCOME_MISSING = "missing"
OUTCOME_FAILED = "failed"
OUTCOME_TIMEOUT = "timeout"

# Kinds emitted by defines/5.
KIND_CLASS = "class"
KIND_FUNC = "func"
KIND_METHOD = "method"

# file_class/2 values.
CLASS_SRC = "src"
CLASS_TEST = "test"
CLASS_SCRIPT = "script"
CLASS_GENERATED = "generated"

# The emitted ontology, by layer, in file order. This is also WHY the facts are
# written grouped by predicate rather than in discovery order: SWI warns
# "clauses are not together in the source-file" for every interleaved run, and
# the freshness layer used to emit two predicates alternately per file —
# measured at ~3,400 warnings on the live tree, enough noise to bury a real
# consult error. Grouping removes the warning class at its source instead of
# suppressing it with a discontiguous declaration.
SECTIONS: tuple[tuple[str, tuple[str, ...]], ...] = (
    (
        "project — the scope every other fact lives inside",
        ("project/2", "source_root/1"),
    ),
    (
        "capture provenance — what the analyzer that wrote the snapshot performed",
        ("snapshot_capture/1",),
    ),
    (
        "structural — from the AST snapshot ONLY (REQ-GATE-001)",
        (
            "module/2",
            "defines/5",
            "imports/5",
            "calls/3",
            "decorates/3",
            "param/4",
            "returns/2",
            "has_docstring/1",
            "subclass_of/2",
            "literal/4",
            "env_access/4",
            "argv_literal/5",
        ),
    ),
    (
        "freshness — hashes, not clocks (REQ-GATE-002)",
        ("snapshot_built_at/1", "content_hash/3"),
    ),
    (
        "policy data — single-sourced from the registries that own it",
        ("infra_key/1", "banned_import/2", "file_class/2", "waiver/3"),
    ),
    (
        "tooling — the linters as fact sources, never re-derived (REQ-GATE-010)",
        ("tool_rule/4", "tool_rule_status/3", "tool_outcome/3", "policy_mechanism/3"),
    ),
)


class SnapshotMissingError(RuntimeError):
    """The snapshot the extractor was pointed at does not exist.

    Raised rather than returning an empty fact base: a gate over zero facts
    finds zero violations and reports success, which is the exact false-green
    class this design exists to close (REQ-GATE-003).
    """


@dataclass
class FactStats:
    """Per-predicate fact counts, for ``gate facts --stats``.

    ``predicates`` counts only predicates the extractor actually emitted.
    ``code_facts.pl`` additionally DECLARES the tooling and Gate-0 predicates
    that later phases will populate; counting declarations here would report a
    fact base larger than the one that exists.
    """

    predicates: Counter[str] = field(default_factory=Counter)
    modules: int = 0
    files_hashed: int = 0
    out_of_scope: int = 0

    @property
    def total(self) -> int:
        return sum(self.predicates.values())

    @property
    def distinct_predicates(self) -> int:
        return len(self.predicates)

    def table(self) -> str:
        """Render the acceptance table P1 asks to be pasted."""
        width = max((len(name) for name in self.predicates), default=9)
        lines = [f"{'predicate':<{width}}  {'facts':>9}"]
        lines.append("-" * (width + 11))
        for name, count in sorted(
            self.predicates.items(), key=lambda kv: (-kv[1], kv[0])
        ):
            lines.append(f"{name:<{width}}  {count:>9,}")
        lines.append("-" * (width + 11))
        lines.append(f"{'TOTAL':<{width}}  {self.total:>9,}")
        lines.append("")
        lines.append(
            f"distinct fact predicates: {self.distinct_predicates}  "
            f"(modules {self.modules:,}, files hashed {self.files_hashed:,})"
        )
        if self.out_of_scope:
            # Announced, never silent: a snapshot describing files outside the
            # project it declares is a configuration defect, and dropping them
            # quietly would read as a smaller codebase rather than a broken
            # scope.
            lines.append(
                f"dropped {self.out_of_scope:,} file(s) outside the declared "
                "project root"
            )
        return "\n".join(lines)


def prolog_atom(value: object) -> str:
    """Quote a Python value as a single-quoted SWI-Prolog atom.

    Backslash and quote are both escape-significant inside SWI single quotes,
    and an un-escaped newline inside an atom is a syntax error that would fail
    the whole consult — one bad default expression in one module would take
    the entire fact base down. Control characters are escaped rather than
    stripped so the atom still round-trips to the source text.
    """
    text = "" if value is None else str(value)
    text = (
        text.replace("\\", "\\\\")
        .replace("'", "\\'")
        .replace("\n", "\\n")
        .replace("\r", "\\r")
        .replace("\t", "\\t")
    )
    return f"'{text}'"


def prolog_list(values: list[str]) -> str:
    """Render a list of strings as a Prolog list of atoms."""
    return "[" + ", ".join(prolog_atom(v) for v in values) + "]"


def _root_admits(root: str, file_path: str) -> bool:
    """Does *root* cover *file_path*?

    ``.`` and ``""`` both mean "the whole repo" and admit everything. That is
    not a nicety: ``Path(x).relative_to(x)`` is ``.``, so a snapshot serialized
    with ``--source .`` normalises its root to ``.``, and a naive prefix test
    against ``.`` matches nothing — the extractor would emit ZERO facts and the
    gate would read a clean tree instead of a broken configuration. Measured:
    it silently emptied seven fixtures.
    """
    if root in ("", "."):
        return True
    return file_path == root or file_path.startswith(f"{root}/")


def classify_file(file_path: str) -> str:
    """Classify a repo-relative path for defeasible severity.

    THE ROOT DIRECTORY DECIDES, and a ``test_`` basename is only a fallback for
    paths outside a known root. Measured on this tree, a basename-first rule
    misclassified five PRODUCTION modules as tests — ``core/threading/
    test_executor.py`` (the parallel test runner), ``test_scheduler.py`` (the
    CLP(FD) module scheduler), ``gateway/handlers/test_metrics.py``,
    ``gateway/test_runner.py`` and ``knowledge_graph/test_results_store.py``.
    Each is shipped code that happens to be ABOUT testing. Since
    ``gate_severity.pl`` demotes findings in test files to advisory, that
    misclassification would have silently exempted five src modules from every
    rule the gate enforces — a hole in exactly the layer meant to close holes.

    Generated code is checked before the rest because the generated test trees
    (``tests/generated_*``) are both, and a rule violated by a generator's
    output is the generator's defect, not the file's.
    """
    parts = Path(file_path).parts
    if any(p.startswith("generated") for p in parts):
        return CLASS_GENERATED
    root = parts[0] if parts else ""
    if root == "src":
        return CLASS_SRC
    if root == "tests":
        return CLASS_TEST
    if root == "scripts":
        return CLASS_SCRIPT
    if "tests" in parts or Path(file_path).name.startswith("test_"):
        return CLASS_TEST
    if "scripts" in parts:
        return CLASS_SCRIPT
    return CLASS_SRC


def _fixability(rule: dict[str, Any]) -> str:
    """Normalise ruff's fix availability to an atom.

    Ruff reports it as prose ("Fix is always available.") in ``fix`` and as an
    enum in ``fix_availability``; the enum is preferred and the prose is the
    fallback, so a ruff version that drops either field still yields a usable
    atom rather than ''.
    """
    availability = rule.get("fix_availability")
    if isinstance(availability, str) and availability:
        return availability.strip().lower()
    prose = str(rule.get("fix", ""))
    if "always" in prose.lower():
        return "always"
    if "sometimes" in prose.lower():
        return "sometimes"
    return "none"


def rule_status(code: str, select: list[str], ignore: list[str]) -> str:
    """Is this ruff rule ENFORCED by this project's config?

    The three values are deliberately distinct. ``ignored`` means the project
    turned the rule off on purpose (E501 is the live example, which is exactly
    why long-line policy has no mechanism). ``unknown`` means the rule was
    never selected, so this config expresses no opinion. Collapsing the two
    would make a deliberate exemption indistinguishable from an oversight,
    and telling those apart is the entire point of REQ-GATE-011.
    """
    if _matches(code, ignore):
        return "ignored"
    if _matches(code, select):
        return "enabled"
    return "unknown"


def _matches(code: str, prefixes: list[str]) -> bool:
    """Ruff selectors are code PREFIXES, with ``ALL`` meaning every rule."""
    return any(prefix == "ALL" or code.startswith(prefix) for prefix in prefixes)


def env_form(form: str, has_default: bool) -> str:
    """Fold the snapshot's (form, has_default) pair into one Form atom.

    ``code_facts.pl`` documents why the two dimensions are one atom: the
    contract fixes ``env_access/4``, and both rules over it need the default's
    presence — ``bare_os_environ`` forbids the subscript form outright while
    ``infra_key_with_default`` inverts the polarity of the default for infra
    keys, so a rule that could not see the default could not be written.
    """
    if form in ("get", "getenv") and has_default:
        return f"{form}_default"
    return form


@dataclass
class FactExtractor:
    """Turn ``plans/ast-ultimate.json`` into a ground Prolog fact base.

    Args:
        snapshot_path: the AST snapshot to read.
        repo_root: root the snapshot's relative file paths resolve against;
            used ONLY by the freshness layer.
        infra_keys: the required infra-endpoint keys. Defaults to
            ``core.infra_env_rule.INFRA_ENDPOINT_KEYS`` — the injection seam
            exists so tests can supply a known set without touching the
            environment (monkeypatch is forbidden project-wide).
        banned_libraries_path: the banned-dependency registry.
        pyproject_path: the config whose ruff select/ignore lists decide
            which rules are enforced.
        gate_rules_path: the NATIVE rulebase (``gate_rules.pl``), read to
            discover which policies it mechanises (REQ-GATE-011). The
            injection seam exists so a test can point it at a fixture
            rulebase without touching the real one.
        ruff_catalogue_command: how to ask ruff for its rule catalogue.
            An injection seam so a test can drive the missing-tool,
            failed-tool and bad-output branches with REAL commands rather
            than a double (RUL-002).
        ruff_catalogue_timeout_s: how long to wait for that catalogue.
    """

    snapshot_path: Path = DEFAULT_SNAPSHOT
    repo_root: Path = Path(".")
    infra_keys: frozenset[str] = INFRA_ENDPOINT_KEYS
    banned_libraries_path: Path = BANNED_LIBRARIES
    pyproject_path: Path = PYPROJECT
    gate_rules_path: Path = GATE_RULES_PATH
    ruff_catalogue_command: tuple[str, ...] = RUFF_CATALOGUE_COMMAND
    ruff_catalogue_timeout_s: float = RUFF_CATALOGUE_TIMEOUT_S

    def __post_init__(self) -> None:
        self.snapshot_path = Path(self.snapshot_path)
        self.repo_root = Path(self.repo_root)
        self.banned_libraries_path = Path(self.banned_libraries_path)
        self.pyproject_path = Path(self.pyproject_path)
        self.gate_rules_path = Path(self.gate_rules_path)
        self._stats = FactStats()
        self._buckets: dict[str, list[str]] = {}
        self._scopes: tuple[str, ...] = ()

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

    def extract(self) -> tuple[list[str], FactStats]:
        """Return (fact lines, stats). Never touches a ``.py`` file's syntax."""
        snapshot = self._load_snapshot()
        modules = snapshot.get("modules", [])
        file_meta = snapshot.get("file_meta", {})

        self._stats = FactStats(modules=len(modules))
        self._buckets = {}
        self._scopes = self._project_facts(snapshot)
        self._capture_facts(snapshot)

        self._structural_facts(modules)
        self._freshness_facts(file_meta, self._scopes)
        self._policy_facts(modules)
        self._tooling_facts()
        return self._render(), self._stats

    def _render(self) -> list[str]:
        """Lay the buckets out grouped by predicate, in ontology order.

        A predicate the extractor produced but SECTIONS does not list would be
        silently dropped, so the leftovers are appended under an explicit
        heading rather than lost — an unlisted predicate is a bug to see, not
        one to hide.
        """
        lines: list[str] = [
            "% GENERATED by orchestrata_ultimate.gates.fact_extractor —"
            " do not hand-edit.",
            f"% Source snapshot: {self.snapshot_path}",
            "% Ontology + provenance:" " src/orchestrata_ultimate/prolog/code_facts.pl",
        ]
        emitted: set[str] = set()
        for heading, predicates in SECTIONS:
            lines.extend(["", f"% ===== {heading} ====="])
            for predicate in predicates:
                facts = self._buckets.get(predicate)
                if not facts:
                    continue
                lines.extend(["", f"% {predicate} — {len(facts):,}"])
                lines.extend(facts)
                emitted.add(predicate)
        leftover = sorted(set(self._buckets) - emitted)
        if leftover:
            lines.extend(["", "% ===== NOT DECLARED IN SECTIONS ====="])
            for predicate in leftover:
                lines.extend(["", f"% {predicate}"])
                lines.extend(self._buckets[predicate])
        return lines

    def banned_registry(self) -> set[str]:
        """The banned package names, normalised for comparison.

        The differential needs to ask "is this package in OUR registry?" about
        a package only the legacy checker flagged. Returning the same set the
        facts were built from is what makes the answer trustworthy.
        """
        return {normalise_package(name) for name, _ in self._banned_imports()}

    def env_index(self) -> dict[tuple[str, int], str]:
        """The env key read at each ``(file, line)``, for scoped files only.

        The differential needs it to say WHY only one side flagged a defaulted
        read: the legacy ``ENV_VAR_KEYS`` list covers 6 of the 10 required
        infra keys, and naming the key is the difference between "our rule is
        broader" and an unexplained disagreement.
        """
        snapshot = self._load_snapshot()
        self._scopes = self._project_facts(snapshot)
        self._buckets = {}

        keys: dict[tuple[str, int], str] = {}
        for mod in snapshot.get("modules", []):
            raw = str(mod.get("file_path", ""))
            if not raw:
                continue
            file_path = self._rel(raw)
            if not self._in_scope(file_path):
                continue
            for access in mod.get("env_accesses", []):
                keys[(file_path, int(access.get("line", 0) or 0))] = str(
                    access.get("key", "")
                )
        return keys

    def import_index(self) -> tuple[set[str], dict[tuple[str, int], str]]:
        """The scoped file set, and the module imported at each ``(file, line)``.

        The P5 differential has to answer "why did only one side flag this
        import?", and the honest answer needs the MODULE — which side's ban
        registry lists it. Parsing the module back out of either checker's
        message would make the comparison depend on wording; reading it from
        the snapshot reads it from the same source both sides consumed.

        It lives here because scope and path normalisation live here. A caller
        re-deriving either would re-open the absolute-vs-relative hole that
        :meth:`_rel` exists to close, and would decide project membership by a
        second rule that could disagree with the one the facts were built with.
        """
        snapshot = self._load_snapshot()
        self._scopes = self._project_facts(snapshot)
        self._buckets = {}

        scoped: set[str] = set()
        modules: dict[tuple[str, int], str] = {}
        for mod in snapshot.get("modules", []):
            raw = str(mod.get("file_path", ""))
            if not raw:
                continue
            file_path = self._rel(raw)
            if not self._in_scope(file_path):
                continue
            scoped.add(file_path)
            for imp in mod.get("imports", []):
                line = int(imp.get("line", 0) or 0)
                # The FULL dotted path, not the top-level package. Folding it
                # here would make `from unittest.mock import patch` look like
                # an import of `unittest`, and the registry bans the submodule
                # while permitting the package — the distinction `ban_target/3`
                # exists for. The fold belongs at the comparison, which knows
                # which granularity the registry entry declared.
                modules[(file_path, line)] = str(imp.get("module", ""))
        return scoped, modules

    def write(self, out_path: Path) -> FactStats:
        """Extract and persist the fact base."""
        lines, stats = self.extract()
        out_path = Path(out_path)
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
        return stats

    # -- snapshot ----------------------------------------------------------

    def _load_snapshot(self) -> dict[str, Any]:
        if not self.snapshot_path.exists():
            raise SnapshotMissingError(
                f"AST snapshot not found at {self.snapshot_path}. The gate fact "
                "base is derived from it and from nothing else (REQ-GATE-001); "
                "an empty fact base would report zero violations and pass. "
                "Regenerate with: uv run ast-sync — the ONE producer, which "
                "covers every tree in ast_pipeline.SOURCE_TREES (src, tests, "
                "scripts). A hand-written serialize command naming one --source "
                "rebuilds a NARROWER snapshot that looks complete."
            )
        loaded = json.loads(self.snapshot_path.read_text(encoding="utf-8"))
        if not isinstance(loaded, dict):
            raise SnapshotMissingError(
                f"AST snapshot at {self.snapshot_path} is not a JSON object "
                f"(got {type(loaded).__name__})."
            )
        return loaded

    def _project_facts(self, snapshot: dict[str, Any]) -> tuple[str, ...]:
        """Declare the project, and return every root a fact may live under.

        Mohamed, 2026-08-08: "make sure to scope facts by project." A bare
        relative path is not an identity — the operational KG already ruled the
        same way for code-structure nodes, because it must hold 100+ projects
        and ``src/cli.py`` names a different file in each.

        The scope is enforced HERE, at the source, rather than by qualifying
        every atom: :meth:`_in_scope` refuses any fact outside the root, so
        every File in the base belongs to the declared project by construction.
        Paths stay repo-relative, which keeps ``file:line`` clickable.
        """
        name = str(snapshot.get("name", "")) or self.snapshot_path.stem
        root = self._rel(str(snapshot.get("root_path", "")))
        self._emit(
            "project/2",
            f"project({prolog_atom(name)}, {prolog_atom(root)}).",
        )
        roots = [root]
        for extra in snapshot.get("extra_roots", []):
            rel = self._rel(str(extra))
            if rel and rel not in roots:
                roots.append(rel)
        for source_root in roots:
            self._emit(
                "source_root/1",
                f"source_root({prolog_atom(source_root)}).",
            )
        return tuple(roots)

    def _capture_facts(self, snapshot: dict[str, Any]) -> None:
        """Emit one ``snapshot_capture/1`` per entry of the snapshot's
        top-level ``captures`` list.

        Provenance, not structure: the list says which per-module captures the
        analyzer that WROTE the snapshot actually performed
        (``orchestrata_ast.analyzer.SNAPSHOT_CAPTURES``). A snapshot written
        before a capture existed holds module dicts without the key, and the
        missing records deserialize indistinguishably from "captured, found
        none" — so a rule over that input matches nothing and reports PASS.
        ``rule_requires_capture/2`` (gate_rules.pl) joined against these facts
        is what turns that false PASS into an INVALID verdict
        (:meth:`orchestrata_ultimate.gates.engine.GateEngine._missing_captures`).
        An absent ``captures`` key emits NOTHING: attesting a capture the
        snapshot never claimed would defeat the very distinction.
        """
        for capture in snapshot.get("captures", []):
            self._emit(
                "snapshot_capture/1",
                f"snapshot_capture({prolog_atom(capture)}).",
            )

    def _in_scope(self, file_path: str) -> bool:
        """Is this path inside ANY tree the snapshot declares?

        A snapshot covers one primary root plus whatever ``extra_roots`` it was
        built with, and a fact is in scope if any one of them admits it.
        Keeping this a disjunction rather than a single prefix is what let the
        snapshot widen to ``tests`` and ``scripts`` without the extractor
        dropping every fact from the new trees as out-of-project.
        """
        return any(_root_admits(scope, file_path) for scope in self._scopes)

    def _rel(self, file_path: str) -> str:
        """Normalise a snapshot path to ONE repo-relative form.

        The snapshot's ``file_path`` values are absolute or relative depending
        on how the serializer was invoked — ``gates.py ast-refresh`` produces
        absolute paths, a direct ``serialize --source src/orchestrata_ultimate``
        produces relative ones. Measured: with absolute paths ``classify_file``
        sees ``parts[0] == '/'``, misses every root check, falls through to the
        basename heuristic, and re-misclassifies the five production ``test_*``
        modules it was fixed to handle. Two path forms also mean a rule reports
        ``file:line`` differently between snapshot generations, and freshness
        compares keys that never match.

        So the form is decided here, once, at the boundary. A path outside the
        repo is passed through unchanged rather than forced — inventing a
        relative form for it would be a lie about where the code lives.
        """
        path = Path(file_path)
        if not path.is_absolute():
            return path.as_posix()
        try:
            return path.relative_to(self.repo_root.resolve()).as_posix()
        except ValueError:
            return path.as_posix()

    def _emit(self, predicate: str, text: str) -> None:
        self._buckets.setdefault(predicate, []).append(text)
        self._stats.predicates[predicate] += 1

    # -- structural (REQ-GATE-001: snapshot only) --------------------------

    def _structural_facts(self, modules: list[dict[str, Any]]) -> None:
        for mod in modules:
            raw = str(mod.get("file_path", ""))
            if not raw:
                continue
            file_path = self._rel(raw)
            if not self._in_scope(file_path):
                self._stats.out_of_scope += 1
                continue
            import_path = str(mod.get("path", ""))
            self._emit(
                "module/2",
                f"module({prolog_atom(import_path)}, {prolog_atom(file_path)}).",
            )
            self._module_level_facts(mod, file_path, import_path)

            for cls in mod.get("classes", []):
                self._class_facts(cls, file_path)
            for func in mod.get("functions", []):
                self._callable_facts(func, file_path, KIND_FUNC)

    def _module_level_facts(
        self,
        mod: dict[str, Any],
        file_path: str,
        import_path: str,
    ) -> None:
        for imp in mod.get("imports", []):
            self._emit(
                "imports/5",
                "imports("
                f"{prolog_atom(file_path)}, "
                f"{int(imp.get('line', 0))}, "
                f"{prolog_atom(imp.get('module', ''))}, "
                f"{int(imp.get('level', 0))}, "
                f"{prolog_list([str(n) for n in imp.get('names', [])])}).",
            )

        for var in mod.get("module_vars", []):
            default = var.get("default")
            if default is None:
                continue
            self._emit(
                "literal/4",
                "literal("
                f"{prolog_atom(file_path)}, "
                f"{prolog_atom(import_path)}, "
                f"{prolog_atom(var.get('name', ''))}, "
                f"{prolog_atom(default)}).",
            )

        for access in mod.get("env_accesses", []):
            self._emit(
                "env_access/4",
                "env_access("
                f"{prolog_atom(file_path)}, "
                f"{int(access.get('line', 0))}, "
                f"{prolog_atom(access.get('key', ''))}, "
                f"{prolog_atom(env_form(str(access.get('form', 'other')), bool(access.get('has_default', False))))})."
                "",
            )

        # The sibling of env_access/4 (ADR-0089, Q2a capture): the literal
        # program token at a spawn-shaped call, verbatim. Classification is
        # the rule's join (toolchain_program/2, gate_rules.pl), never a
        # filter here — an over-captured fact abstains honestly, while a
        # filtered one silently decides policy at the wrong layer.
        for record in mod.get("argv_literals", []):
            self._emit(
                "argv_literal/5",
                "argv_literal("
                f"{prolog_atom(file_path)}, "
                f"{int(record.get('line', 0))}, "
                f"{prolog_atom(record.get('site', '?'))}, "
                f"{prolog_atom(record.get('form', ''))}, "
                f"{prolog_atom(record.get('program', ''))}).",
            )

        if mod.get("docstring"):
            self._emit("has_docstring/1", f"has_docstring({prolog_atom(import_path)}).")

    def _class_facts(self, cls: dict[str, Any], file_path: str) -> None:
        qname = str(cls.get("qualified_name", ""))
        self._emit(
            "defines/5",
            "defines("
            f"{prolog_atom(file_path)}, {prolog_atom(KIND_CLASS)}, "
            f"{prolog_atom(qname)}, {int(cls.get('line', 0))}, "
            f"{int(cls.get('end_line', 0))}).",
        )
        for base in cls.get("bases", []):
            self._emit(
                "subclass_of/2",
                f"subclass_of({prolog_atom(qname)}, {prolog_atom(base)}).",
            )
        for dec in cls.get("decorators", []):
            self._emit(
                "decorates/3",
                f"decorates({prolog_atom(dec)}, {prolog_atom(qname)}, "
                f"{prolog_atom(file_path)}).",
            )
        for var in cls.get("class_vars", []):
            default = var.get("default")
            if default is None:
                continue
            self._emit(
                "literal/4",
                "literal("
                f"{prolog_atom(file_path)}, {prolog_atom(qname)}, "
                f"{prolog_atom(var.get('name', ''))}, {prolog_atom(default)}).",
            )
        if cls.get("docstring"):
            self._emit("has_docstring/1", f"has_docstring({prolog_atom(qname)}).")
        for method in cls.get("methods", []):
            self._callable_facts(method, file_path, KIND_METHOD)

    def _callable_facts(self, fn: dict[str, Any], file_path: str, kind: str) -> None:
        qname = str(fn.get("qualified_name", ""))
        self._emit(
            "defines/5",
            "defines("
            f"{prolog_atom(file_path)}, {prolog_atom(kind)}, "
            f"{prolog_atom(qname)}, {int(fn.get('line', 0))}, "
            f"{int(fn.get('end_line', 0))}).",
        )
        for dec in fn.get("decorators", []):
            self._emit(
                "decorates/3",
                f"decorates({prolog_atom(dec)}, {prolog_atom(qname)}, "
                f"{prolog_atom(file_path)}).",
            )
        for prm in fn.get("parameters", []):
            self._emit(
                "param/4",
                "param("
                f"{prolog_atom(qname)}, {prolog_atom(prm.get('name', ''))}, "
                f"{prolog_atom(prm.get('type_annotation') or '')}, "
                f"{prolog_atom(prm.get('kind', 'positional'))}).",
            )
        return_type = fn.get("return_type")
        if return_type:
            # Absent, not '' — "unannotated" is negation-as-failure
            # (REQ-GATE-009), never a sentinel comparison.
            self._emit(
                "returns/2",
                f"returns({prolog_atom(qname)}, {prolog_atom(return_type)}).",
            )
        if fn.get("docstring"):
            self._emit("has_docstring/1", f"has_docstring({prolog_atom(qname)}).")
        for callee in fn.get("calls", []):
            self._emit(
                "calls/3",
                f"calls({prolog_atom(qname)}, {prolog_atom(callee)}, "
                f"{prolog_atom(file_path)}).",
            )

    # -- freshness (the ONE sanctioned disk access) ------------------------

    def _freshness_facts(
        self, file_meta: dict[str, Any], root_paths: tuple[str, ...]
    ) -> None:
        """Emit the staleness comparison's two sides.

        This is the only method in the module that touches the filesystem for
        anything other than the snapshot, and it reads bytes solely to hash
        them — it never parses. REQ-GATE-002 requires both sides: mtime alone
        misses a same-second rewrite, and a hash alone misses a file the
        snapshot never saw.

        That last case needs a DIRECTORY WALK, not just a per-file stat: a
        brand-new ``.py`` file has no snapshot entry at all, so iterating
        ``file_meta`` can never reach it, and a gate would silently not check
        the newest code in the tree — the worst possible blind spot. Walking the
        snapshot's own root and hashing what it did not describe closes it, and
        needs no new predicate: ``content_hash(disk, F, _)`` with no matching
        ``content_hash(snapshot, F, _)`` IS "missing from snapshot".
        """
        self._emit(
            "snapshot_built_at/1",
            f"snapshot_built_at({self.snapshot_path.stat().st_mtime_ns}).",
        )
        described: set[str] = set()
        for meta in file_meta.values():
            raw = str(meta.get("file_path", ""))
            if not raw:
                continue
            file_path = self._rel(raw)
            if not self._in_scope(file_path):
                self._stats.out_of_scope += 1
                continue
            described.add(file_path)
            self._emit(
                "content_hash/3",
                f"content_hash(snapshot, {prolog_atom(file_path)}, "
                f"{prolog_atom(meta.get('sha1', ''))}).",
            )
            # A file the snapshot describes but disk no longer has emits NO
            # disk side. freshness.pl reads that absence as "the snapshot
            # describes a file that is gone" — a staleness verdict, not a
            # silent pass.
            self._disk_side(file_path)

        for file_path in self._undescribed_disk_files(root_paths, described):
            self._disk_side(file_path)

    def _disk_side(self, file_path: str) -> None:
        """Emit the disk half of the freshness comparison for one file."""
        disk = self.repo_root / file_path
        if not disk.is_file():
            return
        self._emit(
            "content_hash/3",
            f"content_hash(disk, {prolog_atom(file_path)}, "
            f"{prolog_atom(hashlib.sha1(disk.read_bytes()).hexdigest())}).",
        )
        self._stats.files_hashed += 1

    def _undescribed_disk_files(
        self, root_paths: tuple[str, ...], described: set[str]
    ) -> list[str]:
        """Every ``.py`` under the snapshot's own roots that it does not describe.

        Scoped to the trees the snapshot was BUILT from, because that is exactly
        the territory it claims to cover; walking wider would report files the
        snapshot never promised to describe as gaps in it. Every declared tree
        is walked, not just the primary one — otherwise a new module under
        ``tests`` would be missing from a snapshot that claims to cover it and
        nothing would say so.
        """
        found: list[str] = []
        for root_path in root_paths:
            if not root_path:
                continue
            root = Path(root_path)
            if not root.is_absolute():
                root = self.repo_root / root
            if not root.is_dir():
                continue
            # The analyzer's own walk, not a second copy of its rules. This
            # comparison is only meaningful if both sides agree on what counts
            # as a source file: a private rglob here re-derived "skip build
            # output" but never learned "skip nested distributions", so the
            # vendored trees the snapshot correctly omits were reported as
            # gaps in it and the gate went INVALID on a healthy repo.
            found.extend(self._rel(str(path)) for path in CodebaseAnalyzer._walk(root))
        return sorted({path for path in found if path not in described})

    # -- policy data -------------------------------------------------------

    def _policy_facts(self, modules: list[dict[str, Any]]) -> None:
        for key in sorted(self.infra_keys):
            self._emit("infra_key/1", f"infra_key({prolog_atom(key)}).")
        for module_name, reason in self._banned_imports():
            self._emit(
                "banned_import/2",
                f"banned_import({prolog_atom(module_name)}, {prolog_atom(reason)}).",
            )
        for mod in modules:
            raw = str(mod.get("file_path", ""))
            if not raw:
                continue
            file_path = self._rel(raw)
            if not self._in_scope(file_path):
                continue
            self._emit(
                "file_class/2",
                f"file_class({prolog_atom(file_path)}, "
                f"{prolog_atom(classify_file(file_path))}).",
            )
            self._waiver_facts(file_path)

    def _waiver_facts(self, file_path: str) -> None:
        """Emit the reasoned ``# gate-ok:`` declarations one file carries.

        A COMMENT is not in the AST, so the only way a rule can see a
        declaration a reviewer wrote at the line is for the extractor to read it
        — the same boundary read that brings in the ban registry and the ruff
        catalogue (REQ-GATE-010/011). It is policy DATA, not a re-parse: the
        file is scanned line-by-line for one regex and never handed to ast.

        Without it the Prolog layer diverges from the checker it replaces.
        ``test_isolation_audit``'s ``OS_ENVIRON_MUTATION`` honours this
        convention; on 2026-08-08 the first run over tests/ and scripts/ found
        47 env mutations, 37 of them at lines carrying a reasoned waiver that
        someone had already adjudicated in review. A replacement that ignores
        those does not find more, it finds wrong.
        """
        disk = self.repo_root / file_path
        try:
            text = disk.read_text(encoding="utf-8")
        except OSError, UnicodeDecodeError:
            # Unreadable source waives nothing — see core.gate_waivers.is_waived.
            return
        for line, reason in sorted(waiver_reasons(text).items()):
            self._emit(
                "waiver/3",
                f"waiver({prolog_atom(file_path)}, {line}, {prolog_atom(reason)}).",
            )

    # -- tooling (REQ-GATE-010/011) ----------------------------------------

    def _tooling_facts(self) -> None:
        """Record the linter's catalogue and this project's enforcement of it.

        REQ-GATE-010: Prolog does not reimplement a linter. Re-deriving line
        length or trailing whitespace was measured at 845,547 facts for the
        four least valuable rules, three of which black silently REWRITES —
        a rule reporting what the formatter already fixed is noise by
        construction. What no linter can do is reason about its own coverage,
        so the CATALOGUE and the CONFIGURED STATUS become facts, and a
        THIRD mechanism kind (formatter, :meth:`_formatter_mechanism_facts`)
        joins the ruff and native ones so ``covered/1`` (gate_rules.pl) is
        not blind to what a configured formatter genuinely fixes on every
        run. ``uncovered/1`` (P3) derives the gap that remains: this
        project's own pyproject.toml ignores E501 with the comment "handled
        by black", and that claim does NOT hold for every long line — a
        long string literal or a single-token long comment survives black
        untouched (verified reproducibly, see
        :data:`BLACK_UNCONDITIONAL_RUFF_CODES`) — so long-line policy is
        reported as having no mechanism behind it rather than a false
        "covered" manufactured to make the finding disappear.

        ``tool_outcome/3`` is why this cannot quietly return nothing. A missing
        or failed ruff would emit zero catalogue facts, and zero catalogue
        facts make every policy look uncovered (or, read the other way, make a
        coverage query vacuously true). The outcome fact says which happened,
        so a gate can refuse rather than infer over an absence it created.
        """
        self._native_mechanism_facts()

        started = time.time_ns()
        status, rules = self._ruff_catalogue()
        self._emit(
            "tool_outcome/3",
            f"tool_outcome(ruff, {prolog_atom(status)}, {started}).",
        )
        if status != OUTCOME_OK:
            return

        select, ignore = self._ruff_config()
        code_to_name: dict[str, str] = {}
        for rule in rules:
            code = str(rule.get("code", ""))
            if not code:
                continue
            name = str(rule.get("name", ""))
            code_to_name[code] = name
            self._emit(
                "tool_rule/4",
                "tool_rule(ruff, "
                f"{prolog_atom(code)}, "
                f"{prolog_atom(name)}, "
                f"{prolog_atom(_fixability(rule))}).",
            )
            self._emit(
                "tool_rule_status/3",
                f"tool_rule_status(ruff, {prolog_atom(code)}, "
                f"{prolog_atom(rule_status(code, select, ignore))}).",
            )
            # REQ-GATE-011: the ruff half of "no enabled tool rule or native
            # rule behind it". The rule's own catalogue NAME is its policy
            # identity — 'line-too-long' for E501 — so this is a mechanical
            # mirror of tool_rule/4 rather than a second, hand-picked list of
            # "which ruff rules count as policies". covered/1 (gate_rules.pl)
            # joins this to tool_rule_status/3 to decide enabled vs not.
            self._emit(
                "policy_mechanism/3",
                f"policy_mechanism({prolog_atom(name)}, {prolog_atom('ruff')}, "
                f"{prolog_atom(code)}).",
            )
        self._formatter_mechanism_facts(code_to_name)

    def _native_mechanism_facts(self) -> None:
        """Emit policy_mechanism/3 for every NATIVE rule gate_rules.pl implements.

        REQ-GATE-011 needs "no enabled tool rule OR native rule" — the
        tool half alone (the plan's original covered/1 sketch) cannot say a
        policy is covered by ``bare_os_environ`` existing as a real
        ``violation/4`` clause. The native side is exactly those clause
        heads, RIGHT NOW — parsed the identical way
        ``tests/gates/test_gate_rules.py::TestCorpusDiscipline::
        test_every_declared_rule_has_a_corpus`` already parses them, rather
        than a second Python list of rule ids that could drift the moment a
        rule is renamed or added here without that list being touched — the
        exact drift this predicate exists to catch.

        Policy IS the rule id: a native rule's mechanism is its own
        existence, and Tool = native is what lets ``covered/1``
        (gate_rules.pl) tell it apart from a ruff-mechanised policy that
        happens to share a name.
        """
        if not self.gate_rules_path.is_file():
            return
        text = self.gate_rules_path.read_text(encoding="utf-8")
        rule_ids = sorted(
            {
                line.split("(", 1)[1].split(",", 1)[0].strip()
                for line in text.splitlines()
                if line.startswith("violation(")
            }
        )
        for rule_id in rule_ids:
            self._emit(
                "policy_mechanism/3",
                f"policy_mechanism({prolog_atom(rule_id)}, {prolog_atom('native')}, "
                f"{prolog_atom(rule_id)}).",
            )

    def _formatter_mechanism_facts(self, code_to_name: dict[str, str]) -> None:
        """Emit policy_mechanism/3, Tool=formatter, for ruff codes THIS
        project's configured black demonstrably, UNCONDITIONALLY fixes.

        covered/1 needed a THIRD mechanism kind: a formatter can genuinely
        mechanise a policy with no linter rule enabled for it at all. The
        condition is read from real config — ``[tool.black]`` must actually
        be present in this project's pyproject.toml — so removing black
        from the project's tooling stops these facts on the next
        extraction, the same discipline REQ-GATE-010 already applies to the
        ruff select/ignore read.

        Deliberately NARROW: only :data:`BLACK_UNCONDITIONAL_RUFF_CODES`
        qualify — codes verified, reproducibly, to be rewritten by black on
        EVERY run regardless of content. E501/line-too-long is the rule
        this predicate was built to answer and is DELIBERATELY absent: this
        project's own pyproject.toml ignores it with the comment "handled
        by black", and that claim does not hold in general — black only
        reflows constructs it has freedom to reflow, and a long string
        literal or a single-token long comment survives it untouched (the
        constant's own docstring carries the reproduction). Asserting black
        as E501's mechanism here would manufacture a false "covered" —
        exactly the false-green class this predicate exists to catch, not
        produce.

        Reported under the Name ruff already uses for the same code (not
        the bare code) via ``code_to_name`` — the SAME dict ``_tooling_facts``
        built from the live catalogue — so this lines up with the ruff
        mechanism's OWN Policy identity: one real-world policy, one
        identifier, regardless of which mechanism covers it. A code
        ``BLACK_UNCONDITIONAL_RUFF_CODES`` names that the live catalogue did
        not report (ruff down, or a future catalogue drops it) is silently
        skipped rather than asserted under a fabricated name.
        """
        if not self._black_is_configured():
            return
        for code in sorted(BLACK_UNCONDITIONAL_RUFF_CODES):
            name = code_to_name.get(code)
            if not name:
                continue
            self._emit(
                "policy_mechanism/3",
                f"policy_mechanism({prolog_atom(name)}, {prolog_atom('formatter')}, "
                f"{prolog_atom('black')}).",
            )

    def _ruff_catalogue(self) -> tuple[str, list[dict[str, Any]]]:
        """Ask ruff for its rule catalogue. Every failure mode is NAMED."""
        try:
            completed = subprocess.run(  # noqa: S603 — fixed command, no shell
                list(self.ruff_catalogue_command),
                capture_output=True,
                text=True,
                timeout=self.ruff_catalogue_timeout_s,
                check=False,
                cwd=self.repo_root,
            )
        except FileNotFoundError:
            return OUTCOME_MISSING, []
        except subprocess.TimeoutExpired:
            return OUTCOME_TIMEOUT, []
        if completed.returncode != 0:
            return OUTCOME_FAILED, []
        try:
            parsed = json.loads(completed.stdout)
        except json.JSONDecodeError:
            return OUTCOME_FAILED, []
        if not isinstance(parsed, list):
            return OUTCOME_FAILED, []
        return OUTCOME_OK, [r for r in parsed if isinstance(r, dict)]

    def _pyproject_toml(self) -> dict[str, Any]:
        """This project's pyproject.toml, parsed once. Empty dict if absent.

        Shared by :meth:`_ruff_config` and :meth:`_black_is_configured` so
        the file is read and parsed a single time per extraction rather than
        once per tool section.
        """
        if not self.pyproject_path.is_file():
            return {}
        return tomllib.loads(self.pyproject_path.read_text(encoding="utf-8"))

    def _ruff_config(self) -> tuple[list[str], list[str]]:
        """Read this project's select/ignore lists.

        Both the modern ``[tool.ruff.lint]`` location and the legacy top-level
        one are read: a config that moved would otherwise silently report every
        rule ``unknown``, which reads as "nothing is enforced".
        """
        ruff = self._pyproject_toml().get("tool", {}).get("ruff", {})
        lint = ruff.get("lint", {})
        select = lint.get("select", ruff.get("select", []))
        ignore = lint.get("ignore", ruff.get("ignore", []))
        return [str(s) for s in select], [str(i) for i in ignore]

    def _black_is_configured(self) -> bool:
        """Is ``[tool.black]`` present in this project's config at all?

        The condition IS the fact: a formatter mechanism only makes sense
        while the formatter it names is actually configured for this
        project. Removing ``[tool.black]`` stops
        :meth:`_formatter_mechanism_facts` on the next extraction.
        """
        return isinstance(self._pyproject_toml().get("tool", {}).get("black"), dict)

    def _banned_imports(self) -> list[tuple[str, str]]:
        """Read the banned-dependency registry, keeping each entry's rationale.

        Delegates to ``core.banned_libraries``: the registry being shared DATA
        is only half of what matters, and a second parser is a second chance to
        disagree about what an entry means.
        """
        return load_banned_libraries(self.banned_libraries_path)