scripts.factory_evidence

Reference › scripts › factory_evidence

The command-line interface for the SPEC-033 red-first ledger and affected-set mutation testing. It uses the ledger in ledger and the mutation engine in mutation; the factory narrative describes the subsystem. No module or test imports it.

Invocation: main has two subcommands. red-status reads the ledger (default location from _default_ledger) and reports whether the tests for a unit were seen failing before they passed (_cmd_red_status). mutate runs mutation testing over a set of units (_cmd_mutate), choosing the tests to run for each unit with test_selector. Comma-separated option values 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_evidence.py

#!/usr/bin/env python3
"""CLI for the SPEC-033 red-first ledger + affected-set mutation testing
(§9, B6).

Subcommands
-----------
red-status  Classify a test's red-first evidence quality at a commit
            (``red_first`` / ``green_only`` / ``never_red`` / ``unobserved``)
            against the shared B1c event ledger.
mutate      Compute the affected units for ``--changed-files`` and run the
            small closed mutation-operator set against each, reporting
            killed/survived/timeout counts and every surviving mutant.

Test selection for ``mutate``
------------------------------
``factory.evidence.mutation.run_mutations`` takes the unit -> tests mapping
from its caller. ``scripts/add_traces_from_evidence.py``'s ``TestMatcher``
was evaluated for this (per the B6 brief) and does NOT fit here: it needs
a pre-serialized whole-codebase ``ASTCodebase`` pair (source + tests, via
``JSONASTLoader`` over committed multi-MB JSON snapshots that a separate
``serialize`` step must produce and keep fresh) and its strongest strategy
is a ``Test<ClassName>`` naming convention -- both a poor fit for an ad hoc,
possibly-stale ``--changed-files`` invocation. This CLI instead takes an
explicit, low-magic test selection: ``--tests`` (the same test id list
applied to every affected unit) or ``--test-map`` (a JSON file mapping
``"<file>::<unit_name>"`` to a per-unit test id list). Smarter auto-derived
selection (e.g. via TestMatcher once the AST snapshots are guaranteed
fresh) is future work, not required for this task's MVP.

Path defaults mirror ``scripts/factory_enforce.py``: ``--ledger`` ->
``<repo-root>/factory_ledger.jsonl`` (the SAME file B1c governed-write
records live in -- see ``factory.evidence.red_first``'s module docstring
for why that is safe), ``--repo-root`` -> the current directory,
``--worktree`` -> ``str(repo_root)``.

Examples
--------
    python scripts/factory_evidence.py red-status \\
        --test "tests/foo.py::test_bar" --commit HEAD
    python scripts/factory_evidence.py mutate \\
        --changed-files src/pkg/mod.py \\
        --tests "tests/pkg/test_mod.py::test_thing" --limit 20
"""

from __future__ import annotations

import argparse
import dataclasses
import json
import shlex
import sys
from pathlib import Path

# Ensure repo src is on path when run directly (mirrors factory_enforce.py /
# factory_dor_procedure.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))

_DEFAULT_RUNNER_CMD = ["uv", "run", "pytest", "-q", "-p", "no:cacheprovider"]
_DEFAULT_TIMEOUT = 30.0
_DEFAULT_LIMIT = 50

# evidence_status values that make `red-status` exit non-zero: only
# `red_first` is treated as a passing evidence claim (fail-closed, per
# B1a's "no allow-by-default branch" philosophy elsewhere in this factory).
_RED_STATUS_FAILING = {"never_red", "green_only", "unobserved"}


def _default_ledger(repo_root: Path) -> Path:
    return repo_root / "factory_ledger.jsonl"


def _split_csv(value: str) -> list[str]:
    return [v.strip() for v in value.split(",") if v.strip()]


def _cmd_red_status(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.enforcement.ledger import EventLedger
    from orchestrata_ultimate.factory.evidence.red_first import RedFirstLedger

    repo_root = Path(args.repo_root)
    ledger_path = Path(args.ledger) if args.ledger else _default_ledger(repo_root)
    ledger = EventLedger(ledger_path)
    red_first = RedFirstLedger(ledger, repo_root)

    status = red_first.evidence_status(args.test, args.commit)
    print(json.dumps({"test_id": args.test, "commit": args.commit, "status": status}))
    return 1 if status in _RED_STATUS_FAILING else 0


def _cmd_mutate(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.factory.evidence.mutation import (
        UnitRef,
        affected_units,
        run_mutations,
    )

    repo_root = Path(args.repo_root)
    changed_files = _split_csv(args.changed_files)
    units = affected_units(changed_files)

    uniform_tests = _split_csv(args.tests) if args.tests else []
    test_map: dict[str, list[str]] = {}
    if args.test_map:
        test_map = json.loads(Path(args.test_map).read_text(encoding="utf-8"))
    if not uniform_tests and not test_map:
        raise SystemExit("mutate: one of --tests / --test-map is required")

    def test_selector(unit: UnitRef) -> list[str]:
        key = f"{unit.file}::{unit.name}"
        return test_map.get(key, uniform_tests)

    runner_cmd = (
        shlex.split(args.runner_cmd) if args.runner_cmd else list(_DEFAULT_RUNNER_CMD)
    )

    report = run_mutations(
        units,
        test_selector,
        runner_cmd,
        limit=args.limit,
        timeout=args.timeout,
        repo_root=repo_root,
    )

    payload = {
        "total_generated": report.total_generated,
        "killed": report.killed,
        "survived": report.survived,
        "timeout": report.timeout,
        "survivors": [
            {
                "unit": dataclasses.asdict(s.unit),
                "operator": s.operator,
                "diff_snippet": s.diff_snippet,
            }
            for s in report.survivors
        ],
        "skipped": [
            {
                "unit": dataclasses.asdict(s.unit),
                "operator": s.operator,
                "reason": s.reason,
            }
            for s in report.skipped
        ],
    }
    print(json.dumps(payload, indent=2))
    return 1 if report.survivors else 0


def _add_red_status_subparser(sub: argparse._SubParsersAction) -> None:
    p = sub.add_parser(
        "red-status",
        help="Classify a test's red-first evidence quality at a commit.",
    )
    p.add_argument("--test", required=True, help="pytest node id")
    p.add_argument("--commit", required=True)
    p.add_argument("--ledger", default=None)
    p.add_argument("--repo-root", default=".")


def _add_mutate_subparser(sub: argparse._SubParsersAction) -> None:
    p = sub.add_parser(
        "mutate",
        help="Run affected-set mutation testing over --changed-files.",
    )
    p.add_argument("--changed-files", required=True, help="comma-separated paths")
    p.add_argument(
        "--tests",
        default=None,
        help="comma-separated pytest ids/paths run for EVERY affected unit",
    )
    p.add_argument(
        "--test-map",
        default=None,
        help='JSON file: {"<file>::<unit_name>": ["test_id", ...]}',
    )
    p.add_argument("--limit", type=int, default=_DEFAULT_LIMIT)
    p.add_argument("--timeout", type=float, default=_DEFAULT_TIMEOUT)
    p.add_argument("--runner-cmd", default=None, help="shell-quoted argv prefix")
    p.add_argument("--repo-root", default=".")


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="factory_evidence",
        description="SPEC-033 red-first ledger + affected-set mutation testing CLI (B6).",
    )
    sub = parser.add_subparsers(dest="command", required=True)
    _add_red_status_subparser(sub)
    _add_mutate_subparser(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 = {
        "red-status": _cmd_red_status,
        "mutate": _cmd_mutate,
    }
    handler = dispatch[args.command]
    return handler(args)


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