orchestrata_ultimate.factory.conformance.engine

Reference › orchestrata_ultimate › factory › conformance › engine

engine is the AST-to-design conformance engine (SPEC-033 row B2, ADR-0007, ADR-0008). check_conformance takes a repository root and a DesignContract and returns a ConformanceReport. ADR-0008 bounds the engine to structure: it checks that modules, symbols and signatures exist and that cross-component edges are declared, never what the code means. A green report is therefore not a correctness claim; the module's docstring says semantic review is left to humans and to the anchors.

It runs three checks, each returning ConformanceFinding records. _check_module_ownership reports a declared module that is missing from disk and a .py file that sits beside a node's module but belongs to no node. _check_symbols verifies each declared symbol, either name (existence only) or name(sig) (existence plus signature, parsed by _parse_symbol_spec and _expected_signature_key and compared with the signature key of the fast-path classifier), reporting missing symbols and signature mismatches. _check_outbound_edges reports undeclared cross-component outbound edges against the declared_edges mapping, using _declared_dependency_packages to read declared dependencies from pyproject.toml. The report holds the findings, a per-node NodeSummary tally, the contract id, the change reference, the git head from _git_head_sha (or None outside a repository) and a summary string. Each type has to_dict for JSON output. The scripts.factory_conformance CLI imports the module; tests are in tests/factory/test_conformance_engine.py.

At a glance

Class diagram

Dependency diagram

API

Classes

ConformanceFinding (dataclass)

Decorators: dataclass(frozen=True) · Line 122

One structural drift between the contract and the code on disk.

NodeSummary (dataclass)

Decorators: dataclass(frozen=True) · Line 146

Per-node finding tally.

ConformanceReport (dataclass)

Decorators: dataclass(frozen=True) · Line 163

The full result of one :func:check_conformance call.

Functions

Source

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

"""AST≡design conformance engine (SPEC-033 B2, ADR-0007/ADR-0008).

ADR-0008 BOUNDS this engine: it checks STRUCTURE and transcription --
module/symbol/signature existence, and cross-component dependency edges --
never semantic intent. A green :class:`ConformanceReport` is NOT a
correctness claim: a module whose body silently does the wrong thing while
its public signature stays byte-for-byte the one the contract names is, by
design, CONFORMANT (see ``test_conformance_is_structural_not_semantic``).
Semantic review is a human responsibility this engine deliberately does not
attempt (ADR-0008 "Alternatives rejected": deterministic semantic
verification is undecidable in general; the tractable subset is the
anchors, B15a/B15b, not this module).

Checks three axes against a :class:`~orchestrata_ultimate.factory.
contracts.models.DesignContract` (B3):

1. **Module ownership** -- every ``node.modules`` path exists on disk
   (else ``missing_module``). ``DesignContract`` load-time validation
   (:func:`~orchestrata_ultimate.factory.contracts.models.
   validate_contract`) already enforces that no module is claimed by more
   than one node; this engine checks the OTHER direction, the filesystem
   side: a ``.py`` file that lives directly inside a directory some node
   owns a module in, but is not itself declared by ANY node, is
   ``unowned_module`` (``node_id=None`` -- that IS the finding: nobody
   owns it).
2. **Symbols** -- for nodes with a non-empty ``symbols`` tuple, each entry
   names a required top-level (or ``Class.method``) symbol somewhere among
   the node's own ``modules``. The grammar EXTENDS B3's plain names with
   an optional signature spec, entirely inside this module's parsing
   (``factory/contracts/models.py`` is untouched -- ``symbols`` stays a
   plain ``tuple[str, ...]`` there):

       "name"          existence only, no signature check.
       "name(sig)"     existence AND signature match, e.g.
                       "run(x, y=1, *, z)" or "Foo.method(a, b=None)".
                       ``sig`` is compared exactly like B12's classifier
                       compares two real signatures: ordered parameter
                       names, whether each carries a default (defaults
                       COUNT), ``*args``/``**kwargs`` presence, and
                       ``ast.unparse``'d annotations -- drift is
                       ``signature_mismatch``.

   A declared symbol absent from every one of the node's modules is
   ``missing_symbol``. A node with an EMPTY ``symbols`` tuple puts the
   whole module in-contract (no finding at all); a node with a NON-EMPTY
   ``symbols`` tuple additionally flags every OTHER public top-level
   symbol (or public method of a public class) in each of its owned
   modules as ``undeclared_public_symbol`` -- the two are asymmetric on
   purpose, matching the B3 docstring's "empty tuple means the whole
   module is in-contract".
3. **No-new-outbound** -- per node, the top-level packages its owned
   modules import that are neither stdlib, a declared third-party
   dependency (best-effort read of ``pyproject.toml``'s
   ``[project.dependencies]`` at ``repo_root`` -- absent in most test
   fixtures, in which case NOTHING is filtered here and every non-stdlib,
   non-same-node import must be declared), nor a package the SAME node
   owns, are the node's "actual outbound edges". Any actual edge missing
   from ``declared_edges[node_id]`` (default: no edges declared for a
   node not present in the mapping) is ``new_outbound_edge`` -- a
   cross-NODE edge inside the project counts too, that IS the point
   (component boundaries), not just edges leaving the project entirely.

Reuse-vs-mirror decision (recorded per the brief): signature comparison
and public-surface detection IMPORT (do not copy, do not modify)
:mod:`orchestrata_ultimate.factory.fastpath.classifier`'s
``_signature_key``/``_public_surface``/``_import_targets``/
``_own_top_level_package`` helpers -- they are generic (operate on any
``ast.Module``/``ast.FunctionDef``, not on B12's HEAD-vs-BASE diff
specifically), so the two engines -- B12's RELATIVE (HEAD-vs-BASE of the
same file) check and this module's ABSOLUTE (code-vs-contract) check --
can never disagree on what "same signature" or "public surface" means.
No modification was needed in ``factory/fastpath/classifier.py`` to make
this possible.

Where ``codebase_db.pl`` (SPEC-033 §9) could answer an equivalent query in
Prolog, this module still implements in Python per the B2 brief -- a
Prolog round-trip is not required here. A future consolidation pass could
express these as ``owns_module/2``, ``exports_symbol/3``,
``outbound_edge/3`` predicates.

Pure read path: stdlib ``ast`` (plus ``tomllib`` for the best-effort
dependency read, and ``subprocess`` for an optional ``git rev-parse``)
only -- no writes, no graph.
"""

