scripts.prolog_meta_lint

Reference › scripts › prolog_meta_lint

scripts/prolog_meta_lint.py is the Prolog governance-rulebase meta-lint (SPEC-033 section 9, task B17): a linter for the .pl rule files themselves. It is run by a human or a gate, and it is also a library: orchestrata_ultimate.factory.amendment.syntactic, orchestrata_ultimate.factory.projection.analyzer and check_prolog_boundary import it. tests/governance/test_policy_layers.py and tests/scripts/test_prolog_meta_lint.py cover it.

It has no Prolog parser dependency. mask_non_code blanks the content of comments and quoted literals (including SWI-Prolog character-code notation), _split_clauses and _parse_clause split the masked text into clauses, and scan_file produces a per-file scan. The checks are: check_no_cut (ADR-0025, no bare !), check_no_dynamic_goal (ADR-0025, no call(Var, ...) on an unbound variable, including the build-with-univ-then-call shape), check_no_rule_side_effects (ADR-0025, no assert or retract in rules), check_stratification (ADR-0010, no predicate exported by constitutional.pl may reach an extension point, using the call graph from build_predicate_graph) and check_no_redefinition (no two files define the same predicate head). lint_file runs the per-file ones, and classification_gaps reports .pl files that no classification bucket covers.

run_lint scans the governance rulebases, which always gate the exit code. main options: --all additionally scans the allowlist, application-scope and unclassified files for reporting only, and exits 0 regardless; --json prints machine-readable output. Without --all it exits 0 when the report passed and 1 otherwise.

At a glance

Source scripts/prolog_meta_lint.py
Lines of code 1333
Classes 6
Functions 38
Imports (internal) none
Imported by orchestrata_ultimate.factory.amendment.syntactic, orchestrata_ultimate.factory.projection.analyzer, scripts.check_prolog_boundary
Tests that import it tests/governance/test_policy_layers.py, tests/scripts/test_prolog_meta_lint.py

Class diagram

Class diagram scripts.prolog_meta_lint

Dependency diagram

Dependency diagram scripts.prolog_meta_lint

API

Constants

