orchestrata_ultimate.factory.fastpath.classifier

Reference › orchestrata_ultimate › factory › fastpath › classifier

classifier is the fast-path classifier (SPEC-033 row B12, ADR-0023). A change qualifies for the fast path only if it is structurally within the existing contract and the anchors of every touched criterion stay satisfied and unchanged. The anchor half is decided by fast_path_gate in anchors.gate; this module supplies the structural half and combines the two.

structural_within_contract compares each changed file at the base commit with the working tree, using only the standard-library ast module, and returns a StructuralProof listing new and removed public symbols, changed signatures, new outbound imports and non-Python changes. The public surface (_public_surface) is the top-level function and class names plus the public methods of public classes. A signature key (_signature_key) compares ordered parameter names, whether each has a default, *args and **kwargs, and the unparsed annotations. A newly added import of a new top-level package, any non-.py change, and any surface difference all rule the change out. classify runs the structural check first and evaluates anchors only if it passed, because evaluation can parse statements and read the registry; otherwise anchor_verdict and the anchor timing are None. It returns a FastPathDecision that includes a CostTrace recording timings, the number of anchors evaluated and the route (fast or slow). The conformance engine reuses its signature comparison, and the scripts.factory_fastpath CLI imports it. Tests are in tests/factory/test_fastpath_classifier.py.

At a glance

Class diagram

Dependency diagram

API

Classes

StructuralProof (dataclass)

Decorators: dataclass(frozen=True) · Line 61

The structural "within existing contract" half of ADR-0023.

CostTrace (dataclass)

Decorators: dataclass(frozen=True) · Line 309

