scripts.check_prolog_boundary
Reference › scripts › check_prolog_boundary
The Prolog boundary gate. It proves that every Python site that queries Prolog targets a predicate that the shipped .pl rules define, with the right name and arity. It is a large script (about 1,400 lines) that a gate or a human runs; it imports PROLOG_RULES_DIR and related settings from infra_config and the decorator in traces.
Invocation: main takes --source (default src/orchestrata_ultimate), --rules-dir (default PROLOG_RULES_DIR), --json and --list-sites, which prints every discovered site with its status and exits 0. The orphan ratchet, which looks for rules never queried, is a whole-repository invariant and only applies when --source is the default.
The flow has three stages. discover_query_sites parses the Python files and finds query calls, following simple assignments and pass-through helper functions (_find_passthrough_helpers, _resolve_arg), and returns a DiscoveryResult of QuerySite records. build_engine loads the rules directory into a real PrologBridge. _evaluate then checks each site: parse_functor_arity extracts the functor and arity from the query text, _predicate_arities asks the engine which arities exist, and a mismatch becomes a Failure with the nearest existing candidate (_nearest_candidate). Governance functors declared in the rules are checked separately, and _advisory_never_queried lists rules nobody queries. The result is a BoundaryReport.
At a glance
| Source | scripts/check_prolog_boundary.py |
| Lines of code | 1418 |
| Classes | 4 |
| Functions | 27 |
| Imports (internal) | orchestrata_ultimate.core.base.traces, orchestrata_ultimate.core.infra_config, orchestrata_ultimate.governance.prolog_client, orchestrata_ultimate.infra.prolog_bridge, scripts.prolog_meta_lint |
| Imported by | none |
| Tests that import it | none |
Class diagram
Dependency diagram
API
Constants
| Name | Type | Value |
|---|---|---|
WHITELIST |
dict[tuple[str, str], str] |
{('governance/prolog_client.py', 'GovernancePrologClient._run_goals'): _VOLK_FA… |
Classes
QuerySite (dataclass)
Decorators: dataclass(frozen=True) · Line 543
One resolved, governance-shaped Prolog query call site.
| Attribute | Type | Default |
|---|---|---|
relpath |
str |
|
line |
int |
|
functor |
str |
|
arity |
int |
|
module_qualified |
bool |
|
goal_text |
str |
|
whitelist_key |
tuple[str, str] | None |
None |
Failure (dataclass)
Decorators: dataclass(frozen=True) · Line 568
A query site whose functor/arity does not exist in the consulted corpus (or an un-whitelisted dynamic site in a governance-shaped dir).
| Attribute | Type | Default |
|---|---|---|
relpath |
str |
|
line |
int |
|
functor |
str |
|
arity |
int |
|
candidate |
str |
| Method | Signature | Summary |
|---|---|---|
format |
format() -> str |
DiscoveryResult (dataclass)
Decorators: dataclass · Line 583
Everything :func:discover_query_sites found, bucketed.
| Attribute | Type | Default |
|---|---|---|
checked |
list[QuerySite] |
field(default_factory=list) |
whitelisted |
list[tuple[str, int, str]] |
field(default_factory=list) |
unwhitelisted_dynamic |
list[tuple[str, int]] |
field(default_factory=list) |
BoundaryReport (dataclass)
Decorators: dataclass · Line 592
| Attribute | Type | Default |
|---|---|---|
passed |
bool |
|
failures |
list[Failure] |
|
advisory |
list[str] |
|
discovered_sites |
int |
|
checked_sites |
int |
|
whitelist_hits |
list[tuple[str, int, str]] |
| Method | Signature | Summary |
|---|---|---|
to_dict |
to_dict() -> dict[str, Any] |
Functions
| Function | Signature | Summary |
|---|---|---|
_flatten_query_expr |
_flatten_query_expr(node: ast.expr) -> str | None |
Reduce a Constant(str) or JoinedStr node to literal text, with each f-string {...} substitution collapsed to one placeholder character (an f-string placeholder counts as ONE argument… |
_iter_own_scope |
_iter_own_scope(node: ast.AST) -> Iterator[ast.AST] |
Yield every descendant of node WITHOUT crossing into a nested function/class/lambda scope. |
_direct_body_assigns |
_direct_body_assigns(func_node: ast.AST) -> dict[str, ast.expr] |
Collect name = <literal-or-fstring> assignments in func_node's own body — does NOT descend into nested function/class defs, so a nested closure's local variable can't leak into the enclosing… |
_find_passthrough_helpers |
_find_passthrough_helpers(tree: ast.Module) -> dict[str, int] |
Find private helper methods whose OWN body performs a direct Prolog seam call using one of their own bare parameters as the query text — e.g. PolicyEvaluator._execute_with_timeout(self, query)… |
_resolve_arg |
_resolve_arg(arg: ast.expr, assigns: dict[str, ast.expr]) -> str | None |
Resolve a call argument to literal query text: directly if it's a literal/f-string, else via a same-function last-assignment lookup. |
_enclosing_function |
_enclosing_function(node: ast.AST, parents: dict[int, ast.AST]) -> ast.AST | None |
|
_qualified_name |
_qualified_name(node: ast.AST, parents: dict[int, ast.AST]) -> str |
Dotted qualified name of every enclosing class/function around node — e.g. "MultiagentPrologBridge.request_resources" — used as (half of) the WHITELIST key instead of a raw line number. |
_build_parent_map |
_build_parent_map(tree: ast.AST) -> dict[int, ast.AST] |
|
_count_top_level_commas |
_count_top_level_commas(inner: str) -> int |
Count commas at parenthesis/bracket depth 0, outside quotes. |
_find_matching_close |
_find_matching_close(rest: str) -> int | None |
rest starts with '('. |
parse_functor_arity |
parse_functor_arity(text: str) -> tuple[str, int, bool] | None |
Parse a Prolog goal string into (functor, arity, was_module_qualified). |
_in_governance_dir |
_in_governance_dir(relpath: str) -> bool |
|
discover_query_sites |
discover_query_sites(source_root: Path) -> DiscoveryResult |
AST-walk every *.py under source_root for Prolog query seam calls, resolve each to a functor/arity, and bucket into checked / whitelisted / unwhitelisted-dynamic per the module docstring's… |
_included_fragments |
_included_fragments(rules_dir: Path) -> set[str] |
Basenames pulled in by another file's :- include('...').. |
build_engine |
build_engine(rules_dir: Path) -> PrologBridge |
Consult the shipped rules exactly like production (load_governance_rules, the canonical loader) plus every remaining *.pl under rules_dir so cross-domain query sites (agents/, postman/… |
_predicate_arities |
_predicate_arities(bridge: PrologBridge, functor: str) -> list[int] |
|
_known_module_names |
_known_module_names(rules_dir: Path) -> set[str] |
|
_all_predicate_names |
_all_predicate_names(bridge: PrologBridge, rules_dir: Path) -> list[str] |
Enumerate every predicate NAME declared across the consulted modules (for the closest-name fallback). |
_nearest_candidate |
_nearest_candidate(bridge: PrologBridge, rules_dir: Path, functor: str, arities: list[int]) -> str |
|
_declared_governance_functors |
_declared_governance_functors(rules_dir: Path) -> set[str] |
Text-scan (informational only — the pass/fail gate above never uses this) the curated governance rule files for declared predicate indicators: :- dynamic declarations and module export lists. |
_body_reachable_functors |
_body_reachable_functors(rules_dir: Path, entry_functors: set[str]) -> set[str] |
Every predicate NAME reachable from entry_functors by following rule bodies across the whole consulted corpus. |
_advisory_never_queried |
_advisory_never_queried(rules_dir: Path, queried_functors: set[str]) -> list[str] |
Declared governance predicates that NOTHING can reach — not a Python query site, and not another rule's body either. |
_evaluate |
_evaluate(bridge: PrologBridge, rules_dir: Path, discovery: DiscoveryResult, *, whole_repo: bool = True) -> BoundaryReport |
Check every discovered site's functor/arity against bridge's consulted corpus and assemble the final report. |
run_boundary_check traces |
run_boundary_check(source_root: Path, rules_dir: Path) -> BoundaryReport |
Run full discovery + declaration-side existence check. |
_print_report |
_print_report(report: BoundaryReport) -> None |
|
_print_site_listing |
_print_site_listing(report: BoundaryReport, discovery: DiscoveryResult) -> None |
|
main |
main(argv: list[str] | None = None) -> int |
Source
#!/usr/bin/env python3
"""Prolog boundary gate — proves every Python Prolog query site targets a
predicate/arity that actually exists in the consulted rule base.
Phase 1 of ``reports/proposal-wire-prolog-governance.md`` §7: a mechanical
contract between the 42+ hand-written Python query strings across
``src/orchestrata_ultimate`` and the hand-written ``.pl`` declarations they
target. Eight of those sites were found querying predicates that do not
exist at the queried arity/argument order (F2/F3/F4/F6 in the report) —
four of them failing silently OPEN. This gate closes that residue.
Design (binding, from the brief):
* **Query-site discovery is an AST walk, never a regex over source.** A site
is a call whose receiver method is one of the known query seams
(``query``/``query_once``/``query_bool``/``query_one``) on a
PrologBridge/PrologEngine/janus-shaped receiver, extracted either directly
(``bridge.query_bool(f"...")``) or one hop indirect — a local variable
assigned an f-string/literal a few lines earlier, then passed BY NAME to
the seam call, OR passed as a bare (uncalled) method reference into
``asyncio.to_thread``/``loop.run_in_executor`` alongside the query text
(``kernel/policy_evaluator.py`` and ``governance/escalation_handler/``
both use these two shapes for their real query sites — a same-call-site
literal-only rule would miss the very sites this gate exists to catch).
* **Declaration side is the real engine, not a second parser.** The shipped
``.pl`` files are consulted into a real janus-swi engine (starting from
``governance.prolog_client.load_governance_rules``, the exact production
loader, then every remaining ``*.pl`` under ``PROLOG_RULES_DIR`` so
cross-domain sites — e.g. ``agents/prolog_bridge.py`` querying
``multiagent.pl`` — are checked against their OWN real rules instead of
producing false-positive "missing" reports). Existence is decided with
``current_predicate(_:F/A)`` — module-aware, free module variable, so a
predicate declared in ``governance``, ``pam``, ``multiagent``, or the
default ``user`` module all resolve.
* **A site only "counts"** (is checked, can fail, appears in the report) if
it is governance-shaped: its functor exists at ANY arity anywhere in the
consulted corpus, OR its file lives under one of the governance-shaped
directories (``governance/``, ``kernel/``, ``orchestrator/``,
``postman/``, ``agents/``, ``infra/``). Everything else (Neo4j ``.query()`` calls,
Cypher text, unrelated domains) is silently out of scope.
* **Non-literal/dynamic functors** (the query text can't be resolved to a
string at all — e.g. built via a helper function call, not a plain
f-string) that land inside a governance-shaped directory MUST have an
explicit, justified entry in :data:`WHITELIST` below; an un-whitelisted
one is itself a hard failure. Starts empty; every entry present here was
added because a real site could not be statically resolved.
* **Known limitation (documented, not solved, per the brief's own escape
hatch):** F3-class argument-ORDER swaps (same functor, same arity,
different argument semantics — e.g. ``has_permission(Actor, Action,
Resource)`` vs. the shipped ``has_permission(Volk, Resource,
Operation)``) are invisible to an arity-only check. This gate does not
attempt a generic name/position heuristic for that class (too fragile to
be trustworthy) and instead documents it as an accepted gap — see the
ADVISORY section of the report.
* **WHITELIST is keyed by (path, qualified function name), not line
number.** An earlier revision keyed by line number; that drifted TWICE
during this gate's own development (a concurrent fix to
kernel/policy_evaluator.py shifted whitelisted lines by +9, then again
later the same session) — a line-keyed suppression in a gate that runs
on every suite run is a recurring-false-failure generator. The qualified
name (``_qualified_name``: the dotted enclosing class/function, or
``"<module>"`` at module scope) is stable across unrelated edits
(imports, docstrings, blank lines) and only needs updating if the site
is actually renamed or moved to a different function/class.
Exit code: 1 if any governance-shaped site queries a predicate/arity that
does not exist (or is an un-whitelisted dynamic site) in the consulted
corpus; 0 otherwise. The advisory section (shipped-but-never-queried
predicates) never affects the exit code.
CLI:
python scripts/check_prolog_boundary.py # human report
python scripts/check_prolog_boundary.py --json # machine report
python scripts/check_prolog_boundary.py --list-sites # debug listing
python scripts/check_prolog_boundary.py --source <dir> # override tree
"""
from __future__ import annotations
import argparse
import ast
import difflib
import json
import re
import sys
from collections.abc import Iterator
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.infra_config import PROLOG_RULES_DIR
from orchestrata_ultimate.governance.prolog_client import (
GOVERNANCE_RULE_FILES,
load_governance_rules,
)
from orchestrata_ultimate.infra.prolog_bridge import PrologBridge
REPO_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_SOURCE = REPO_ROOT / "src" / "orchestrata_ultimate"
#: Ceiling on declared-but-unreachable governance predicates. Only ever
#: lowered — see the ratchet in :func:`_evaluate`.
#:
#: 2026-08-21 127 -> 79 reachability: 48 of the 127 were reached from a rule
#: body and had been libelled as dead for months
#: 79 -> 76 the constitutional bridge wired constitutional_article/4;
#: postman's permission tables became rule-base reads,
#: wiring agent_permission/2, topic_owner/2, queue_owner/2
#: 76 -> 71 the four authority goals became literals, so
#: send/topic/queue/admin_authority and permission_matches
#: are visible to this gate instead of being built at
#: runtime and reported as never-queried
#: 71 -> 63 deleted 8 dynamic declarations that occurred exactly
#: once in the tree — their own line. No writer, no
#: reader, no test: custom_{file,network,api,db}_policy/3
#: (resources.pl), pii_data/2, bias_report/3,
#: audit_trail_exists/1 (constitutional.pl, each a
#: superseded duplicate of the predicate its article
#: actually reads), ethical_override/2 (ethics.pl)
#: 63 -> 62 human_override_active/3 became load-bearing: a
#: recorded override now crosses every article
#: (Mohamed's ruling), so the fact production had been
#: asserting into a void is finally read
#: 62 -> 54 SessionController now queries can_close_session/1 and
#: can_open_session/1 instead of re-implementing the
#: policy as five Python conditions. The lifecycle chain
#: had been dead at BOTH ends: the rules were queried by
#: nothing and their facts asserted by nothing, because
#: PrologBridge.start_session/end_session — the only
#: asserters — had zero production callers.
ORPHAN_CEILING = 54
# Repo root on sys.path so `scripts` resolves regardless of how this file is
# invoked (direct `python scripts/check_prolog_boundary.py`, `-m`, or a test
# importing it by path) — same pattern as scripts/check_wiring.py.
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from scripts.prolog_meta_lint import ( # noqa: E402
build_predicate_graph,
scan_file,
)
# Attribute names that identify a Prolog query seam. ``query_one`` is
# PrologBridge's name for "first solution"; some governance surfaces
# (governance/escalation_handler/handler.py) call it via
# ``loop.run_in_executor(None, self._prolog.query_one, query)`` — a bare
# method reference, not a direct call — which the indirect-resolution path
# below handles.
SEAM_ATTRS = frozenset({"query", "query_once", "query_bool", "query_one"})
# The directories treated as governance-shaped by location, even when a
# query's functor cannot be resolved to a known-in-corpus name (e.g. F4's
# multi_hop_escalation/2, terminal_authority/1 — declared in NO shipped
# .pl at all). Relative to DEFAULT_SOURCE / --source. The brief named the
# first five; ``infra/`` joined when the PrologBridge seam relocated from
# orchestrator/ to infra/ (K1, L4 -> L1).
GOVERNANCE_SHAPED_DIRS = (
"governance/",
"kernel/",
"orchestrator/",
"postman/",
"agents/",
"infra/",
)
# Non-literal/dynamic query sites that live in a governance-shaped
# directory but cannot be statically resolved to a functor/arity by this
# AST-only gate — each entry is a deliberate, justified exception, not a
# blind spot. Keyed by (POSIX-relative path from the source root, dotted
# qualified name of the enclosing class/function — see _qualified_name),
# NOT by line number: line numbers drift under unrelated edits (a new
# import, a docstring change), which would silently un-whitelist a site
# every time the file changes nearby. An unresolvable dynamic site whose
# (path, qualname) is NOT listed here is itself a hard failure.
_MODULE_QUERY_JUSTIFICATION = (
"MultiagentPrologBridge.{method}: query text built via the "
"_module_query() helper (an ast.Call, not a literal/f-string) — the "
"resolved goal is 'multiagent:{predicate}(...)', outside SPEC-007 "
"governance scope (this bridge's domain is multiagent.pl coordination). "
"This AST-only gate does not unwrap an arbitrary helper-function call "
"without risking false confidence about the actual functor; the site is "
"still discovered (governance-shaped by directory: agents/) so it is "
"accounted for here rather than silently dropped."
)
# The volk-fact sync sites. These assert/retract the `governance:volk_*`
# facts that make `may_execute/2` able to deny (before 2026-08-10 nothing
# asserted them, so its first clause never fired and a permissive second
# clause allowed every agent).
_VOLK_FACT_JUSTIFICATION = "the goal text is built by governance.volk_facts, whose predicate names are a CLOSED, module-level tuple (OPERATIONAL_FACTS + BLOCKING_FACTS) with the `governance:` qualifier baked into every goal — statically knowable, just not literal at the call site. Both engines (PrologBridge here, PrologEngine in prolog_gate.py) run the identical goals, which is why the strings are built in one place rather than written out twice; inlining them to satisfy this gate would reintroduce exactly the duplication that let the two gates drift apart."
WHITELIST: dict[tuple[str, str], str] = {
(
"governance/prolog_client.py",
"GovernancePrologClient._run_goals",
): _VOLK_FACT_JUSTIFICATION,
(
"governance/prolog_gate.py",
"PrologGate._run_volk_goals",
): _VOLK_FACT_JUSTIFICATION,
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.can_register_volk",
): _MODULE_QUERY_JUSTIFICATION.format(
method="can_register_volk", predicate="resources"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.volk_type_valid",
): _MODULE_QUERY_JUSTIFICATION.format(
method="volk_type_valid", predicate="volk_type_valid"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.can_start_volk",
): _MODULE_QUERY_JUSTIFICATION.format(
method="can_start_volk", predicate="can_start_volk"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.can_stop_volk",
): _MODULE_QUERY_JUSTIFICATION.format(
method="can_stop_volk", predicate="can_stop_volk"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.can_force_stop_volk",
): _MODULE_QUERY_JUSTIFICATION.format(
method="can_force_stop_volk", predicate="can_force_stop_volk"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.can_assign_task",
): _MODULE_QUERY_JUSTIFICATION.format(
method="can_assign_task", predicate="can_assign_task"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.find_best_volk_for_task",
): _MODULE_QUERY_JUSTIFICATION.format(
method="find_best_volk_for_task", predicate="find_best_volk_for_task"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.task_assignment_priority",
): _MODULE_QUERY_JUSTIFICATION.format(
method="task_assignment_priority", predicate="task_assignment_priority"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.can_become_leader",
): _MODULE_QUERY_JUSTIFICATION.format(
method="can_become_leader", predicate="can_become_leader"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.consensus_required",
): _MODULE_QUERY_JUSTIFICATION.format(
method="consensus_required", predicate="consensus_required"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.quorum_met",
): _MODULE_QUERY_JUSTIFICATION.format(method="quorum_met", predicate="quorum_met"),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.majority_reached",
): _MODULE_QUERY_JUSTIFICATION.format(
method="majority_reached", predicate="majority_reached"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.voting_complete",
): _MODULE_QUERY_JUSTIFICATION.format(
method="voting_complete", predicate="voting_complete"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.volk_healthy",
): _MODULE_QUERY_JUSTIFICATION.format(
method="volk_healthy", predicate="volk_healthy"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.should_quarantine",
): _MODULE_QUERY_JUSTIFICATION.format(
method="should_quarantine", predicate="should_quarantine"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.health_check_due",
): _MODULE_QUERY_JUSTIFICATION.format(
method="health_check_due", predicate="health_check_due"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.can_restart_child",
): _MODULE_QUERY_JUSTIFICATION.format(
method="can_restart_child", predicate="can_restart_child"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.recovery_strategy",
): _MODULE_QUERY_JUSTIFICATION.format(
method="recovery_strategy", predicate="recovery_strategy"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.can_recover_volk",
): _MODULE_QUERY_JUSTIFICATION.format(
method="can_recover_volk", predicate="can_recover_volk"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.fan_out_decision",
): _MODULE_QUERY_JUSTIFICATION.format(
method="fan_out_decision", predicate="fan_out_decision"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.volk_load_factor",
): _MODULE_QUERY_JUSTIFICATION.format(
method="volk_load_factor", predicate="volk_load_factor"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.least_loaded_volk",
): _MODULE_QUERY_JUSTIFICATION.format(
method="least_loaded_volk", predicate="least_loaded_volk"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.can_acquire_lock",
): _MODULE_QUERY_JUSTIFICATION.format(
method="can_acquire_lock", predicate="can_acquire_lock"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.can_release_lock",
): _MODULE_QUERY_JUSTIFICATION.format(
method="can_release_lock", predicate="can_release_lock"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.can_communicate",
): _MODULE_QUERY_JUSTIFICATION.format(
method="can_communicate", predicate="can_communicate"
),
(
"agents/prolog_bridge.py",
"MultiagentPrologBridge.can_broadcast_to_group",
): _MODULE_QUERY_JUSTIFICATION.format(
method="can_broadcast_to_group", predicate="can_broadcast_to_group"
),
("governance/prolog_client.py", "GovernancePrologClient._query_bool_safe"): (
"GovernancePrologClient._query_bool_safe: a GENERIC passthrough — "
"``query_str`` is the method's own parameter, supplied by every "
"public method on this class (may_execute, tool_allowed, ...). Each "
"of THOSE call sites is itself a literal/f-string and is checked "
"independently; this helper's body is not a second query site."
),
("governance/prolog_client.py", "GovernancePrologClient._query_safe"): (
"GovernancePrologClient._query_safe: same generic-passthrough shape "
"as _query_bool_safe — see that entry."
),
("governance/prolog_gate.py", "PrologGate.may_complete"): (
"PrologGate.may_complete: iterates a locally-built list of "
"(gate_name, goal) tuples (`gate_checks`) — each goal IS a literal "
"f-string, but this gate's per-function resolver only tracks plain "
"`name = <literal>` assignments, not for-loop unpacking over an "
"inline list-of-tuples literal. Manually verified present at the "
"shipped arity (2026-07-21, Phase 1 build): "
"all_acceptance_criteria_met/2, all_outputs_verified/2, "
"no_open_questions/2 — all exist in governance.pl at that arity."
),
("kernel/policy_evaluator.py", "PolicyEvaluator._execute_with_timeout"): (
"PolicyEvaluator._execute_with_timeout: a GENERIC passthrough — "
"``query`` is the method's own parameter. Every call SITE that "
"builds a query string (tool_allowed, policy_permits, "
"check_constraint, escalation_target, ...) is itself a literal "
"f-string and is checked independently at its own call site. Keyed "
"by qualified method name (not line number) precisely because this "
"file's line numbers shifted twice already during Phase-1 "
"development from concurrent P2 edits — see the module docstring's "
"'known limitation' history note."
),
("kernel/policy_evaluator.py", "PolicyEvaluator._execute_sync"): (
"PolicyEvaluator._execute_sync: same generic-passthrough shape as "
"_execute_with_timeout — see that entry."
),
(
"orchestrator/impact/dependency_analyzer.py",
"DependencyAnalyzer._query",
): (
"DependencyAnalyzer._query: ``client`` here is a Neo4j async graph "
"client, not a Prolog bridge — ``client.query(cypher, params)`` "
"matches the seam attribute name ('query') coincidentally, and "
"``cypher`` is a parameter (unresolvable), which is what routes "
"this site here. Confirmed not Prolog-related by reading the call "
"site; this file lives under orchestrator/ (governance-shaped by "
"directory) purely because of the KG impact-analysis package split, "
"not governance."
),
("orchestrator/intent_router.py", "IntentRouter.route"): (
"IntentRouter.route: the goal is built via "
"``_ROUTE_INTENT_QUERY.format(action=..., target=...)`` — a "
"module-level string-template '.format()' call, not an f-string. "
"This gate resolves f-strings and same-function literal "
"assignments but does not additionally implement '.format()'-call "
"template parsing (a real, scoped-out gap — see module docstring). "
"Manually verified present at the shipped arity (2026-07-21): "
"route_intent/3 exists in routing.pl (26 shipped facts)."
),
("postman/governance.py", "PostmanGovernanceManager._query_authority"): (
"PostmanGovernanceManager._query_authority: the four authority "
"families this method exists for — send_authority/3, "
"topic_authority/3, queue_authority/3, admin_authority/2 — are each "
"issued as an f-string LITERAL in this same method and are checked "
"by this gate (they are why `checked sites` went 108 -> 112 on "
"2026-08-21; before that they were built at runtime from a "
"(predicate, args) pair and the whole surface was invisible here). "
"Whitelisted is the final `engine.query(generic_goal)` tail, reached "
"only for a predicate OUTSIDE those four. `_prolog_audit_goal` — the "
"sole production producer of (predicate, args) — can return nothing "
"else, so the tail is unreachable in production; it is kept as the "
"fail-closed backstop that "
"TestQueryPrologReal::test_unknown_predicate_fails_closed drives "
"with a genuinely undefined predicate to prove an existence_error "
"denies rather than raising or allowing."
),
("infra/prolog_bridge.py", "PrologBridge.query"): (
"PrologBridge.query: this IS the generic seam definition itself — "
"``query_str`` is the method's own parameter. Every CALLER of "
"PrologBridge.query() supplies a literal/f-string at ITS OWN call "
"site, which is what this gate actually checks."
),
("infra/prolog_bridge.py", "PrologBridge.query_one"): (
"PrologBridge.query_one: delegates to self.query(query_str) — same "
"generic-seam shape as PrologBridge.query."
),
("infra/prolog_bridge.py", "PrologBridge.query_bool"): (
"PrologBridge.query_bool: delegates to self.query(query_str) — same "
"generic-seam shape as PrologBridge.query."
),
("orchestrator/prolog_worker.py", "PrologWorker._execute"): (
"PrologWorker._execute: the generic janus-swi executor — "
"``req.payload`` is an attribute of the request object dispatched "
"through the worker queue, not a literal at this call site. Every "
"caller that builds ``req.payload`` does so from its own literal/"
"f-string, checked at ITS call site."
),
(
"postman/governance.py",
"PostmanGovernanceManager._evaluate_prolog_goal",
): (
"PostmanGovernanceManager._evaluate_prolog_goal: postman's "
"goal-dispatch table — ``goal = self._build_prolog_goal(predicate, "
"args)`` builds the functor NAME dynamically from a caller-supplied "
"``predicate`` argument by design (one dispatch method serves many "
"postman_governance.pl predicates). This is the one seam the brief "
"explicitly names as a known exception (see module docstring); the "
"predicates it dispatches to are declared in postman_governance.pl "
"and are exercised by tests/postman/test_governance.py."
),
(
"orchestrator/ambiguity_resolver.py",
"AmbiguityResolver._try_prolog_resolution",
): (
"AmbiguityResolver._try_prolog_resolution: queries "
"resolve_ambiguity/3, which no shipped .pl file declares — but the "
"entire AmbiguityResolver class has ZERO production constructors "
"(AST refs: 1 def, 0 uses, 2026-07-21), so this Prolog tier is "
"unreachable dead code. Defining the predicate would fabricate "
"semantics with no consumer to validate them. Owner: ORCH; "
"trigger: the wire-or-delete work item for AmbiguityResolver "
"(REQ-003-US-003-012); expiry: that item's resolution removes this "
"entry either way."
),
# W1-P6 (proof-tree explanations, deliverable 3): explain_may_execute_deny
# / explain_policy_permits_deny (governance/verdict.py) are queried via a
# Callable[[str], bool] passed IN by the caller (query_bool bound method
# or an adapter closure) rather than a literal at the call site itself --
# the actual query text is built from fixed f-string templates against
# volk_operational/1, volk_exists/1, agent_action_blocked/2,
# volk_blacklisted/1 (all declared ``:- dynamic`` in governance.pl and
# independently checked elsewhere in this corpus against real query
# sites). A cross-module AST walk to prove that would be a second,
# fragile parser; these three call sites are the ones that hand the
# callable across the module boundary.
(
"governance/prolog_client.py",
"GovernancePrologClient.may_execute",
): (
"GovernancePrologClient.may_execute: passes self._bridge.query_bool "
"(a bound method, not a call) into "
"governance.verdict.explain_may_execute_deny, which builds its own "
"literal query strings against volk_operational/1, volk_exists/1, "
"agent_action_blocked/2 -- see the W1-P6 note above."
),
(
"governance/prolog_gate.py",
"PrologGate._query_bool",
): (
"PrologGate._query_bool: an adapter (Callable[[str], bool] over "
"PrologEngine.query) passed into explain_may_execute_deny by "
"PrologGate.may_execute; ``goal`` is the bare parameter this "
"adapter forwards, not a literal at this call site -- see the "
"W1-P6 note above."
),
(
"kernel/policy_evaluator.py",
"PolicyEvaluator.evaluate_routing",
): (
"PolicyEvaluator.evaluate_routing: passes self._prolog.query_bool "
"(a bound method, not a call) into "
"governance.verdict.explain_policy_permits_deny, which builds its "
"own literal query strings against volk_exists/1, "
"volk_blacklisted/1, agent_action_blocked/2 -- see the W1-P6 note "
"above."
),
}
_PLACEHOLDER = "" # stands in for one f-string {...} substitution
_MODULE_PREFIX_RE = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*):(?=[A-Za-z_])")
_FUNCTOR_RE = re.compile(r"^([a-z][A-Za-z0-9_]*)")
_MODULE_DECL_RE = re.compile(
r":-\s*module\(([a-z_][a-zA-Z0-9_]*)\s*,\s*\[(.*?)\]\s*\)", re.S
)
_INDICATOR_RE = re.compile(r"\b([a-z_][a-zA-Z0-9_]*)/(\d+)")
_DYNAMIC_RE = re.compile(r":-\s*dynamic\s+([a-z_][a-zA-Z0-9_]*/\d+)")
# `:- include('intent_lexicon.gen.pl').` — quotes may be single or double.
_INCLUDE_RE = re.compile(r""":-\s*include\(\s*['"]([^'"]+)['"]\s*\)""")
# ─────────────────────────────────────────────────────────────────────────
# Data model
# ─────────────────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class QuerySite:
"""One resolved, governance-shaped Prolog query call site."""
relpath: str
line: int
functor: str
arity: int
module_qualified: bool
goal_text: str
# (relpath, qualified enclosing function/class name) WHITELIST key, or
# None outside a governance-shaped directory. A site whose functor
# resolved to a LITERAL (not just a dynamic/unresolvable one) can still
# be a deliberate, justified exception -- e.g. a call chain that is
# dead code with zero production constructors, so defining the
# predicate would fabricate semantics with no consumer to validate them
# (see the ambiguity_resolver.py WHITELIST entry). Consulted as a
# fallback ONLY when the existence check below would otherwise FAIL, so
# a real, independently-verifiable predicate (assertz/1, retract/1, ...)
# is never silently pulled out of the checked/verified set just because
# its enclosing function happens to share a WHITELIST key with an
# unrelated dynamic-site entry.
whitelist_key: tuple[str, str] | None = None
@dataclass(frozen=True)
class Failure:
"""A query site whose functor/arity does not exist in the consulted
corpus (or an un-whitelisted dynamic site in a governance-shaped dir)."""
relpath: str
line: int
functor: str
arity: int
candidate: str
def format(self) -> str:
return f"{self.relpath}:{self.line} → {self.functor}/{self.arity} → {self.candidate}"
@dataclass
class DiscoveryResult:
"""Everything :func:`discover_query_sites` found, bucketed."""
checked: list[QuerySite] = field(default_factory=list)
whitelisted: list[tuple[str, int, str]] = field(default_factory=list)
unwhitelisted_dynamic: list[tuple[str, int]] = field(default_factory=list)
@dataclass
class BoundaryReport:
passed: bool
failures: list[Failure]
advisory: list[str]
discovered_sites: int
checked_sites: int
whitelist_hits: list[tuple[str, int, str]]
def to_dict(self) -> dict[str, Any]:
return {
"passed": self.passed,
"failures": [
{
"file": f.relpath,
"line": f.line,
"functor": f.functor,
"arity": f.arity,
"candidate": f.candidate,
}
for f in self.failures
],
"advisory": self.advisory,
"discovered_sites": self.discovered_sites,
"checked_sites": self.checked_sites,
"whitelist_hits": [
{"file": p, "line": ln, "justification": j}
for p, ln, j in self.whitelist_hits
],
"limitations": [
"F3-class argument-ORDER swaps (same functor/arity, different "
"argument semantics) are not detected — documented gap, see "
"module docstring.",
],
}
# ─────────────────────────────────────────────────────────────────────────
# Discovery — AST walk
# ─────────────────────────────────────────────────────────────────────────
def _flatten_query_expr(node: ast.expr) -> str | None:
"""Reduce a ``Constant(str)`` or ``JoinedStr`` node to literal text, with
each f-string ``{...}`` substitution collapsed to one placeholder
character (an f-string placeholder counts as ONE argument term).
Returns ``None`` for any other shape (not statically resolvable)."""
if isinstance(node, ast.Constant) and isinstance(node.value, str):
return node.value
if isinstance(node, ast.JoinedStr):
parts: list[str] = []
for value in node.values:
if isinstance(value, ast.Constant) and isinstance(value.value, str):
parts.append(value.value)
elif isinstance(value, ast.FormattedValue):
parts.append(_PLACEHOLDER)
else: # pragma: no cover - JoinedStr values are always one of the above
return None
return "".join(parts)
return None
def _iter_own_scope(node: ast.AST) -> Iterator[ast.AST]:
"""Yield every descendant of ``node`` WITHOUT crossing into a nested
function/class/lambda scope. Shared by the local-assignment collector
and the passthrough-helper detector below, both of which must reason
about "this function's own body" only."""
for child in ast.iter_child_nodes(node):
yield child
if isinstance(
child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)
):
continue
yield from _iter_own_scope(child)
def _direct_body_assigns(func_node: ast.AST) -> dict[str, ast.expr]:
"""Collect ``name = <literal-or-fstring>`` assignments in ``func_node``'s
own body — does NOT descend into nested function/class defs, so a
nested closure's local variable can't leak into the enclosing scope's
resolution. Last assignment wins (straight-line code in these files)."""
assigns: dict[str, ast.expr] = {}
for child in _iter_own_scope(func_node):
if (
isinstance(child, ast.Assign)
and len(child.targets) == 1
and isinstance(child.targets[0], ast.Name)
):
text = _flatten_query_expr(child.value)
if text is not None:
assigns[child.targets[0].id] = child.value
return assigns
def _find_passthrough_helpers(tree: ast.Module) -> dict[str, int]:
"""Find private helper methods whose OWN body performs a direct Prolog
seam call using one of their own bare parameters as the query text —
e.g. ``PolicyEvaluator._execute_with_timeout(self, query)`` calling
``asyncio.to_thread(self._prolog.query, query)``. These are the
multi-hop sites the brief's F2 defects live behind: the CALLER builds
the literal f-string and hands it to the helper by name; the helper
itself only ever sees a bare parameter.
Returns ``{helper_method_name: query_param_index}`` where the index
counts positional parameters AFTER ``self``/``cls`` (0 = first).
Discovery then resolves the matching positional argument at each
``self.<helper_method_name>(...)`` call site using that CALLER's own
local assignments — recovering the real functor/arity."""
helpers: dict[str, int] = {}
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
params = [a.arg for a in node.args.args if a.arg not in ("self", "cls")]
if not params:
continue
param_set = set(params)
for child in _iter_own_scope(node):
if not isinstance(child, ast.Call):
continue
if (
isinstance(child.func, ast.Attribute)
and child.func.attr in SEAM_ATTRS
and child.args
and isinstance(child.args[0], ast.Name)
and child.args[0].id in param_set
):
helpers[node.name] = params.index(child.args[0].id)
break
for idx, arg in enumerate(child.args):
if isinstance(arg, ast.Attribute) and arg.attr in SEAM_ATTRS:
for j, other in enumerate(child.args):
if j == idx:
continue
if isinstance(other, ast.Constant) and other.value is None:
continue
if isinstance(other, ast.Name) and other.id in param_set:
helpers[node.name] = params.index(other.id)
break
break
if node.name in helpers:
break
return helpers
def _resolve_arg(arg: ast.expr, assigns: dict[str, ast.expr]) -> str | None:
"""Resolve a call argument to literal query text: directly if it's a
literal/f-string, else via a same-function last-assignment lookup."""
text = _flatten_query_expr(arg)
if text is not None:
return text
if isinstance(arg, ast.Name) and arg.id in assigns:
return _flatten_query_expr(assigns[arg.id])
return None
def _enclosing_function(node: ast.AST, parents: dict[int, ast.AST]) -> ast.AST | None:
cur = node
while id(cur) in parents:
cur = parents[id(cur)]
if isinstance(cur, (ast.FunctionDef, ast.AsyncFunctionDef)):
return cur
return None
def _qualified_name(node: ast.AST, parents: dict[int, ast.AST]) -> str:
"""Dotted qualified name of every enclosing class/function around
``node`` — e.g. ``"MultiagentPrologBridge.request_resources"`` — used
as (half of) the WHITELIST key instead of a raw line number. Stable
against line-number drift from unrelated edits elsewhere in the file
(a new import, a docstring change, a blank line) — a real hazard for a
line-keyed whitelist proven during this gate's own development, when a
concurrent edit to kernel/policy_evaluator.py shifted two whitelisted
lines by +9. Returns ``"<module>"`` if ``node`` has no enclosing
function at all (a call at module top level)."""
parts: list[str] = []
cur = node
while id(cur) in parents:
cur = parents[id(cur)]
if isinstance(cur, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
parts.append(cur.name)
if not parts:
return "<module>"
return ".".join(reversed(parts))
def _build_parent_map(tree: ast.AST) -> dict[int, ast.AST]:
parents: dict[int, ast.AST] = {}
for node in ast.walk(tree):
for child in ast.iter_child_nodes(node):
parents[id(child)] = node
return parents
def _count_top_level_commas(inner: str) -> int:
"""Count commas at parenthesis/bracket depth 0, outside quotes."""
depth = 0
in_quote: str | None = None
count = 0
skip_next = False
for ch in inner:
if skip_next:
skip_next = False
continue
if in_quote:
if ch == "\\":
skip_next = True
elif ch == in_quote:
in_quote = None
continue
if ch in ("'", '"'):
in_quote = ch
elif ch in "([{":
depth += 1
elif ch in ")]}":
depth -= 1
elif ch == "," and depth == 0:
count += 1
return count
def _find_matching_close(rest: str) -> int | None:
"""``rest`` starts with '('. Return the index of the MATCHING ')',
respecting nesting and quotes, or None if unbalanced."""
depth = 0
in_quote: str | None = None
skip_next = False
for i, ch in enumerate(rest):
if skip_next:
skip_next = False
continue
if in_quote:
if ch == "\\":
skip_next = True
elif ch == in_quote:
in_quote = None
continue
if ch in ("'", '"'):
in_quote = ch
continue
if ch == "(":
depth += 1
elif ch == ")":
depth -= 1
if depth == 0:
return i
return None
def parse_functor_arity(text: str) -> tuple[str, int, bool] | None:
"""Parse a Prolog goal string into ``(functor, arity,
was_module_qualified)``. Strips an optional leading ``module:``
qualifier, then a functor name, then counts TOP-LEVEL (paren/quote
respecting) comma-separated arguments. A bare atom (no parens) is arity
0. Returns ``None`` if the text doesn't look like a Prolog goal at all —
deliberately conservative so non-goal text (e.g. Cypher) is treated as
"not a query site" rather than a false match."""
stripped = text.strip()
module_qualified = False
m = _MODULE_PREFIX_RE.match(stripped)
if m:
stripped = stripped[m.end() :]
module_qualified = True
fm = _FUNCTOR_RE.match(stripped)
if not fm:
return None
functor = fm.group(1)
rest = stripped[fm.end() :]
if not rest.startswith("("):
if rest.strip():
return None # trailing non-paren garbage — not a clean goal
return functor, 0, module_qualified
end = _find_matching_close(rest)
if end is None:
return None
trailer = rest[end + 1 :].strip()
if trailer:
return None
inner = rest[1:end]
if not inner.strip():
return functor, 0, module_qualified
arity = _count_top_level_commas(inner) + 1
return functor, arity, module_qualified
def _in_governance_dir(relpath: str) -> bool:
return any(relpath.startswith(prefix) for prefix in GOVERNANCE_SHAPED_DIRS)
def discover_query_sites(source_root: Path) -> DiscoveryResult:
"""AST-walk every ``*.py`` under ``source_root`` for Prolog query seam
calls, resolve each to a functor/arity, and bucket into checked /
whitelisted / unwhitelisted-dynamic per the module docstring's rules."""
result = DiscoveryResult()
for py_file in sorted(source_root.rglob("*.py")):
relpath = py_file.relative_to(source_root).as_posix()
try:
source = py_file.read_text(encoding="utf-8")
tree = ast.parse(source, filename=str(py_file))
except OSError, SyntaxError: # pragma: no cover - unreadable file
continue
parents = _build_parent_map(tree)
passthrough_helpers = _find_passthrough_helpers(tree)
func_assigns_cache: dict[int, dict[str, ast.expr]] = {}
def assigns_for(
call_node: ast.AST,
_cache: dict[int, dict[str, ast.expr]] = func_assigns_cache,
) -> dict[str, ast.expr]:
enclosing = _enclosing_function(call_node, parents)
if enclosing is None:
return {}
key = id(enclosing)
if key not in _cache:
_cache[key] = _direct_body_assigns(enclosing)
return _cache[key]
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
candidate_arg: ast.expr | None = None
lineno = node.lineno
if (
isinstance(node.func, ast.Attribute)
and node.func.attr in SEAM_ATTRS
and node.args
):
candidate_arg = node.args[0]
elif (
isinstance(node.func, ast.Attribute)
and node.func.attr in passthrough_helpers
and len(node.args) > passthrough_helpers[node.func.attr]
):
# Multi-hop: this call passes a (presumably literal, built
# HERE) query string BY NAME into a private helper method
# whose own body performs the actual seam call using a bare
# parameter (see _find_passthrough_helpers) — e.g.
# PolicyEvaluator.tool_allowed building the f-string then
# calling ``self._execute_with_timeout(query)``.
candidate_arg = node.args[passthrough_helpers[node.func.attr]]
else:
for idx, arg in enumerate(node.args):
if isinstance(arg, ast.Attribute) and arg.attr in SEAM_ATTRS:
for j, other in enumerate(node.args):
if j == idx:
continue
if isinstance(other, ast.Constant) and other.value is None:
continue
candidate_arg = other
break
break
if candidate_arg is None:
continue
resolved = _resolve_arg(candidate_arg, assigns_for(node))
if resolved is None:
if _in_governance_dir(relpath):
# Keyed by qualified function name, not line number —
# stable against unrelated edits shifting lines (see
# _qualified_name docstring).
key = (relpath, _qualified_name(node, parents))
if key in WHITELIST:
result.whitelisted.append((relpath, lineno, WHITELIST[key]))
else:
result.unwhitelisted_dynamic.append((relpath, lineno))
continue
parsed = parse_functor_arity(resolved)
if parsed is None:
continue # not shaped like a Prolog goal at all — skip
functor, arity, module_qualified = parsed
whitelist_key = (
(relpath, _qualified_name(node, parents))
if (_in_governance_dir(relpath))
else None
)
result.checked.append(
QuerySite(
relpath,
lineno,
functor,
arity,
module_qualified,
resolved,
whitelist_key,
)
)
return result
# ─────────────────────────────────────────────────────────────────────────
# Declaration side — consult, don't parse
# ─────────────────────────────────────────────────────────────────────────
def _included_fragments(rules_dir: Path) -> set[str]:
"""Basenames pulled in by another file's ``:- include('...').``.
An included file is a FRAGMENT, not a loadable unit: consulting it in its
own right loads its clauses a second time, on top of the copy the including
file already pasted in. SWI reports that as
Warning: Redefined static procedure target_word_raw/2
Warning: Previously defined at .../intent_lexicon.gen.pl:163
which is noise here (a static redefinition REPLACES the predicate rather
than appending to it, and intent_grammar.pl rebuilds its stem index from
scratch on every load) — but it is noise that hides real load errors in the
gate's captured output, which is exactly how the cxp/knowledge ``owns/2``
export collision survived unnoticed. Derived from the sources rather than
hardcoded so a newly generated fragment needs no edit here.
"""
fragments: set[str] = set()
for pl_file in sorted(Path(rules_dir).glob("*.pl")):
text = pl_file.read_text(encoding="utf-8")
fragments.update(Path(name).name for name in _INCLUDE_RE.findall(text))
return fragments
def build_engine(rules_dir: Path) -> PrologBridge:
"""Consult the shipped rules exactly like production
(``load_governance_rules``, the canonical loader) plus every remaining
``*.pl`` under ``rules_dir`` so cross-domain query sites (agents/,
postman/, orchestrator/) are checked against their own real rules
instead of producing false-positive "missing" reports for predicates
the narrow governance-domain loader never consults.
Files that another file ``include``s are skipped — they arrive with their
includer, and consulting them again only duplicates the load."""
bridge = PrologBridge()
load_governance_rules(bridge, rules_dir)
fragments = _included_fragments(rules_dir)
for pl_file in sorted(Path(rules_dir).glob("*.pl")):
if pl_file.name in GOVERNANCE_RULE_FILES or pl_file.suffix != ".pl":
continue
if pl_file.name in fragments:
continue
bridge.load_rules(pl_file)
return bridge
def _predicate_arities(bridge: PrologBridge, functor: str) -> list[int]:
result = bridge.query(f"current_predicate(_:'{functor}'/A)")
arities: set[int] = set()
for sol in result.solutions:
value = sol.get("A")
if isinstance(value, int):
arities.add(value)
return sorted(arities)
def _known_module_names(rules_dir: Path) -> set[str]:
modules = {"user"}
for pl_file in Path(rules_dir).glob("*.pl"):
try:
text = pl_file.read_text(encoding="utf-8")
except OSError: # pragma: no cover - unreadable file
continue
for match in _MODULE_DECL_RE.finditer(text):
modules.add(match.group(1))
return modules
def _all_predicate_names(bridge: PrologBridge, rules_dir: Path) -> list[str]:
"""Enumerate every predicate NAME declared across the consulted
modules (for the closest-name fallback). Only called on a FAIL with
zero same-name arities, so the cost is rare and bounded to the
project's own modules (never ``system``/library builtins)."""
modules = sorted(_known_module_names(rules_dir))
names: set[str] = set()
for module in modules:
result = bridge.query(f"current_predicate({module}:N/_)")
for sol in result.solutions:
value = sol.get("N")
if isinstance(value, str):
names.add(value)
return sorted(names)
def _nearest_candidate(
bridge: PrologBridge, rules_dir: Path, functor: str, arities: list[int]
) -> str:
if arities:
others = ", ".join(f"{functor}/{a}" for a in arities)
return f"same functor shipped at different arity: {others}"
names = _all_predicate_names(bridge, rules_dir)
close = difflib.get_close_matches(functor, names, n=1, cutoff=0.0)
if close:
return f"no '{functor}' at any arity — closest shipped name: '{close[0]}'"
return f"no '{functor}' at any arity found in the consulted rule base"
# ─────────────────────────────────────────────────────────────────────────
# Advisory — shipped-but-never-queried
# ─────────────────────────────────────────────────────────────────────────
def _declared_governance_functors(rules_dir: Path) -> set[str]:
"""Text-scan (informational only — the pass/fail gate above never uses
this) the curated governance rule files for declared predicate
indicators: ``:- dynamic`` declarations and module export lists."""
functors: set[str] = set()
for filename in GOVERNANCE_RULE_FILES:
path = Path(rules_dir) / filename
try:
text = path.read_text(encoding="utf-8")
except OSError: # pragma: no cover - file missing
continue
for match in _DYNAMIC_RE.finditer(text):
name, _arity = match.group(1).split("/")
functors.add(name)
functors.update(
name
for _module_name, block in _MODULE_DECL_RE.findall(text)
for name, _arity in _INDICATOR_RE.findall(block)
)
return functors
def _body_reachable_functors(rules_dir: Path, entry_functors: set[str]) -> set[str]:
"""Every predicate NAME reachable from ``entry_functors`` by following
rule bodies across the whole consulted corpus.
Prolog is a language with calls in it. A helper reached only from another
rule's body is never queried from Python and never will be, yet it runs on
every query that reaches it — so comparing declarations against Python
query sites alone libels live code. Measured 2026-08-21 on the shipped
corpus: 47 of 127 advised names were body-reachable, a 37% false-positive
rate, which is why nobody could act on the list.
The graph comes from :func:`scripts.prolog_meta_lint.build_predicate_graph`
— this repo's existing masked-text scanner — rather than a second parser
here. It is name-only and deliberately OVER-inclusive (its own docstring:
"over-inclusive on arity mismatch is safe for a gate, under-inclusive
would not be"), which is the correct direction for a list whose entries
are candidates for deletion: an extra edge hides an orphan, a missing edge
manufactures one.
Cross-checked against an independent graph built with SWI's own
``read_term/3`` over the same corpus: despite carrying 387 fewer edges, it
produced an IDENTICAL orphan set, in both directions. The one blind spot
both share — ``_CALL_NAME_RE`` needs an open paren, so a 0-arity goal call
(``retract_facts,``) and a meta-call closure (``maplist(param_text, ...)``)
are invisible — was checked by hand against every surviving name and hits
none of them. It DOES hit ``tools/codebase_db.pl``, which is outside the
governance corpus this advisory covers.
``scripts.projection.analyzer`` is the arity-aware alternative and was not
used: it scopes to ONE rulebase file at a time, and raises
``ProjectionUndecidable`` on a dynamic goal. An advisory must span the
corpus (``governance.pl`` is the only caller of half of
``constitutional.pl``) and must never refuse to answer.
"""
scans = [scan_file(path) for path in sorted(Path(rules_dir).glob("*.pl"))]
graph = {
head: {edge.to_name for edge in edges}
for head, edges in build_predicate_graph(scans).items()
}
reachable: set[str] = set()
stack = list(entry_functors)
while stack:
current = stack.pop()
if current in reachable:
continue
reachable.add(current)
stack.extend(graph.get(current, ()))
return reachable
def _advisory_never_queried(rules_dir: Path, queried_functors: set[str]) -> list[str]:
"""Declared governance predicates that NOTHING can reach — not a Python
query site, and not another rule's body either.
A name surviving both filters is a genuine finding: it is shipped,
documented, usually unit-tested, and cannot influence any decision. That
is the shape every governance defect in this subsystem has taken — the
12 constitutional articles were repaired one at a time to be capable of
denial while no caller asked them anything.
"""
declared = _declared_governance_functors(rules_dir)
reachable = _body_reachable_functors(rules_dir, queried_functors)
never_queried = sorted(declared - queried_functors - reachable)
return [
f"{name} — shipped in a curated governance rule file, never queried "
"by any discovered site AND not reached from any rule body "
"(informational; not a failure)"
for name in never_queried
]
# ─────────────────────────────────────────────────────────────────────────
# Orchestration
# ─────────────────────────────────────────────────────────────────────────
def _evaluate(
bridge: PrologBridge,
rules_dir: Path,
discovery: DiscoveryResult,
*,
whole_repo: bool = True,
) -> BoundaryReport:
"""Check every discovered site's functor/arity against ``bridge``'s
consulted corpus and assemble the final report. Shared by
:func:`run_boundary_check` and the CLI so there is exactly one
pass/fail decision path.
``whole_repo`` says whether discovery covered this repo's real source
root. It gates the orphan ratchet only — see the RATCHET block below.
"""
failures: list[Failure] = []
checked_functors: set[str] = set()
checked_count = 0
whitelist_hits = list(discovery.whitelisted)
for site in discovery.checked:
exact = bridge.query_bool(f"current_predicate(_:'{site.functor}'/{site.arity})")
arities = _predicate_arities(bridge, site.functor)
governance_shaped = exact or bool(arities) or _in_governance_dir(site.relpath)
if not governance_shaped:
continue
if exact:
checked_functors.add(site.functor)
checked_count += 1
continue
# A site whose functor resolved to a LITERAL can still be a
# deliberate, justified exception (dead code with zero production
# constructors, etc.) -- fall back to the same WHITELIST a dynamic
# site would use, but ONLY here, on an existence-check FAILURE, so a
# real verifiable predicate (assertz/1, retract/1, ...) that merely
# shares an enclosing function with an unrelated WHITELIST entry is
# never pulled out of the independently-checked set.
if site.whitelist_key is not None and site.whitelist_key in WHITELIST:
whitelist_hits.append(
(site.relpath, site.line, WHITELIST[site.whitelist_key])
)
continue
checked_functors.add(site.functor)
checked_count += 1
candidate = _nearest_candidate(bridge, rules_dir, site.functor, arities)
failures.append(
Failure(site.relpath, site.line, site.functor, site.arity, candidate)
)
failures.extend(
Failure(
relpath,
lineno,
"<dynamic>",
-1,
"non-literal query site in a governance-shaped directory with "
"no WHITELIST justification — add one to scripts/"
"check_prolog_boundary.py or make the query text statically "
"resolvable",
)
for relpath, lineno in discovery.unwhitelisted_dynamic
)
advisory = _advisory_never_queried(rules_dir, checked_functors)
# RATCHET. The advisory was informational for as long as it was
# untrustworthy — 38% of its names were live code reached from a rule
# body, so nobody could act on it and nobody did. It is now measured, so
# it gets a floor: the count may fall, never rise.
#
# A ratchet rather than a per-name waiver list. Seventy-one waivers would
# be seventy-one rubber stamps nobody reads; one number that can only go
# down states the same thing and cannot rot. Shrink ORPHAN_CEILING with
# every batch — that is the point of it.
#
# WHOLE-REPO ONLY. "Unreachable" means no discovered query site AND no
# rule body reaches it, and the first half of that is measured from
# whatever ``--source`` names. Narrow the source to a fixture tree and
# discovery finds ZERO sites, so every predicate in the rulebase looks
# orphaned at once: 158 against a ceiling of 54.
#
# That is not a finding about the tree being checked, it is an artefact
# of asking a whole-repo question with a partial input.
# tests/scripts/test_run_tests.py::test_run_gates_runs_real_static_analysis
# drives exactly that path -- the ``--source`` seam exists so the three
# tree-walking gates can be exercised against a two-file fixture in
# milliseconds -- and it therefore failed on EVERY broad run. Verified
# 2026-08-23 at 143310ec, before any of this session's commits: same
# fixture, same "discovered sites: 0", same 158-vs-54 failure. It could
# never have passed as written.
#
# So the ratchet applies only when discovery covered the real source
# root. The per-site existence checks above still run for a narrowed
# source, which is the property that seam is actually there to test.
if whole_repo and len(advisory) > ORPHAN_CEILING:
failures.append(
Failure(
"src/orchestrata_ultimate/prolog/",
0,
"<orphan-ratchet>",
len(advisory),
f"{len(advisory)} governance predicates are declared but "
f"unreachable — above the ceiling of {ORPHAN_CEILING}. A new "
f"one means a rule was shipped with no caller and no query, "
f"which is how this subsystem accumulated 127 of them. Wire "
f"it, delete it, or lower nothing: the ceiling only moves "
f"down.",
)
)
return BoundaryReport(
passed=not failures,
failures=sorted(failures, key=lambda f: (f.relpath, f.line)),
advisory=advisory,
discovered_sites=len(discovery.checked)
+ len(discovery.whitelisted)
+ len(discovery.unwhitelisted_dynamic),
checked_sites=checked_count,
whitelist_hits=whitelist_hits,
)
@traces(spec="007", story="US-PGOV-BOUNDARY", reqs=["REQ-007-FR10-034"])
def run_boundary_check(source_root: Path, rules_dir: Path) -> BoundaryReport:
"""Run full discovery + declaration-side existence check. This is the
single entry point both the CLI and the three wired gates call."""
bridge = build_engine(rules_dir)
discovery = discover_query_sites(source_root)
# Resolved, so a caller passing the default by a different spelling
# (relative path, symlink, trailing slash) still gets the ratchet.
whole_repo = source_root.resolve() == DEFAULT_SOURCE.resolve()
return _evaluate(bridge, rules_dir, discovery, whole_repo=whole_repo)
# ─────────────────────────────────────────────────────────────────────────
# CLI
# ─────────────────────────────────────────────────────────────────────────
def _print_report(report: BoundaryReport) -> None:
print("=" * 70)
print("PROLOG BOUNDARY GATE")
print("=" * 70)
print(f"discovered sites: {report.discovered_sites}")
print(f"checked sites: {report.checked_sites}")
print(f"whitelisted: {len(report.whitelist_hits)}")
print()
if report.failures:
print(f"FAIL — {len(report.failures)} mismatch(es):")
for failure in report.failures:
print(f" {failure.format()}")
else:
print(
"PASS — every checked site's predicate/arity exists in the "
"consulted rule base."
)
if report.advisory:
print()
print(
f"ADVISORY — {len(report.advisory)} shipped-but-never-queried "
"predicate(s) (does not fail the gate):"
)
for line in report.advisory:
print(f" {line}")
print()
print(
"LIMITATION: F3-class argument-ORDER swaps (same functor/arity, "
"different argument semantics) are not detected by this gate — "
"documented gap, see module docstring."
)
print("=" * 70)
def _print_site_listing(report: BoundaryReport, discovery: DiscoveryResult) -> None:
print("=" * 70)
print("PROLOG BOUNDARY GATE — SITE LISTING")
print("=" * 70)
for site in discovery.checked:
print(f"CHECKED {site.relpath}:{site.line} -> {site.functor}/{site.arity}")
for relpath, lineno, justification in discovery.whitelisted:
print(f"WHITELISTED {relpath}:{lineno} -> {justification}")
for relpath, lineno in discovery.unwhitelisted_dynamic:
print(f"UNWHITELISTED-DYNAMIC {relpath}:{lineno}")
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description="Prolog boundary gate — Python query sites vs. shipped .pl rules"
)
parser.add_argument(
"--source",
default=str(DEFAULT_SOURCE),
help="Source root to walk (default: src/orchestrata_ultimate)",
)
parser.add_argument(
"--rules-dir",
default=None,
help="Prolog rules directory (default: PROLOG_RULES_DIR)",
)
parser.add_argument("--json", action="store_true", help="Machine-readable output")
parser.add_argument(
"--list-sites",
action="store_true",
help="Debug listing of every discovered site (checked/whitelisted/unwhitelisted)",
)
args = parser.parse_args(argv)
source_root = Path(args.source)
rules_dir = Path(args.rules_dir) if args.rules_dir else Path(PROLOG_RULES_DIR)
bridge = build_engine(rules_dir)
discovery = discover_query_sites(source_root)
# Same rule as run_boundary_check: the orphan ratchet is a whole-repo
# invariant and is meaningless when --source narrows discovery. This path
# is the one the gates actually shell out to, so omitting it here would
# leave the fix working only for in-process callers.
report = _evaluate(
bridge,
rules_dir,
discovery,
whole_repo=source_root.resolve() == DEFAULT_SOURCE.resolve(),
)
if args.list_sites:
_print_site_listing(report, discovery)
return 0
if args.json:
print(json.dumps(report.to_dict(), indent=2))
else:
_print_report(report)
return 0 if report.passed else 1
if __name__ == "__main__":
sys.exit(main())