orchestrata_ultimate.factory.amendment.coupling

Reference › orchestrata_ultimate › factory › amendment › coupling

coupling supplies the coupling signal of the amendment classifier (SPEC-033 row B16, ADR-0021). ADR-0021 names three ways naive edge counting can be laundered; this module closes two of them and leaves the third (cumulative budget refill) out of scope, since it measures only one before-and-after change. boundary_change treats any change to the set of components (a merge, split or move) as loosening by definition. transitive_coupling computes the reachability closure of the edge set with Warshall's algorithm, so a new indirect dependency shows up as closure growth. import_edges builds the cross-component import edges among a list of changed files by parsing imports; _module_to_candidates and _resolve_import place an import on a file, and imports they cannot place are skipped, never guessed. component_of is a documented heuristic: the first two path segments under src/. coupling_verdict turns old and new edges plus the boundary flag into a Direction, and coupling_note gives a one-line summary for the CLI. classify and scripts.factory_amend import it; tests are in tests/factory/test_amendment_coupling.py.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""ADR-0021 coupling accounting: transitive closure + boundary-change detection.

ADR-0021's three laundering attacks on naive direct-edge counting, and why
this module closes them:

1. **Boundary redefinition** -- merging two components makes a
   cross-component edge vanish with identical underlying coupling.
   :func:`boundary_change` treats ANY change to the component SET (merge,
   split, move) as loosening BY DEFINITION, independent of edge count.
2. **Indirection laundering** -- A->B and B->C already exist; A gaining a
   dependency on C introduces zero NEW direct edges. :func:`transitive_coupling`
   counts over the reachability CLOSURE, so A->C via B is already present in
   the closure before the change and any genuinely new indirection shows up
   as closure growth.
