recall_guard.harness

recall_guard.harness

Public API for the harness layer of the honest-model-ranking harness.

Re-exports the consumer-facing surface (Req 12.1) so that the qualification notebook (Req 12.2) and any external scripts can import every orchestration helper, evaluator type, ranker primitive, report writer, plotting helper, and runner entry point from the package root::

from recall_guard.harness import (
    # smoke
    SmokeOutcome, Shortlist, smoke_test,
    # evaluator
    Record, CIBound, ModelEvalResult, evaluate_model,
    compute_majority_baseline,
    # ranker
    CompositeScore, COMPOSITE_FORMULA, GATES,
    composite_score, write_top3,
    # report
    render_terminal, write_records, write_summary_csv,
    print_artifact_paths,
    # plots
    configure_paper_style,
    plot_mia_feature_distributions, plot_mcs_calibration,
    plot_accuracy_with_ci, plot_mcs_auc_with_ci,
    plot_composite_ranking,
    # runner
    run, build_parser,
)

The notebook in notebooks/qualification.ipynb consumes this surface verbatim; any drift here breaks tests/harness/test_notebook.py's "public API imports succeed" smoke test (Req 12.1).

CIBound dataclass

A bootstrap point estimate plus 95% percentile bounds.

Source code in recall_guard/harness/evaluator.py

@dataclass(frozen=True)
class CIBound:
    """A bootstrap point estimate plus 95% percentile bounds."""

    point: float
    lo: float
    hi: float

ModelEvalResult dataclass

Aggregate evaluation result for one model on the eval set.

Attributes:

Name	Type	Description
model	str	NVIDIA model ID.
raw_accuracy	CIBound	Bootstrap CI on predicted_direction == target_direction over parse-OK rows (Req 6.1, 7.3).
memguard_accuracy	CIBound	Same accuracy denominator as raw_accuracy: in this spec the MemGuard penalty discounts confidence but does not change the predicted direction, so the two CIs coincide. The field exists so a future confidence-thresholded variant can diverge without breaking downstream consumers.
mcs_auc	CIBound	Bootstrap CI over the (p_memorized, label) pairs in holdout_records. Falls back to mcs.holdout_auc (point only) when no holdout records are supplied (design § Implementation Notes).
parse_success_rate	float	Fraction of rows with parse_ok=True (Req 7.2). 1.0 for empty eval sets (vacuously true).
parse_failures	int	Count of rows with parse_ok=False (Req 7.1).
warnings	list[str]	Subset of {"temperature-not-honoured"}. Other warnings (weak-calibration, parse-unreliable, not-better-than-baseline, uncalibrated) are added by the ranker.
records	list[Record]	Per-row records in eval-set order.

Source code in recall_guard/harness/evaluator.py

@dataclass(frozen=True)
class ModelEvalResult:
    """Aggregate evaluation result for one model on the eval set.

    Attributes
    ----------
    model:
        NVIDIA model ID.
    raw_accuracy:
        Bootstrap CI on ``predicted_direction == target_direction`` over
        parse-OK rows (Req 6.1, 7.3).
    memguard_accuracy:
        Same accuracy denominator as ``raw_accuracy``: in this spec the
        MemGuard penalty discounts confidence but does not change the
        predicted direction, so the two CIs coincide. The field exists so a
        future confidence-thresholded variant can diverge without breaking
        downstream consumers.
    mcs_auc:
        Bootstrap CI over the ``(p_memorized, label)`` pairs in
        ``holdout_records``. Falls back to ``mcs.holdout_auc`` (point only)
        when no holdout records are supplied (design § Implementation Notes).
    parse_success_rate:
        Fraction of rows with ``parse_ok=True`` (Req 7.2). ``1.0`` for empty
        eval sets (vacuously true).
    parse_failures:
        Count of rows with ``parse_ok=False`` (Req 7.1).
    warnings:
        Subset of ``{"temperature-not-honoured"}``. Other warnings
        (weak-calibration, parse-unreliable, not-better-than-baseline,
        uncalibrated) are added by the ranker.
    records:
        Per-row records in eval-set order.
    """

    model: str
    raw_accuracy: CIBound
    memguard_accuracy: CIBound
    mcs_auc: CIBound
    parse_success_rate: float
    parse_failures: int
    warnings: list[str]
    records: list[Record]