from __future__ import annotations

import ast
import re
import subprocess
import sys
import tomllib
from collections.abc import Mapping
from collections.abc import Set as AbstractSet
from dataclasses import dataclass
from pathlib import Path
from typing import Literal

from orchestrata_ultimate.factory.contracts.models import DesignContract
from orchestrata_ultimate.factory.fastpath.classifier import (
    _import_targets,
    _own_top_level_package,
    _public_surface,
    _signature_key,
    _SignatureKey,
)

FindingKind = Literal[
    "missing_module",
    "unowned_module",
    "missing_symbol",
    "undeclared_public_symbol",
    "signature_mismatch",
    "new_outbound_edge",
]

_SYMBOL_SIG_RE = re.compile(r"^(?P<name>[A-Za-z_][\w.]*)\((?P<sig>.*)\)$", re.DOTALL)
_DEP_NAME_RE = re.compile(r"[<>=!~\[; ]")


@dataclass(frozen=True)
class ConformanceFinding:
    """One structural drift between the contract and the code on disk.

    ``node_id`` is ``None`` exactly for ``unowned_module`` -- the finding
    IS that no node claims the file. ``detail`` always names the exact
    offending symbol/edge/path, never a generic message.
    """

    kind: FindingKind
    node_id: str | None
    path: str
    detail: str

    def to_dict(self) -> dict[str, object]:
        """A plain, JSON-serializable ``dict`` of this finding."""
        return {
            "kind": self.kind,
            "node_id": self.node_id,
            "path": self.path,
            "detail": self.detail,
        }


@dataclass(frozen=True)
class NodeSummary:
    """Per-node finding tally."""

    node_id: str
    finding_count: int
    conformant: bool

    def to_dict(self) -> dict[str, object]:
        """A plain, JSON-serializable ``dict`` of this summary."""
        return {
            "node_id": self.node_id,
            "finding_count": self.finding_count,
            "conformant": self.conformant,
        }