A worked cost trace for one :func:classify call (§9 B12's "ROI" point).

FastPathDecision (dataclass)

Decorators: dataclass(frozen=True) · Line 335

The combined ADR-0023 fast-path decision: structural AND anchors.

Functions

Source

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

"""The ADR-0023 fast-path classifier (B12): structural half + combined decision.

ADR-0023, verbatim: "A change qualifies for the fast path only if, in
addition to the structural within-contract proof: the anchors of every
touched acceptance criterion remain **satisfied**, and those anchors remain
**unchanged**." B15b's :func:`~orchestrata_ultimate.factory.anchors.gate.
fast_path_gate` already decides the ANCHOR half; this module supplies the
STRUCTURAL half ("within the existing contract": same public surface, no
new cross-component edges) and :func:`classify` combines the two into one
decision with a recorded :class:`CostTrace`.

**Short-circuit, by design:** structural is the cheap, stdlib-``ast``-only
check; anchor evaluation may run evaluators, parse statements, and touch a
registry file. So :func:`classify` runs structural FIRST and only pays the
anchor-evaluation cost when structural already passed -- when it fails,
``anchor_verdict`` is ``None`` and ``anchors_seconds`` is ``None``. This
ordering is the entire cost argument for this row (§9 B12: "ROI").

Public-surface + signature comparison (stdlib ``ast`` only, no new deps):
a changed ``.py`` file's public surface is its top-level ``def``/``class``
names and (for a public class) its public method names, all lacking a
leading underscore. A signature is compared by ordered parameter names,
whether each carries a default (defaults COUNT -- promoting/demoting a
parameter to/from required is a signature change), ``*args``/``**kwargs``
presence, and the textual annotation of every parameter and the return
(via ``ast.unparse``). Outbound imports (module- and function-level
``import``/``from``, found via ``ast.walk``) are resolved to their
top-level package; a newly-added import whose top-level package differs
from the changed file's own is a new cross-component edge. A file absent
at ``base_commit`` is a brand-new module -- never within contract,
regardless of what its (empty) diff would otherwise show. Any changed
non-``.py`` file is a conservative floor: never within contract.
"""

from __future__ import annotations

import ast
import subprocess
import time
from collections.abc import Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Literal

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.factory.anchors.evaluation import (
    AnchorEvaluator,
    AnchorEvidence,
)
from orchestrata_ultimate.factory.anchors.gate import AnchorGateVerdict, fast_path_gate
from orchestrata_ultimate.factory.anchors.store import AnchorRegistry

_NEW_MODULE_MARKER = "<new-module>"

# ---------------------------------------------------------------------------
# StructuralProof
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class StructuralProof:
    """The structural "within existing contract" half of ADR-0023.

    ``within_contract`` is ``True`` iff every reason list below is empty.
    Each reason list is sorted and holds ``"<changed-file>:<detail>"``
    entries so a caller can see, per changed file, exactly what broke the
    contract.
    """

    within_contract: bool
    new_public_symbols: list[str]
    removed_public_symbols: list[str]
    changed_signatures: list[str]
    new_outbound_imports: list[str]
    non_python_changes: list[str]


def _own_top_level_package(rel_path: str) -> str:
    """The changed file's own top-level package (or its own name, if none).

    ``pkg/sub/mod.py`` -> ``"pkg"``. A module living directly at the repo
    root, ``mod.py`` -> ``"mod"`` (the module IS its own single-name
    "package" for this comparison -- there is no enclosing directory).
    """
    parts = Path(rel_path).with_suffix("").parts
    return parts[0] if parts else rel_path


def _public_surface(tree: ast.Module) -> set[str]:
    """Top-level ``def``/``class`` names, and public methods of public classes.

    A name "lacks a leading underscore" is the sole test (ADR-0023's
    "same public signatures"). A private (underscore-prefixed) class's
    methods are not part of the module's public surface even if the
    method name itself lacks an underscore -- an inaccessible class's
    methods are not part of the contract.
    """
    surface: set[str] = set()
    for node in tree.body:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            if not node.name.startswith("_"):
                surface.add(node.name)
        elif isinstance(node, ast.ClassDef):
            if node.name.startswith("_"):
                continue
            surface.add(node.name)
            for member in node.body:
                if isinstance(
                    member, (ast.FunctionDef, ast.AsyncFunctionDef)
                ) and not member.name.startswith("_"):
                    surface.add(f"{node.name}.{member.name}")
    return surface


_SignatureKey = tuple[tuple[tuple[str, bool, str], ...], str | None, str | None, str]


def _signature_key(node: ast.FunctionDef | ast.AsyncFunctionDef) -> _SignatureKey:
    """Normalize ``node``'s signature: ordered params, defaults, *a/**kw, annotations.

    Params are (name, has_default, annotation-text) in ``posonlyargs +
    args`` order followed by ``kwonlyargs`` order -- reordering across
    those groups (e.g. moving a param behind a bare ``*``) is therefore a
    detectable signature change, matching "positional/keyword parameter
    names in order". Defaults are right-aligned onto ``posonlyargs +
    args`` per Python's own ``ast`` shape.
    """
    args = node.args
    combined = args.posonlyargs + args.args
    num_no_default = len(combined) - len(args.defaults)
    params: list[tuple[str, bool, str]] = []
    for index, arg in enumerate(combined):
        has_default = index >= num_no_default
        annotation = ast.unparse(arg.annotation) if arg.annotation is not None else ""
        params.append((arg.arg, has_default, annotation))
    for index, arg in enumerate(args.kwonlyargs):
        has_default = args.kw_defaults[index] is not None
        annotation = ast.unparse(arg.annotation) if arg.annotation is not None else ""
        params.append((arg.arg, has_default, annotation))
    vararg = args.vararg.arg if args.vararg is not None else None
    kwarg = args.kwarg.arg if args.kwarg is not None else None
    returns = ast.unparse(node.returns) if node.returns is not None else ""
    return (tuple(params), vararg, kwarg, returns)


def _signatures(tree: ast.Module) -> dict[str, _SignatureKey]:
    """Signature keys for every public top-level function and public-class method."""
    sigs: dict[str, _SignatureKey] = {}
    for node in tree.body:
        if isinstance(
            node, (ast.FunctionDef, ast.AsyncFunctionDef)
        ) and not node.name.startswith("_"):
            sigs[node.name] = _signature_key(node)
        elif isinstance(node, ast.ClassDef) and not node.name.startswith("_"):
            for member in node.body:
                if isinstance(
                    member, (ast.FunctionDef, ast.AsyncFunctionDef)
                ) and not member.name.startswith("_"):
                    sigs[f"{node.name}.{member.name}"] = _signature_key(member)
    return sigs


def _import_targets(tree: ast.Module, own_package: str) -> dict[str, str]:
    """Every import target in ``tree`` (module- or function-level, via ``ast.walk``).

    Maps a target string (``"a.b.c"`` for ``import a.b.c``; the module
    path for ``from a.b import c``; ``"." * level + module`` for a
    relative ``from``) to its resolved top-level package. A relative
    import (``level > 0``) can only ever reach within the same top-level
    package, so it is mapped to ``own_package`` -- never a new outbound
    edge regardless of what it imports.
    """
    targets: dict[str, str] = {}
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                targets[alias.name] = alias.name.split(".")[0]
        elif isinstance(node, ast.ImportFrom):
            if node.level and node.level > 0:
                target = f"{'.' * node.level}{node.module or ''}"
                targets[target] = own_package
            else:
                module = node.module or ""
                targets[module] = module.split(".")[0] if module else ""
    return targets


def _read_head_source(repo_root: Path, rel_path: str) -> str:
    """The working-tree content of ``rel_path``."""
    return (repo_root / rel_path).read_text(encoding="utf-8")


def _read_base_source(repo_root: Path, base_commit: str, rel_path: str) -> str | None:
    """``git show {base_commit}:{rel_path}``, or ``None`` if absent at that commit."""
    result = subprocess.run(
        ["git", "show", f"{base_commit}:{rel_path}"],
        cwd=repo_root,
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        return None
    return result.stdout


def _classify_python_file(
    repo_root: Path,
    base_commit: str,
    rel_path: str,
    new_public_symbols: set[str],
    removed_public_symbols: set[str],
    changed_signatures: set[str],
    new_outbound_imports: set[str],
) -> None:
    """Diff one changed ``.py`` file's public surface/signatures/imports, in place."""
    head_source = _read_head_source(repo_root, rel_path)
    base_source = _read_base_source(repo_root, base_commit, rel_path)
    existed_at_base = base_source is not None

    head_tree = ast.parse(head_source, filename=rel_path)
    base_tree = ast.parse(base_source or "", filename=rel_path)
    own_package = _own_top_level_package(rel_path)

    head_public = _public_surface(head_tree)
    base_public = _public_surface(base_tree)
    added = head_public - base_public
    removed = base_public - head_public
    if not existed_at_base:
        # A brand-new module is never within contract, even in the
        # (unusual) case its own diff would otherwise show nothing added.
        added.add(_NEW_MODULE_MARKER)

    new_public_symbols.update(f"{rel_path}:{name}" for name in added)
    removed_public_symbols.update(f"{rel_path}:{name}" for name in removed)

    head_sigs = _signatures(head_tree)
    base_sigs = _signatures(base_tree)
    for name in head_public & base_public:
        if (
            name in head_sigs
            and name in base_sigs
            and head_sigs[name] != base_sigs[name]
        ):
            changed_signatures.add(f"{rel_path}:{name}")

    head_imports = _import_targets(head_tree, own_package)
    base_imports = _import_targets(base_tree, own_package)
    for target, top_package in head_imports.items():
        if target not in base_imports and top_package and top_package != own_package:
            new_outbound_imports.add(f"{rel_path}:{target}")


@traces(spec="033", reqs=["REQ-033-C1"])
def structural_within_contract(
    repo_root: Path, base_commit: str, changed_files: Sequence[str]
) -> StructuralProof:
    """The structural half of ADR-0023: same public surface, no new outbound edges.

    Pure read path: reads the working tree and ``git show {base_commit}:
    <path>`` for every entry in ``changed_files`` -- never writes anything.
    Any changed non-``.py`` file is a conservative floor (recorded in
    ``non_python_changes``, always forces ``within_contract=False``); every
    changed ``.py`` file is diffed BASE-vs-HEAD by
    :func:`_classify_python_file`.
    """
    new_public_symbols: set[str] = set()
    removed_public_symbols: set[str] = set()
    changed_signatures: set[str] = set()
    new_outbound_imports: set[str] = set()
    non_python_changes: set[str] = set()

    for rel_path in changed_files:
        if not rel_path.endswith(".py"):
            non_python_changes.add(rel_path)
            continue
        _classify_python_file(
            repo_root,
            base_commit,
            rel_path,
            new_public_symbols,
            removed_public_symbols,
            changed_signatures,
            new_outbound_imports,
        )

    within_contract = not (
        new_public_symbols
        or removed_public_symbols
        or changed_signatures
        or new_outbound_imports
        or non_python_changes
    )
    return StructuralProof(
        within_contract=within_contract,
        new_public_symbols=sorted(new_public_symbols),
        removed_public_symbols=sorted(removed_public_symbols),
        changed_signatures=sorted(changed_signatures),
        new_outbound_imports=sorted(new_outbound_imports),
        non_python_changes=sorted(non_python_changes),
    )


# ---------------------------------------------------------------------------
# CostTrace + FastPathDecision
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class CostTrace:
    """A worked cost trace for one :func:`classify` call (§9 B12's "ROI" point).

    ``anchors_seconds`` is ``None`` exactly when anchor evaluation was
    short-circuited (structural already failed) -- absence of a number
    IS the signal that the expensive half never ran.
    """

    structural_seconds: float
    anchors_seconds: float | None
    anchors_evaluated: int
    files_classified: int
    route: Literal["fast", "slow"]

    def to_dict(self) -> dict[str, object]:
        """A plain, JSON-serializable ``dict`` of this trace."""
        return {
            "structural_seconds": self.structural_seconds,
            "anchors_seconds": self.anchors_seconds,
            "anchors_evaluated": self.anchors_evaluated,
            "files_classified": self.files_classified,
            "route": self.route,
        }


@dataclass(frozen=True)
class FastPathDecision:
    """The combined ADR-0023 fast-path decision: structural AND anchors.

    ``eligible = structural.within_contract and anchor_verdict.
    fast_path_eligible``. When ``structural.within_contract`` is ``False``
    the anchor half is never evaluated (the cheap check runs first) --
    ``anchor_verdict`` is ``None`` in that case, which is itself the proof
    the short-circuit fired.
    """

    eligible: bool
    structural: StructuralProof
    anchor_verdict: AnchorGateVerdict | None
    cost_trace: CostTrace


@traces(spec="033", reqs=["REQ-033-C1"])
def classify(
    repo_root: Path,
    base_commit: str,
    changed_files: Sequence[str],
    requirement_ids: list[str],
    registry: AnchorRegistry,
    evidence: AnchorEvidence,
    evaluators: Sequence[AnchorEvaluator],
    *,
    changed_anchor_ids: set[str],
) -> FastPathDecision:
    """Decide ADR-0023 fast-path eligibility: structural first, anchors second.

    Pure read path -- never writes ``registry``. Structural
    (:func:`structural_within_contract`) always runs; anchor evaluation
    (:func:`~orchestrata_ultimate.factory.anchors.gate.fast_path_gate`)
    runs ONLY when structural already passed, since it is the more
    expensive of the two checks (evaluator dispatch, statement parsing,
    a registry read) and ADR-0023 requires both halves to pass anyway --
    there is nothing to gain by paying that cost when structural has
    already disqualified the change.
    """
    structural_start = time.perf_counter()
    structural = structural_within_contract(repo_root, base_commit, changed_files)
    structural_seconds = time.perf_counter() - structural_start

    if not structural.within_contract:
        cost_trace = CostTrace(
            structural_seconds=structural_seconds,
            anchors_seconds=None,
            anchors_evaluated=0,
            files_classified=len(changed_files),
            route="slow",
        )
        return FastPathDecision(
            eligible=False,
            structural=structural,
            anchor_verdict=None,
            cost_trace=cost_trace,
        )

    anchors_start = time.perf_counter()
    anchor_verdict = fast_path_gate(
        requirement_ids,
        registry,
        evidence,
        evaluators,
        changed_anchor_ids=changed_anchor_ids,
    )
    anchors_seconds = time.perf_counter() - anchors_start

    eligible = structural.within_contract and anchor_verdict.fast_path_eligible
    cost_trace = CostTrace(
        structural_seconds=structural_seconds,
        anchors_seconds=anchors_seconds,
        anchors_evaluated=len(anchor_verdict.results),
        files_classified=len(changed_files),
        route="fast" if eligible else "slow",
    )
    return FastPathDecision(
        eligible=eligible,
        structural=structural,
        anchor_verdict=anchor_verdict,
        cost_trace=cost_trace,
    )