Record dataclass

Per-(model, prompt) record produced by evaluate_model.

Attributes:

Name	Type	Description
model	str	NVIDIA model ID this record was scored for.
prompt_hash	str	First 16 hex chars of sha256(prompt); keeps records.jsonl readable without leaking full prompts. Uniquely keys the record together with model (design § Data Models).
parse_ok	bool	True when Direction: and Confidence: both parsed strictly and MIA features were computable.
predicted_direction	int | None	Parsed integer in {-1, 0, 1}; None when parse_ok is False.
raw_confidence	float | None	Parsed confidence in [0, 1]; None on parse failure.
penalized_confidence	float | None	raw_confidence * (1 - p_memorized) (Req 5.4); None on failure.
target_direction	int	Ground-truth direction copied from the eval row (always populated).
features_raw	MiaFeatures | None	:class:MiaFeatures instance; None when the LM call failed before feature computation could run.
features_standardised	dict[str, float | None] | None	Per-feature standardised values (z-score against the model's baseline); None whenever features_raw is None.
p_memorized	float | None	p(memorized | features) ∈ [0, 1]; None when features_raw is None.
fail_reason	str | None	One of "timeout" / "no_logprobs" / "parse_failure" / "error" on failure; None on success.
raw_response_excerpt	str | None	First ~400 chars of the raw model response when the row failed to parse. Always None for parse-OK rows (no need to bloat the artifact). Use this to inspect why a parse failed without re-running the model.

Source code in recall_guard/harness/evaluator.py

@dataclass(frozen=True)
class Record:
    """Per-(model, prompt) record produced by ``evaluate_model``.

    Attributes
    ----------
    model:
        NVIDIA model ID this record was scored for.
    prompt_hash:
        First 16 hex chars of ``sha256(prompt)``; keeps records.jsonl readable
        without leaking full prompts. Uniquely keys the record together with
        ``model`` (design § Data Models).
    parse_ok:
        ``True`` when ``Direction:`` and ``Confidence:`` both parsed strictly
        and MIA features were computable.
    predicted_direction:
        Parsed integer in ``{-1, 0, 1}``; ``None`` when ``parse_ok`` is False.
    raw_confidence:
        Parsed confidence in ``[0, 1]``; ``None`` on parse failure.
    penalized_confidence:
        ``raw_confidence * (1 - p_memorized)`` (Req 5.4); ``None`` on failure.
    target_direction:
        Ground-truth direction copied from the eval row (always populated).
    features_raw:
        :class:`MiaFeatures` instance; ``None`` when the LM call failed before
        feature computation could run.
    features_standardised:
        Per-feature standardised values (z-score against the model's baseline);
        ``None`` whenever ``features_raw`` is ``None``.
    p_memorized:
        ``p(memorized | features) ∈ [0, 1]``; ``None`` when ``features_raw``
        is ``None``.
    fail_reason:
        One of ``"timeout"`` / ``"no_logprobs"`` / ``"parse_failure"`` /
        ``"error"`` on failure; ``None`` on success.
    raw_response_excerpt:
        First ~400 chars of the raw model response when the row failed to
        parse. Always ``None`` for parse-OK rows (no need to bloat the
        artifact). Use this to inspect *why* a parse failed without re-running
        the model.
    """

    model: str
    prompt_hash: str
    parse_ok: bool
    predicted_direction: int | None
    raw_confidence: float | None
    penalized_confidence: float | None
    target_direction: int
    features_raw: MiaFeatures | None
    features_standardised: dict[str, float | None] | None
    p_memorized: float | None
    fail_reason: str | None
    raw_response_excerpt: str | None = None

CompositeScore dataclass

Composite rank score for one model with gate verdict + warnings.

Attributes:

Name	Type	Description
model	str	NVIDIA model ID this score belongs to.
score	float	Multiplicative composite memguard_acc_lo * mcs_auc_point * parse_success_rate if all gates pass; 0.0 otherwise (Req 8.1).
components	dict[str, float]	The three component values that fed the formula. Keys are stable: "memguard_acc_lo", "mcs_auc_point", "parse_success_rate".
survives_gates	bool	True iff none of weak-calibration, parse-unreliable, not-better-than-baseline, uncalibrated are in warnings. temperature-not-honoured does not affect this flag.
warnings	list[str]	Subset of the ranker warning vocabulary, including any informational warnings (temperature-not-honoured) passed through from the evaluator.

Source code in recall_guard/harness/ranker.py

@dataclass(frozen=True)
class CompositeScore:
    """Composite rank score for one model with gate verdict + warnings.

    Attributes
    ----------
    model:
        NVIDIA model ID this score belongs to.
    score:
        Multiplicative composite ``memguard_acc_lo * mcs_auc_point *
        parse_success_rate`` if all gates pass; ``0.0`` otherwise (Req 8.1).
    components:
        The three component values that fed the formula. Keys are stable:
        ``"memguard_acc_lo"``, ``"mcs_auc_point"``, ``"parse_success_rate"``.
    survives_gates:
        ``True`` iff none of ``weak-calibration``, ``parse-unreliable``,
        ``not-better-than-baseline``, ``uncalibrated`` are in ``warnings``.
        ``temperature-not-honoured`` does not affect this flag.
    warnings:
        Subset of the ranker warning vocabulary, including any
        informational warnings (``temperature-not-honoured``) passed through
        from the evaluator.
    """

    model: str
    score: float
    components: dict[str, float]
    survives_gates: bool
    warnings: list[str]

Shortlist dataclass

Result of the smoke-test gate.

selected contains the passing models in candidate order, capped at max_size. outcomes contains one entry per candidate (regardless of pass/fail) so the runner can persist a reproducible artifact (Req 1.4).

Source code in recall_guard/harness/smoke.py

@dataclass(frozen=True)
class Shortlist:
    """Result of the smoke-test gate.

    `selected` contains the passing models in candidate order, capped at
    `max_size`. `outcomes` contains one entry per candidate (regardless of
    pass/fail) so the runner can persist a reproducible artifact (Req 1.4).
    """

    selected: list[str]
    outcomes: list[SmokeOutcome]

SmokeOutcome dataclass

Per-candidate smoke-test outcome.

fail_reason is None on pass and one of "timeout", "no_logprobs", "parse_failure", or "error" on fail.

Source code in recall_guard/harness/smoke.py

@dataclass(frozen=True)
class SmokeOutcome:
    """Per-candidate smoke-test outcome.

    `fail_reason` is `None` on pass and one of `"timeout"`, `"no_logprobs"`,
    `"parse_failure"`, or `"error"` on fail.
    """

    model: str
    passed: bool
    fail_reason: str | None

compute_majority_baseline

compute_majority_baseline(
    eval_set, bootstrap_n=1000, seed=0
)

Bootstrap CI on the majority-class baseline accuracy (Req 6.2).

Returns CIBound(0.0, 0.0, 0.0) for an empty eval set.

Source code in recall_guard/harness/evaluator.py

def compute_majority_baseline(
    eval_set: EvalSet,
    bootstrap_n: int = 1000,
    seed: int = 0,
) -> CIBound:
    """Bootstrap CI on the majority-class baseline accuracy (Req 6.2).

    Returns ``CIBound(0.0, 0.0, 0.0)`` for an empty eval set.
    """
    rows = eval_set.rows
    if not rows:
        return CIBound(0.0, 0.0, 0.0)

    counts = Counter(r.target_direction for r in rows)
    majority_class, _ = counts.most_common(1)[0]
    indicators: list[int] = [1 if r.target_direction == majority_class else 0 for r in rows]

    def _mean(samples: list[int]) -> float:
        return sum(samples) / len(samples)

    point, lo, hi = bootstrap_ci(
        samples=indicators,
        statistic=_mean,
        n_resamples=bootstrap_n,
        seed=seed,
    )
    return CIBound(point=point, lo=lo, hi=hi)

evaluate_model

evaluate_model(
    model_lm,
    eval_set,
    baseline,
    mcs,
    ref_lm,
    holdout_records=None,
    bootstrap_n=1000,
    seed=0,
    max_workers=1,
)

Score one model against eval_set and assemble a ModelEvalResult.

With max_workers > 1 the per-row primary + reference LM calls fan out via concurrent.futures.ThreadPoolExecutor (results are paired with rows by index, so order is preserved). Post-processing (parsing, MIA feature compute, MCS scoring) runs serially.

See module docstring for the row-level pipeline. The function performs no I/O beyond the model HTTP calls; all artifact writing is owned by harness.report and harness.runner.

Source code in recall_guard/harness/evaluator.py

def evaluate_model(
    model_lm: NvidiaLM,
    eval_set: EvalSet,
    baseline: ControlBaseline,
    mcs: MCSCalibrator,
    ref_lm: NvidiaLM | None,
    holdout_records: list[Record] | None = None,
    bootstrap_n: int = 1000,
    seed: int = 0,
    max_workers: int = 1,
) -> ModelEvalResult:
    """Score one model against ``eval_set`` and assemble a ``ModelEvalResult``.

    With ``max_workers > 1`` the per-row primary + reference LM calls
    fan out via ``concurrent.futures.ThreadPoolExecutor`` (results are
    paired with rows by index, so order is preserved). Post-processing
    (parsing, MIA feature compute, MCS scoring) runs serially.

    See module docstring for the row-level pipeline. The function performs no
    I/O beyond the model HTTP calls; all artifact writing is owned by
    ``harness.report`` and ``harness.runner``.
    """
    model_id = getattr(model_lm, "model", "<unknown>")

    prompts = [row.prompt for row in eval_set.rows]
    primary_results = generate_many(model_lm, prompts, max_workers=max_workers)
    ref_results: list = (
        generate_many(ref_lm, prompts, max_workers=max_workers)
        if ref_lm is not None else [None] * len(prompts)
    )

    records: list[Record] = []
    temperature_violated = False
    for row, primary, ref_res in zip(eval_set.rows, primary_results, ref_results, strict=True):
        record, row_temp_violated = _score_row(
            model_id=model_id, row=row, primary=primary, ref_res=ref_res,
            ref_lm=ref_lm, baseline=baseline, mcs=mcs,
        )
        records.append(record)
        if row_temp_violated:
            temperature_violated = True

    n_rows = len(records)
    parse_failures = sum(1 for r in records if not r.parse_ok)
    parse_ok_records = [r for r in records if r.parse_ok]
    parse_success_rate = (
        1.0 if n_rows == 0 else (n_rows - parse_failures) / n_rows
    )

    raw_accuracy = _accuracy_ci(parse_ok_records, n=bootstrap_n, seed=seed)
    # MemGuard accuracy uses the same parse-OK denominator. The penalty
    # affects confidence only (predicted_direction is unchanged in this
    # spec), so the bootstrap statistic is identical. The dataclass field is
    # kept distinct so a future confidence-threshold variant may diverge
    # without breaking the report schema.
    memguard_accuracy = _accuracy_ci(parse_ok_records, n=bootstrap_n, seed=seed)
    mcs_auc = _mcs_auc_ci(holdout_records, mcs, n=bootstrap_n, seed=seed)

    warnings: list[str] = []
    if temperature_violated:
        warnings.append(WARNING_TEMPERATURE_NOT_HONOURED)

    return ModelEvalResult(
        model=model_id,
        raw_accuracy=raw_accuracy,
        memguard_accuracy=memguard_accuracy,
        mcs_auc=mcs_auc,
        parse_success_rate=parse_success_rate,
        parse_failures=parse_failures,
        warnings=warnings,
        records=records,
    )

composite_score

composite_score(
    results,
    majority_baseline,
    formula=COMPOSITE_FORMULA,
    gates=GATES,
)

Convert per-model evaluation results into composite scores.

The formula argument is persisted alongside the gates in top3.md (Req 8.4) but is not parsed at runtime; the multiplicative formula is the only one defined in this spec. A future variant would change both the component dict keys and the formula string in lockstep.

Returns the scores in the input order of results; ordering is deferred to write_top3 (or other downstream consumers) so the canonical record stream stays aligned with the eval-set order (Req 9.3).

Source code in recall_guard/harness/ranker.py

def composite_score(
    results: list[ModelEvalResult],
    majority_baseline: CIBound,
    formula: str = COMPOSITE_FORMULA,
    gates: dict[str, float] = GATES,
) -> list[CompositeScore]:
    """Convert per-model evaluation results into composite scores.

    The ``formula`` argument is persisted alongside the gates in ``top3.md``
    (Req 8.4) but is not parsed at runtime; the multiplicative formula is the
    only one defined in this spec. A future variant would change both the
    component dict keys and the formula string in lockstep.

    Returns the scores in the *input order* of ``results``; ordering is
    deferred to ``write_top3`` (or other downstream consumers) so the
    canonical record stream stays aligned with the eval-set order (Req 9.3).
    """
    del formula  # Persisted via write_top3; not interpreted here.

    scores: list[CompositeScore] = []
    for result in results:
        components: dict[str, float] = {
            "memguard_acc_lo": float(result.memguard_accuracy.lo),
            "mcs_auc_point": float(result.mcs_auc.point),
            "parse_success_rate": float(result.parse_success_rate),
        }

        warnings = _gate_warnings(result, majority_baseline, gates)
        survives = not any(w in _BLOCKING_WARNINGS for w in warnings)

        if survives:
            score_value = (
                components["memguard_acc_lo"]
                * components["mcs_auc_point"]
                * components["parse_success_rate"]
            )
        else:
            score_value = 0.0

        scores.append(
            CompositeScore(
                model=result.model,
                score=float(score_value),
                components=components,
                survives_gates=survives,
                warnings=warnings,
            )
        )

    return scores

write_top3

write_top3(
    scores, path, formula=COMPOSITE_FORMULA, gates=GATES
)

Write top3.md to path.

The file always contains the # Top 3 Models heading and the ## Composite score formula footer; the explanatory section is included whenever fewer than three models survive (Req 8.3). The parent directory is created if missing so callers do not have to mkdir first.

Source code in recall_guard/harness/ranker.py

def write_top3(
    scores: list[CompositeScore],
    path: Path,
    formula: str = COMPOSITE_FORMULA,
    gates: dict[str, float] = GATES,
) -> None:
    """Write ``top3.md`` to ``path``.

    The file always contains the ``# Top 3 Models`` heading and the
    ``## Composite score formula`` footer; the explanatory section is
    included whenever fewer than three models survive (Req 8.3). The parent
    directory is created if missing so callers do not have to mkdir first.
    """
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)

    # Stable sort: descending by score, ties broken by input order.
    # ``sorted`` in CPython is stable, so reversing the indexed pairs is not
    # needed; we sort by negated score directly.
    indexed: list[tuple[int, CompositeScore]] = list(enumerate(scores))
    indexed.sort(key=lambda pair: -pair[1].score)
    sorted_scores = [s for _, s in indexed]

    survivors = [s for s in sorted_scores if s.survives_gates]
    nonsurvivors_in_input_order = [s for s in scores if not s.survives_gates]
    top_survivors = survivors[:3]

    lines: list[str] = ["# Top 3 Models", ""]

    if not top_survivors:
        lines.append("_No models passed all gates._")
        lines.append("")
    else:
        for rank, score in enumerate(top_survivors, start=1):
            lines.extend(_render_survivor_block(rank, score))
            lines.append("")

    if len(top_survivors) < 3:
        lines.extend(
            _render_short_list_explanation(
                total_evaluated=len(scores),
                nonsurvivors=nonsurvivors_in_input_order,
            )
        )

    lines.extend(_render_formula_footer(formula, gates))

    path.write_text("\n".join(lines), encoding="utf-8")