@dataclass(frozen=True)
class ConformanceReport:
    """The full result of one :func:`check_conformance` call.

    ``conformant`` is ``True`` iff ``findings`` is empty. ``summary``
    always carries the ADR-0008 boundary wording ("structure, not
    intent") -- a green report is not, and must never be read as, a
    correctness claim.
    """

    findings: tuple[ConformanceFinding, ...]
    conformant: bool
    node_summaries: tuple[NodeSummary, ...]
    contract_id: str
    change_ref: str
    head_sha: str | None
    summary: str

    def to_dict(self) -> dict[str, object]:
        """A plain, JSON-serializable ``dict`` of this report."""
        return {
            "conformant": self.conformant,
            "summary": self.summary,
            "provenance": {
                "contract_id": self.contract_id,
                "change_ref": self.change_ref,
                "head_sha": self.head_sha,
            },
            "node_summaries": [n.to_dict() for n in self.node_summaries],
            "findings": [f.to_dict() for f in self.findings],
        }


def _parse_symbol_spec(entry: str) -> tuple[str, str | None]:
    """Parse a B2 ``symbols`` grammar entry: ``"name"`` or ``"name(sig)"``.

    Returns ``(name, sig_text)``; ``sig_text`` is ``None`` for a bare name
    (existence check only). The regex is greedy on the signature body so a
    default value that itself contains parens (``"fn(x=Foo())"``) still
    resolves correctly -- the trailing ``\\)$`` anchors to the string's
    actual last character.
    """
    match = _SYMBOL_SIG_RE.match(entry.strip())
    if match is None:
        return entry.strip(), None
    return match.group("name"), match.group("sig")


def _expected_signature_key(name: str, sig_text: str) -> _SignatureKey | None:
    """Parse a contract signature spec into B12's normalized signature key.

    Synthesizes a throwaway ``def <simple-name>(<sig_text>): pass`` and
    parses it with stdlib ``ast``, then hands the resulting
    ``FunctionDef`` to the IMPORTED (not copied) ``_signature_key`` --
    the same normalisation function governs both sides of the comparison,
    the contract's declared signature and the module's actual one, so
    they can never silently disagree. Returns ``None`` if ``sig_text`` is
    not parseable as a parameter list.
    """
    simple_name = name.rsplit(".", 1)[-1]
    try:
        synthetic = ast.parse(f"def {simple_name}({sig_text}): pass")
    except SyntaxError:
        return None
    func = synthetic.body[0]
    if not isinstance(func, (ast.FunctionDef, ast.AsyncFunctionDef)):
        return None
    return _signature_key(func)


def _find_function_node(
    tree: ast.Module, qualified_name: str
) -> ast.FunctionDef | ast.AsyncFunctionDef | None:
    """Locate the actual function/method node for ``name`` or ``Class.method``."""
    if "." in qualified_name:
        class_name, method_name = qualified_name.split(".", 1)
        for node in tree.body:
            if isinstance(node, ast.ClassDef) and node.name == class_name:
                for member in node.body:
                    if (
                        isinstance(member, (ast.FunctionDef, ast.AsyncFunctionDef))
                        and member.name == method_name
                    ):
                        return member
        return None
    for node in tree.body:
        if (
            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
            and node.name == qualified_name
        ):
            return node
    return None


def _declared_dependency_packages(repo_root: Path) -> set[str]:
    """Best-effort top-level import names declared in ``pyproject.toml``.

    Reads ``[project.dependencies]`` at ``repo_root`` if a
    ``pyproject.toml`` exists there, stripping version specifiers/extras
    and normalizing ``-`` to ``_`` as an approximation of the importable
    top-level name. Most tmp_path test fixtures have no ``pyproject.toml``
    at all -- in that case this returns an empty set, and every non-
    stdlib, non-same-node import is treated as an edge that must be
    declared via ``declared_edges``.
    """
    pyproject = repo_root / "pyproject.toml"
    if not pyproject.is_file():
        return set()
    try:
        data = tomllib.loads(pyproject.read_text(encoding="utf-8"))
    except tomllib.TOMLDecodeError, OSError:
        return set()
    deps = data.get("project", {}).get("dependencies", [])
    packages: set[str] = set()
    for dep in deps:
        name = _DEP_NAME_RE.split(dep, maxsplit=1)[0].strip()
        if name:
            packages.add(name.replace("-", "_"))
    return packages