Name Type Value
GOVERNANCE_RULEBASES tuple[RuleFile, ...] (RuleFile('src/orchestrata_ultimate/prolog/agents.pl', 'Constitutional validati…
ALLOWLIST tuple[RuleFile, ...] (RuleFile('src/orchestrata_ultimate/prolog/explain.pl', 'ADR-0031: the proof-tr…
APPLICATION_SCOPE tuple[RuleFile, ...] (RuleFile('src/orchestrata_ultimate/prolog/intent_grammar.pl', 'DCG natural-lan…
UNCLASSIFIED_BY_LANE0 tuple[RuleFile, ...] (RuleFile('src/orchestrata_ultimate/prolog/kg_relevance.pl', 'Likely applicatio…

Classes

RuleFile (dataclass)

Decorators: dataclass(frozen=True) · Line 95

One rulebase file with a reviewed prose reason for its classification.

Attribute Type Default
path str
reason str

Violation (dataclass)

Decorators: dataclass(frozen=True) · Line 300

One lint finding: file:line: rule: evidence.

Attribute Type Default
file str
line int
rule str
evidence str
Method Signature Summary
format format() -> str Render as file:line: rule: evidence, per the brief's format.

ClauseInfo (dataclass)

Decorators: dataclass(frozen=True) · Line 314

One parsed clause (fact, rule, or directive) from a governance file.

Attribute Type Default
relpath str
start int
head_name str | None
body_start int | None
body_text str | None

EdgeInfo (dataclass)

Decorators: dataclass · Line 325

One predicate-call edge discovered in a clause body.

Attribute Type Default
to_name str
is_naf bool
origin_relpath str
origin_line int

FileScan (dataclass)

Decorators: dataclass · Line 335

One .pl file: its masked text, line-start index, and clauses.

Attribute Type Default
relpath str
masked str
line_starts list[int]
clauses list[ClauseInfo]

LintReport (dataclass)

Decorators: dataclass · Line 345

The full result of one :func:run_lint call.

Attribute Type Default
gate_violations list[Violation] field(default_factory=list)
all_violations dict[str, list[Violation]] field(default_factory=dict)
allowlist_lines list[str] field(default_factory=list)
unclassified list[str] field(default_factory=list)
classification_gaps list[str] field(default_factory=list)
Method Signature Summary
passed property passed() -> bool Gate verdict: True iff there are zero gate-scoped violations.
to_dict to_dict() -> dict[str, object] JSON-serializable rendering, for --json.

Functions

Function Signature Summary
_mask_line_comment _mask_line_comment(text: str, out: list[str], i: int) -> int Blank a % line comment in-place; return the index just past it.
_mask_block_comment _mask_block_comment(text: str, out: list[str], i: int) -> int Blank a /* ... */ block comment in-place (newlines preserved); return the index just past it (or EOF if unterminated).
_is_char_code_prefix _is_char_code_prefix(text: str, i: int) -> bool True if text[i] == "'" is preceded by an unbound 0 digit, forming SWI-Prolog's 0'c character-code notation (e.g. 0'a, 0'\n, 0''') rather than opening a quoted atom.
_mask_char_code_literal _mask_char_code_literal(text: str, out: list[str], i: int) -> int Blank SWI-Prolog's 0'c character-code notation (text[i] is the opening '); return the index just past the encoded character.
_mask_quoted_literal _mask_quoted_literal(text: str, out: list[str], i: int) -> int Blank a single/double-quoted literal in-place (\-escapes and the doubled-quote escape both handled; newlines preserved); return the index just past the closing quote (or EOF if unterminated).
mask_non_code mask_non_code(text: str) -> str Return text with the CONTENT of every % line comment, /* ... */ block comment, and single/double-quoted literal replaced by spaces (newlines preserved) — same length, same line numbering…
_line_starts _line_starts(text: str) -> list[int]
_line_of _line_of(line_starts: list[int], pos: int) -> int
_split_clauses _split_clauses(masked: str) -> list[tuple[int, int]] Return (start, end) char ranges for each clause in masked (end excludes the terminating '.').
_find_top_level_neck _find_top_level_neck(clause: str) -> int Find the index of the clause's own :- (paren depth 0), or -1.
_parse_clause _parse_clause(relpath: str, masked: str, start: int, end: int) -> ClauseInfo
discover_all_pl_files discover_all_pl_files(prolog_dir: Path | None = None, codebase_db_pl: Path | None = None) -> list[Path] Every .pl file this meta-lint's scope could possibly need to classify: prolog/*.pl (skipping .bak) plus tools/codebase_db.pl — exactly the Lane-0 inventory's stated survey scope.…
_known_classified_paths _known_classified_paths(governance_rulebases: tuple[RuleFile, ...] | None, allowlist: tuple[RuleFile, ...] | None, application_scope: tuple[RuleFile, ...] | None, unclassified_by_lane0: tuple[RuleFile, ...] | None) -> set[str] The union of paths across all four classification buckets, each defaulting to its real shipped constant when not overridden (the None-means-"use the shipped default" injection-seam convention…
classification_gaps classification_gaps(*, root: Path | None = None, prolog_dir: Path | None = None, codebase_db_pl: Path | None = None, governance_rulebases: tuple[RuleFile, ...] | None = None, allowlist: tuple[RuleFile, ...] | None = None, application_scope: tuple[RuleFile, ...] | None = None, unclassified_by_lane0:… Relpaths of any .pl file not accounted for by ANY of GOVERNANCE_RULEBASES / ALLOWLIST / APPLICATION_SCOPE / UNCLASSIFIED_BY_LANE0.
scan_file scan_file(path: Path, root: Path | None = None) -> FileScan Read + mask + clause-split one .pl file. root is the base a relative display path is computed against (defaults to REPO_ROOT) — an injection seam so tests can scan a synthetic corpus under…
check_no_cut check_no_cut(scan: FileScan) -> list[Violation] ADR-0025's no-cut rule: every bare ! outside comments/strings.
_scan_call_var_violations _scan_call_var_violations(scan: FileScan, pattern: re.Pattern[str], message: str) -> list[Violation] Shared body for the two call(...)-on-unbound-variable shapes check_no_dynamic_goal checks — same violation shape, only the regex and evidence-message template differ.
_scan_univ_then_call_violations _scan_univ_then_call_violations(scan: FileScan) -> list[Violation] The Var =.. [...] then call(Var, ...) construction-then-call shape — a genuinely per-clause scan (the univ assignment and the later call must be correlated within the SAME clause body).
check_no_dynamic_goal check_no_dynamic_goal(scan: FileScan) -> list[Violation] ADR-0025's no-dynamic-goal rule: call(Var, ...) on an unbound variable (bare or module-qualified), plus the Var =.. [...] then call(Var, ...) shape.
check_no_rule_side_effects check_no_rule_side_effects(scan: FileScan) -> list[Violation] ADR-0025's no-rule-side-effects rule: assert*/retract* inside a rule BODY (a :- dynamic ... declaration is unaffected).
_extract_naf_span _extract_naf_span(body: str, after: int) -> tuple[int, int] The (start, end) char span of the goal negated by a '+' found ending at after — either a balanced parenthesized goal, or a bare term up to the next top-level ',' / ';' / '->' / ')' / end.
_module_exports _module_exports(masked_text: str) -> set[str] Predicate NAMES declared in a :- module(Name, [ind/arity, ...]). directive (arity dropped — see the module docstring's name-only-graph note).
_declares_module _declares_module(masked_text: str) -> bool True iff FILE opens with SWI's own two-argument module directive (:- module(Name, [ind/arity, ...]). — REQ-114-FR2-004 clause (b)'s "declares :- module with an export list").
build_predicate_graph build_predicate_graph(scans: list[FileScan]) -> dict[str, list[EdgeInfo]] Build the predicate call graph across ALL given governance scans — edges are (name-only, per this module's deliberately conservative graph-keying: over-inclusive on arity mismatch is safe for a…
_bfs_root_to_extension_point _bfs_root_to_extension_point(root: str, graph: dict[str, list[EdgeInfo]], fallback_origin: tuple[str, int]) -> Violation | None BFS from one constitutional-invariant root predicate, looking for a path to extension_point/1 that crosses a \+.
check_stratification check_stratification(scans: list[FileScan]) -> list[Violation] ADR-0010: no predicate exported by constitutional.pl may reach extension_point/1 via a path crossing a \+.
check_no_redefinition check_no_redefinition(scans: list[FileScan]) -> list[Violation] No two of scans may define the same predicate head unless at least one of them declares :- module(Name, [...]).
lint_file lint_file(scan: FileScan) -> list[Violation] Every per-file check (no-cut, no-dynamic-goal, no-rule-side-effects).
_allowlist_line _allowlist_line(rule_file: RuleFile) -> str
_lint_governance_scope _lint_governance_scope(base: Path, governance: tuple[RuleFile, ...], report: LintReport) -> None Scan + lint every governance rulebase (always gates), plus the cross-file stratification and redefinition checks over that same scan set.
_lint_reporting_only_bucket _lint_reporting_only_bucket(base: Path, files: tuple[RuleFile, ...], report: LintReport) -> list[str] Scan a non-gating bucket (application/unclassified) purely for the --all full report — never touches gate_violations.
_resolve_scope_buckets _resolve_scope_buckets(governance_rulebases: tuple[RuleFile, ...] | None, allowlist: tuple[RuleFile, ...] | None, application_scope: tuple[RuleFile, ...] | None, unclassified_by_lane0: tuple[RuleFile, ...] | None) -> tuple[tuple[RuleFile, ...], tuple[RuleFile, ...], tuple[RuleFile, ...], tuple[Rule… Resolve the four None-means-"use the shipped default" scope buckets run_lint accepts as an injection seam.
run_lint run_lint(*, all_files: bool = False, root: Path | None = None, governance_rulebases: tuple[RuleFile, ...] | None = None, allowlist: tuple[RuleFile, ...] | None = None, application_scope: tuple[RuleFile, ...] | None = None, unclassified_by_lane0: tuple[RuleFile, ...] | None = None, check_classificat… Run the meta-lint.
_print_classification_gaps _print_classification_gaps(report: LintReport) -> None
_print_all_files_report _print_all_files_report(report: LintReport) -> None
_print_gate_report _print_gate_report(report: LintReport) -> None
_print_report _print_report(report: LintReport, *, all_files: bool) -> None
main main(argv: list[str] | None = None) -> int CLI entry point — see the module docstring for the two modes.

Source

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

#!/usr/bin/env python3
"""Prolog governance-rulebase meta-lint (SPEC-033 §9, task B17).

Design authority: ADR-0010 (stratified negation), ADR-0025 (construct ban),
ADR-0031 (explain.pl is the single allowlisted exception, BY NAME).

A deterministic, dependency-free (stdlib-only) linter over the GOVERNANCE
rulebase — the ``.pl`` files a gate decision can depend on (constitution,
patterns, policy/permission/completion rules). It enforces five checks:

* ``no-cut`` — ADR-0025 bans the cut (``!``): it makes rule meaning
  order-dependent, defeating reordering/caching/independent testability.
* ``no-dynamic-goal`` — ADR-0025 bans ``call/N`` on an unbound goal, bare or
  module-qualified (``call(Goal)`` and ``call(module:Goal)`` alike), and the
  ``Var =.. [...]`` + ``call(Var, ...)`` construction-then-call shape: the
  call graph must be statically knowable or ADR-0006's projection-completeness
  check and this file's own stratification check are both unsound.
* ``no-rule-side-effects`` — ADR-0025 bans ``assert*``/``retract*`` inside a
  rule BODY (a ``:- dynamic ...`` declaration is fine; ground-fact assertz
  from the Python side is a different mechanism entirely and is untouched by
  this check — it never appears inside a ``.pl`` clause body).
* ``stratification`` — ADR-0010: no constitutional invariant (a predicate
  exported by ``constitutional.pl``) may reach ``extension_point/1`` via a
  path that crosses a ``\\+`` (negation-as-failure). That would let a
  lower-layer addition — one that only touches sanctioned extension points —
  flip a constitutional check from block to allow (the ``safe_override``
  scenario ADR-0010 documents), which the old syntactic
  extension-point-only lint could not see.
* ``redefinition`` — REQ-114-FR2-004 clause (b): no two consulted governance rulebases
  may define the same predicate head unless at least one of them declares
  ``:- module(Name, [...])``. This is the STATIC half of the defence against
  a live hole demonstrated 2026-08-19 with real swipl: consulting a second
  module-less ``.pl`` file that defines an already-loaded predicate (e.g.
  ``violation/4`` in ``gate_rules.pl``) silently REPLACES the earlier
  clauses instead of adding to them — SWI's only diagnostic is a Warning
  (``Redefined static procedure``), which nothing checked before this rule.
  ``gates/engine.py::GateEngine._evaluate``'s clause-count invariant is the
  RUNTIME half, for whatever slips past this static one.

**Allowlist**: ``explain.pl`` is the one file ADR-0031 sanctions as exempt —
it is the proof-tree meta-interpreter, whose cuts/dynamic ``call/N`` are the
sanctioned machinery that makes explanation possible at all. The allowlist is
a **visible constant**, not a skipped file: every run of this tool prints
``allowlisted: explain.pl (ADR-0031)`` — its violations are still detected
internally, just never counted against the gate's exit code.

Scope: ``GOVERNANCE_RULEBASES`` below is the Lane-0 inventory's classification
(agents/constitutional/cxp/multiagent/ethics/governance/knowledge/completion/
intent_policy.pl) — files where scope=governance and at least one construct
site exists. ``APPLICATION_SCOPE`` documents files the wave-A sweep already
ruled out-of-scope (``intent_grammar.pl`` — DCG NL parsing; ``codebase_db.pl``
— AST/codebase-query tooling, no permission semantics), carried forward per
this task's instructions. ``UNCLASSIFIED_BY_LANE0`` lists ``.pl`` files under
``prolog/`` that the Lane-0 inventory handed to this task did NOT classify at
all — several of them (``pam_rules.pl``, ``postman_governance.pl``,
``resources.pl``) are governance-shaped by this very module's own scope
definition (ownership/permission/resource-access rules a gate decision can
depend on) and ``resources.pl`` alone has 11 ``\\+`` occurrences never
triaged for stratification-safety. This is a real coverage gap in
``GOVERNANCE_RULEBASES``, reported (not silently closed) — see
``--all`` output and the module docstring of ``tests/scripts/
test_prolog_meta_lint.py``.

CLI:
    python scripts/prolog_meta_lint.py       # GATE: governance scope only, exit 0/1
    python scripts/prolog_meta_lint.py --all # REPORT: every .pl file, always exits 0
    python scripts/prolog_meta_lint.py --json  # machine-readable (either mode)
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from bisect import bisect_right
from collections import deque
from dataclasses import dataclass, field
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
PROLOG_DIR = REPO_ROOT / "src" / "orchestrata_ultimate" / "prolog"
CODEBASE_DB_PL = REPO_ROOT / "src" / "orchestrata_ultimate" / "tools" / "codebase_db.pl"

EXTENSION_POINT_NAME = "extension_point"

NO_CUT = "no-cut"
NO_DYNAMIC_GOAL = "no-dynamic-goal"
NO_RULE_SIDE_EFFECTS = "no-rule-side-effects"
STRATIFICATION = "stratification"
REDEFINITION = "redefinition"


@dataclass(frozen=True)
class RuleFile:
    """One rulebase file with a reviewed prose reason for its classification."""

    path: str  # POSIX path relative to REPO_ROOT
    reason: str


# The Lane-0 inventory's governance classification (SPEC-033 B17 Lane 0):
# every file below has scope=governance AND at least one construct site
# (cut/naf) that Lane 0 enumerated a verdict for.
GOVERNANCE_RULEBASES: tuple[RuleFile, ...] = (
    RuleFile(
        "src/orchestrata_ultimate/prolog/agents.pl",
        "Constitutional validation, role hierarchy, agent scope and escalation "
        "— a gate decision (constitutional_valid/2) depends on this file.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/constitutional.pl",
        "The 11 constitutional articles themselves — the immutable floor "
        "(SPEC-033 §1.3).",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/cxp.pl",
        "CXP packet governance: evaluate_action/3, validate_packet/2 — read/"
        "modify/send permission decisions.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/multiagent.pl",
        "Multi-agent coordination policy: fan_out_decision/6 and related "
        "resourcing/permission rules.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/ethics.pl",
        "Ethical boundary checks (harm/deception/privacy/autonomy) consulted "
        "by the governance gate.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/governance.pl",
        "The main policy/permission rulebase: session, tool, volk, resource, "
        "and compliance gates.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/knowledge.pl",
        "Ownership/access-scope predicates (owns/2, in_scope/2) a gate can "
        "depend on. NOTE: the ownership_chain block was just remediated "
        "(ADR-0026) and is off-limits for further edits per this task's "
        "boundaries — it is still linted, never rewritten by this tool.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/completion.pl",
        "Completion-gate predicates (has_open_question/2) — a gate decision " "input.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/intent_policy.pl",
        "Intent permission policy (permission_denied/2) — a gate decision " "input.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/policy_layers.pl",
        "SPEC-033 B27 defeasible policy layering (layered_verdict/3, "
        "layer_verdict/4) — delegates to constitutional_check/3, ethics:, "
        "resources:, and a team_norms ground-fact layer; a gate decision "
        "can depend on it once adopted.",
    ),
    # The Prolog gate layer (REQ-GATE-001..012). Classified as GOVERNANCE
    # rather than APPLICATION_SCOPE deliberately: these clauses decide
    # whether a change may land, which is a permission semantics even though
    # the subject is code quality rather than an agent. Holding them to the
    # construct ban and stratification check is the point of having one.
    RuleFile(
        "src/orchestrata_ultimate/prolog/gate_rules.pl",
        "The gate's ONE entry predicate (violation/4) and its rule clauses "
        "— the decision itself.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/gate_severity.pl",
        "Defeasible severity + severity_exception/3 — decides whether a "
        "violation blocks or advises.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/code_facts.pl",
        "The gate's fact-base declarations (dynamic/1 for every extracted "
        "predicate) — a gate decision input.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/constraints.pl",
        "Constraint-satisfaction semantics for SPEC-007 verification (constraint_satisfied/3). "
        "Governance-classified because it returns a FINAL verdict a VerificationResult is "
        "built from -- SPEC-007 Key Principle #1 -- so the ADR-0025 cut ban must apply: a cut "
        "would make a constraint's verdict depend on clause ORDER rather than on its guards.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/commit_review.pl",
        "The pre-commit review verdict (commit_status/1): decides from each "
        "reviewer's reported facts whether a commit may land, and names every "
        "way a review can fail to be a verdict (review_defect/3).",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/freshness.pl",
        "Freshness-as-inference: decides the three-valued pass/fail/invalid "
        "verdict, so a stale fact base cannot read as PASS.",
    ),
)

# ADR-0031: the ONE sanctioned exception. Its cuts/dynamic call/N are the
# proof-tree meta-interpreter's own sanctioned machinery, not a hole in the
# ban — still scanned, never counted in the gate's exit code.
ALLOWLIST: tuple[RuleFile, ...] = (
    RuleFile(
        "src/orchestrata_ultimate/prolog/explain.pl",
        "ADR-0031: the proof-tree meta-interpreter — the sanctioned "
        "exception to the construct ban (cuts + qualified-goal call/N are "
        "its own machinery, not application logic).",
    ),
)

# Wave-A sweep rulings, carried forward per this task's instructions: NOT
# governance-scoped, so out of the gate entirely.
APPLICATION_SCOPE: tuple[RuleFile, ...] = (
    RuleFile(
        "src/orchestrata_ultimate/prolog/intent_grammar.pl",
        "DCG natural-language parsing — application-scoped, no permission "
        "semantics.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/tools/codebase_db.pl",
        "AST/codebase-query tooling consumed by scripts/"
        "serialize_codebase_ast.py — non-decision, no permission semantics.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/test_schedule.pl",
        "CLP(FD) test-scheduling solver (SPEC-033 B28) — pure constraint "
        "optimisation, no permission/policy semantics; the construct ban "
        "does not apply.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/history.pl",
        "Temporal reasoning over git-derived feature/symbol-loss history "
        "(WHAT/WHEN/HOW/WHY a symbol disappeared) — diagnostics over "
        "commit/ADR facts, no permission/gate semantics; the construct ban "
        "does not apply.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/codemod_review.pl",
        "SPEC-110 §9.9 semantic-safety predicates for the PrologSemanticReview "
        "codemod pass (orchestrata_ultimate.refactor.codemod.review) — "
        "non-decision refactoring tooling over Python-asserted ground facts, "
        "no permission/gate semantics; the construct ban does not apply.",
    ),
)

# Reported, NOT silently resolved: .pl files under prolog/ the Lane-0
# inventory handed to this task never classified at all. Several are
# governance-shaped by this module's own definition (permission/ownership/
# resource-access rules) — see the module docstring. --all surfaces this
# list every run so the gap stays visible until Lane 0 is extended.
UNCLASSIFIED_BY_LANE0: tuple[RuleFile, ...] = (
    RuleFile(
        "src/orchestrata_ultimate/prolog/kg_relevance.pl",
        "Likely application-scoped (KG relevance SCORING is explicitly named "
        "application in the brief) — not formally classified by Lane 0.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/pam_rules.pl",
        "PAM++ ownership/permission rules — looks governance-shaped by name "
        "and content; not classified by Lane 0.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/postman_governance.pl",
        "Postman message-broker GOVERNANCE rules (literally named) — looks "
        "governance-shaped; not classified by Lane 0.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/resources.pl",
        "Resource access control policies — 11 '\\+' occurrences never "
        "triaged for stratification-safety; looks governance-shaped; not "
        "classified by Lane 0.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/routing.pl",
        "Ground-fact intent routing table — no cuts/naf present today; "
        "scope unclear, not classified by Lane 0.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/rtm_constraints.pl",
        "RTM/traceability constraints — likely diagnostics (application); "
        "not classified by Lane 0.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/spec_completeness.pl",
        "Spec-completeness evaluation — likely diagnostics (application); "
        "not classified by Lane 0.",
    ),
    RuleFile(
        "src/orchestrata_ultimate/prolog/intent_lexicon.gen.pl",
        "Generated lexicon FACTS only (no cuts/naf) — not classified by " "Lane 0.",
    ),
)


# ─────────────────────────────────────────────────────────────────────────
# Data model
# ─────────────────────────────────────────────────────────────────────────


@dataclass(frozen=True)
class Violation:
    """One lint finding: ``file:line: rule: evidence``."""

    file: str
    line: int
    rule: str
    evidence: str

    def format(self) -> str:
        """Render as ``file:line: rule: evidence``, per the brief's format."""
        return f"{self.file}:{self.line}: {self.rule}: {self.evidence}"


@dataclass(frozen=True)
class ClauseInfo:
    """One parsed clause (fact, rule, or directive) from a governance file."""

    relpath: str
    start: int  # absolute char offset of the clause in the file's masked text
    head_name: str | None  # None only for a bare directive with no head
    body_start: int | None  # absolute offset of body start, None if no body
    body_text: str | None  # masked body slice, None if no body (fact/directive)


@dataclass
class EdgeInfo:
    """One predicate-call edge discovered in a clause body."""

    to_name: str
    is_naf: bool
    origin_relpath: str
    origin_line: int


@dataclass
class FileScan:
    """One ``.pl`` file: its masked text, line-start index, and clauses."""

    relpath: str
    masked: str
    line_starts: list[int]
    clauses: list[ClauseInfo]


@dataclass
class LintReport:
    """The full result of one :func:`run_lint` call."""

    gate_violations: list[Violation] = field(default_factory=list)
    all_violations: dict[str, list[Violation]] = field(default_factory=dict)
    allowlist_lines: list[str] = field(default_factory=list)
    unclassified: list[str] = field(default_factory=list)
    classification_gaps: list[str] = field(default_factory=list)

    @property
    def passed(self) -> bool:
        """Gate verdict: True iff there are zero gate-scoped violations."""
        return not self.gate_violations

    def to_dict(self) -> dict[str, object]:
        """JSON-serializable rendering, for ``--json``."""
        return {
            "passed": self.passed,
            "gate_violations": [v.format() for v in self.gate_violations],
            "all_violations": {
                relpath: [v.format() for v in vs]
                for relpath, vs in self.all_violations.items()
            },
            "allowlist": self.allowlist_lines,
            "unclassified_by_lane0": self.unclassified,
            "classification_gaps": self.classification_gaps,
        }


# ─────────────────────────────────────────────────────────────────────────
# Tokenizing — mask comments/strings so patterns only match real code
# ─────────────────────────────────────────────────────────────────────────


def _mask_line_comment(text: str, out: list[str], i: int) -> int:
    """Blank a ``%`` line comment in-place; return the index just past it."""
    n = len(text)
    j = i
    while j < n and text[j] != "\n":
        out[j] = " "
        j += 1
    return j


def _mask_block_comment(text: str, out: list[str], i: int) -> int:
    """Blank a ``/* ... */`` block comment in-place (newlines preserved);
    return the index just past it (or EOF if unterminated)."""
    n = len(text)
    j = i
    out[j] = " "
    out[j + 1] = " "
    j += 2
    while j < n and not (text[j] == "*" and j + 1 < n and text[j + 1] == "/"):
        out[j] = " " if text[j] != "\n" else "\n"
        j += 1
    if j < n:
        out[j] = " "
        if j + 1 < n:
            out[j + 1] = " "
        j += 2
    return j


def _is_char_code_prefix(text: str, i: int) -> bool:
    """True if ``text[i] == "'"`` is preceded by an unbound ``0`` digit,
    forming SWI-Prolog's ``0'c`` character-code notation (e.g. ``0'a``,
    ``0'\\n``, ``0'''``) rather than opening a quoted atom. Without this
    check the ``'`` in ``0'!`` would be mistaken for the start of a quoted
    literal, desyncing quote-matching for the rest of the file."""
    if i == 0 or text[i - 1] != "0":
        return False
    prev2 = text[i - 2] if i >= 2 else ""
    return not (prev2.isalnum() or prev2 == "_")


def _mask_char_code_literal(text: str, out: list[str], i: int) -> int:
    """Blank SWI-Prolog's ``0'c`` character-code notation (``text[i]`` is the
    opening ``'``); return the index just past the encoded character. Handles
    a bare char (``0'a``), a backslash escape incl. hex (``0'\\n``,
    ``0'\\x21\\``), and the doubled-quote escape (``0'''`` — the code of the
    quote character itself)."""
    n = len(text)
    out[i] = " "  # the "'" that opens the notation
    i += 1
    if i >= n:
        return i
    if text[i] == "'" and i + 1 < n and text[i + 1] == "'":
        out[i] = " "
        out[i + 1] = " "
        return i + 2
    if text[i] == "\\":
        out[i] = " "
        i += 1
        if i < n and text[i] in "xX":
            out[i] = " "
            i += 1
            while i < n and text[i] not in ("\\", "\n"):
                out[i] = " "
                i += 1
            if i < n and text[i] == "\\":
                out[i] = " "
                i += 1
            return i
        if i < n:
            out[i] = " " if text[i] != "\n" else "\n"
            i += 1
        return i
    out[i] = " " if text[i] != "\n" else "\n"
    return i + 1


def _mask_quoted_literal(text: str, out: list[str], i: int) -> int:
    """Blank a single/double-quoted literal in-place (``\\``-escapes and the
    doubled-quote escape both handled; newlines preserved); return the index
    just past the closing quote (or EOF if unterminated)."""
    n = len(text)
    quote = text[i]
    out[i] = " "
    i += 1
    while i < n:
        c = text[i]
        if c == "\\" and i + 1 < n:
            out[i] = " "
            out[i + 1] = " " if text[i + 1] != "\n" else "\n"
            i += 2
            continue
        if c == quote:
            if i + 1 < n and text[i + 1] == quote:
                out[i] = " "
                out[i + 1] = " "
                i += 2
                continue
            out[i] = " "
            i += 1
            break
        out[i] = " " if c != "\n" else "\n"
        i += 1
    return i


def mask_non_code(text: str) -> str:
    """Return ``text`` with the CONTENT of every ``%`` line comment,
    ``/* ... */`` block comment, and single/double-quoted literal replaced by
    spaces (newlines preserved) — same length, same line numbering as the
    input, so every subsequent regex/char scan only ever sees real code."""
    out = list(text)
    n = len(text)
    i = 0
    while i < n:
        ch = text[i]
        if ch == "%":
            i = _mask_line_comment(text, out, i)
        elif ch == "/" and i + 1 < n and text[i + 1] == "*":
            i = _mask_block_comment(text, out, i)
        elif ch == "'" and _is_char_code_prefix(text, i):
            i = _mask_char_code_literal(text, out, i)
        elif ch in ("'", '"'):
            i = _mask_quoted_literal(text, out, i)
        else:
            i += 1
    return "".join(out)


def _line_starts(text: str) -> list[int]:
    starts = [0]
    for i, c in enumerate(text):
        if c == "\n":
            starts.append(i + 1)
    return starts


def _line_of(line_starts: list[int], pos: int) -> int:
    return bisect_right(line_starts, pos)


# ─────────────────────────────────────────────────────────────────────────
# Clause splitting — top-level '.' terminated, paren/quote aware
# ─────────────────────────────────────────────────────────────────────────

_HEAD_NAME_RE = re.compile(r"^\s*(?:[a-z_][A-Za-z0-9_]*:)?([a-z][A-Za-z0-9_]*)")


def _split_clauses(masked: str) -> list[tuple[int, int]]:
    """Return ``(start, end)`` char ranges for each clause in ``masked``
    (``end`` excludes the terminating '.'). A clause ends at a '.' at paren
    depth 0 immediately followed by whitespace or end-of-file — the standard
    Prolog convention, which also correctly leaves decimal literals like
    ``0.5`` alone (the char after their '.' is a digit, not whitespace)."""
    spans: list[tuple[int, int]] = []
    depth = 0
    n = len(masked)
    clause_start = 0
    i = 0
    while i < n:
        c = masked[i]
        if c in "([{":
            depth += 1
        elif c in ")]}":
            depth -= 1
        elif c == "." and depth == 0:
            nxt = masked[i + 1] if i + 1 < n else ""
            # The univ operator '=..' ends in two literal dots; its SECOND
            # dot must never be mistaken for a clause terminator just
            # because it happens to be followed by whitespace (e.g.
            # "Term =.. [Name | Args],").
            is_univ_second_dot = i >= 2 and masked[i - 2 : i] == "=."
            if (nxt == "" or nxt.isspace()) and not is_univ_second_dot:
                clause_text = masked[clause_start:i]
                if clause_text.strip():
                    spans.append((clause_start, i))
                clause_start = i + 1
        i += 1
    tail = masked[clause_start:n]
    if tail.strip():
        spans.append((clause_start, n))
    return spans


def _find_top_level_neck(clause: str) -> int:
    """Find the index of the clause's own ``:-`` (paren depth 0), or -1."""
    depth = 0
    n = len(clause)
    i = 0
    while i < n:
        c = clause[i]
        if c in "([{":
            depth += 1
        elif c in ")]}":
            depth -= 1
        elif depth == 0 and clause[i : i + 2] == ":-":
            return i
        i += 1
    return -1


def _parse_clause(relpath: str, masked: str, start: int, end: int) -> ClauseInfo:
    clause = masked[start:end]
    stripped = clause.lstrip()
    if stripped.startswith(":-"):
        # A bare directive (":- dynamic ...", ":- module(...)", etc.) has no
        # head and, per this module's docstring, is out of scope for
        # no-rule-side-effects (which is defined over a Head :- Body shape).
        return ClauseInfo(relpath, start, None, None, None)
    neck = _find_top_level_neck(clause)
    if neck == -1:
        m = _HEAD_NAME_RE.match(clause)
        head_name = m.group(1) if m else None
        return ClauseInfo(relpath, start, head_name, None, None)
    head_text = clause[:neck]
    body_text = clause[neck + 2 :]
    m = _HEAD_NAME_RE.match(head_text)
    head_name = m.group(1) if m else None
    return ClauseInfo(relpath, start, head_name, start + neck + 2, body_text)


# ─────────────────────────────────────────────────────────────────────────
# Classification completeness — every .pl file must land in exactly one of
# the four buckets above (governance / allowlist / application /
# unclassified-by-lane0); a brand-new file that lands in none of them is a
# real coverage gap, not a pass-by-omission.
# ─────────────────────────────────────────────────────────────────────────


def discover_all_pl_files(
    prolog_dir: Path | None = None, codebase_db_pl: Path | None = None
) -> list[Path]:
    """Every ``.pl`` file this meta-lint's scope could possibly need to
    classify: ``prolog/*.pl`` (skipping ``.bak``) plus ``tools/codebase_db.pl``
    — exactly the Lane-0 inventory's stated survey scope. ``prolog_dir``/
    ``codebase_db_pl`` are an injection seam for tests; both default to the
    real shipped locations."""
    p_dir = prolog_dir if prolog_dir is not None else PROLOG_DIR
    db_pl = codebase_db_pl if codebase_db_pl is not None else CODEBASE_DB_PL
    files = sorted(p for p in p_dir.glob("*.pl") if not p.name.endswith(".bak"))
    if db_pl.exists():
        files.append(db_pl)
    return files


def _known_classified_paths(
    governance_rulebases: tuple[RuleFile, ...] | None,
    allowlist: tuple[RuleFile, ...] | None,
    application_scope: tuple[RuleFile, ...] | None,
    unclassified_by_lane0: tuple[RuleFile, ...] | None,
) -> set[str]:
    """The union of paths across all four classification buckets, each
    defaulting to its real shipped constant when not overridden (the
    ``None``-means-"use the shipped default" injection-seam convention
    ``run_lint``'s docstring documents)."""
    return {
        rf.path
        for rf in (
            *(
                governance_rulebases
                if governance_rulebases is not None
                else GOVERNANCE_RULEBASES
            ),
            *(allowlist if allowlist is not None else ALLOWLIST),
            *(
                application_scope
                if application_scope is not None
                else APPLICATION_SCOPE
            ),
            *(
                unclassified_by_lane0
                if unclassified_by_lane0 is not None
                else UNCLASSIFIED_BY_LANE0
            ),
        )
    }


def classification_gaps(
    *,
    root: Path | None = None,
    prolog_dir: Path | None = None,
    codebase_db_pl: Path | None = None,
    governance_rulebases: tuple[RuleFile, ...] | None = None,
    allowlist: tuple[RuleFile, ...] | None = None,
    application_scope: tuple[RuleFile, ...] | None = None,
    unclassified_by_lane0: tuple[RuleFile, ...] | None = None,
) -> list[str]:
    """Relpaths of any ``.pl`` file not accounted for by ANY of
    GOVERNANCE_RULEBASES / ALLOWLIST / APPLICATION_SCOPE /
    UNCLASSIFIED_BY_LANE0. Should always be empty for the real repo — a new
    file must be triaged into one of the four buckets before this passes.
    Every keyword is an injection seam (see ``run_lint``'s docstring);
    defaults to the real shipped classification and locations."""
    base = root if root is not None else REPO_ROOT
    known = _known_classified_paths(
        governance_rulebases, allowlist, application_scope, unclassified_by_lane0
    )
    gaps = []
    for path in discover_all_pl_files(
        prolog_dir=prolog_dir, codebase_db_pl=codebase_db_pl
    ):
        relpath = path.resolve().relative_to(base.resolve()).as_posix()
        if relpath not in known:
            gaps.append(relpath)
    return gaps


def scan_file(path: Path, root: Path | None = None) -> FileScan:
    """Read + mask + clause-split one ``.pl`` file. ``root`` is the base a
    relative display path is computed against (defaults to REPO_ROOT) — an
    injection seam so tests can scan a synthetic corpus under ``tmp_path``
    without touching the real repo tree."""
    base = root if root is not None else REPO_ROOT
    text = path.read_text(encoding="utf-8")
    masked = mask_non_code(text)
    try:
        relpath = path.resolve().relative_to(base.resolve()).as_posix()
    except ValueError:
        relpath = path.name
    line_starts = _line_starts(masked)
    clauses = [
        _parse_clause(relpath, masked, start, end)
        for start, end in _split_clauses(masked)
    ]
    return FileScan(relpath, masked, line_starts, clauses)


# ─────────────────────────────────────────────────────────────────────────
# Check: no-cut
# ─────────────────────────────────────────────────────────────────────────

_CUT_RE = re.compile(r"!")


def check_no_cut(scan: FileScan) -> list[Violation]:
    """ADR-0025's ``no-cut`` rule: every bare ``!`` outside comments/strings."""
    violations = []
    for m in _CUT_RE.finditer(scan.masked):
        pos = m.start()
        line = _line_of(scan.line_starts, pos)
        violations.append(
            Violation(
                scan.relpath,
                line,
                NO_CUT,
                "cut ('!') in a governance file — ADR-0025 bans the cut "
                "(rule meaning must not be order-dependent)",
            )
        )
    return violations


# ─────────────────────────────────────────────────────────────────────────
# Check: no-dynamic-goal
# ─────────────────────────────────────────────────────────────────────────

_CALL_VAR_RE = re.compile(r"\bcall\(\s*([A-Z_][A-Za-z0-9_]*)")
# call(module:Goal) with a lowercase (bound, literal) module atom qualifying
# an unbound Goal variable — the ``:`` module-qualification shifts the
# unbound variable past the first token, which _CALL_VAR_RE alone never
# inspects; a genuinely dynamic module-qualified call must still be flagged.
_CALL_MODULE_VAR_RE = re.compile(
    r"\bcall\(\s*[a-z][A-Za-z0-9_]*\s*:\s*([A-Z_][A-Za-z0-9_]*)"
)
_UNIV_ASSIGN_RE = re.compile(r"\b([A-Z_][A-Za-z0-9_]*)\s*=\.\.")


def _scan_call_var_violations(
    scan: FileScan, pattern: re.Pattern[str], message: str
) -> list[Violation]:
    """Shared body for the two ``call(...)``-on-unbound-variable shapes
    ``check_no_dynamic_goal`` checks — same violation shape, only the regex
    and evidence-message template differ."""
    violations = []
    for m in pattern.finditer(scan.masked):
        var = m.group(1)
        violations.append(
            Violation(
                scan.relpath,
                _line_of(scan.line_starts, m.start()),
                NO_DYNAMIC_GOAL,
                message.format(var=var),
            )
        )
    return violations


def _scan_univ_then_call_violations(scan: FileScan) -> list[Violation]:
    """The ``Var =.. [...]`` then ``call(Var, ...)`` construction-then-call
    shape — a genuinely per-clause scan (the univ assignment and the later
    ``call`` must be correlated within the SAME clause body)."""
    violations = []
    for clause in scan.clauses:
        if clause.body_text is None or clause.body_start is None:
            continue
        for am in _UNIV_ASSIGN_RE.finditer(clause.body_text):
            var = am.group(1)
            call_re = re.compile(rf"\bcall\(\s*{re.escape(var)}\b")
            cm = call_re.search(clause.body_text, am.end())
            if cm:
                pos = clause.body_start + cm.start()
                violations.append(
                    Violation(
                        scan.relpath,
                        _line_of(scan.line_starts, pos),
                        NO_DYNAMIC_GOAL,
                        f"{var} =.. [...] then call({var}, ...) — dynamic "
                        f"predicate construction via univ (ADR-0025)",
                    )
                )
    return violations


def check_no_dynamic_goal(scan: FileScan) -> list[Violation]:
    """ADR-0025's ``no-dynamic-goal`` rule: ``call(Var, ...)`` on an unbound
    variable (bare or module-qualified), plus the ``Var =.. [...]`` then
    ``call(Var, ...)`` shape."""
    return [
        *_scan_call_var_violations(
            scan,
            _CALL_VAR_RE,
            "call({var}, ...) — dynamic goal construction on an unbound "
            "variable (ADR-0025 bans call/N on a non-ground goal)",
        ),
        *_scan_call_var_violations(
            scan,
            _CALL_MODULE_VAR_RE,
            "call(module:{var}, ...) — dynamic goal construction on an "
            "unbound module-qualified variable (ADR-0025 bans call/N on a "
            "non-ground goal)",
        ),
        *_scan_univ_then_call_violations(scan),
    ]


# ─────────────────────────────────────────────────────────────────────────
# Check: no-rule-side-effects
# ─────────────────────────────────────────────────────────────────────────

_SIDE_EFFECT_RE = re.compile(r"\b(assertz|asserta|assert|retractall|retract)\s*\(")


def check_no_rule_side_effects(scan: FileScan) -> list[Violation]:
    """ADR-0025's ``no-rule-side-effects`` rule: ``assert*``/``retract*``
    inside a rule BODY (a ``:- dynamic ...`` declaration is unaffected)."""
    violations = []
    for clause in scan.clauses:
        if clause.body_text is None or clause.body_start is None:
            continue
        for m in _SIDE_EFFECT_RE.finditer(clause.body_text):
            pos = clause.body_start + m.start()
            name = m.group(1)
            head = clause.head_name or "<unknown>"
            violations.append(
                Violation(
                    scan.relpath,
                    _line_of(scan.line_starts, pos),
                    NO_RULE_SIDE_EFFECTS,
                    f"{name}(...) in the body of {head}/? — ADR-0025 bans "
                    f"side effects in governance rule bodies (a ':- dynamic "
                    f"...' declaration is fine; Python-side GROUND-fact "
                    f"assertz is a separate mechanism and untouched by this "
                    f"check)",
                )
            )
    return violations


# ─────────────────────────────────────────────────────────────────────────
# Check: stratification (ADR-0010)
# ─────────────────────────────────────────────────────────────────────────

_CALL_NAME_RE = re.compile(r"\b([a-z][A-Za-z0-9_]*)\s*\(")
_NAF_TOKEN = "\\+"


def _extract_naf_span(body: str, after: int) -> tuple[int, int]:
    """The ``(start, end)`` char span of the goal negated by a '\\+' found
    ending at ``after`` — either a balanced parenthesized goal, or a bare
    term up to the next top-level ',' / ';' / '->' / ')' / end. Returns the
    SPAN (not a re-located substring) so a repeated identical goal text
    elsewhere in the body can never be mistaken for this one."""
    i = after
    n = len(body)
    while i < n and body[i].isspace():
        i += 1
    if i < n and body[i] == "(":
        depth = 0
        j = i
        while j < n:
            if body[j] == "(":
                depth += 1
            elif body[j] == ")":
                depth -= 1
                if depth == 0:
                    return i, j + 1
            j += 1
        return i, n
    depth = 0
    j = i
    while j < n:
        c = body[j]
        if c in "([{":
            depth += 1
        elif c in ")]}":
            if depth == 0:
                break
            depth -= 1
        elif depth == 0 and c in (",", ";"):
            break
        elif depth == 0 and body[j : j + 2] == "->":
            break
        j += 1
    return i, j


_MODULE_DIRECTIVE_RE = re.compile(
    r":-\s*module\(\s*[a-z_][A-Za-z0-9_]*\s*,\s*\[(.*?)\]\s*\)", re.S
)


def _module_exports(masked_text: str) -> set[str]:
    """Predicate NAMES declared in a ``:- module(Name, [ind/arity, ...]).``
    directive (arity dropped — see the module docstring's name-only-graph
    note)."""
    m = _MODULE_DIRECTIVE_RE.search(masked_text)
    if not m:
        return set()
    return {
        name
        for name, _arity in re.findall(r"([a-z_][A-Za-z0-9_]*)\s*/\s*(\d+)", m.group(1))
    }


def _declares_module(masked_text: str) -> bool:
    """True iff FILE opens with SWI's own two-argument module directive
    (``:- module(Name, [ind/arity, ...]).`` — REQ-114-FR2-004 clause (b)'s "declares
    :- module with an export list").

    A module-scoped predicate lives in ITS OWN SWI module namespace instead
    of the shared ``user`` module, so a same-named predicate consulted from
    another file cannot silently redefine it: SWI's ``Redefined static
    procedure`` warning (and the clause loss it announces) only fires within
    ONE shared namespace. This is why the fix for the demonstrated hole is
    module boundary, not merely "a different file" — file separation ALONE
    was demonstrated to still clobber (see ``check_no_redefinition``)."""
    return _MODULE_DIRECTIVE_RE.search(masked_text) is not None


def build_predicate_graph(
    scans: list[FileScan],
) -> dict[str, list[EdgeInfo]]:
    """Build the predicate call graph across ALL given governance scans —
    edges are (name-only, per this module's deliberately conservative
    graph-keying: over-inclusive on arity mismatch is safe for a gate,
    under-inclusive would not be)."""
    edges: dict[str, list[EdgeInfo]] = {}
    for scan in scans:
        for clause in scan.clauses:
            if clause.head_name is None or clause.body_text is None:
                continue
            body = clause.body_text
            body_line = _line_of(scan.line_starts, clause.body_start or 0)
            naf_spans: list[tuple[int, int]] = []
            idx = 0
            while True:
                naf_pos = body.find(_NAF_TOKEN, idx)
                if naf_pos == -1:
                    break
                naf_spans.append(_extract_naf_span(body, naf_pos + len(_NAF_TOKEN)))
                idx = naf_pos + len(_NAF_TOKEN)

            def _under_naf(pos: int, spans: list[tuple[int, int]] = naf_spans) -> bool:
                return any(s <= pos < e for s, e in spans)

            bucket = edges.setdefault(clause.head_name, [])
            for cm in _CALL_NAME_RE.finditer(body):
                to_name = cm.group(1)
                pos = (clause.body_start or 0) + cm.start()
                line = _line_of(scan.line_starts, pos)
                is_naf = _under_naf(cm.start())
                bucket.append(
                    EdgeInfo(
                        to_name, is_naf, scan.relpath, line if is_naf else body_line
                    )
                )
    return edges


def _bfs_root_to_extension_point(
    root: str,
    graph: dict[str, list[EdgeInfo]],
    fallback_origin: tuple[str, int],
) -> Violation | None:
    """BFS from one constitutional-invariant ``root`` predicate, looking for
    a path to ``extension_point/1`` that crosses a ``\\+``. Returns the first
    such violation found (or ``None``) — the state ``(node, naf_used)`` is
    visited at most once, so this always terminates on a finite graph."""
    visited: set[tuple[str, bool]] = set()
    queue: deque[tuple[str, bool, list[str], tuple[str, int] | None]] = deque(
        [(root, False, [root], None)]
    )
    while queue:
        node, naf_used, path, naf_origin = queue.popleft()
        if node == EXTENSION_POINT_NAME and naf_used:
            origin_file, origin_line = naf_origin or fallback_origin
            return Violation(
                origin_file,
                origin_line,
                STRATIFICATION,
                f"{' -> '.join(path)}: constitutional invariant '{root}' "
                f"reaches extension_point/1 via a path crossing '\\+' — "
                f"ADR-0010 forbids this (a lower-layer extension-point "
                f"addition could flip the invariant)",
            )
        state = (node, naf_used)
        if state in visited:
            continue
        visited.add(state)
        for edge in graph.get(node, ()):
            new_naf_used = naf_used or edge.is_naf
            new_origin = naf_origin
            if edge.is_naf and naf_origin is None:
                new_origin = (edge.origin_relpath, edge.origin_line)
            queue.append(
                (edge.to_name, new_naf_used, path + [edge.to_name], new_origin)
            )
    return None


def check_stratification(
    scans: list[FileScan],
) -> list[Violation]:
    """ADR-0010: no predicate exported by ``constitutional.pl`` may reach
    ``extension_point/1`` via a path crossing a ``\\+``. Reports the path.

    Coverage note: as of this writing ``extension_point`` appears in NO real
    governance ``.pl`` file (only in ADRs/spec prose), so this BFS can never
    reach its target against the live corpus and a real-repo PASS here is
    not evidence the mechanism works on real files — only the toy-corpus
    unit tests (``test_check_stratification_*``) exercise the traversal
    itself. This is expected and correct (no violation is possible today),
    not a defect."""
    constitutional_scan = next(
        (s for s in scans if s.relpath.endswith("constitutional.pl")), None
    )
    if constitutional_scan is None:
        return []
    roots = _module_exports(constitutional_scan.masked)
    if not roots:
        return []
    graph = build_predicate_graph(scans)
    fallback_origin = (constitutional_scan.relpath, 1)
    violations: list[Violation] = []
    for root in sorted(roots):
        violation = _bfs_root_to_extension_point(root, graph, fallback_origin)
        if violation is not None:
            violations.append(violation)
    return violations


# ─────────────────────────────────────────────────────────────────────────
# Check: redefinition (REQ-114-FR2-004 clause (b)) — the static half of the two-part
# defence against the live clobbering hole demonstrated 2026-08-19 with
# real swipl (see ``gates/engine.py::GateEngine._expected_rule_count`` for
# the runtime half).
# ─────────────────────────────────────────────────────────────────────────


def check_no_redefinition(scans: list[FileScan]) -> list[Violation]:
    """No two of ``scans`` may define the same predicate head unless at
    least one of them declares ``:- module(Name, [...])``.

    THE HOLE THIS CLOSES. Consulting two module-less ``.pl`` files that both
    define ``violation/4`` does not ADD their clauses together: SWI treats
    the second consult as a REDEFINITION and keeps only the second file's
    clauses, emitting nothing stronger than a Warning
    (``Redefined static procedure violation/4``) — which nothing in this
    project checked before this rule. Reproduced 2026-08-19: consulting
    ``gate_rules.pl`` (5 ``violation/4`` clauses) then a second, module-less
    file with 1 more clause of the same predicate leaves exactly 1 clause
    loaded, and the gate reported a clean PASS over a real, undetected
    violation. Declaring ``:- dynamic violation/4.`` in the first-loaded file
    does NOT prevent this (tested as a variant) — only a module boundary
    does, because a module-scoped predicate lives in its own SWI namespace
    rather than the shared ``user`` module a same-named clause elsewhere
    would redefine.

    DESIGN — keyed by predicate NAME ONLY, not name/arity: the same
    over-inclusive-is-safe-for-a-gate choice :func:`build_predicate_graph`
    above already makes, for the identical reason (a name-only false
    positive is a reviewable false alarm; a name+arity false negative is a
    silently unmechanised hole). Measured against the real governance
    corpus (2026-08-19): four predicate names already repeat by name alone
    across files (``can_modify``, ``constraint_satisfied``, ``escalation_
    path``, ``volk_healthy``) and every one of them already has at least one
    module-scoped owner, so this check is zero-noise on the corpus that
    exists today — see ``test_check_no_redefinition_clean_on_real_
    governance_corpus``.
    """
    owners: dict[str, list[FileScan]] = {}
    for scan in scans:
        heads = {c.head_name for c in scan.clauses if c.head_name is not None}
        for head in heads:
            owners.setdefault(head, []).append(scan)

    violations: list[Violation] = []
    for head, owning_scans in sorted(owners.items()):
        if len(owning_scans) < 2:
            continue
        if any(_declares_module(s.masked) for s in owning_scans):
            continue
        other_files = [s.relpath for s in owning_scans]
        for scan in owning_scans:
            first_pos = min(c.start for c in scan.clauses if c.head_name == head)
            elsewhere = ", ".join(f for f in other_files if f != scan.relpath)
            violations.append(
                Violation(
                    scan.relpath,
                    _line_of(scan.line_starts, first_pos),
                    REDEFINITION,
                    f"{head}/? is also defined in {elsewhere}, and none of "
                    f"the {len(owning_scans)} file(s) declares "
                    f"':- module(Name, [...])' — consulting both into the "
                    f"shared SWI namespace silently REPLACES one file's "
                    f"clauses with the other's ('Redefined static "
                    f"procedure') instead of adding to them (REQ-114-FR2-"
                    f"004)",
                )
            )
    return violations


# ─────────────────────────────────────────────────────────────────────────
# Orchestration
# ─────────────────────────────────────────────────────────────────────────


def lint_file(scan: FileScan) -> list[Violation]:
    """Every per-file check (no-cut, no-dynamic-goal, no-rule-side-effects).
    Stratification and redefinition are cross-file and run separately by the
    caller."""
    violations: list[Violation] = []
    violations.extend(check_no_cut(scan))
    violations.extend(check_no_dynamic_goal(scan))
    violations.extend(check_no_rule_side_effects(scan))
    return sorted(violations, key=lambda v: (v.file, v.line, v.rule))


def _allowlist_line(rule_file: RuleFile) -> str:
    name = Path(rule_file.path).name
    return f"allowlisted: {name} (ADR-0031)"


def _lint_governance_scope(
    base: Path, governance: tuple[RuleFile, ...], report: LintReport
) -> None:
    """Scan + lint every governance rulebase (always gates), plus the
    cross-file stratification and redefinition checks over that same scan
    set."""
    governance_scans = [scan_file(base / rf.path, root=base) for rf in governance]
    for scan in governance_scans:
        violations = lint_file(scan)
        report.all_violations[scan.relpath] = violations
        report.gate_violations.extend(violations)
    strat_violations = check_stratification(governance_scans)
    if strat_violations:
        report.gate_violations.extend(strat_violations)
        for v in strat_violations:
            report.all_violations.setdefault(v.file, []).append(v)
    redefinition_violations = check_no_redefinition(governance_scans)
    if redefinition_violations:
        report.gate_violations.extend(redefinition_violations)
        for v in redefinition_violations:
            report.all_violations.setdefault(v.file, []).append(v)


def _lint_reporting_only_bucket(
    base: Path, files: tuple[RuleFile, ...], report: LintReport
) -> list[str]:
    """Scan a non-gating bucket (application/unclassified) purely for the
    ``--all`` full report — never touches ``gate_violations``. Returns the
    relpaths actually scanned (files that exist), so a caller tracking
    "unclassified by Lane 0" membership doesn't need to re-scan."""
    scanned: list[str] = []
    for rf in files:
        path = base / rf.path
        if path.exists():
            scan = scan_file(path, root=base)
            report.all_violations[scan.relpath] = lint_file(scan)
            scanned.append(scan.relpath)
    return scanned


def _resolve_scope_buckets(
    governance_rulebases: tuple[RuleFile, ...] | None,
    allowlist: tuple[RuleFile, ...] | None,
    application_scope: tuple[RuleFile, ...] | None,
    unclassified_by_lane0: tuple[RuleFile, ...] | None,
) -> tuple[
    tuple[RuleFile, ...],
    tuple[RuleFile, ...],
    tuple[RuleFile, ...],
    tuple[RuleFile, ...],
]:
    """Resolve the four ``None``-means-"use the shipped default" scope
    buckets ``run_lint`` accepts as an injection seam."""
    return (
        (
            governance_rulebases
            if governance_rulebases is not None
            else GOVERNANCE_RULEBASES
        ),
        allowlist if allowlist is not None else ALLOWLIST,
        application_scope if application_scope is not None else APPLICATION_SCOPE,
        (
            unclassified_by_lane0
            if unclassified_by_lane0 is not None
            else UNCLASSIFIED_BY_LANE0
        ),
    )


def run_lint(
    *,
    all_files: bool = False,
    root: Path | None = None,
    governance_rulebases: tuple[RuleFile, ...] | None = None,
    allowlist: tuple[RuleFile, ...] | None = None,
    application_scope: tuple[RuleFile, ...] | None = None,
    unclassified_by_lane0: tuple[RuleFile, ...] | None = None,
    check_classification_gaps: bool = True,
) -> LintReport:
    """Run the meta-lint. Governance scope is always scanned + gates the
    exit code; ``all_files=True`` additionally scans the allowlist,
    application-scope, and unclassified buckets for REPORTING (never gates).

    The ``root``/``*_rulebases`` keyword parameters are an injection seam —
    every one defaults to this module's real, shipped classification, but a
    test can supply a synthetic corpus (its own ``RuleFile`` tuples resolved
    against its own ``root``) without touching the real repo tree.
    ``check_classification_gaps=False`` skips the real-repo completeness scan
    for the same reason (it always walks the real ``PROLOG_DIR``)."""
    base = root if root is not None else REPO_ROOT
    governance, allow, application, unclassified = _resolve_scope_buckets(
        governance_rulebases, allowlist, application_scope, unclassified_by_lane0
    )
    report = LintReport()

    _lint_governance_scope(base, governance, report)

    for rf in allow:
        report.allowlist_lines.append(_allowlist_line(rf))
    if all_files:
        _lint_reporting_only_bucket(base, allow, report)
        _lint_reporting_only_bucket(base, application, report)
        report.unclassified = _lint_reporting_only_bucket(base, unclassified, report)

    report.gate_violations = sorted(
        report.gate_violations, key=lambda v: (v.file, v.line, v.rule)
    )
    if check_classification_gaps:
        report.classification_gaps = classification_gaps()
    return report


def _print_classification_gaps(report: LintReport) -> None:
    if not report.classification_gaps:
        return
    print()
    print(
        f"CLASSIFICATION GAP — {len(report.classification_gaps)} .pl file(s) "
        "under prolog/ are not in ANY of GOVERNANCE_RULEBASES / ALLOWLIST / "
        "APPLICATION_SCOPE / UNCLASSIFIED_BY_LANE0 (a brand-new file):"
    )
    for relpath in report.classification_gaps:
        print(f"  {relpath}")


def _print_all_files_report(report: LintReport) -> None:
    print(
        f"Scanned {len(report.all_violations)} file(s) "
        "(--all: full report, non-gating)."
    )
    if report.unclassified:
        print()
        print(
            f"UNCLASSIFIED BY LANE 0 — {len(report.unclassified)} file(s) not "
            "reviewed for governance scope (see module docstring):"
        )
        for relpath in report.unclassified:
            print(f"  {relpath}")
    print()
    for relpath, violations in sorted(report.all_violations.items()):
        if not violations:
            continue
        print(f"{relpath}: {len(violations)} finding(s)")
        for v in violations:
            print(f"  {v.format()}")


def _print_gate_report(report: LintReport) -> None:
    print(f"Governance scope: {len(GOVERNANCE_RULEBASES)} file(s).")
    if report.gate_violations:
        print(f"FAIL — {len(report.gate_violations)} violation(s):")
        for v in report.gate_violations:
            print(f"  {v.format()}")
    else:
        print("PASS — no construct-ban or stratification violations.")


def _print_report(report: LintReport, *, all_files: bool) -> None:
    print("=" * 70)
    print("PROLOG GOVERNANCE META-LINT (SPEC-033 B17)")
    print("=" * 70)
    for line in report.allowlist_lines:
        print(line)
    _print_classification_gaps(report)
    print()
    if all_files:
        _print_all_files_report(report)
    else:
        _print_gate_report(report)
    print("=" * 70)


def main(argv: list[str] | None = None) -> int:
    """CLI entry point — see the module docstring for the two modes."""
    parser = argparse.ArgumentParser(
        description="Prolog governance-rulebase meta-lint (SPEC-033 B17)"
    )
    parser.add_argument(
        "--all",
        action="store_true",
        help="Lint every .pl file (governance + allowlist + application + "
        "unclassified) for REPORTING; never gates the exit code.",
    )
    parser.add_argument("--json", action="store_true", help="Machine-readable output")
    args = parser.parse_args(argv)

    report = run_lint(all_files=args.all)

    if args.json:
        print(json.dumps(report.to_dict(), indent=2))
    else:
        _print_report(report, all_files=args.all)

    if args.all:
        return 0
    return 0 if report.passed else 1


if __name__ == "__main__":
    sys.exit(main())