print_artifact_paths

print_artifact_paths(paths, console=None)

Print the final Artifacts: summary block (Req 9.4).

Each key/value pair is rendered as <name> <path> so the operator can copy paths directly out of the terminal.

Source code in recall_guard/harness/report.py

def print_artifact_paths(
    paths: dict[str, Path], console: Console | None = None
) -> None:
    """Print the final ``Artifacts:`` summary block (Req 9.4).

    Each key/value pair is rendered as ``<name>  <path>`` so the
    operator can copy paths directly out of the terminal.
    """
    target = console or Console()
    target.print("Artifacts:")
    for name, path in paths.items():
        target.print(f"  {name}\t{path}")

render_terminal

render_terminal(results, majority, scores, console=None)

Print one table row per model plus a majority-baseline row (Req 9.1, 9.2).

Rows for surviving + non-surviving models are sorted by score descending (stable within ties on input order); the __majority_baseline__ row always renders last so a reader can compare every model against it visually.

The majority row populates only the Raw Acc CI column; the other cells are em-dashes since MemGuard accuracy, MCS-AUC, and the composite score are not defined for the baseline.

Parameters:

Name	Type	Description	Default
results	list[ModelEvalResult]	Aligned by model ID via lookup (not by index) so a missing score does not silently misalign rows.	required
scores	list[ModelEvalResult]	Aligned by model ID via lookup (not by index) so a missing score does not silently misalign rows.	required
majority	CIBound	Bootstrap CI on the majority-class baseline accuracy from compute_majority_baseline.	required
console	Console | None	Optional rich.console.Console injection point for tests; defaults to Console() (writes to stdout).	None