def _git_head_sha(repo_root: Path) -> str | None:
    """``git rev-parse HEAD`` at ``repo_root``, or ``None`` if not a git repo.

    Most conformance test fixtures are plain tmp_path package trees, not
    git repos -- provenance degrades gracefully rather than raising.
    """
    try:
        result = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            cwd=repo_root,
            capture_output=True,
            text=True,
            check=False,
        )
    except OSError:
        return None
    if result.returncode != 0:
        return None
    return result.stdout.strip()


def _check_module_ownership(
    contract: DesignContract, repo_root: Path
) -> list[ConformanceFinding]:
    """Deliverable check 1: every declared module exists; no unowned modules.

    Equivalent Prolog predicate for a future consolidation pass:
    ``owns_module(NodeId, Path)`` plus a ``file_exists/1`` fact base.
    """
    findings: list[ConformanceFinding] = []
    all_modules: dict[str, str] = {}
    for node in contract.nodes:
        for module in node.modules:
            all_modules[module] = node.node_id

    for node in contract.nodes:
        for module in node.modules:
            if not (repo_root / module).is_file():
                findings.append(
                    ConformanceFinding(
                        kind="missing_module",
                        node_id=node.node_id,
                        path=module,
                        detail=(
                            f"node {node.node_id!r} declares module {module!r} "
                            f"which does not exist under {repo_root}"
                        ),
                    )
                )

    owned_dirs = sorted({str(Path(m).parent) for m in all_modules}, key=str)
    for directory in owned_dirs:
        abs_dir = repo_root / directory
        if not abs_dir.is_dir():
            continue
        for entry in sorted(abs_dir.glob("*.py")):
            rel = str(entry.relative_to(repo_root))
            if rel in all_modules:
                continue
            findings.append(
                ConformanceFinding(
                    kind="unowned_module",
                    node_id=None,
                    path=rel,
                    detail=(
                        f"{rel} exists under a node-owned directory but is "
                        "not declared by any node"
                    ),
                )
            )
    return findings


def _check_symbols(
    contract: DesignContract, repo_root: Path
) -> list[ConformanceFinding]:
    """Deliverable check 2: declared symbols exist, match signature, own the surface.

    Equivalent Prolog predicate: ``exports_symbol(NodeId, Path, Name,
    Sig)``.
    """
    findings: list[ConformanceFinding] = []
    for node in contract.nodes:
        if not node.symbols:
            continue
        declared_specs = [_parse_symbol_spec(entry) for entry in node.symbols]
        declared_names = {name for name, _sig in declared_specs}
        found_names: set[str] = set()

        for module in node.modules:
            module_path = repo_root / module
            if not module_path.is_file():
                continue  # already reported as missing_module
            tree = ast.parse(module_path.read_text(encoding="utf-8"), filename=module)
            public = _public_surface(tree)

            for entry_name, sig_text in declared_specs:
                if entry_name not in public:
                    continue
                found_names.add(entry_name)
                if sig_text is None:
                    continue
                actual_node = _find_function_node(tree, entry_name)
                if actual_node is None:
                    continue  # a class name with a signature spec -- not callable
                expected_key = _expected_signature_key(entry_name, sig_text)
                if expected_key is None:
                    continue  # unparsable spec -- defensive no-op, not a false positive
                if _signature_key(actual_node) != expected_key:
                    findings.append(
                        ConformanceFinding(
                            kind="signature_mismatch",
                            node_id=node.node_id,
                            path=module,
                            detail=(
                                f"{module}:{entry_name} signature does not match "
                                f"contract spec {entry_name}({sig_text})"
                            ),
                        )
                    )

            for name in sorted(public):
                if name not in declared_names:
                    findings.append(
                        ConformanceFinding(
                            kind="undeclared_public_symbol",
                            node_id=node.node_id,
                            path=module,
                            detail=(
                                f"{module}:{name} is public but not declared in "
                                f"node {node.node_id!r}'s symbols"
                            ),
                        )
                    )

        for entry_name, _sig in declared_specs:
            if entry_name not in found_names:
                missing_path = node.modules[0] if node.modules else ""
                findings.append(
                    ConformanceFinding(
                        kind="missing_symbol",
                        node_id=node.node_id,
                        path=missing_path,
                        detail=(
                            f"node {node.node_id!r} declares symbol {entry_name!r} "
                            "which was not found in any of its modules"
                        ),
                    )
                )
    return findings