3. **Refill drift** -- a per-change budget refill masks cumulative growth.
   Out of scope here (ADR-0021: "the budget is cumulative against trunk,
   with an explicit repayment rule"); this module only measures one
   before/after delta, it does not track a cumulative ledger.

``component_of`` is a documented HEURISTIC (first two path segments under
``src/...``), not a registry lookup -- there is no canonical component
registry in this codebase yet. ``import_edges`` is a best-effort AST-based
import resolver: unresolvable imports (external packages, dynamic imports,
imports the resolver's relative-import arithmetic can't place) are silently
skipped, never guessed at.
"""

from __future__ import annotations

import ast
from pathlib import Path

from orchestrata_ultimate.factory.amendment.models import Direction


def component_of(module_path: str | Path) -> str:
    """The first two path segments under ``src/...`` as the "component" --
    e.g. ``src/orchestrata_ultimate/gateway/handlers/auth.py`` ->
    ``"orchestrata_ultimate/gateway"``.

    Heuristic, not a registry: a path with no ``src`` segment falls back to
    its own first two segments (best-effort for non-``src`` inputs such as
    bare dotted module names already split into path form by the caller).
    """
    parts = Path(module_path).parts
    if "src" in parts:
        idx = parts.index("src")
        segs = parts[idx + 1 : idx + 3]
    else:
        segs = parts[:2]
    return "/".join(segs)


def _module_to_candidates(base: Path, dotted: str) -> list[Path]:
    """Candidate files a dotted module name (relative to ``base``, a
    directory) could resolve to: ``name.py`` or ``name/__init__.py``."""
    if not dotted:
        return [base / "__init__.py"]
    target = base.joinpath(*dotted.split("."))
    return [target.with_suffix(".py"), target / "__init__.py"]


def _resolve_import(
    repo_root: Path, importing_file: Path, module: str | None, level: int
) -> Path | None:
    """Best-effort resolution of one import statement to an in-repo file.

    Absolute imports (``level == 0``) resolve against ``repo_root/src``.
    Relative imports (``level > 0``) resolve against the importing file's
    own package directory, walking up ``level - 1`` additional parents (the
    standard relative-import arithmetic: ``level == 1`` is "this package").
    Returns ``None`` when no candidate file exists (an external package, a
    dynamic import, or an unresolvable relative import) -- never guessed.
    """
    if level == 0:
        if not module:
            return None
        src_root = repo_root / "src"
        for candidate in _module_to_candidates(src_root, module):
            if candidate.is_file():
                return candidate
        return None
    pkg_dir = importing_file.parent
    for _ in range(level - 1):
        pkg_dir = pkg_dir.parent
    for candidate in _module_to_candidates(pkg_dir, module or ""):
        if candidate.is_file():
            return candidate
    return None


def import_edges(
    repo_root: str | Path, changed_files: list[str]
) -> set[tuple[str, str]]:
    """Cross-component import edges among ``changed_files``.

    Parses each changed ``.py`` file with :mod:`ast`, resolves every
    ``import``/``from ... import`` target that lands inside ``repo_root/src``
    to a file, and emits an edge ``(component_of(changed_file),
    component_of(resolved_target))`` whenever the two components differ.
    Non-Python changed files and unresolvable imports contribute nothing.
    """
    root = Path(repo_root)
    edges: set[tuple[str, str]] = set()
    for raw in changed_files:
        path = Path(raw)
        full_path = path if path.is_absolute() else root / path
        if full_path.suffix != ".py" or not full_path.is_file():
            continue
        try:
            tree = ast.parse(full_path.read_text(encoding="utf-8"))
        except SyntaxError, UnicodeDecodeError, OSError:
            continue
        from_component = component_of(full_path)
        for node in ast.walk(tree):
            resolved: Path | None = None
            if isinstance(node, ast.Import):
                for alias in node.names:
                    resolved = _resolve_import(root, full_path, alias.name, 0)
                    if resolved is not None:
                        to_component = component_of(resolved)
                        if to_component != from_component:
                            edges.add((from_component, to_component))
                continue
            if isinstance(node, ast.ImportFrom):
                resolved = _resolve_import(root, full_path, node.module, node.level)
                if resolved is not None:
                    to_component = component_of(resolved)
                    if to_component != from_component:
                        edges.add((from_component, to_component))
    return edges


def transitive_coupling(edges: set[tuple[str, str]]) -> set[tuple[str, str]]:
    """The reachability closure over ``edges`` (Warshall's algorithm).

    Coupling is counted over the CLOSURE, not the direct edges, so an
    indirection through an existing hop (A->B, B->C already present) is
    visible as A->C (ADR-0021 attack #2).
    """
    nodes = sorted({n for edge in edges for n in edge})
    reach: dict[str, dict[str, bool]] = {
        i: {j: (i, j) in edges for j in nodes} for i in nodes
    }
    for k in nodes:
        for i in nodes:
            if reach[i][k]:
                for j in nodes:
                    if reach[k][j]:
                        reach[i][j] = True
    return {(i, j) for i in nodes for j in nodes if i != j and reach[i][j]}


def boundary_change(old_components: set[str], new_components: set[str]) -> bool:
    """True iff the component SET changed (a merge, split, or move) --
    ADR-0021: boundary changes are loosening BY DEFINITION, regardless of
    how the edge count moves (attack #1)."""
    return old_components != new_components


def coupling_note(
    old_edges: set[tuple[str, str]],
    new_edges: set[tuple[str, str]],
    boundary_changed: bool,
) -> str:
    """A one-line human-readable summary of the coupling signal, for the CLI
    report -- not itself part of the classification contract."""
    if boundary_changed:
        return (
            "component boundary changed (merge/split/move) -- loosening by definition"
        )
    old_closure = transitive_coupling(old_edges)
    new_closure = transitive_coupling(new_edges)
    added = new_closure - old_closure
    removed = old_closure - new_closure
    if not added and not removed:
        return "coupling closure unchanged"
    if added and removed:
        return (
            f"coupling closure mixed: +{len(added)} edge(s), -{len(removed)} edge(s) "
            "-- conflicting signal"
        )
    if added:
        return f"coupling closure grew by {len(added)} edge(s): {sorted(added)}"
    return f"coupling closure shrank by {len(removed)} edge(s): {sorted(removed)}"


def coupling_verdict(
    old_edges: set[tuple[str, str]],
    new_edges: set[tuple[str, str]],
    boundary_changed: bool,
) -> Direction:
    """The coupling-based direction signal (ADR-0021).

    - A boundary change is loosening, full stop.
    - Otherwise, compare the transitive closures: a PURE shrink or a PURE
      growth is treated as compatible with a syntactic tightening verdict
      (closure growth is real but its budgeting is a separate, out-of-scope
      mechanism -- ADR-0021's cumulative edge budget -- so this classifier
      does not itself turn coupling growth into an auto-loosening call).
      A MIXED change (some edges added, some removed -- the closure grew in
      one place and shrank in another) is undetermined: the two sub-signals
      conflict and neither dominates.

    Coupling is an INPUT to the human decision (or to
    :func:`~orchestrata_ultimate.factory.amendment.classify.classify_amendment`'s
    fail-closed downgrade), never a standalone gate in its own right.
    """
    if boundary_changed:
        return Direction.LOOSENING
    old_closure = transitive_coupling(old_edges)
    new_closure = transitive_coupling(new_edges)
    added = new_closure - old_closure
    removed = old_closure - new_closure
    if added and removed:
        return Direction.UNDETERMINED
    return Direction.TIGHTENING