Source code in recall_guard/harness/report.py

def render_terminal(
    results: list[ModelEvalResult],
    majority: CIBound,
    scores: list[CompositeScore],
    console: Console | None = None,
) -> None:
    """Print one table row per model plus a majority-baseline row (Req 9.1, 9.2).

    Rows for surviving + non-surviving models are sorted by ``score``
    descending (stable within ties on input order); the
    ``__majority_baseline__`` row always renders last so a reader can compare
    every model against it visually.

    The majority row populates only the Raw Acc CI column; the other cells
    are em-dashes since MemGuard accuracy, MCS-AUC, and the composite score
    are not defined for the baseline.

    Parameters
    ----------
    results, scores:
        Aligned by model ID via lookup (not by index) so a missing score does
        not silently misalign rows.
    majority:
        Bootstrap CI on the majority-class baseline accuracy from
        ``compute_majority_baseline``.
    console:
        Optional ``rich.console.Console`` injection point for tests; defaults
        to ``Console()`` (writes to stdout).
    """
    table = Table(show_header=True, header_style="bold")
    table.add_column("Model")
    table.add_column("Raw Acc (CI)")
    table.add_column("MemGuard Acc (CI)")
    table.add_column("MCS-AUC (CI)")
    table.add_column("Parse %")
    table.add_column("Score")
    table.add_column("Warnings")

    # Stable sort: descending by score (None / missing scores treated as 0.0
    # so they fall to the bottom but stay above the majority row).
    indexed = list(enumerate(results))

    def _sort_key(pair: tuple[int, ModelEvalResult]) -> tuple[float, int]:
        idx, result = pair
        score = _score_for_model(result.model, scores)
        score_value = score.score if score is not None else 0.0
        # Negate score for descending order while keeping idx ascending for
        # stability on ties.
        return (-score_value, idx)

    indexed.sort(key=_sort_key)
    sorted_results = [r for _, r in indexed]

    for result in sorted_results:
        score = _score_for_model(result.model, scores)
        # Warnings shown in the terminal merge evaluator + ranker warnings;
        # the ranker passes through informational ones so we deduplicate while
        # preserving order.
        warning_set: list[str] = []
        for w in (result.warnings or []) + (score.warnings if score else []):
            if w not in warning_set:
                warning_set.append(w)

        table.add_row(
            result.model,
            _format_ci(result.raw_accuracy),
            _format_ci(result.memguard_accuracy),
            _format_ci(result.mcs_auc),
            _format_percent(result.parse_success_rate),
            _format_score(score),
            _format_warnings(warning_set),
        )

    # Majority-baseline row: only the Raw Acc CI is meaningful.
    em_dash = "—"
    table.add_row(
        MAJORITY_LABEL,
        _format_ci(majority),
        em_dash,
        em_dash,
        em_dash,
        em_dash,
        "",
    )

    # Auto-detect terminal width when stdout is a TTY; fall back to a wide
    # 200-col console for redirected/captured stdout (pytest, pipes) so the
    # model column and warning strings do not get truncated. Callers can
    # inject their own ``Console`` to override this.
    target = console or Console(width=shutil.get_terminal_size((200, 20)).columns)
    target.print(table)

