orchestrata_ultimate.factory.projection.analyzer

Reference › orchestrata_ultimate › factory › projection › analyzer

analyzer derives, by static analysis, which predicates a Prolog governance rulebase needs (SPEC-033 row B5, ADR-0006, ADR-0025). Negation as failure is only sound under a closed world, and a hand-picked projection of facts is an open world: an under-selected projection makes a negated goal succeed because the fact was never present. The fix is to derive the required set from the rule text, transitively through the call graph, instead of trusting a manual declaration.

required_predicates scans a rulebase with scan_file from scripts.prolog_meta_lint (the shared comment masking and clause splitting), then builds an arity-aware call graph (_build_graph, _extract_goal_calls, _arity_at_paren), takes the transitive closure from the given root predicates (_closure) and partitions it into a RequiredSet of intensional, extensional and builtins sets of PredId (name/arity) values. Any name( shape in a clause body counts as a call, which over-approximates the set; this is the safe direction for a gate. The call graph is only sound if it is decidable, so _check_undecidable raises ProjectionUndecidable on constructs that ADR-0025 bans in the rulebase, such as call/N on a non-ground goal, assert or retract in a rule body, and goal construction with =... verify_declaration cross-checks an optional manual %! requires: header against the derived set. completeness and the scripts.factory_projection CLI import it; tests are in tests/factory/test_projection_analyzer.py.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

ProjectionUndecidable

Bases: Exception · Line 66

Raised when a rule body contains an ADR-0025-banned construct.

PredId (dataclass)

Decorators: dataclass(frozen=True, slots=True) · Line 78

One Prolog predicate identity: name/arity.

RequiredSet (dataclass)

Decorators: dataclass(frozen=True, slots=True) · Line 106

The result of :func:required_predicates: a total partition of every predicate reachable (transitively) from the given roots.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/factory/projection/analyzer.py

"""Rule-driven projection: derived required-predicates (B5, ADR-0006/0025).

ADR-0006's thesis: Prolog's negation-as-failure is only sound under a
**closed world**. A hand-picked "relevant" projection is an *open* world
unless it is provably complete for every predicate the rules consult -- an
under-selected projection makes NAF silently report success (the fact was
never there to fail on), the worst failure class in the system (a silent
false-green, ADR-0006 calls it out by name). The fix: **derive** the
required-predicate set from the rule ASTs by static analysis, transitively
through the call graph, rather than trust a hand-maintained declaration
(which can under-declare and reintroduce the exact same hole).

That static analysis is only sound if the call graph is **decidable** --
which is what ADR-0025 buys by banning, in the governance rulebase: `call/N`
on a non-ground goal, `assert`/`retract` in a rule body, and goal
construction via `=../2`. This module does not silently mis-handle those
constructs (a mis-handled dynamic goal would silently under-project, same
failure class ADR-0006 exists to close) -- it raises
:class:`ProjectionUndecidable` instead.

Parsing layer: REUSED, not reimplemented. ``scan_file``/``FileScan``/
``ClauseInfo`` (comment/quote masking, clause splitting, head/body
extraction) come from ``scripts.prolog_meta_lint`` (task B17), which already
parses governance rulebases into clause structures. This module builds NEW
logic on top of that shared layer for the B5-specific concerns the meta-lint
does not need: per-goal ARITY (meta-lint's own stratification graph is
deliberately name-only -- "over-inclusive on arity mismatch is safe for a
gate" -- but a completeness check needs exact pred/arity to name what is
missing) and the derived/manual cross-check. Reusing rather than
reimplementing the parser keeps exactly one masking/clause-splitting
algorithm in the codebase; forking it here would let the two drift.

The goal-call extraction below shares meta-lint's own conservative
simplification: any ``name(`` shape in a clause body is treated as a
predicate-call edge, including one that is actually a compound DATA term
(e.g. ``Context = access(Type, Res, Mode)`` -- ``access`` is data, not a
call). This over-approximates the required set. Per ADR-0006 that is the
SAFE direction (a spurious extra requirement can only make completeness
stricter, never let an under-selected projection through), so it is kept
rather than attempting full goal-vs-data disambiguation, which needs a real
Prolog parser this module deliberately does not build.
"""

from __future__ import annotations

import re
from collections import deque
from collections.abc import Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from scripts.prolog_meta_lint import FileScan

# NOTE: `scan_file`/`FileScan` live in scripts/prolog_meta_lint.py and are
# imported LAZILY inside the functions that need them -- a module-level
# `from scripts...` makes this src module unimportable unless the repo root
# is on sys.path (it is not under a plain `PYTHONPATH=<repo>/src`), coupling
# a src package to a top-level script at import time. The lazy import
# confines that coupling to the two call sites that actually need it
# (mirrors `factory/amendment/syntactic.py` and
# `factory/contracts/trace_validation.py`'s existing precedent).


class ProjectionUndecidable(Exception):
    """Raised when a rule body contains an ADR-0025-banned construct.

    Non-ground ``call/N``, ``assert``/``retract`` in a rule body, or goal
    construction via ``=../2`` make the predicate call graph impossible to
    determine by static analysis alone -- guessing here would silently
    under-project (ADR-0006's exact failure mode), so this is raised
    instead of guessed past.
    """


@dataclass(frozen=True, slots=True)
class PredId:
    """One Prolog predicate identity: ``name/arity``.

    ``name`` carries a module qualifier verbatim when the call site used one
    (e.g. ``"constitutional:requires_human_consent"``) -- a projection
    caller needs to know exactly which module's predicate to assert facts
    for (ADR-0005), not just a bare local name.
    """

    name: str
    arity: int

    def __str__(self) -> str:
        return f"{self.name}/{self.arity}"

    @classmethod
    def parse(cls, spec: str) -> PredId:
        """Parse a ``"name/arity"`` string (also accepts ``module:name/arity``)."""
        name, sep, arity_text = spec.rpartition("/")
        if not sep or not name or not arity_text.isdigit():
            raise ValueError(
                f"invalid predicate spec {spec!r} -- expected 'name/arity' "
                f"(e.g. 'foo/2' or 'module:foo/2')"
            )
        return cls(name, int(arity_text))


@dataclass(frozen=True, slots=True)
class RequiredSet:
    """The result of :func:`required_predicates`: a total partition of every
    predicate reachable (transitively) from the given roots.

    - ``intensional``: defined by a clause head in this rulebase.
    - ``extensional``: consulted but not defined here -- these MUST come
      from the projection (ADR-0006's completeness obligation).
    - ``builtins``: matched the ISO/janus builtin allowlist.

    Every predicate found by the call-graph walk lands in exactly one of
    the three sets -- nothing is silently dropped (mirrors the meta-lint's
    own "classification gap" discipline: an unclassified functor is a real
    gap to report, not a pass-by-omission).
    """

    intensional: frozenset[PredId]
    extensional: frozenset[PredId]
    builtins: frozenset[PredId]


# ISO + SWI-Prolog + janus builtins that may appear in a governance rule
# body. Matched by NAME ONLY (arity varies across builtins, e.g. format/1,
# format/2, format/3) -- an explicit, visible constant per the brief, never
# an implicit "anything not user-defined is fine" fallback.
BUILTIN_NAMES: frozenset[str] = frozenset(
    {
        "true",
        "fail",
        "false",
        "is",
        "=",
        "\\=",
        "==",
        "\\==",
        "@<",
        "@>",
        "@=<",
        "@>=",
        "<",
        ">",
        "=<",
        ">=",
        "=:=",
        "=\\=",
        "var",
        "nonvar",
        "atom",
        "atomic",
        "number",
        "integer",
        "float",
        "compound",
        "callable",
        "is_list",
        "ground",
        "functor",
        "arg",
        "copy_term",
        "findall",
        "forall",
        "bagof",
        "setof",
        "aggregate_all",
        "member",
        "memberchk",
        "append",
        "length",
        "reverse",
        "last",
        "nth0",
        "nth1",
        "sort",
        "msort",
        "predsort",
        "list_to_set",
        "sum_list",
        "max_list",
        "min_list",
        "maplist",
        "foldl",
        "include",
        "exclude",
        "partition",
        "atom_concat",
        "atomic_list_concat",
        "string_concat",
        "split_string",
        "sub_atom",
        "upcase_atom",
        "downcase_atom",
        "atom_string",
        "atom_number",
        "number_string",
        "string_to_atom",
        "string_chars",
        "atom_chars",
        "atom_codes",
        "char_code",
        "format",
        "writeln",
        "write",
        "print",
        "nl",
        "tab",
        "print_message",
        "catch",
        "throw",
        "assertion",
        "must_be",
        "between",
        "succ",
        "plus",
        "abs",
        "nb_getval",
        "nb_setval",
        "b_getval",
        "b_setval",
        "not",
        "once",
        "ignore",
        "tabled",
        "dynamic",
    }
)

# ── ADR-0025 banned-construct detection (independent of, and cross-checked
# by, B17's meta-lint gate -- see module docstring) ─────────────────────
_CALL_VAR_RE = re.compile(r"\bcall\(\s*([A-Z_][A-Za-z0-9_]*)")
_CALL_MODULE_VAR_RE = re.compile(
    r"\bcall\(\s*[a-z][A-Za-z0-9_]*\s*:\s*([A-Z_][A-Za-z0-9_]*)"
)
_SIDE_EFFECT_RE = re.compile(r"\b(assertz|asserta|assert|retractall|retract)\s*\(")
_UNIV_RE = re.compile(r"=\.\.")

# ── goal-call + head extraction over already-masked text ───────────────
_GOAL_CALL_RE = re.compile(r"\b((?:[a-z][A-Za-z0-9_]*:)?[a-z][A-Za-z0-9_]*)\s*\(")
_HEAD_RE = re.compile(r"\s*(?:[a-z_][A-Za-z0-9_]*:)?([a-z][A-Za-z0-9_]*)")

# ``%! requires: pred/arity, ...`` manual cross-check header (a COMMENT --
# read from the RAW file text, never the masked text, since mask_non_code
# blanks comment content).
_REQUIRES_HEADER_RE = re.compile(r"%!\s*requires:\s*(?P<preds>.+)")


def _matching_close(text: str, open_idx: int) -> int:
    """Index of the ``)``/``]``/``}`` matching the bracket at ``text[open_idx]``,
    or ``len(text)`` if unterminated."""
    depth = 0
    n = len(text)
    i = open_idx
    while i < n:
        c = text[i]
        if c in "([{":
            depth += 1
        elif c in ")]}":
            depth -= 1
            if depth == 0:
                return i
        i += 1
    return n


def _top_level_split(text: str) -> list[str]:
    """Split ``text`` on top-level commas (depth 0 across ``()``/``[]``/``{}``)."""
    parts: list[str] = []
    depth = 0
    start = 0
    for i, c in enumerate(text):
        if c in "([{":
            depth += 1
        elif c in ")]}":
            depth -= 1
        elif c == "," and depth == 0:
            parts.append(text[start:i])
            start = i + 1
    parts.append(text[start:])
    return parts


def _arity_at_paren(text: str, open_idx: int) -> int:
    """Arity of the argument list opening at ``text[open_idx] == "("``."""
    close_idx = _matching_close(text, open_idx)
    inner = text[open_idx + 1 : close_idx]
    if not inner.strip():
        return 0
    return len(_top_level_split(inner))


def _extract_goal_calls(body_text: str) -> list[PredId]:
    """Every ``name(...)``-shaped call in a (masked) clause body, with arity.

    Scans the WHOLE body regardless of ``\\+``/``findall``/``forall``
    wrapping -- a goal consulted only under negation-as-failure, or only as
    a ``findall/3``/``forall/2`` inner goal, is still a REQUIRED read (the
    exact ADR-0006 false-green scenario: a NAF-negated goal that is silently
    dropped from the required set would let an under-projected fact make
    the negation vacuously succeed)."""
    calls: list[PredId] = []
    for m in _GOAL_CALL_RE.finditer(body_text):
        name = m.group(1)
        open_idx = m.end() - 1
        calls.append(PredId(name, _arity_at_paren(body_text, open_idx)))
    return calls


def _head_arity(masked: str, clause_start: int) -> int:
    """Arity of the clause head starting at ``masked[clause_start]``."""
    m = _HEAD_RE.match(masked, clause_start)
    if not m:
        return 0
    pos = m.end()
    n = len(masked)
    while pos < n and masked[pos].isspace():
        pos += 1
    if pos < n and masked[pos] == "(":
        return _arity_at_paren(masked, pos)
    return 0


def _check_undecidable(scan: FileScan) -> None:
    """Raise :class:`ProjectionUndecidable` on any ADR-0025 banned construct
    in a rule body: non-ground ``call/N`` (bare or module-qualified),
    ``assert*``/``retract*``, or ``=../2`` goal construction."""
    for clause in scan.clauses:
        if clause.body_text is None:
            continue
        body = clause.body_text
        head = clause.head_name or "<directive>"
        m = _CALL_VAR_RE.search(body) or _CALL_MODULE_VAR_RE.search(body)
        if m:
            raise ProjectionUndecidable(
                f"{scan.relpath}: non-ground call({m.group(1)}, ...) in the "
                f"body of {head}/? -- ADR-0025 bans call/N on a non-ground "
                f"goal; the call graph is not statically decidable"
            )
        m = _SIDE_EFFECT_RE.search(body)
        if m:
            raise ProjectionUndecidable(
                f"{scan.relpath}: {m.group(1)}(...) in the body of {head}/? "
                f"-- ADR-0025 bans assert*/retract* in a governance rule "
                f"body"
            )
        if _UNIV_RE.search(body):
            raise ProjectionUndecidable(
                f"{scan.relpath}: '=..' (univ) in the body of {head}/? -- "
                f"ADR-0025 bans dynamic goal construction via univ"
            )


def _build_graph(scan: FileScan) -> tuple[dict[PredId, set[PredId]], set[PredId]]:
    """Build the (arity-aware) predicate call graph for one rulebase scan.

    Returns ``(graph, defined)`` where ``graph[head] = {called, ...}`` and
    ``defined`` is every predicate with at least one clause head in this
    file (the "intensional" candidates)."""
    graph: dict[PredId, set[PredId]] = {}
    defined: set[PredId] = set()
    for clause in scan.clauses:
        if clause.head_name is None:
            continue  # a bare directive (":- module(...)", ":- dynamic ...")
        head_id = PredId(clause.head_name, _head_arity(scan.masked, clause.start))
        defined.add(head_id)
        if clause.body_text is None:
            continue
        graph.setdefault(head_id, set()).update(_extract_goal_calls(clause.body_text))
    return graph, defined


def _closure(roots: Sequence[PredId], graph: dict[PredId, set[PredId]]) -> set[PredId]:
    """BFS transitive closure from ``roots`` over ``graph`` (terminates: the
    graph is finite and each node is visited at most once)."""
    seen: set[PredId] = set()
    queue: deque[PredId] = deque(roots)
    while queue:
        node = queue.popleft()
        if node in seen:
            continue
        seen.add(node)
        queue.extend(n for n in graph.get(node, ()) if n not in seen)
    return seen


def _is_builtin(pred: PredId) -> bool:
    """Match by bare name (module qualifier stripped), arity-agnostic."""
    bare = pred.name.rsplit(":", 1)[-1]
    return bare in BUILTIN_NAMES


def _partition(
    closure: set[PredId], defined: set[PredId]
) -> tuple[set[PredId], set[PredId], set[PredId]]:
    intensional = {p for p in closure if p in defined}
    builtins = {p for p in closure if p not in defined and _is_builtin(p)}
    extensional = closure - intensional - builtins
    return intensional, extensional, builtins


def _scan_rulebase(rulebase_path: Path) -> FileScan:
    from scripts.prolog_meta_lint import scan_file

    return scan_file(rulebase_path)


def required_predicates(rulebase_path: Path, roots: Sequence[str]) -> RequiredSet:
    """Derive the required-predicate set for ``rulebase_path``, transitively
    closed from ``roots`` (``"name/arity"`` strings, e.g. ``"layered_verdict/3"``).

    Raises :class:`ProjectionUndecidable` if any rule body in the file
    contains an ADR-0025-banned construct -- the call graph is then not
    statically decidable and guessing would silently under-project.
    """
    scan = _scan_rulebase(rulebase_path)
    _check_undecidable(scan)
    graph, defined = _build_graph(scan)
    root_ids = [PredId.parse(r) for r in roots]
    closure = _closure(root_ids, graph) | set(root_ids)
    intensional, extensional, builtins = _partition(closure, defined)
    return RequiredSet(
        frozenset(intensional), frozenset(extensional), frozenset(builtins)
    )


def _parse_requires_header(raw_text: str) -> set[PredId]:
    declared: set[PredId] = set()
    for m in _REQUIRES_HEADER_RE.finditer(raw_text):
        for spec in m.group("preds").split(","):
            spec = spec.strip()
            if not spec:
                continue
            try:
                declared.add(PredId.parse(spec))
            except ValueError:
                continue
    return declared


def verify_declaration(rulebase_path: Path) -> list[str]:
    """Cross-check an optional manual ``%! requires: pred/arity, ...``
    header comment against the DERIVED extensional set -- a cross-check
    only, never authoritative (ADR-0006: a hand declaration can
    under-declare and reintroduce the hole one level up).

    Design note: with no caller-supplied root, the derived side is computed
    as the closure over EVERY predicate this file itself defines (each
    clause head is its own root) -- the widest reasonable default for a
    whole-file header, and the only one that needs no extra parameter per
    the brief's stated signature.

    Returns one message per mismatch, in both directions:
    - "under-declared": required (derived) but missing from the header.
    - "over-declared": in the header but not actually required.
    The derived set (:func:`required_predicates`) is unaffected either way.
    """
    scan = _scan_rulebase(rulebase_path)
    _check_undecidable(scan)
    graph, defined = _build_graph(scan)
    closure = _closure(list(defined), graph) | defined
    _intensional, extensional, _builtins = _partition(closure, defined)

    raw_text = rulebase_path.read_text(encoding="utf-8")
    declared = _parse_requires_header(raw_text)

    mismatches: list[str] = [
        f"under-declared: {missing} is required but absent from the "
        f"'%! requires:' header"
        for missing in sorted(extensional - declared, key=str)
    ]
    mismatches.extend(
        f"over-declared: {extra} is in the '%! requires:' header but "
        f"not actually required"
        for extra in sorted(declared - extensional, key=str)
    )
    return mismatches