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
Dependency diagram
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
#!/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())