write_records

write_records(results, path)

Stream every Record from every result to records.jsonl (Req 9.3).

Memory stays bounded for long runs because the writer opens the file once and emits one json.dumps line per record before moving to the next; no all-records list is built in memory.

The schema is documented in :func:_record_to_jsonable and audited by tests/harness/test_report.py::test_write_records_includes_all_required_fields.

Source code in recall_guard/harness/report.py

def write_records(results: Iterable[ModelEvalResult], path: Path) -> None:
    """Stream every ``Record`` from every result to ``records.jsonl`` (Req 9.3).

    Memory stays bounded for long runs because the writer opens the file once
    and emits one ``json.dumps`` line per record before moving to the next;
    no all-records list is built in memory.

    The schema is documented in :func:`_record_to_jsonable` and audited by
    ``tests/harness/test_report.py::test_write_records_includes_all_required_fields``.
    """
    target = Path(path)
    _ensure_parent(target)
    with target.open("w", encoding="utf-8") as fh:
        for result in results:
            for record in result.records:
                payload = _record_to_jsonable(record)
                fh.write(json.dumps(payload, ensure_ascii=False))
                fh.write("\n")

write_summary_csv

write_summary_csv(results, scores, majority, path)

Write one CSV row per model plus a majority-baseline row (Req 9.3).

