scripts.factory_fastpath

Reference › scripts › factory_fastpath

The command-line interface for the SPEC-033 B12 fast-path classifier (ADR-0023). It shares its evaluator and evidence handling with factory_anchors, through evaluation and gate. The factory narrative describes the subsystem. No module or test imports it.

Invocation: main has two subcommands. structural computes a structural proof for a change (_cmd_structural, printed through _structural_to_dict). classify decides whether a change can take the fast path, loading an anchor evidence file (_load_evidence) and running the default evaluators (_default_evaluators); the anchor gate verdict is included in the output (_anchor_verdict_to_dict). Comma-separated lists of options are split by _split_csv. Each handler returns the exit code.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/factory_fastpath.py

#!/usr/bin/env python3
"""CLI for the SPEC-033 B12 fast-path classifier (ADR-0023).

Subcommands
-----------
structural   Run only the structural half (within-contract: same public
             surface, no new outbound edges) -- cheap, stdlib ``ast`` only.
classify     Run the combined ADR-0023 decision: structural first, then
             (only if structural passed) the B15b anchor gate. Emits a
             JSON ``FastPathDecision`` including the worked ``CostTrace``.

Examples
--------
    python scripts/factory_fastpath.py structural --repo-root . \\
        --base HEAD~1 --files src/pkg/mod.py

    python scripts/factory_fastpath.py classify --repo-root . \\
        --base HEAD~1 --files src/pkg/mod.py --requirements REQ-033-C5 \\
        --registry anchors.json --evidence evidence.json \\
        --changed-anchors ANC-xxxxxxxxxxxx

WORKED EXAMPLE (real numbers, captured while building this classifier by
actually running the two commands above against a scratch git repo seeded
from this project's own ``orchestrata_ultimate.factory.anchors.models``
module -- not fabricated)
-------------------------------------------------------------------------
Base commit ``d50e24b5``; HEAD is a body-only edit (the module docstring
changed, public surface untouched); one RATIFIED, SATISFIED, unchanged
EXAMPLE anchor on the touched requirement. The real ``classify()`` JSON
output (structural passed, so anchors were evaluated too -- the fast path
fired, exit code 0):

    {
      "eligible": true,
      "structural": {
        "within_contract": true,
        "new_public_symbols": [],
        "removed_public_symbols": [],
        "changed_signatures": [],
        "new_outbound_imports": [],
        "non_python_changes": []
      },
      "anchor_verdict": {
        "fast_path_eligible": true,
        "blocking": [],
        "edited_anchor_ids": [],
        "results": [
          {
            "anchor_id": "ANC-379c0437113d",
            "klass": "example",
            "satisfaction": "satisfied",
            "detail": "k='1' matches expected '1'"
          }
        ]
      },
      "cost_trace": {
        "structural_seconds": 0.003735624995897524,
        "anchors_seconds": 0.0002003879999392666,
        "anchors_evaluated": 1,
        "files_classified": 1,
        "route": "fast"
      }
    }

The number that matters for the ROI argument: structural (~3.7ms, one
``git show`` subprocess + two ``ast.parse`` calls) dominates anchor
evaluation (~0.2ms, a single in-memory comparison) by ~19x on this file --
and on the slow path (structural fails), the anchor number is never even
paid at all (``anchors_seconds`` is ``None``, not merely small).
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any

# Ensure repo src is on path when run directly (mirrors factory_anchors.py /
# kg_schema.py / run_ast_pipeline.py).
_repo_src = Path(__file__).resolve().parent.parent / "src"
if str(_repo_src) not in sys.path:
    sys.path.insert(0, str(_repo_src))

from orchestrata_ultimate.factory.anchors.evaluation import (  # noqa: E402
    AnchorEvaluator,
    AnchorEvidence,
    ExampleEvaluator,
    OracleEvaluator,
    PropertyEvaluator,
)
from orchestrata_ultimate.factory.anchors.gate import AnchorGateVerdict  # noqa: E402
from orchestrata_ultimate.factory.anchors.store import AnchorRegistry  # noqa: E402
from orchestrata_ultimate.factory.fastpath.classifier import (  # noqa: E402
    StructuralProof,
    classify,
    structural_within_contract,
)


def _default_evaluators() -> list[AnchorEvaluator]:
    """The three built-in evaluators (one per anchor class), same as factory_anchors.py."""
    return [ExampleEvaluator(), PropertyEvaluator(), OracleEvaluator()]


def _load_evidence(evidence_path: Path) -> AnchorEvidence:
    """Load an ``AnchorEvidence`` from a JSON file.

    Expected shape: ``{"observed_outputs": {...}, "differential": {...}}``
    (both optional, default to ``{}``).
    """
    with open(evidence_path, encoding="utf-8") as f:
        data = json.load(f)
    return AnchorEvidence(
        observed_outputs=data.get("observed_outputs", {}),
        differential=data.get("differential", {}),
    )


def _split_csv(raw: str) -> list[str]:
    """Split a comma-separated CLI value into a list, dropping empty entries."""
    return [item for item in raw.split(",") if item]


def _structural_to_dict(proof: StructuralProof) -> dict[str, Any]:
    return {
        "within_contract": proof.within_contract,
        "new_public_symbols": proof.new_public_symbols,
        "removed_public_symbols": proof.removed_public_symbols,
        "changed_signatures": proof.changed_signatures,
        "new_outbound_imports": proof.new_outbound_imports,
        "non_python_changes": proof.non_python_changes,
    }


def _anchor_verdict_to_dict(verdict: AnchorGateVerdict | None) -> dict[str, Any] | None:
    if verdict is None:
        return None
    return {
        "fast_path_eligible": verdict.fast_path_eligible,
        "blocking": verdict.blocking,
        "edited_anchor_ids": verdict.edited_anchor_ids,
        "results": [
            {
                "anchor_id": result.anchor_id,
                "klass": result.klass.value,
                "satisfaction": result.satisfaction.value,
                "detail": result.detail,
            }
            for result in verdict.results
        ],
    }


def _cmd_structural(args: argparse.Namespace) -> int:
    repo_root = Path(args.repo_root)
    files = _split_csv(args.files)
    proof = structural_within_contract(repo_root, args.base, files)
    print(json.dumps(_structural_to_dict(proof), indent=2))
    return 0 if proof.within_contract else 1


def _cmd_classify(args: argparse.Namespace) -> int:
    repo_root = Path(args.repo_root)
    files = _split_csv(args.files)
    requirements = _split_csv(args.requirements)
    changed_anchor_ids = set(_split_csv(args.changed_anchors or ""))
    registry = AnchorRegistry(Path(args.registry))
    try:
        evidence = _load_evidence(Path(args.evidence))
    except (OSError, ValueError) as exc:
        print(f"error: {exc}")
        return 1
    decision = classify(
        repo_root,
        args.base,
        files,
        requirements,
        registry,
        evidence,
        _default_evaluators(),
        changed_anchor_ids=changed_anchor_ids,
    )
    output = {
        "eligible": decision.eligible,
        "structural": _structural_to_dict(decision.structural),
        "anchor_verdict": _anchor_verdict_to_dict(decision.anchor_verdict),
        "cost_trace": decision.cost_trace.to_dict(),
    }
    print(json.dumps(output, indent=2))
    return 0 if decision.eligible else 1


def _add_structural_parser(sub: argparse._SubParsersAction) -> None:
    structural_p = sub.add_parser(
        "structural", help="Run only the ADR-0023 structural (within-contract) half."
    )
    structural_p.add_argument("--repo-root", required=True)
    structural_p.add_argument("--base", required=True)
    structural_p.add_argument("--files", required=True, help="comma-separated paths")


def _add_classify_parser(sub: argparse._SubParsersAction) -> None:
    classify_p = sub.add_parser(
        "classify",
        help="Combined ADR-0023 decision: structural first, anchors second.",
    )
    classify_p.add_argument("--repo-root", required=True)
    classify_p.add_argument("--base", required=True)
    classify_p.add_argument("--files", required=True, help="comma-separated paths")
    classify_p.add_argument(
        "--requirements", required=True, help="comma-separated requirement ids"
    )
    classify_p.add_argument("--registry", required=True)
    classify_p.add_argument("--evidence", required=True)
    classify_p.add_argument(
        "--changed-anchors",
        dest="changed_anchors",
        default="",
        help="comma-separated anchor ids edited by this change",
    )


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="factory_fastpath",
        description="SPEC-033 B12 fast-path classifier CLI (ADR-0023).",
    )
    sub = parser.add_subparsers(dest="command", required=True)
    _add_structural_parser(sub)
    _add_classify_parser(sub)
    return parser


def main(argv: list[str] | None = None) -> int:
    """Parse ``argv`` and dispatch to the matching subcommand handler."""
    parser = _build_parser()
    args = parser.parse_args(argv)

    dispatch = {
        "structural": _cmd_structural,
        "classify": _cmd_classify,
    }
    handler = dispatch[args.command]
    return handler(args)


if __name__ == "__main__":
    sys.exit(main())