orchestrata_ultimate.factory.contracts.trace_validation

Reference › orchestrata_ultimate › factory › contracts › trace_validation

trace_validation checks @traces claims structurally (SPEC-033 row B3, ADR-0007, ADR-0008). Because the @traces decorator is written by the model being governed, the validator never reads it. validate_traces takes a DesignContract, a list of changed files and a repository root, and returns a TraceVerdict with a passed flag and a list of TraceViolation records. Two clauses must hold. First, every changed .py file must be claimed by a design node (_unit_in_contract_violations); files under EXEMPT_PREFIXES (tests/, docs/, specs/, tasks/, plans/) are exempt through _is_exempt, and scripts/ is deliberately not exempt. Second, every requirement of a node with changed modules must be cited by a test module that imports the changed unit, directly or through one intermediate import (_reached_by_asserting_test_violations, _node_reaches_requirement, _test_reaches_module, _parse_imported_modules). The citation harvest harvest_test_citations is reused from scripts.generate_rtm_evidence and imported lazily inside the function that needs it, so the module imports without the repository root on the path. The scripts.factory_trace_check CLI imports it; tests are in tests/factory/test_trace_validation.py.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

TraceViolation (dataclass)

Decorators: dataclass(frozen=True) · Line 49

One structural trace-validation failure.

TraceVerdict (dataclass)

Decorators: dataclass(frozen=True) · Line 65

The full result of :func:validate_traces -- every violation named.

Functions

Source

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

"""Structural trace validation (SPEC-033 B3; ADR-0007, ADR-0008).

ADR-0007's thesis: an ``@traces(...)`` decorator is written by the model
being governed, so a check that only diffs requirement-ID *strings* against
the spec (the pre-existing ``verification/traceability.py::validate_traces``)
never confirms the annotated unit actually implements the requirement, nor
that the trace was attributed to the correct unit. A trace is valid only if
both hold, checked **structurally**:

1. **Unit-in-contract** -- every changed unit (a ``.py`` file, per this
   brief's minimal contract granularity) is claimed by exactly one design
   node in the :class:`~orchestrata_ultimate.factory.contracts.models.DesignContract`.
2. **Reached-by-asserting-test** -- every requirement a node's changed
   modules implement is cited by at least one test module (the citation
   harvest already built for the RTM, reused here, not reimplemented), and
   at least one citing test module actually imports the changed unit
   (directly, or transitively through one intermediate import) -- the
   structural stand-in for "a test asserts on it".

This validator consumes ONLY structural facts -- file lists, AST import
statements, and the citation scan -- never the ``@traces`` decorator itself:
the decorator is exactly the model-declared channel ADR-0007 distrusts, so
reading it here would smuggle the circularity straight back in.
"""

from __future__ import annotations

import ast
from dataclasses import dataclass, field
from pathlib import Path

from orchestrata_ultimate.factory.contracts.models import DesignContract, DesignNode

# NOTE: `harvest_test_citations` lives in scripts/generate_rtm_evidence.py and is
# imported LAZILY inside _reached_by_asserting_test_violations — 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 one function that actually needs it.

# Changed files under these prefixes are exempt from the unit-in-contract
# check (test/docs/spec/task/plan authoring isn't "a unit" the contract has
# to claim). ``scripts/`` is deliberately NOT in this list -- a changed
# script is a unit too and must be claimable via a node's ``modules``.
EXEMPT_PREFIXES: tuple[str, ...] = ("tests/", "docs/", "specs/", "tasks/", "plans/")


@dataclass(frozen=True)
class TraceViolation:
    """One structural trace-validation failure.

    ``kind`` is one of the three exact strings the brief specifies:
    ``"unit_outside_contract"`` (``path`` set, ``node_id``/``requirement_id``
    blank), ``"requirement_untested"`` or ``"citing_test_does_not_reach"``
    (``node_id``/``requirement_id`` set, ``path`` blank).
    """

    kind: str
    path: str = ""
    node_id: str = ""
    requirement_id: str = ""


@dataclass(frozen=True)
class TraceVerdict:
    """The full result of :func:`validate_traces` -- every violation named."""

    passed: bool
    violations: list[TraceViolation] = field(default_factory=list)


def _is_exempt(path: str) -> bool:
    """True iff ``path`` falls under one of :data:`EXEMPT_PREFIXES`."""
    return any(path.startswith(prefix) for prefix in EXEMPT_PREFIXES)


def _path_to_dotted(path: str) -> str | None:
    """Convert a repo-relative ``.py`` path to its importable dotted name.

    ``src/orchestrata_ultimate/foo/bar.py`` -> ``orchestrata_ultimate.foo.bar``
    (the ``src/`` layout root is stripped); ``scripts/foo.py`` ->
    ``scripts.foo`` (a top-level package, per the repo's own convention of
    importing ``scripts.*`` directly). ``__init__.py`` maps to its package's
    own dotted name. Returns ``None`` for anything not a ``.py`` path.
    """
    if not path.endswith(".py"):
        return None
    rel = path[: -len(".py")]
    if rel.startswith("src/"):
        rel = rel[len("src/") :]
    if rel.endswith("/__init__"):
        rel = rel[: -len("/__init__")]
    elif rel == "__init__":
        return None
    if not rel:
        return None
    return rel.replace("/", ".")