The 15-column schema is fixed in :data:SUMMARY_CSV_COLUMNS; every CSV consumer (e.g. the qualification notebook) can rely on it.

The majority row only fills the raw-accuracy CI cells; the rest are blank because there is no MemGuard accuracy / MCS-AUC / score notion for the baseline (this matches the design's "majority row alongside model rows" interpretation).

Notes

The function signature includes majority even though the design's Service Interface lists only three arguments; Task 4.4's observable requires the majority row in the CSV, which forces the parameter through.

Source code in recall_guard/harness/report.py

def write_summary_csv(
    results: list[ModelEvalResult],
    scores: list[CompositeScore],
    majority: CIBound,
    path: Path,
) -> None:
    """Write one CSV row per model plus a majority-baseline row (Req 9.3).

    The 15-column schema is fixed in :data:`SUMMARY_CSV_COLUMNS`; every CSV
    consumer (e.g. the qualification notebook) can rely on it.

    The majority row only fills the raw-accuracy CI cells; the rest are blank
    because there is no MemGuard accuracy / MCS-AUC / score notion for the
    baseline (this matches the design's "majority row alongside model rows"
    interpretation).

    Notes
    -----
    The function signature includes ``majority`` even though the design's
    Service Interface lists only three arguments; Task 4.4's observable
    requires the majority row in the CSV, which forces the parameter through.
    """
    target = Path(path)
    _ensure_parent(target)

    with target.open("w", encoding="utf-8", newline="") as fh:
        writer = csv.DictWriter(
            fh, fieldnames=SUMMARY_CSV_COLUMNS, lineterminator="\n"
        )
        writer.writeheader()

        for result in results:
            score = _score_for_model(result.model, scores)
            writer.writerow(_result_row(result, score))

        writer.writerow(_majority_row(majority))

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

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

smoke_test

smoke_test(
    candidates,
    api_key,
    smoke_prompts,
    max_size=10,
    timeout_s=DEFAULT_TIMEOUT_S,
    lm_factory=None,
)

Run the smoke-test gate over candidates and return a Shortlist.

Parameters:

Name	Type	Description	Default
candidates	list[str]	Ordered candidate model IDs to evaluate.	required
api_key	str	NVIDIA API key forwarded to the LM factory.	required
smoke_prompts	list[str]	The fixed smoke prompts. Every candidate runs every prompt unless an exclusion fires earlier (Req 1.2).	required
max_size	int	Hard cap on Shortlist.selected (Req 1.1; default 10).	10
timeout_s	float	Per-call timeout forwarded to the LM client (Req 1.2).	DEFAULT_TIMEOUT_S
lm_factory	LMFactory | None	Optional (api_key, model, timeout_s) -> NvidiaLM factory used for test injection. Defaults to the real NvidiaLM constructor.	None

Returns:

Type	Description
Shortlist	selected capped at max_size, plus one SmokeOutcome per candidate. The function performs no I/O; the runner persists outcomes to shortlist.json (Req 1.4).

Source code in recall_guard/harness/smoke.py

def smoke_test(
    candidates: list[str],
    api_key: str,
    smoke_prompts: list[str],
    max_size: int = 10,
    timeout_s: float = DEFAULT_TIMEOUT_S,
    lm_factory: LMFactory | None = None,
) -> Shortlist:
    """Run the smoke-test gate over `candidates` and return a Shortlist.

    Parameters
    ----------
    candidates:
        Ordered candidate model IDs to evaluate.
    api_key:
        NVIDIA API key forwarded to the LM factory.
    smoke_prompts:
        The fixed smoke prompts. Every candidate runs every prompt unless an
        exclusion fires earlier (Req 1.2).
    max_size:
        Hard cap on `Shortlist.selected` (Req 1.1; default 10).
    timeout_s:
        Per-call timeout forwarded to the LM client (Req 1.2).
    lm_factory:
        Optional `(api_key, model, timeout_s) -> NvidiaLM` factory used for
        test injection. Defaults to the real `NvidiaLM` constructor.

    Returns
    -------
    Shortlist
        `selected` capped at `max_size`, plus one `SmokeOutcome` per
        candidate. The function performs no I/O; the runner persists
        `outcomes` to `shortlist.json` (Req 1.4).
    """
    factory: LMFactory = lm_factory or _default_lm_factory
    outcomes: list[SmokeOutcome] = []
    selected: list[str] = []

    for model in candidates:
        outcome = _smoke_one(model, api_key, smoke_prompts, timeout_s, factory)
        outcomes.append(outcome)
        if outcome.passed and len(selected) < max_size:
            selected.append(model)

    return Shortlist(selected=selected, outcomes=outcomes)