def _check_outbound_edges(
    contract: DesignContract,
    repo_root: Path,
    declared_edges: Mapping[str, AbstractSet[str]],
) -> list[ConformanceFinding]:
    """Deliverable check 3: no undeclared cross-component outbound edges.

    Equivalent Prolog predicate: ``outbound_edge(NodeId, TopPackage)``
    compared against a ``declared_edge/2`` fact base.
    """
    findings: list[ConformanceFinding] = []
    stdlib_names = sys.stdlib_module_names
    declared_deps = _declared_dependency_packages(repo_root)

    for node in contract.nodes:
        own_packages = {_own_top_level_package(m) for m in node.modules}
        edge_sources: dict[str, str] = {}
        for module in node.modules:
            module_path = repo_root / module
            if not module_path.is_file():
                continue  # already reported as missing_module
            tree = ast.parse(module_path.read_text(encoding="utf-8"), filename=module)
            own_package = _own_top_level_package(module)
            for _target, top_package in _import_targets(tree, own_package).items():
                if not top_package:
                    continue
                if top_package in stdlib_names:
                    continue
                if top_package in declared_deps:
                    continue
                if top_package in own_packages:
                    continue
                edge_sources.setdefault(top_package, module)

        allowed = set(declared_edges.get(node.node_id, set()))
        findings.extend(
            ConformanceFinding(
                kind="new_outbound_edge",
                node_id=node.node_id,
                path=edge_sources[edge],
                detail=(
                    f"{edge_sources[edge]} imports {edge!r}, an undeclared "
                    f"outbound edge from node {node.node_id!r}"
                ),
            )
            for edge in sorted(set(edge_sources) - allowed)
        )
    return findings


def check_conformance(
    repo_root: Path,
    contract: DesignContract,
    *,
    declared_edges: Mapping[str, AbstractSet[str]] | None = None,
) -> ConformanceReport:
    """Check ``contract`` against the code tree at ``repo_root``.

    Pure read path -- never writes. Runs, independently, the three
    deliverable-1 checks (module ownership, symbols, no-new-outbound) and
    concatenates their findings; a change is ``conformant`` iff none of
    the three produced anything.

    ADR-0008: a ``conformant=True`` result is a STRUCTURAL claim only
    (existence, signatures, dependency edges) -- see this module's
    docstring and :data:`ConformanceReport.summary`. It is never evidence
    of semantic correctness.
    """
    declared_edges = declared_edges or {}
    findings = (
        _check_module_ownership(contract, repo_root)
        + _check_symbols(contract, repo_root)
        + _check_outbound_edges(contract, repo_root, declared_edges)
    )

    node_summaries = tuple(
        NodeSummary(
            node_id=node.node_id,
            finding_count=sum(1 for f in findings if f.node_id == node.node_id),
            conformant=not any(f.node_id == node.node_id for f in findings),
        )
        for node in contract.nodes
    )
    conformant = not findings
    summary = (
        "Conformance check is STRUCTURAL, not semantic (ADR-0008: structure, "
        f"not intent) -- {len(findings)} finding(s) across {len(contract.nodes)} "
        f"node(s); conformant={conformant}."
    )
    return ConformanceReport(
        findings=tuple(findings),
        conformant=conformant,
        node_summaries=node_summaries,
        contract_id=contract.contract_id,
        change_ref=contract.change_ref,
        head_sha=_git_head_sha(repo_root),
        summary=summary,
    )


__all__ = [
    "ConformanceFinding",
    "ConformanceReport",
    "FindingKind",
    "NodeSummary",
    "check_conformance",
]