def _dotted_to_candidate_paths(dotted: str) -> list[str]:
    """Repo-relative file-path candidates a dotted import name could resolve to."""
    slashes = dotted.replace(".", "/")
    return [
        f"src/{slashes}.py",
        f"src/{slashes}/__init__.py",
        f"{slashes}.py",
        f"{slashes}/__init__.py",
    ]


def _parse_imported_modules(text: str) -> set[str]:
    """Dotted module names a Python source's ``import``/``from`` statements name.

    For ``from a.b import c`` this adds BOTH ``a.b`` and ``a.b.c`` -- the
    latter covers importing a submodule via ``from package import submodule``
    (the common project convention, e.g. ``from orchestrata_ultimate.factory
    import contracts``); accepting the ambiguity with a plain attribute
    import of the same name is a deliberate structural over-approximation,
    not a precision claim (ADR-0008's determinism boundary is exactly this
    kind of structural fact, not semantic certainty).
    """
    dotted: set[str] = set()
    try:
        tree = ast.parse(text)
    except SyntaxError:
        return dotted
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            dotted.update(alias.name for alias in node.names)
        elif isinstance(node, ast.ImportFrom):
            if node.module is None:
                continue  # relative import -- no absolute dotted base to record
            dotted.add(node.module)
            dotted.update(f"{node.module}.{alias.name}" for alias in node.names)
    return dotted


def _test_reaches_module(test_text: str, target_dotted: str, repo_root: Path) -> bool:
    """True iff ``test_text`` imports ``target_dotted`` directly, or through
    exactly one intermediate import (ADR-0007's "reached" link)."""
    direct_imports = _parse_imported_modules(test_text)
    if target_dotted in direct_imports:
        return True
    for imported in direct_imports:
        for candidate in _dotted_to_candidate_paths(imported):
            candidate_path = repo_root / candidate
            if not candidate_path.is_file():
                continue
            try:
                intermediate_text = candidate_path.read_text(encoding="utf-8")
            except OSError:
                continue
            if target_dotted in _parse_imported_modules(intermediate_text):
                return True
    return False


def _unit_in_contract_violations(
    changed_set: set[str], module_owner: dict[str, str]
) -> list[TraceViolation]:
    """Clause 1: every non-exempt changed ``.py`` file must be claimed."""
    violations: list[TraceViolation] = []
    for path in sorted(changed_set):
        if not path.endswith(".py") or _is_exempt(path):
            continue
        if path not in module_owner:
            violations.append(TraceViolation(kind="unit_outside_contract", path=path))
    return violations


def _node_reaches_requirement(
    node: DesignNode,
    req_id: str,
    changed_modules: list[str],
    citations: dict[str, list[dict]],
    repo_root: Path,
) -> TraceViolation | None:
    """Clause 2 for one (node, requirement) pair, or ``None`` if satisfied."""
    cites = citations.get(req_id, [])
    if not cites:
        return TraceViolation(
            kind="requirement_untested", node_id=node.node_id, requirement_id=req_id
        )

    citing_files = sorted({c["file"] for c in cites})
    for citing_file in citing_files:
        try:
            citing_text = Path(citing_file).read_text(encoding="utf-8")
        except OSError:
            continue
        for changed_module in changed_modules:
            target_dotted = _path_to_dotted(changed_module)
            if target_dotted is None:
                continue
            if _test_reaches_module(citing_text, target_dotted, repo_root):
                return None

    return TraceViolation(
        kind="citing_test_does_not_reach",
        node_id=node.node_id,
        requirement_id=req_id,
    )


def _reached_by_asserting_test_violations(
    contract: DesignContract, changed_set: set[str], repo_root: Path
) -> list[TraceViolation]:
    """Clause 2: every requirement of a node with changed modules must be
    cited by a test that also reaches (imports) the changed unit."""
    from scripts.generate_rtm_evidence import harvest_test_citations

    tests_dir = repo_root / "tests"
    # harvest_test_citations (via core.tracing.scanner.collect_traces) raises
    # if its root isn't a directory -- a repo/tmp-fixture with no tests/ tree
    # yet has zero citations, not an error.
    citations = harvest_test_citations(tests_dir) if tests_dir.is_dir() else {}

    violations: list[TraceViolation] = []
    for node in contract.nodes:
        changed_modules = [m for m in node.modules if m in changed_set]
        if not changed_modules:
            continue
        for req_id in node.requirements:
            violation = _node_reaches_requirement(
                node, req_id, changed_modules, citations, repo_root
            )
            if violation is not None:
                violations.append(violation)
    return violations


def validate_traces(
    contract: DesignContract, changed_files: list[str], repo_root: Path | str
) -> TraceVerdict:
    """Check ``changed_files`` against ``contract`` per ADR-0007's conjunction.

    ``changed_files`` are repo-relative paths (as ``git diff --name-only``
    reports them). ``repo_root`` is the repository root those paths are
    relative to.
    """
    repo_root = Path(repo_root)
    changed_set = set(changed_files)

    module_owner: dict[str, str] = {}
    for node in contract.nodes:
        for module in node.modules:
            module_owner[module] = node.node_id

    violations = _unit_in_contract_violations(changed_set, module_owner)
    violations += _reached_by_asserting_test_violations(
        contract, changed_set, repo_root
    )

    return TraceVerdict(passed=not violations, violations=violations)


__all__ = ["EXEMPT_PREFIXES", "TraceVerdict", "TraceViolation", "validate_traces"]