scripts.factory_trace_check

Reference › scripts › factory_trace_check

The command-line interface for the SPEC-033 B3 structural trace validator (ADR-0007 and ADR-0008). It uses the contract models in models and the validator in trace_validation; the factory narrative describes the subsystem. No module or test imports it.

Invocation: main has two subcommands. template writes a skeleton trace contract with a slug derived from a title (_slugify, _cmd_template). check validates a contract against the files changed between two git revisions (_changed_files_from_git, _cmd_check), printing each TraceViolation through _format_violation. With no subcommand it prints help and exits 1, and an exception prints an error and exits 1.

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

#!/usr/bin/env python3
"""CLI for the SPEC-033 B3 structural trace validator (ADR-0007, ADR-0008).

Subcommands
-----------
template    Write a skeleton :class:`DesignContract` JSON file (empty
            ``nodes`` -- structurally valid, ready to be filled in by hand).
check       Run :func:`validate_traces` against a contract file and a set of
            changed files (from ``git diff`` between two refs, or an
            explicit list); prints every violation and exits non-zero if any
            exist.

Examples
--------
    python scripts/factory_trace_check.py template \\
        --change-ref feat/x --out contract.json
    python scripts/factory_trace_check.py check \\
        --contract contract.json --base main --head HEAD
    python scripts/factory_trace_check.py check \\
        --contract contract.json --files src/a.py src/b.py
"""

from __future__ import annotations

import argparse
import re
import subprocess
import sys
from pathlib import Path

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

from orchestrata_ultimate.factory.contracts.models import (  # noqa: E402
    DesignContract,
    load_contract,
    save_contract,
)
from orchestrata_ultimate.factory.contracts.trace_validation import (  # noqa: E402
    TraceViolation,
    validate_traces,
)

_SCHEMA_VERSION = 1
_SLUG_RE = re.compile(r"[^a-z0-9]+")


def _slugify(text: str) -> str:
    """Lowercase ``text`` and collapse anything non-alphanumeric to ``-``."""
    slug = _SLUG_RE.sub("-", text.lower()).strip("-")
    return slug or "contract"


def _changed_files_from_git(repo_root: Path, base: str, head: str) -> list[str]:
    """Repo-relative changed ``.py``-and-everything-else paths between two refs."""
    result = subprocess.run(
        ["git", "-C", str(repo_root), "diff", "--name-only", base, head],
        capture_output=True,
        text=True,
        check=True,
    )
    return [line for line in result.stdout.splitlines() if line.strip()]


def _format_violation(violation: TraceViolation) -> str:
    if violation.kind == "unit_outside_contract":
        return f"unit_outside_contract: {violation.path}"
    return (
        f"{violation.kind}: node={violation.node_id} "
        f"requirement={violation.requirement_id}"
    )


def _cmd_template(args: argparse.Namespace) -> int:
    contract = DesignContract(
        contract_id=f"contract-{_slugify(args.change_ref)}",
        change_ref=args.change_ref,
        nodes=(),
        schema_version=_SCHEMA_VERSION,
    )
    save_contract(contract, args.out)
    print(f"Wrote skeleton contract to {args.out}")
    return 0


def _cmd_check(args: argparse.Namespace) -> int:
    repo_root = Path(args.repo_root) if args.repo_root else Path.cwd()
    contract = load_contract(args.contract)

    if args.files is not None:
        changed_files = list(args.files)
    elif args.base and args.head:
        changed_files = _changed_files_from_git(repo_root, args.base, args.head)
    else:
        print(
            "ERROR: 'check' requires either --files, or both --base and --head",
            file=sys.stderr,
        )
        return 1

    verdict = validate_traces(contract, changed_files, repo_root)
    if verdict.passed:
        print("PASS: all traces structurally valid.")
        return 0

    print(f"FAIL: {len(verdict.violations)} trace violation(s):")
    for violation in verdict.violations:
        print(f"  - {_format_violation(violation)}")
    return 1


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="factory_trace_check",
        description=(
            "SPEC-033 B3 structural trace validation "
            "(unit-in-contract AND reached-by-asserting-test)."
        ),
    )
    sub = parser.add_subparsers(dest="command", required=True)

    template_p = sub.add_parser(
        "template", help="Write a skeleton DesignContract JSON file."
    )
    template_p.add_argument(
        "--change-ref", required=True, help="Branch/commit context for the contract."
    )
    template_p.add_argument("--out", required=True, help="Output path.")

    check_p = sub.add_parser(
        "check", help="Validate traces for a contract against changed files."
    )
    check_p.add_argument(
        "--contract", required=True, help="Path to the DesignContract JSON."
    )
    check_p.add_argument("--base", default=None, help="Git ref to diff from.")
    check_p.add_argument("--head", default=None, help="Git ref to diff to.")
    check_p.add_argument(
        "--files", nargs="*", default=None, help="Explicit changed-file list."
    )
    check_p.add_argument(
        "--repo-root",
        default=None,
        help="Repository root the contract's paths are relative to "
        "(default: current directory).",
    )

    return parser


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

    dispatch = {
        "template": _cmd_template,
        "check": _cmd_check,
    }
    handler = dispatch.get(args.command)
    if handler is None:
        parser.print_help()
        sys.exit(1)

    try:
        exit_code = handler(args)
    except Exception as exc:  # noqa: BLE001
        print(f"ERROR: {exc}", file=sys.stderr)
        sys.exit(1)

    sys.exit(exit_code)


if __name__ == "__main__":
    main()