recall_guard.harness.runner

recall_guard.harness.runner

End-to-end run orchestrator for the honest-model-ranking harness.

This module drives the current build-only harness flow:

• resolve the shortlist (--shortlist directly or --candidates via the smoke test)
• enforce the cutoff guard before any model calls
• build the control baseline and per-model MCS classifier
• evaluate the shortlisted models on the eval set
• rank the results and write the run artifacts

The successful run writes records.jsonl, summary.csv, top3.md, and manifest.json, then prints the artifact paths. When the run starts from --candidates, it also writes shortlist.json.

Key behavior:

• --shortlist skips the smoke test and does not write shortlist.json.
• The cutoff guard runs immediately after shortlist resolution and before any HTTP call to a candidate model.
• The manifest records input hashes, the seed, the resolved shortlist, the composite-score formula, MCS hyperparameters, the bootstrap count, and the artifact path map.
• Temperature-0 problems are surfaced through evaluator warnings rather than a runner-specific enforcement layer.

Pipeline summary:

1. Load .env and read NVIDIA_API_KEY. Missing key -> exit code 2.
2. Load the eval set and cutoff registry. Missing eval-set file -> exit code 2.
3. Resolve the shortlist.
4. Run assert_cutoff_safe(eval_set, shortlist, cutoffs). Any CutoffViolation aborts with exit code 3.
5. Load the IS and OOS calibration corpora.
6. Construct the optional reference-model LM via the injected lm_factory.
7. For each shortlisted model, build the control baseline. If it is not calibrated, append a stub ModelEvalResult with the uncalibrated warning. Otherwise train the MCS classifier and evaluate the model on the eval set.
8. Compute the majority baseline, rank the models, and write the run artifacts.
9. Render the terminal table and print the artifact-path summary.

run(args, *, lm_factory=...) accepts a factory (api_key, model, timeout_s) -> NvidiaLM so tests can inject a fake LM that records calls and returns scripted CompletionResult objects.

build_parser

build_parser()

Top-level CLI parser. Single build flow.

Source code in recall_guard/harness/runner.py

def build_parser() -> argparse.ArgumentParser:
    """Top-level CLI parser. Single ``build`` flow."""
    parser = argparse.ArgumentParser(
        prog="harness",
        description=(
            "Honest model ranking harness. Loads a (prompt, target_direction) "
            "JSONL, calibrates each shortlisted NVIDIA-hosted model with the "
            "paper's full MIA feature set, and produces a defensible top-3 "
            "ranking with bootstrap CIs."
        ),
    )
    _add_build_arguments(parser)
    return parser

parse_argv

parse_argv(argv)

Parse CLI arguments. build is the only mode now.

Accepts an optional leading build token for back-compat with older scripts that wrote harness build --eval-set X; it gets stripped before parsing.

Source code in recall_guard/harness/runner.py

def parse_argv(argv: list[str]) -> argparse.Namespace:
    """Parse CLI arguments. ``build`` is the only mode now.

    Accepts an optional leading ``build`` token for back-compat with
    older scripts that wrote ``harness build --eval-set X``; it gets
    stripped before parsing.
    """
    parser = build_parser()
    if argv and argv[0] == "build":
        argv = argv[1:]
    args = parser.parse_args(argv)
    if args.bootstrap_n < 1:
        parser.error("--bootstrap-n must be >= 1")
    return args

run

run(args, *, lm_factory=None)

Execute one harness run.

Returns:

Type	Description
int	Process exit code: 0 on success, 2 on missing/invalid input (eval set, API key), 3 on cutoff violation, 1 on any other unrecovered error.

Source code in recall_guard/harness/runner.py

def run(
    args: argparse.Namespace,
    *,
    lm_factory: LMFactory | None = None,
) -> int:
    """Execute one harness run.

    Returns
    -------
    int
        Process exit code: ``0`` on success, ``2`` on missing/invalid input
        (eval set, API key), ``3`` on cutoff violation, ``1`` on any other
        unrecovered error.
    """
    if lm_factory is not None:
        factory: LMFactory = lm_factory
    else:
        pace = float(getattr(args, "min_call_interval", 0.0) or 0.0)
        factory = _make_paced_factory(pace) if pace > 0 else _default_lm_factory

    load_dotenv()
    api_key = os.environ.get("NVIDIA_API_KEY")
    if not api_key:
        sys.stderr.write(
            "ERROR: NVIDIA_API_KEY is not set in the environment. "
            "Add it to your shell or .env file.\n"
        )
        return 2

    loaded = _load_all_inputs(args)
    if isinstance(loaded, int):
        return loaded

    out_dir = _resolve_out_dir(args.out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    try:
        shortlist_models, shortlist_path = _resolve_shortlist(
            args, api_key, out_dir, lm_factory=factory
        )
    except (FileNotFoundError, ValueError) as exc:
        sys.stderr.write(f"ERROR: shortlist resolution failed: {exc}\n")
        return 2
    if not shortlist_models:
        sys.stderr.write(
            "ERROR: smoke gate selected no models; aborting run before evaluation.\n"
        )
        return 2

    # Cutoff guard MUST run BEFORE the main eval/baseline/MCS work (Req 2.5).
    try:
        assert_cutoff_safe(loaded.eval_set, shortlist_models, loaded.cutoffs)
    except CutoffViolation as exc:
        sys.stderr.write(f"ERROR: cutoff violation: {exc}\n")
        return 3

    ref_lm: NvidiaLM | None = None
    if not args.no_reference:
        ref_lm = factory(api_key, args.reference_model, DEFAULT_TIMEOUT_S)

    try:
        results = _evaluate_all_models(
            shortlist_models=shortlist_models, api_key=api_key, inputs=loaded,
            ref_lm=ref_lm, factory=factory, args=args,
        )

        majority = compute_majority_baseline(
            loaded.eval_set, bootstrap_n=args.bootstrap_n, seed=args.seed
        )
        scores = composite_score(results, majority)

        artifacts = _write_run_artifacts(
            out_dir=out_dir, results=results, scores=scores, majority=majority,
            inputs=loaded, shortlist_models=shortlist_models,
            shortlist_path=shortlist_path, args=args,
        )
    except Exception as exc:  # pragma: no cover - top-level run guard
        logger.exception("runner: unrecovered error; aborting run.")
        sys.stderr.write(f"ERROR: harness run failed: {exc!r}\n")
        return 1

    try:
        render_terminal(results, majority, scores)
    except Exception:  # pragma: no cover - terminal rendering must never fail the run
        logger.exception("runner: render_terminal failed; continuing.")

    print_artifact_paths(artifacts)
    return 0