scripts.factory_anchors

Reference › scripts › factory_anchors

The command-line interface for the SPEC-033 semantic anchor registry (B15a; ADR-0012, ADR-0013 and ADR-0016). It uses the anchor evaluation and gate code in evaluation and gate; the factory narrative describes the subsystem. No module or test imports it.

Invocation: main takes a subcommand and dispatches to its handler. The subcommands are propose (add an anchor proposal), import-dor (import anchors from a definition-of-ready source), ratify, supersede, list, evaluate and fast-path. evaluate loads an evidence file (_load_evidence) and runs the default evaluators (_default_evaluators). fast-path is handled by _cmd_fast_path; compare the separate factory_fastpath script. Each handler returns the process exit code, and every subcommand has its own parser builder (_add_propose_parser and its siblings).

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_anchors.py

#!/usr/bin/env python3
"""CLI for the SPEC-033 semantic anchor registry (B15a, ADR-0012/0013/0016).

Subcommands
-----------
propose       Propose a new anchor (status is always forced to PROPOSED).
import-dor    Import B23 DoR anchor candidates as PROPOSED EXAMPLE anchors.
ratify        PROPOSED -> RATIFIED (the human act -- ADR-0013).
supersede     RATIFIED -> SUPERSEDED, replaced by another anchor (the only
              edit path for a ratified anchor -- ADR-0013).
list          Print every anchor (optionally filtered), a per-class count
              line, and the human-ratification-pending count (ADR-0016's
              price rule made visible).
evaluate      Evaluate a requirement's RATIFIED anchors against observed
              evidence (B15b, pure comparison, no code execution).
fast-path     The ADR-0023 fast-path eligibility gate: satisfied AND
              unchanged anchors, or it is blocked (B15b).

Examples
--------
    python scripts/factory_anchors.py propose --registry anchors.json \\
        --requirement REQ-033-C5 --class example --statement "..." \\
        --source "authored:mohamed"
    python scripts/factory_anchors.py import-dor --registry anchors.json \\
        --from candidates.json
    python scripts/factory_anchors.py ratify --registry anchors.json \\
        --id ANC-xxxxxxxxxxxx --by mohamed
    python scripts/factory_anchors.py list --registry anchors.json
    python scripts/factory_anchors.py evaluate --registry anchors.json \\
        --requirement REQ-033-C5 --evidence evidence.json
    python scripts/factory_anchors.py fast-path --registry anchors.json \\
        --requirement REQ-033-C5 --evidence evidence.json \\
        --changed ANC-xxxxxxxxxxxx
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

# Ensure repo src is on path when run directly (mirrors kg_schema.py /
# factory_dor_procedure.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,
    Satisfaction,
    evaluate_all,
)
from orchestrata_ultimate.factory.anchors.gate import fast_path_gate  # noqa: E402
from orchestrata_ultimate.factory.anchors.intake import (  # noqa: E402
    import_dor_candidates,
)
from orchestrata_ultimate.factory.anchors.models import (  # noqa: E402
    Anchor,
    AnchorClass,
    AnchorStatus,
)
from orchestrata_ultimate.factory.anchors.store import AnchorRegistry  # noqa: E402


def _default_evaluators() -> list[AnchorEvaluator]:
    """The three built-in evaluators (one per :class:`AnchorClass`)."""
    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 _cmd_propose(args: argparse.Namespace) -> int:
    registry = AnchorRegistry(Path(args.registry))
    try:
        anchor = Anchor.new(
            requirement_id=args.requirement,
            klass=AnchorClass(args.klass.lower()),
            statement=args.statement,
            boundary=args.boundary or "",
            source=args.source or "",
        )
        proposed = registry.propose(anchor)
    except ValueError as exc:
        print(f"error: {exc}")
        return 1
    print(proposed.anchor_id)
    return 0


def _cmd_import_dor(args: argparse.Namespace) -> int:
    registry = AnchorRegistry(Path(args.registry))
    try:
        proposed = import_dor_candidates(Path(args.from_path), registry)
    except (ValueError, OSError) as exc:
        print(f"error: {exc}")
        return 1
    for anchor in proposed:
        print(anchor.anchor_id)
    print(f"imported {len(proposed)} anchor candidate(s)")
    return 0


def _cmd_ratify(args: argparse.Namespace) -> int:
    registry = AnchorRegistry(Path(args.registry))
    try:
        anchor = registry.ratify(args.id, args.by)
    except ValueError as exc:
        print(f"error: {exc}")
        return 1
    print(anchor.anchor_id)
    return 0


def _cmd_supersede(args: argparse.Namespace) -> int:
    registry = AnchorRegistry(Path(args.registry))
    try:
        anchor = registry.supersede(args.id, args.replacement, args.by)
    except ValueError as exc:
        print(f"error: {exc}")
        return 1
    print(anchor.anchor_id)
    return 0


def _cmd_list(args: argparse.Namespace) -> int:
    registry = AnchorRegistry(Path(args.registry))
    status_filter = AnchorStatus(args.status.lower()) if args.status else None
    anchors = registry.anchors(requirement_id=args.requirement, status=status_filter)
    for anchor in anchors:
        print(
            f"{anchor.anchor_id} {anchor.status.value} klass={anchor.klass.value} "
            f"requirement={anchor.requirement_id} source={anchor.source!r} "
            f"statement={anchor.statement!r}"
        )
    per_class: dict[str, int] = {}
    for anchor in anchors:
        per_class[anchor.klass.value] = per_class.get(anchor.klass.value, 0) + 1
    counts = ", ".join(f"{klass}={count}" for klass, count in sorted(per_class.items()))
    print(f"by-class: {counts or '(none)'}")
    pending = sum(1 for a in anchors if a.status == AnchorStatus.PROPOSED)
    print(f"human-ratification pending: {pending} proposed anchors")
    return 0


def _cmd_evaluate(args: argparse.Namespace) -> int:
    registry = AnchorRegistry(Path(args.registry))
    try:
        evidence = _load_evidence(Path(args.evidence))
    except (OSError, ValueError) as exc:
        print(f"error: {exc}")
        return 1
    anchors = registry.anchors(
        requirement_id=args.requirement, status=AnchorStatus.RATIFIED
    )
    results = evaluate_all(anchors, evidence, _default_evaluators())
    for result in results:
        print(
            f"{result.anchor_id} {result.satisfaction.value} "
            f"klass={result.klass.value} detail={result.detail!r}"
        )
    if not results:
        print(f"no RATIFIED anchors for requirement {args.requirement}")
    unsatisfied = [r for r in results if r.satisfaction == Satisfaction.UNSATISFIED]
    return 1 if unsatisfied else 0


def _cmd_fast_path(args: argparse.Namespace) -> int:
    registry = AnchorRegistry(Path(args.registry))
    try:
        evidence = _load_evidence(Path(args.evidence))
    except (OSError, ValueError) as exc:
        print(f"error: {exc}")
        return 1
    changed_anchor_ids = set(args.changed or [])
    verdict = fast_path_gate(
        [args.requirement],
        registry,
        evidence,
        _default_evaluators(),
        changed_anchor_ids=changed_anchor_ids,
    )
    for result in verdict.results:
        print(
            f"{result.anchor_id} {result.satisfaction.value} "
            f"klass={result.klass.value} detail={result.detail!r}"
        )
    if verdict.fast_path_eligible:
        print("fast-path: ELIGIBLE")
        return 0
    reasons = []
    if verdict.blocking:
        reasons.append(f"blocking anchors: {', '.join(verdict.blocking)}")
    if verdict.edited_anchor_ids:
        reasons.append(f"edited anchors: {', '.join(verdict.edited_anchor_ids)}")
    print(f"fast-path: BLOCKED ({'; '.join(reasons) or 'no ratified anchors'})")
    return 1


def _add_propose_parser(sub: argparse._SubParsersAction) -> None:
    propose_p = sub.add_parser(
        "propose", help="Propose a new anchor (always PROPOSED)."
    )
    propose_p.add_argument("--registry", required=True)
    propose_p.add_argument("--requirement", required=True)
    propose_p.add_argument(
        "--class",
        dest="klass",
        required=True,
        choices=[c.value for c in AnchorClass],
    )
    propose_p.add_argument("--statement", required=True)
    propose_p.add_argument("--boundary", default="")
    propose_p.add_argument("--source", default="")


def _add_import_dor_parser(sub: argparse._SubParsersAction) -> None:
    import_p = sub.add_parser("import-dor", help="Import B23 DoR anchor candidates.")
    import_p.add_argument("--registry", required=True)
    import_p.add_argument("--from", dest="from_path", required=True)


def _add_ratify_parser(sub: argparse._SubParsersAction) -> None:
    ratify_p = sub.add_parser(
        "ratify", help="Ratify a PROPOSED anchor (human act -- ADR-0013)."
    )
    ratify_p.add_argument("--registry", required=True)
    ratify_p.add_argument("--id", required=True)
    ratify_p.add_argument("--by", required=True)


def _add_supersede_parser(sub: argparse._SubParsersAction) -> None:
    supersede_p = sub.add_parser(
        "supersede",
        help="Supersede a RATIFIED anchor (the only edit path -- ADR-0013).",
    )
    supersede_p.add_argument("--registry", required=True)
    supersede_p.add_argument("--id", required=True)
    supersede_p.add_argument("--replacement", required=True)
    supersede_p.add_argument("--by", required=True)


def _add_list_parser(sub: argparse._SubParsersAction) -> None:
    list_p = sub.add_parser(
        "list", help="Print anchors + per-class + pending-ratification counts."
    )
    list_p.add_argument("--registry", required=True)
    list_p.add_argument("--requirement", default=None)
    list_p.add_argument("--status", default=None)


def _add_evaluate_parser(sub: argparse._SubParsersAction) -> None:
    evaluate_p = sub.add_parser(
        "evaluate",
        help=(
            "Evaluate a requirement's RATIFIED anchors against observed "
            "evidence (B15b)."
        ),
    )
    evaluate_p.add_argument("--registry", required=True)
    evaluate_p.add_argument("--requirement", required=True)
    evaluate_p.add_argument("--evidence", required=True)


def _add_fast_path_parser(sub: argparse._SubParsersAction) -> None:
    fast_path_p = sub.add_parser(
        "fast-path",
        help="ADR-0023 fast-path eligibility gate: satisfied AND unchanged (B15b).",
    )
    fast_path_p.add_argument("--registry", required=True)
    fast_path_p.add_argument("--requirement", required=True)
    fast_path_p.add_argument("--evidence", required=True)
    fast_path_p.add_argument(
        "--changed", nargs="*", default=[], metavar="ANCHOR_ID", dest="changed"
    )


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="factory_anchors",
        description="SPEC-033 semantic anchor registry CLI (B15a/B15b).",
    )
    sub = parser.add_subparsers(dest="command", required=True)
    _add_propose_parser(sub)
    _add_import_dor_parser(sub)
    _add_ratify_parser(sub)
    _add_supersede_parser(sub)
    _add_list_parser(sub)
    _add_evaluate_parser(sub)
    _add_fast_path_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 = {
        "propose": _cmd_propose,
        "import-dor": _cmd_import_dor,
        "ratify": _cmd_ratify,
        "supersede": _cmd_supersede,
        "list": _cmd_list,
        "evaluate": _cmd_evaluate,
        "fast-path": _cmd_fast_path,
    }
    handler = dispatch[args.command]
    return handler(args)


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