recall_guard.core

recall_guard.core

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

Re-exports the consumer-facing surface (Req 12.1) so that callers (the qualification notebook, future external scripts, and the harness layers themselves) can import every primitive from the package root without ever touching internal module paths::

from recall_guard.core import (
    NvidiaLM, CompletionResult, TokenLogprob,
    EvalRow, EvalSet, load_eval_set, load_cutoffs,
    assert_cutoff_safe, CutoffViolation,
    bootstrap_ci,
    Manifest, write_manifest, read_manifest, compute_file_hash,
)

The __all__ list pins the documented names so from recall_guard.core import * behaves predictably and so a typo in a re-exported name fails fast at import time rather than at the first downstream lookup.

MultimodalVerdict dataclass

Outcome of the separated-cluster check for one component.

separated is the gate: when it is true the component holds two clusters with a genuine gap between them, there is no single location to estimate, and a location estimator must not be run at all.

Source code in recall_guard/core/consensus.py

@dataclass(frozen=True)
class MultimodalVerdict:
    """Outcome of the separated-cluster check for one component.

    ``separated`` is the gate: when it is true the component holds two clusters
    with a genuine gap between them, there is no single location to estimate,
    and a location estimator must not be run at all.
    """

    separated: bool
    lower_mass: float
    upper_mass: float
    trough_mass: float
    gap: tuple[float, float] | None

Tail

Bases: StrEnum

Which tail an interval's confidence level refers to.

This must be declared rather than assumed: the draw count needed to certify an agreement target differs substantially between the two conventions, so a feasibility check evaluated against the wrong one is meaningless.

Source code in recall_guard/core/consensus.py

class Tail(StrEnum):
    """Which tail an interval's confidence level refers to.

    This must be declared rather than assumed: the draw count needed to certify
    an agreement target differs substantially between the two conventions, so a
    feasibility check evaluated against the wrong one is meaningless.
    """

    ONE_SIDED = "one_sided"
    TWO_SIDED = "two_sided"

CostEstimate dataclass

What an ensemble would cost, computed without issuing anything.

Source code in recall_guard/core/ensemble.py

@dataclass(frozen=True)
class CostEstimate:
    """What an ensemble would cost, computed without issuing anything."""

    worst_case_requests: int
    estimated_seconds: float | None

EnsembleResult dataclass

One ensemble's reduced answer plus the evidence behind it.

component_verdicts carries the separated-cluster check for every component, not only the flagged ones. A verdict of separated=False with masses near the threshold is a very different situation from one with no mass on either side, and only the caller can judge which matters -- so the result reports what the test saw rather than only its boolean conclusion. A None verdict means the check did not run at all.

max_tokens and temperature record the settings the draws were taken under. An ensemble is an audit artifact, and "under what generation settings" belongs next to the draw-set digest: a consensus sampled at a different token budget than production is not measuring the production decision.

sampled_at records when the draws were taken, and is None for a result produced by replaying a stored draw set -- which is the honest answer, because a replay was not sampled. It exists because the sampled distribution moves between sessions as well as within one: the same prompt against the same model id has been observed to shift a component's median materially over two days. So a consensus has a shelf life, draws_sha256 pins which draws produced it but nothing else pins when, and a stored corpus is not ground truth against which to judge a fresh ensemble.

Source code in recall_guard/core/ensemble.py

@dataclass(frozen=True)
class EnsembleResult:
    """One ensemble's reduced answer plus the evidence behind it.

    ``component_verdicts`` carries the separated-cluster check for **every**
    component, not only the flagged ones. A verdict of ``separated=False`` with
    masses near the threshold is a very different situation from one with no
    mass on either side, and only the caller can judge which matters -- so the
    result reports what the test saw rather than only its boolean conclusion. A
    ``None`` verdict means the check did not run at all.

    ``max_tokens`` and ``temperature`` record the settings the draws were taken
    under. An ensemble is an audit artifact, and "under what generation settings"
    belongs next to the draw-set digest: a consensus sampled at a different token
    budget than production is not measuring the production decision.

    ``sampled_at`` records when the draws were taken, and is ``None`` for a
    result produced by replaying a stored draw set -- which is the honest
    answer, because a replay was not sampled. It exists because the sampled
    distribution moves *between* sessions as well as within one: the same prompt
    against the same model id has been observed to shift a component's median
    materially over two days. So a consensus has a shelf life, ``draws_sha256``
    pins *which* draws produced it but nothing else pins *when*, and a stored
    corpus is not ground truth against which to judge a fresh ensemble.
    """

    consensus: CompletionResult
    location: Mapping[str, float]
    location_snapped: Mapping[str, float] | None
    grid_adherence: Mapping[str, float] | None
    multimodal: tuple[str, ...]
    component_verdicts: tuple[tuple[str, MultimodalVerdict | None], ...]
    agreement: float
    agreement_ci: tuple[float, float] | None
    draw_dependence: float | None
    max_tokens: int | None
    temperature: float | None
    sampled_at: str | None
    n_requested: int
    n_parsed: int
    fail_counts: tuple[tuple[str, int], ...]
    draws_sha256: str
    draws: tuple[CompletionResult, ...] = field(default=())

EnsembleSpec dataclass

Opt-in ensemble configuration.

Every default here is provisional: all of them were calibrated against a single measurement date at a crisis onset, chosen because it was the hard case. Whether they generalise to calmer regimes is unmeasured, which is why each threshold is a field rather than a literal.

max_tokens and temperature default to None, meaning the client's own defaults. Set them to whatever production uses. An ensemble drawn at a different token budget is not measuring the production decision -- and on a reasoning model the budget is not a detail, because the chain of thought consumes it and truncates the reply before the payload a caller parses. Measured on one such model, dropping from a 2048-token production budget to the 512-token client default took the parse rate from 95% to 48%.

draws is sized for agreement precision, not for component-split detection; those are different numbers and the second is larger. See :func:~recall_guard.core.consensus.smallest_detectable_split_n.

Source code in recall_guard/core/ensemble.py

@dataclass(frozen=True)
class EnsembleSpec:
    """Opt-in ensemble configuration.

    Every default here is **provisional**: all of them were calibrated against a
    single measurement date at a crisis onset, chosen because it was the hard
    case. Whether they generalise to calmer regimes is unmeasured, which is why
    each threshold is a field rather than a literal.

    ``max_tokens`` and ``temperature`` default to ``None``, meaning the client's
    own defaults. **Set them to whatever production uses.** An ensemble drawn at
    a different token budget is not measuring the production decision -- and on a
    reasoning model the budget is not a detail, because the chain of thought
    consumes it and truncates the reply before the payload a caller parses.
    Measured on one such model, dropping from a 2048-token production budget to
    the 512-token client default took the parse rate from 95% to 48%.

    ``draws`` is sized for **agreement precision**, not for component-split
    detection; those are different numbers and the second is larger. See
    :func:`~recall_guard.core.consensus.smallest_detectable_split_n`.
    """

    draws: int = 64
    max_workers: int = 8
    min_parsed: int = 24
    grid: float | None = None
    confidence: float = 0.95
    tail: Tail = Tail.ONE_SIDED
    agreement_target: float | None = None
    location_mode: LocationMode = LocationMode.MEDIAN
    trim: float = 0.25
    multimodal_action: MultimodalAction = MultimodalAction.FLAG
    mass_min: float = 0.25
    trough_steps: int = 3
    density_ratio: float = 10.0
    min_cluster_draws: int = 8
    min_cluster_density: float = 1.5
    max_total_requests: int | None = None
    max_transport_failure_ratio: float = 0.25
    retain_draws: bool = False
    reference_mode: ReferenceMode = ReferenceMode.FIXED
    max_tokens: int | None = None
    temperature: float | None = None

    def __post_init__(self) -> None:
        if self.draws < 1:
            raise ValueError(f"draws must be >= 1; got {self.draws}")
        if not 1 <= self.min_parsed <= self.draws:
            raise ValueError(
                f"min_parsed must satisfy 1 <= min_parsed <= draws; "
                f"got min_parsed={self.min_parsed}, draws={self.draws}"
            )
        if self.max_workers < 1:
            raise ValueError(f"max_workers must be >= 1; got {self.max_workers}")
        if self.grid is not None and not (self.grid > 0 and math.isfinite(self.grid)):
            raise ValueError(f"grid must be a positive finite number; got {self.grid!r}")
        if not 0.0 < self.confidence < 1.0:
            raise ValueError(f"confidence must be in (0, 1); got {self.confidence}")
        if not 0.0 < self.mass_min <= 0.5:
            raise ValueError(f"mass_min must be in (0, 0.5]; got {self.mass_min}")
        if self.trough_steps < 1:
            raise ValueError(f"trough_steps must be >= 1; got {self.trough_steps}")
        if self.density_ratio < 1.0:
            raise ValueError(f"density_ratio must be >= 1; got {self.density_ratio}")
        if self.min_cluster_draws < 2:
            raise ValueError(
                f"min_cluster_draws must be >= 2; got {self.min_cluster_draws}"
            )
        if self.min_cluster_density < 1.0:
            raise ValueError(
                f"min_cluster_density must be >= 1; got {self.min_cluster_density}"
            )
        if not 0.0 <= self.trim < 0.5:
            raise ValueError(f"trim must be in [0, 0.5); got {self.trim}")
        if not 0.0 <= self.max_transport_failure_ratio <= 1.0:
            raise ValueError(
                f"max_transport_failure_ratio must be in [0, 1]; "
                f"got {self.max_transport_failure_ratio}"
            )
        if self.max_tokens is not None and self.max_tokens < 1:
            raise ValueError(f"max_tokens must be >= 1 when set; got {self.max_tokens}")
        if self.temperature is not None and not 0.0 <= self.temperature <= 2.0:
            raise ValueError(
                f"temperature must be in [0, 2] when set; got {self.temperature}"
            )
        if self.max_total_requests is not None and self.max_total_requests < 1:
            raise ValueError(
                f"max_total_requests must be >= 1 when set; got {self.max_total_requests}"
            )

        floor = self.smallest_certifiable_n
        if floor is not None and floor > self.draws:
            raise ValueError(
                f"agreement_target={self.agreement_target} cannot be certified at "
                f"draws={self.draws} under {self.tail.value} confidence "
                f"{self.confidence}: at least {floor} unanimous draws are required. "
                "Raise draws, lower the target, or drop it."
            )

    @property
    def smallest_certifiable_n(self) -> int | None:
        """Draws needed to certify ``agreement_target``, or ``None`` if unset.

        Unanimity is the best case, so this is a hard floor -- below it no
        observed agreement can clear the target, whatever the model returns.
        """
        if self.agreement_target is None:
            return None
        return smallest_certifiable_n(
            self.agreement_target, confidence=self.confidence, tail=self.tail
        )

smallest_certifiable_n property

smallest_certifiable_n

Draws needed to certify agreement_target, or None if unset.

Unanimity is the best case, so this is a hard floor -- below it no observed agreement can clear the target, whatever the model returns.

LocationMode

Bases: StrEnum

Which location estimator to apply to an unflagged component.

Source code in recall_guard/core/ensemble.py

class LocationMode(StrEnum):
    """Which location estimator to apply to an unflagged component."""

    MEAN = "mean"
    MEDIAN = "median"
    TRIMMED = "trimmed"

MultimodalAction

Bases: StrEnum

What to do with a component that holds separated clusters.

Silently averaging across one is the single behaviour that must never be available: it launders a real disagreement into false precision, returning a value the model effectively never emitted.

Source code in recall_guard/core/ensemble.py

class MultimodalAction(StrEnum):
    """What to do with a component that holds separated clusters.

    Silently averaging across one is the single behaviour that must never be
    available: it launders a real disagreement into false precision, returning a
    value the model effectively never emitted.
    """

    FLAG = "flag"
    RAISE = "raise"

ReferenceMode

Bases: StrEnum

Whether the optional reference draw varies per ensemble draw.

Source code in recall_guard/core/ensemble.py

class ReferenceMode(StrEnum):
    """Whether the optional reference draw varies per ensemble draw."""

    FIXED = "fixed"
    PER_DRAW = "per_draw"

CutoffViolation

Bases: Exception

Raised when shortlisted models post-date the eval set's cutoff.

Source code in recall_guard/core/loader.py

class CutoffViolation(Exception):
    """Raised when shortlisted models post-date the eval set's cutoff."""

EvalRow dataclass

One evaluation row: prompt + ground-truth direction + opaque metadata.

Source code in recall_guard/core/loader.py

@dataclass(frozen=True)
class EvalRow:
    """One evaluation row: prompt + ground-truth direction + opaque metadata."""

    prompt: str
    target_direction: int  # in {-1, 0, 1}
    metadata: dict[str, str] = field(default_factory=dict)

EvalSet dataclass

A loaded JSONL eval set plus its cutoff header and content hash.

Source code in recall_guard/core/loader.py

@dataclass(frozen=True)
class EvalSet:
    """A loaded JSONL eval set plus its cutoff header and content hash."""

    rows: list[EvalRow]
    cutoff_date: date | None
    path_hash: str  # sha256 hex digest of the file bytes

Manifest dataclass

Per-run reproducibility manifest written to <out_dir>/manifest.json.

Fields mirror the design's core.manifest Service Interface verbatim. The dataclass is frozen so callers cannot mutate a manifest after it has been hashed/written, which keeps the persisted manifest.json faithful to whatever the runner actually saw.

Source code in recall_guard/core/manifest.py

@dataclass(frozen=True)
class Manifest:
    """Per-run reproducibility manifest written to ``<out_dir>/manifest.json``.

    Fields mirror the design's ``core.manifest`` Service Interface verbatim.
    The dataclass is frozen so callers cannot mutate a manifest after it has
    been hashed/written, which keeps the persisted ``manifest.json`` faithful
    to whatever the runner actually saw.
    """

    harness_version: str
    seed: int
    eval_set_hash: str
    control_corpus_hash: str
    is_memorized_hash: str
    cutoffs_hash: str
    shortlist: list[str]
    composite_score: dict  # {"formula": str, "weights": dict[str, float] | None}
    mcs_hyperparams: dict
    bootstrap_n: int
    artifacts: dict[str, str]  # name -> path
    # Optional cmmd-backtest extension (Req 7.5, 8.2). When ``None`` the
    # manifest serialises to the pre-existing 11-key schema byte-identically;
    # ``write_manifest`` deliberately omits the key in that case so old runs
    # remain bit-stable. When a backtest block is supplied it must record the
    # fields listed in design.md § Manifest extension (signal_model, universe,
    # cash_ticker, cmmd_quantile, cmmd_threshold_value, fees_one_way,
    # init_cash, seed, bootstrap_n, n_is_rows, n_oos_rows, artifacts).
    backtest: dict | None = None

CompletionResult dataclass

Single chat-completion response with logprobs.

Attributes:

Name	Type	Description
content	str	Assistant message content.
logprobs	list[TokenLogprob]	Per-token logprob entries. top_logprobs length is enforced to be non-empty by the client (raises on missing data).
raw_temperature_observed	float | None	The temperature the API reported as honoured, when exposed. None when the API does not echo the temperature back.

Source code in recall_guard/core/nvidia_lm.py

@dataclass(frozen=True)
class CompletionResult:
    """Single chat-completion response with logprobs.

    Attributes
    ----------
    content:
        Assistant message content.
    logprobs:
        Per-token logprob entries. ``top_logprobs`` length is enforced to be
        non-empty by the client (raises on missing data).
    raw_temperature_observed:
        The temperature the API reported as honoured, when exposed. ``None``
        when the API does not echo the temperature back.
    """

    content: str
    logprobs: list[TokenLogprob]
    raw_temperature_observed: float | None

NvidiaLM

Thin HTTP client around the NVIDIA OpenAI-compatible chat endpoint.

Always sends logprobs=True and top_logprobs=20. The default temperature is 0.0 (per Req 10.3) and can be overridden per call.

Source code in recall_guard/core/nvidia_lm.py

class NvidiaLM:
    """Thin HTTP client around the NVIDIA OpenAI-compatible chat endpoint.

    Always sends ``logprobs=True`` and ``top_logprobs=20``. The default
    ``temperature`` is 0.0 (per Req 10.3) and can be overridden per call.
    """

    def __init__(
        self,
        api_key: str,
        model: str,
        timeout_s: float = DEFAULT_TIMEOUT_S,
        max_retries: int = DEFAULT_MAX_RETRIES,
        retry_backoff_s: float = DEFAULT_RETRY_BACKOFF_S,
        min_call_interval_s: float = 0.0,
    ) -> None:
        if not api_key:
            raise ValueError("api_key must be a non-empty string")
        if not model:
            raise ValueError("model must be a non-empty string")
        if max_retries < 0:
            raise ValueError("max_retries must be >= 0")
        if retry_backoff_s < 0:
            raise ValueError("retry_backoff_s must be >= 0")
        if min_call_interval_s < 0:
            raise ValueError("min_call_interval_s must be >= 0")
        self.api_key = api_key
        self.model = model
        self.timeout_s = timeout_s
        self.max_retries = max_retries
        self.retry_backoff_s = retry_backoff_s
        self.min_call_interval_s = min_call_interval_s
        self._last_call_t: float | None = None
        self._pace_lock = Lock()
        self.api_base = NVIDIA_CHAT_COMPLETIONS_URL

    def _reserve_call_slot(self) -> float:
        """Reserve the next paced send slot; return seconds to wait before POST.

        The lock covers only this bookkeeping -- never the network round trip.
        Holding it across the request serialises every concurrent call through
        one client, which is what ``max_workers`` used to run into.

        ``min_call_interval_s`` is defined as the spacing between the *starts*
        of successive requests. Recording the reserved send time (rather than
        the observed completion time) makes that spacing independent of
        endpoint latency; stamping the current clock instead would let a
        still-sleeping thread's slot be handed out twice.

        ``max(now, ...)`` stops an idle client from banking credit and then
        issuing a burst.
        """
        with self._pace_lock:
            now = time.monotonic()
            if self.min_call_interval_s <= 0 or self._last_call_t is None:
                slot = now
            else:
                slot = max(now, self._last_call_t + self.min_call_interval_s)
            self._last_call_t = slot
        return slot - now

    def generate(
        self,
        prompt: str,
        temperature: float = 0.0,
        max_tokens: int = 512,
    ) -> CompletionResult:
        """Send a single chat completion and return parsed logprobs.

        Caps response length at ``max_tokens`` (default 512) so reasoning
        models (gpt-oss-*, nemotron-nano-*) have enough budget to finish
        their reasoning chain AND emit the final ``Direction:`` /
        ``Confidence:`` lines. Non-reasoning models stop early on EOS so
        the higher cap costs nothing for them.

        Raises
        ------
        TimeoutError
            If the underlying HTTP call times out.
        RuntimeError
            If the response body lacks ``logprobs.content`` or any token entry
            is missing its ``top_logprobs`` list.
        """
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        payload: dict[str, Any] = {
            "model": self.model,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": temperature,
            "max_tokens": max_tokens,
            "logprobs": True,
            "top_logprobs": TOP_LOGPROBS,
        }
        def _paced_post() -> requests.Response:
            wait = self._reserve_call_slot()
            if wait > 0:
                time.sleep(wait)
            return requests.post(
                self.api_base,
                headers=headers,
                json=payload,
                timeout=self.timeout_s,
            )

        last_timeout_exc: Exception | None = None
        last_runtime_exc: Exception | None = None
        last_status: int | None = None
        for attempt in range(self.max_retries + 1):
            retry_after: float | None = None
            try:
                response = _paced_post()
                response.raise_for_status()
                return self._parse_response(response.json())
            except requests.exceptions.Timeout as exc:
                last_timeout_exc = exc
                last_runtime_exc = None
                retryable = True
            except requests.exceptions.HTTPError as exc:
                status = exc.response.status_code if exc.response is not None else None
                last_runtime_exc = exc
                last_timeout_exc = None
                last_status = status
                retryable = status in RETRYABLE_HTTP_STATUS
                if not retryable:
                    raise LMHTTPError(
                        f"Model {self.model} request failed: {exc}",
                        status_code=status,
                    ) from exc
                retry_after = _retry_after_seconds(exc.response)
            except requests.exceptions.RequestException as exc:
                last_runtime_exc = exc
                last_timeout_exc = None
                retryable = True

            if attempt < self.max_retries and retryable:
                backoff = self._retry_delay(attempt, retry_after)
                # Logged at DEBUG so a parallel run (8 workers * 50 prompts) does
                # not spam stderr. Final failures still surface via the
                # TimeoutError/RuntimeError raised below, which the evaluator
                # converts into a fail_reason on the row.
                _log.debug(
                    "NvidiaLM transient failure for %s (attempt %d/%d); retrying in %.1fs",
                    self.model, attempt + 1, self.max_retries + 1, backoff,
                )
                time.sleep(backoff)
                continue
            break

        if last_timeout_exc is not None:
            raise TimeoutError(
                f"Model {self.model} timed out after {self.timeout_s} seconds "
                f"(after {self.max_retries + 1} attempt(s))."
            ) from last_timeout_exc
        raise LMHTTPError(
            f"Model {self.model} request failed after {self.max_retries + 1} attempt(s): "
            f"{last_runtime_exc}",
            status_code=last_status,
        ) from last_runtime_exc

    def _retry_delay(self, attempt: int, retry_after: float | None) -> float:
        """Seconds to wait before the next attempt.

        An endpoint-supplied ``Retry-After`` wins outright. Otherwise the
        exponential backoff is fully jittered: rate-limited responses come back
        fast, so without jitter every concurrent draw would sleep for exactly
        the same interval and retry in unison, reproducing the burst that
        triggered the limit.
        """
        if retry_after is not None:
            return retry_after
        ceiling = self.retry_backoff_s * (2 ** attempt)
        return random.uniform(0.0, ceiling) if ceiling > 0 else 0.0

    def _parse_response(self, data: dict[str, Any]) -> CompletionResult:
        try:
            choice = data["choices"][0]
        except (KeyError, IndexError, TypeError) as exc:
            raise RuntimeError(
                f"Model {self.model} response missing 'choices[0]': {data!r}"
            ) from exc
        if not isinstance(choice, dict):
            raise RuntimeError(
                f"Model {self.model} response has malformed 'choices[0]': {choice!r}"
            )

        message = choice.get("message", {}) or {}
        if not isinstance(message, dict):
            raise RuntimeError(
                f"Model {self.model} response has malformed 'message': {message!r}"
            )
        content = message.get("content")
        if not content:
            # Reasoning models (gpt-oss-*, nemotron-nano-*) put output under
            # 'reasoning_content' until reasoning completes. If the answer
            # field is empty, fall back to reasoning_content so the parser
            # can still find Direction:/Confidence: lines.
            content = message.get("reasoning_content") or ""

        logprobs_section = choice.get("logprobs")
        if not isinstance(logprobs_section, dict) or "content" not in logprobs_section:
            raise RuntimeError(
                f"Model {self.model} response missing 'logprobs.content'; "
                "cannot compute MIA features without per-token logprobs."
            )

        token_entries = logprobs_section["content"]
        if not isinstance(token_entries, list) or not token_entries:
            raise RuntimeError(
                f"Model {self.model} response has empty or malformed 'logprobs.content'; "
                "cannot compute MIA features without per-token logprobs."
            )
        parsed: list[TokenLogprob] = []
        for idx, entry in enumerate(token_entries):
            if not isinstance(entry, dict):
                raise RuntimeError(
                    f"Model {self.model} response token #{idx} is malformed: {entry!r}"
                )
            if "top_logprobs" not in entry:
                raise RuntimeError(
                    f"Model {self.model} response token #{idx} is missing "
                    "'top_logprobs'; required for MIA feature computation."
                )
            top_logprobs = entry["top_logprobs"]
            if not isinstance(top_logprobs, list) or not top_logprobs:
                raise RuntimeError(
                    f"Model {self.model} response token #{idx} has empty or malformed "
                    "'top_logprobs'; required for MIA feature computation."
                )
            if "logprob" not in entry:
                raise RuntimeError(
                    f"Model {self.model} response token #{idx} is missing 'logprob'; "
                    "required for MIA feature computation."
                )
            try:
                logprob = float(entry["logprob"])
            except (TypeError, ValueError) as exc:
                raise RuntimeError(
                    f"Model {self.model} response token #{idx} has non-numeric 'logprob'."
                ) from exc
            parsed.append(
                TokenLogprob(
                    token=str(entry.get("token", "")),
                    logprob=logprob,
                    top_logprobs=list(top_logprobs),
                )
            )

        raw_temp = choice.get("temperature")
        if raw_temp is None:
            raw_temp = data.get("temperature")
        observed = float(raw_temp) if isinstance(raw_temp, (int, float)) else None

        return CompletionResult(
            content=content,
            logprobs=parsed,
            raw_temperature_observed=observed,
        )

generate

generate(prompt, temperature=0.0, max_tokens=512)

Send a single chat completion and return parsed logprobs.

Caps response length at max_tokens (default 512) so reasoning models (gpt-oss-, nemotron-nano-) have enough budget to finish their reasoning chain AND emit the final Direction: / Confidence: lines. Non-reasoning models stop early on EOS so the higher cap costs nothing for them.

Raises:

Type	Description
TimeoutError	If the underlying HTTP call times out.
RuntimeError	If the response body lacks logprobs.content or any token entry is missing its top_logprobs list.

Source code in recall_guard/core/nvidia_lm.py

def generate(
    self,
    prompt: str,
    temperature: float = 0.0,
    max_tokens: int = 512,
) -> CompletionResult:
    """Send a single chat completion and return parsed logprobs.

    Caps response length at ``max_tokens`` (default 512) so reasoning
    models (gpt-oss-*, nemotron-nano-*) have enough budget to finish
    their reasoning chain AND emit the final ``Direction:`` /
    ``Confidence:`` lines. Non-reasoning models stop early on EOS so
    the higher cap costs nothing for them.

    Raises
    ------
    TimeoutError
        If the underlying HTTP call times out.
    RuntimeError
        If the response body lacks ``logprobs.content`` or any token entry
        is missing its ``top_logprobs`` list.
    """
    headers = {
        "Authorization": f"Bearer {self.api_key}",
        "Content-Type": "application/json",
    }
    payload: dict[str, Any] = {
        "model": self.model,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": temperature,
        "max_tokens": max_tokens,
        "logprobs": True,
        "top_logprobs": TOP_LOGPROBS,
    }
    def _paced_post() -> requests.Response:
        wait = self._reserve_call_slot()
        if wait > 0:
            time.sleep(wait)
        return requests.post(
            self.api_base,
            headers=headers,
            json=payload,
            timeout=self.timeout_s,
        )

    last_timeout_exc: Exception | None = None
    last_runtime_exc: Exception | None = None
    last_status: int | None = None
    for attempt in range(self.max_retries + 1):
        retry_after: float | None = None
        try:
            response = _paced_post()
            response.raise_for_status()
            return self._parse_response(response.json())
        except requests.exceptions.Timeout as exc:
            last_timeout_exc = exc
            last_runtime_exc = None
            retryable = True
        except requests.exceptions.HTTPError as exc:
            status = exc.response.status_code if exc.response is not None else None
            last_runtime_exc = exc
            last_timeout_exc = None
            last_status = status
            retryable = status in RETRYABLE_HTTP_STATUS
            if not retryable:
                raise LMHTTPError(
                    f"Model {self.model} request failed: {exc}",
                    status_code=status,
                ) from exc
            retry_after = _retry_after_seconds(exc.response)
        except requests.exceptions.RequestException as exc:
            last_runtime_exc = exc
            last_timeout_exc = None
            retryable = True

        if attempt < self.max_retries and retryable:
            backoff = self._retry_delay(attempt, retry_after)
            # Logged at DEBUG so a parallel run (8 workers * 50 prompts) does
            # not spam stderr. Final failures still surface via the
            # TimeoutError/RuntimeError raised below, which the evaluator
            # converts into a fail_reason on the row.
            _log.debug(
                "NvidiaLM transient failure for %s (attempt %d/%d); retrying in %.1fs",
                self.model, attempt + 1, self.max_retries + 1, backoff,
            )
            time.sleep(backoff)
            continue
        break

    if last_timeout_exc is not None:
        raise TimeoutError(
            f"Model {self.model} timed out after {self.timeout_s} seconds "
            f"(after {self.max_retries + 1} attempt(s))."
        ) from last_timeout_exc
    raise LMHTTPError(
        f"Model {self.model} request failed after {self.max_retries + 1} attempt(s): "
        f"{last_runtime_exc}",
        status_code=last_status,
    ) from last_runtime_exc

TokenLogprob dataclass

Per-token logprob record returned by the NVIDIA OpenAI-compatible API.

Source code in recall_guard/core/nvidia_lm.py

@dataclass(frozen=True)
class TokenLogprob:
    """Per-token logprob record returned by the NVIDIA OpenAI-compatible API."""

    token: str
    logprob: float
    top_logprobs: list[dict[str, Any]]

bootstrap_ci

bootstrap_ci(
    samples,
    statistic,
    n_resamples=1000,
    confidence=0.95,
    seed=0,
)

Compute a percentile bootstrap CI for statistic over samples.

Args: samples: Sequence of arbitrary objects (may be ints, floats, tuples, dicts, dataclasses, etc.). Resampling uses index sampling, so the element type does not need to be numpy-friendly. statistic: Callable taking a resampled sequence and returning a float. n_resamples: Number of bootstrap resamples (>=1). Default 1000 matches the harness's Req 6.1 minimum. confidence: Two-sided confidence level in (0, 1). Default 0.95. seed: Seed for numpy.random.default_rng; same seed -> same output.

Returns: Tuple (point, lo, hi):

- ``point = statistic(samples)`` (computed once on the original).
- ``lo``, ``hi`` are the lower / upper percentile bounds.
- Postcondition: ``lo <= point <= hi`` (clamped on tiny float drift).

Raises: ValueError: if samples is empty, n_resamples < 1, or confidence is outside (0, 1).

Source code in recall_guard/core/bootstrap.py

def bootstrap_ci(
    samples: Sequence[T],
    statistic: Callable[[Sequence[T]], float],
    n_resamples: int = 1000,
    confidence: float = 0.95,
    seed: int = 0,
) -> tuple[float, float, float]:
    """Compute a percentile bootstrap CI for ``statistic`` over ``samples``.

    Args:
        samples: Sequence of arbitrary objects (may be ints, floats, tuples,
            dicts, dataclasses, etc.). Resampling uses index sampling, so the
            element type does not need to be numpy-friendly.
        statistic: Callable taking a resampled sequence and returning a float.
        n_resamples: Number of bootstrap resamples (>=1). Default 1000 matches
            the harness's Req 6.1 minimum.
        confidence: Two-sided confidence level in (0, 1). Default 0.95.
        seed: Seed for ``numpy.random.default_rng``; same seed -> same output.

    Returns:
        Tuple ``(point, lo, hi)``:

        - ``point = statistic(samples)`` (computed once on the original).
        - ``lo``, ``hi`` are the lower / upper percentile bounds.
        - Postcondition: ``lo <= point <= hi`` (clamped on tiny float drift).

    Raises:
        ValueError: if ``samples`` is empty, ``n_resamples < 1``, or
            ``confidence`` is outside ``(0, 1)``.
    """
    n = len(samples)
    if n == 0:
        raise ValueError("bootstrap_ci: 'samples' must contain at least one element.")
    if n_resamples < 1:
        raise ValueError(
            f"bootstrap_ci: 'n_resamples' must be >= 1, got {n_resamples}."
        )
    if not (0.0 < confidence < 1.0):
        raise ValueError(
            f"bootstrap_ci: 'confidence' must be in (0, 1), got {confidence}."
        )

    point = float(statistic(samples))

    # Single-sample short-circuit: every resample is identical, so the CI is
    # degenerate by construction. Skip work and return immediately (Req 6.1
    # precondition note).
    if n == 1:
        return point, point, point

    rng = np.random.default_rng(seed)

    stats: list[float] = []
    dropped = 0
    for _ in range(n_resamples):
        # Index sampling keeps the element type unconstrained: T may be any
        # arbitrary Python object, not necessarily a numpy scalar.
        idx = rng.integers(low=0, high=n, size=n)
        resample = [samples[int(i)] for i in idx]
        try:
            value = statistic(resample)
        except ValueError:
            # Statistic refused this resample (e.g., AUC with a single class).
            # Drop and continue rather than crashing the whole CI computation.
            dropped += 1
            continue
        if not _is_finite(value):
            dropped += 1
            continue
        stats.append(float(value))

    if dropped > 0:
        # Exactly one warning per call, regardless of how many resamples we
        # dropped, so logs do not flood under heavy degeneracy.
        logger.warning(
            "bootstrap_ci: dropped %d / %d resamples where the statistic "
            "raised or returned a non-finite value.",
            dropped,
            n_resamples,
        )

    if not stats:
        # All resamples failed: the CI is undefined. Surface the degeneracy
        # via a separate warning and collapse to the point estimate so callers
        # do not get NaN bounds.
        logger.warning(
            "bootstrap_ci: every resample failed; collapsing CI to the point "
            "estimate. Statistic likely undefined for this sample distribution."
        )
        return point, point, point

    arr = np.asarray(stats, dtype=float)
    alpha = 1.0 - confidence
    lo_pct = (alpha / 2.0) * 100.0
    hi_pct = (1.0 - alpha / 2.0) * 100.0
    lo = float(np.percentile(arr, lo_pct))
    hi = float(np.percentile(arr, hi_pct))

    # Clamp tiny floating-point inversions so the documented postcondition
    # (lo <= point <= hi) always holds without surprising the caller.
    if lo > point:
        lo = point
    if hi < point:
        hi = point

    return point, lo, hi

detect_multimodal

detect_multimodal(
    values,
    *,
    grid,
    mass_min=0.25,
    trough_steps=3,
    density_ratio=10.0,
    min_draws=8,
    min_cluster_density=1.5,
)

Detect two clusters separated by a sparse gap.

Returns None when the check could not run -- never a silent "found nothing". There are three such cases, and a caller who needs to tell them apart should read the lattice adherence alongside this:

• no lattice was declared (a continuous quantity has none);
• fewer than min_draws draws, which cannot evidence two clusters;
• the clusters are too thinly populated for a gap between them to mean anything (see below).

The rule is defined directly on the lattice rather than by a classical unimodality test. Those assume a continuous distribution, and on heavily tied lattice data they measure tie mass instead of modality -- badly enough that on the measured corpus the most sharply converged component scores as more multimodal than the genuinely split one.

The sparsity guard matters. trough_steps counts lattice steps, not draws, so on a lattice much finer than the sampling, empty runs occur everywhere by chance; the density test is vacuous there because an empty gap has no peak to compare against. Measured, a unimodal normal at a 0.001 lattice flagged on every single subsample.

What separates the two regimes is not how wide the gap is but how dense the clusters are: a real cluster stacks many draws onto few lattice positions, while a spurious one is a scatter of singletons whose gaps are ordinary spacing. So each side of the split must average at least min_cluster_density draws per occupied position.

Every threshold is a parameter because all of them were tuned against a single measurement date. Detects separated clusters only: two overlapping modes with no gap between them are invisible to it, a known and accepted false-negative class.

Parameters:

Name	Type	Description	Default
mass_min	float	Minimum share of draws each cluster must hold.	0.25
trough_steps	int	Minimum width of the gap, in lattice steps.	3
density_ratio	float	How much denser the taller cluster peak must be than the busiest bin inside the gap. Note this binds only when the gap is non-empty; at realistic draw counts most detections win on a completely empty gap.	10.0
min_draws	int	Below this many draws the check does not run.	8
min_cluster_density	float	Minimum draws per occupied lattice position within each cluster.	1.5

Source code in recall_guard/core/consensus.py

def detect_multimodal(
    values: Sequence[float],
    *,
    grid: float | None,
    mass_min: float = 0.25,
    trough_steps: int = 3,
    density_ratio: float = 10.0,
    min_draws: int = 8,
    min_cluster_density: float = 1.5,
) -> MultimodalVerdict | None:
    """Detect two clusters separated by a sparse gap.

    Returns ``None`` when the check **could not run** -- never a silent "found
    nothing". There are three such cases, and a caller who needs to tell them
    apart should read the lattice adherence alongside this:

    * no lattice was declared (a continuous quantity has none);
    * fewer than ``min_draws`` draws, which cannot evidence two clusters;
    * the clusters are too thinly populated for a gap between them to mean
      anything (see below).

    The rule is defined directly on the lattice rather than by a classical
    unimodality test. Those assume a continuous distribution, and on heavily
    tied lattice data they measure tie mass instead of modality -- badly enough
    that on the measured corpus the most sharply converged component scores as
    *more* multimodal than the genuinely split one.

    **The sparsity guard matters.** ``trough_steps`` counts lattice steps, not
    draws, so on a lattice much finer than the sampling, empty runs occur
    everywhere by chance; the density test is vacuous there because an empty gap
    has no peak to compare against. Measured, a unimodal normal at a 0.001
    lattice flagged on every single subsample.

    What separates the two regimes is not how wide the gap is but how *dense the
    clusters are*: a real cluster stacks many draws onto few lattice positions,
    while a spurious one is a scatter of singletons whose gaps are ordinary
    spacing. So each side of the split must average at least
    ``min_cluster_density`` draws per occupied position.

    Every threshold is a parameter because all of them were tuned against a
    single measurement date. Detects separated clusters only: two overlapping
    modes with no gap between them are invisible to it, a known and accepted
    false-negative class.

    Parameters
    ----------
    mass_min:
        Minimum share of draws each cluster must hold.
    trough_steps:
        Minimum width of the gap, in lattice steps.
    density_ratio:
        How much denser the taller cluster peak must be than the busiest bin
        inside the gap. Note this binds only when the gap is non-empty; at
        realistic draw counts most detections win on a completely empty gap.
    min_draws:
        Below this many draws the check does not run.
    min_cluster_density:
        Minimum draws per occupied lattice position within each cluster.
    """
    if len(values) == 0:
        raise ValueError("values must be non-empty")
    if grid is None:
        return None
    grid = _checked_grid(grid)
    if not 0.0 < mass_min <= 0.5:
        raise ValueError(f"mass_min must be in (0, 0.5]; got {mass_min!r}")
    if trough_steps < 1:
        raise ValueError(f"trough_steps must be >= 1; got {trough_steps!r}")
    if density_ratio < 0:
        raise ValueError(f"density_ratio must be >= 0; got {density_ratio!r}")
    if min_draws < 2:
        raise ValueError(f"min_draws must be >= 2; got {min_draws!r}")
    if min_cluster_density < 1.0:
        raise ValueError(
            f"min_cluster_density must be >= 1; got {min_cluster_density!r}"
        )

    total = len(values)
    if total < min_draws:
        return None

    positions, counts = _occupancy(_finite_floats(values, what="values"), grid)
    if len(positions) < 2:
        return _unseparated()
    span_steps = round((positions[-1] - positions[0]) / grid)
    if span_steps < 1:
        return _unseparated()

    return _search_split(
        positions,
        counts,
        grid=grid,
        need=mass_min * total,
        trough_steps=trough_steps,
        density_ratio=density_ratio,
        min_cluster_density=min_cluster_density,
    )

grid_adherence

grid_adherence(values, grid, *, tolerance=1e-09)

Fraction of values that actually lie on the declared lattice.

Reported so a caller who declares a lattice the data does not follow finds out, instead of silently receiving mis-snapped results. A continuous quantity scores near zero here at any lattice.

A non-finite draw counts as off-lattice rather than raising: this reports on data quality, so it has to survive the bad data it exists to describe.

Source code in recall_guard/core/consensus.py

def grid_adherence(values: Sequence[float], grid: float, *, tolerance: float = 1e-9) -> float:
    """Fraction of ``values`` that actually lie on the declared lattice.

    Reported so a caller who declares a lattice the data does not follow finds
    out, instead of silently receiving mis-snapped results. A continuous
    quantity scores near zero here at any lattice.

    A non-finite draw counts as off-lattice rather than raising: this reports on
    data quality, so it has to survive the bad data it exists to describe.
    """
    grid = _checked_grid(grid)
    if not (tolerance >= 0) or not math.isfinite(tolerance):
        raise ValueError(f"tolerance must be non-negative and finite; got {tolerance!r}")
    if len(values) == 0:
        raise ValueError("values must be non-empty")
    on = 0
    for raw in values:
        v = float(raw)
        if not math.isfinite(v):
            continue
        if abs(v - snap_to_grid(v, grid)) <= tolerance:
            on += 1
    return on / len(values)

lag_dependence

lag_dependence(labels, groups)

How much more alike draws are within a collection group than across them.

Returns None when there are fewer than two groups, or too few pairs to compare -- never a misleading zero.

The statistic is the probability that two draws from the same group carry the same label, minus the probability for two draws from different groups. Zero means the grouping carries no information, which is what independence looks like; positive means draws collected together agree more than draws collected apart.

This exists because the reported agreement interval assumes independent draws, and that is precisely the assumption a serving stack violates -- batching, cache reuse, and node affinity all couple requests issued together. Positive dependence makes every interval narrower than its label, in the one direction that matters. Measuring it does not correct the interval; it makes the assumption falsifiable instead of merely disclaimed.

Depends only on the stored labels and group tags, never on arrival order, so it replays identically from a persisted draw set.

Source code in recall_guard/core/consensus.py

def lag_dependence(
    labels: Sequence[object],
    groups: Sequence[int],
) -> float | None:
    """How much more alike draws are within a collection group than across them.

    Returns ``None`` when there are fewer than two groups, or too few pairs to
    compare -- never a misleading zero.

    The statistic is the probability that two draws from the *same* group carry
    the same label, minus the probability for two draws from *different*
    groups. Zero means the grouping carries no information, which is what
    independence looks like; positive means draws collected together agree more
    than draws collected apart.

    This exists because the reported agreement interval assumes independent
    draws, and that is precisely the assumption a serving stack violates --
    batching, cache reuse, and node affinity all couple requests issued
    together. Positive dependence makes every interval narrower than its label,
    in the one direction that matters. Measuring it does not correct the
    interval; it makes the assumption falsifiable instead of merely disclaimed.

    Depends only on the stored labels and group tags, never on arrival order, so
    it replays identically from a persisted draw set.
    """
    if len(labels) != len(groups):
        raise ValueError(
            f"labels and groups must be the same length; got {len(labels)} and {len(groups)}"
        )
    if len(set(groups)) < 2:
        return None

    same_group_pairs = same_group_matches = 0
    diff_group_pairs = diff_group_matches = 0
    for i in range(len(labels)):
        for j in range(i + 1, len(labels)):
            match = labels[i] == labels[j]
            if groups[i] == groups[j]:
                same_group_pairs += 1
                same_group_matches += match
            else:
                diff_group_pairs += 1
                diff_group_matches += match

    if same_group_pairs == 0 or diff_group_pairs == 0:
        return None
    return same_group_matches / same_group_pairs - diff_group_matches / diff_group_pairs

robust_location

robust_location(values, *, mode='median', trim=0.25)

Reduce values to one location, independent of their order.

mode is one of "mean", "median", or "trimmed".

Note what this cannot do: on a component whose draws form two separated clusters there is no single location to estimate, and no symmetric trim fraction escapes the gap between them -- trimming converges toward the median, not toward a mode. Callers must run the multimodality check first and skip this entirely for a flagged component.

Source code in recall_guard/core/consensus.py

def robust_location(
    values: Sequence[float],
    *,
    mode: str = "median",
    trim: float = 0.25,
) -> float:
    """Reduce ``values`` to one location, independent of their order.

    ``mode`` is one of ``"mean"``, ``"median"``, or ``"trimmed"``.

    Note what this cannot do: on a component whose draws form two separated
    clusters there is no single location to estimate, and no symmetric trim
    fraction escapes the gap between them -- trimming converges toward the
    median, not toward a mode. Callers must run the multimodality check first
    and skip this entirely for a flagged component.
    """
    if len(values) == 0:
        raise ValueError("values must be non-empty")
    values = _finite_floats(values, what="values")

    if mode == "mean":
        return _exact_mean(sorted(values))
    if mode == "median":
        return _median(sorted(values))
    if mode != "trimmed":
        raise ValueError(f"unknown mode {mode!r}; expected mean, median, or trimmed")

    if not 0.0 <= trim < 0.5:
        raise ValueError(f"trim must be in [0, 0.5); got {trim!r}")
    ordered = sorted(values)
    cut = math.floor(len(ordered) * trim)
    core = ordered[cut : len(ordered) - cut] or ordered
    return _exact_mean(core)

scale_floor

scale_floor(values, *, grid=None)

Robust scale estimate, floored at the lattice's resolution limit.

The floor exists because a concentrated lattice-valued component can drive the median absolute deviation to exactly zero, leaving the usual robust scale undefined. It is an identifiability floor, not an estimate of quantization noise: any true dispersion far below one lattice step produces observations on one or two lattice points and is indistinguishable from zero, so an estimate below that level carries no information.

Two caveats worth knowing before relying on it. It binds whenever 1.4826 * MAD < grid / sqrt(12) -- that is, for every MAD below roughly 0.195 * grid, not only when the deviation is exactly zero. So on a sharply concentrated component it can inflate a small but perfectly well-defined estimate. And it only helps at all because the declared lattice is coarser than the emitted one; declare the true lattice and the undefined case returns.

Source code in recall_guard/core/consensus.py

def scale_floor(values: Sequence[float], *, grid: float | None = None) -> float:
    """Robust scale estimate, floored at the lattice's resolution limit.

    The floor exists because a concentrated lattice-valued component can drive
    the median absolute deviation to exactly zero, leaving the usual robust
    scale undefined. It is an *identifiability* floor, not an estimate of
    quantization noise: any true dispersion far below one lattice step produces
    observations on one or two lattice points and is indistinguishable from
    zero, so an estimate below that level carries no information.

    Two caveats worth knowing before relying on it. It binds whenever
    ``1.4826 * MAD < grid / sqrt(12)`` -- that is, for every ``MAD`` below
    roughly ``0.195 * grid``, not only when the deviation is exactly zero. So on
    a sharply concentrated component it can inflate a small but perfectly
    well-defined estimate. And it only helps at all because the declared lattice
    is coarser than the emitted one; declare the true lattice and the undefined
    case returns.
    """
    if len(values) == 0:
        raise ValueError("values must be non-empty")
    values = _finite_floats(values, what="values")
    ordered = sorted(values)
    centre = _median(ordered)
    mad = _median(sorted(abs(v - centre) for v in values))
    sigma = MAD_TO_SIGMA * mad
    if grid is None:
        return sigma
    return max(sigma, _checked_grid(grid) / _QUANTIZATION_DIVISOR)

smallest_certifiable_n

smallest_certifiable_n(
    target,
    *,
    confidence=0.95,
    tail=Tail.TWO_SIDED,
    limit=100000,
)

Fewest unanimous draws whose interval's lower bound reaches target.

Unanimity is the best case, so this is a hard floor: below it no observed agreement can certify the target, and a configuration requesting fewer draws can never succeed no matter what the model returns. Surfacing it at construction turns a silently-unreachable setting into an error.

Source code in recall_guard/core/consensus.py

def smallest_certifiable_n(
    target: float,
    *,
    confidence: float = 0.95,
    tail: Tail = Tail.TWO_SIDED,
    limit: int = 100_000,
) -> int:
    """Fewest unanimous draws whose interval's lower bound reaches ``target``.

    Unanimity is the best case, so this is a hard floor: below it no observed
    agreement can certify the target, and a configuration requesting fewer draws
    can never succeed no matter what the model returns. Surfacing it at
    construction turns a silently-unreachable setting into an error.
    """
    if not 0.0 < target < 1.0:
        raise ValueError(f"target must be in (0, 1); got {target!r}")
    for n in range(1, limit + 1):
        bounds = wilson_interval(n, n, confidence=confidence, tail=tail)
        if bounds is not None and bounds[0] >= target:
            return n
    raise ValueError(
        f"target {target} is not certifiable within {limit} draws at "
        f"confidence={confidence} ({tail.value})"
    )

smallest_detectable_split_n

smallest_detectable_split_n(
    *,
    cluster_positions,
    min_cluster_draws=8,
    min_cluster_density=1.5,
)

Fewest draws at which :func:detect_multimodal can flag a split.

The companion to :func:smallest_certifiable_n, and the reason both exist: agreement precision and split detection need different sample sizes, and sizing for the first silently under-sizes for the second. Agreement is where the reported confidence lives; component splits are where a silently wrong answer lives.

This is a necessary condition, not a sufficient one -- exactly as its companion is a floor under unanimity rather than a promise. Below the value returned here the density guard cannot be satisfied at all, so a split of that shape is undetectable no matter how clean the data. Above it, detection becomes possible; whether it fires still depends on sampling noise in the trough and in the cluster masses.

Measured on a corpus whose split is unambiguous at full size, detection still missed ~3.6% of 64-draw bootstrap resamples -- the closer analogue to a fresh ensemble, which draws independently rather than from a fixed pool. Reaching 99% detection on that corpus took ~128 draws.

The figure tracks parsed draws, not the configured count. The test only ever sees replies that survived transport, parsing, and the caller's projection. A prompt with a 5% failure rate configured at draws=64 is really operating at n=61, where the same measurement gives 5.5% rather than 3.6% -- so size against n_parsed, not against draws.

Parameters:

Name	Type	Description	Default
cluster_positions	int	How many lattice positions the two clusters together occupy. Wider clusters need more draws to reach the same density.	required

Source code in recall_guard/core/consensus.py

def smallest_detectable_split_n(
    *,
    cluster_positions: int,
    min_cluster_draws: int = 8,
    min_cluster_density: float = 1.5,
) -> int:
    """Fewest draws at which :func:`detect_multimodal` *can* flag a split.

    The companion to :func:`smallest_certifiable_n`, and the reason both exist:
    **agreement precision and split detection need different sample sizes**, and
    sizing for the first silently under-sizes for the second. Agreement is where
    the reported confidence lives; component splits are where a silently wrong
    answer lives.

    This is a **necessary condition, not a sufficient one** -- exactly as its
    companion is a floor under unanimity rather than a promise. Below the value
    returned here the density guard cannot be satisfied at all, so a split of
    that shape is undetectable no matter how clean the data. Above it, detection
    becomes *possible*; whether it fires still depends on sampling noise in the
    trough and in the cluster masses.

    Measured on a corpus whose split is unambiguous at full size, detection
    still missed **~3.6% of 64-draw bootstrap resamples** -- the closer analogue
    to a fresh ensemble, which draws independently rather than from a fixed
    pool. Reaching 99% detection on that corpus took ~128 draws.

    **The figure tracks *parsed* draws, not the configured count.** The test
    only ever sees replies that survived transport, parsing, and the caller's
    projection. A prompt with a 5% failure rate configured at ``draws=64`` is
    really operating at n=61, where the same measurement gives 5.5% rather than
    3.6% -- so size against ``n_parsed``, not against ``draws``.

    Parameters
    ----------
    cluster_positions:
        How many lattice positions the two clusters together occupy. Wider
        clusters need more draws to reach the same density.
    """
    if cluster_positions < 2:
        raise ValueError(
            f"cluster_positions must be >= 2 for a split; got {cluster_positions}"
        )
    if min_cluster_draws < 2:
        raise ValueError(f"min_cluster_draws must be >= 2; got {min_cluster_draws}")
    if min_cluster_density < 1.0:
        raise ValueError(
            f"min_cluster_density must be >= 1; got {min_cluster_density}"
        )
    return max(min_cluster_draws, math.ceil(min_cluster_density * cluster_positions))

snap_to_grid

snap_to_grid(value, grid)

Round value onto a lattice of step grid, half away from zero.

Deliberately does not divide by grid. That division is inexact in binary in a value-dependent way -- 0.85 / 0.1 is exactly 8.5 while 0.95 / 0.1 is 9.499999999999998 -- so the tie direction ends up depending on the value rather than on the rule, and the obvious spellings disagree with one another. Working in integer lattice units avoids it.

Source code in recall_guard/core/consensus.py

def snap_to_grid(value: float, grid: float) -> float:
    """Round ``value`` onto a lattice of step ``grid``, half away from zero.

    Deliberately does **not** divide by ``grid``. That division is inexact in
    binary in a value-dependent way -- ``0.85 / 0.1`` is exactly ``8.5`` while
    ``0.95 / 0.1`` is ``9.499999999999998`` -- so the tie direction ends up
    depending on the value rather than on the rule, and the obvious spellings
    disagree with one another. Working in integer lattice units avoids it.
    """
    grid = _checked_grid(grid)
    value = float(value)
    if not math.isfinite(value):
        raise ValueError(f"value must be finite; got {value!r}")
    units = (Decimal(repr(abs(value))) / Decimal(repr(grid))).quantize(
        Decimal(1), rounding=ROUND_HALF_UP
    )
    return math.copysign(float(units * Decimal(repr(grid))), value)

wilson_interval

wilson_interval(
    k,
    n,
    *,
    confidence=0.95,
    tail=Tail.TWO_SIDED,
    continuity=False,
)

Score interval for a binomial proportion, or None when n is zero.

Inverting the score test rather than the Wald test keeps the interval inside [0, 1] and, critically, non-degenerate at k == n. The Wald interval collapses to zero width exactly there, which at high agreement is the typical case -- it would report certainty from a couple of dozen draws.

Parameters:

Name	Type	Description	Default
k	int	Successes observed. Must satisfy 0 <= k <= n.	required
n	int	Draws observed.	required
continuity	bool	Apply the Newcombe continuity correction, widening the interval.	False

Source code in recall_guard/core/consensus.py

def wilson_interval(
    k: int,
    n: int,
    *,
    confidence: float = 0.95,
    tail: Tail = Tail.TWO_SIDED,
    continuity: bool = False,
) -> tuple[float, float] | None:
    """Score interval for a binomial proportion, or ``None`` when ``n`` is zero.

    Inverting the score test rather than the Wald test keeps the interval inside
    ``[0, 1]`` and, critically, non-degenerate at ``k == n``. The Wald interval
    collapses to zero width exactly there, which at high agreement is the
    typical case -- it would report certainty from a couple of dozen draws.

    Parameters
    ----------
    k:
        Successes observed. Must satisfy ``0 <= k <= n``.
    n:
        Draws observed.
    continuity:
        Apply the Newcombe continuity correction, widening the interval.
    """
    if n < 0:
        raise ValueError(f"n must be non-negative; got {n}")
    if not 0 <= k <= n:
        raise ValueError(f"k must satisfy 0 <= k <= n; got k={k}, n={n}")
    if n == 0:
        return None

    z = _z(confidence, tail)
    p = k / n
    z2 = z * z
    denominator = 1.0 + z2 / n
    centre = (p + z2 / (2 * n)) / denominator
    spread = z * math.sqrt(p * (1 - p) / n + z2 / (4 * n * n)) / denominator
    lo, hi = centre - spread, centre + spread

    if continuity:
        # Newcombe's correction adjusts the pivot *before* inversion, which
        # changes the radicand. Applying it as a constant shift of the already
        # inverted bound is a different, anti-conservative interval: it sits
        # inside the exact one across most of the parameter space, and its true
        # coverage dips below nominal -- failing the one job a continuity
        # correction exists to do.
        span = 2.0 * (n + z2)
        if k == 0:
            lo = 0.0
        else:
            radicand = z2 - 2 - 1.0 / n + 4 * p * (n * (1 - p) + 1)
            lo = (2 * n * p + z2 - 1 - z * math.sqrt(max(radicand, 0.0))) / span
        if k == n:
            hi = 1.0
        else:
            radicand = z2 + 2 - 1.0 / n + 4 * p * (n * (1 - p) - 1)
            hi = (2 * n * p + z2 + 1 + z * math.sqrt(max(radicand, 0.0))) / span

    return (max(0.0, lo), min(1.0, hi))

canonical_draw_hash

canonical_draw_hash(contents)

SHA-256 over the draw set's reply text, independent of arrival order.

Covers reply text only. Logprob structures are excluded because their key ordering comes from the provider's JSON and is not stable across servers or library versions, and timing and thread identity are excluded because they are not properties of the answer.

Sorting before hashing is what makes the digest a property of the draw set; the tie-break rules elsewhere in this module recover a deterministic order from content alone, so nothing depends on how the draws arrived.

Source code in recall_guard/core/ensemble.py

def canonical_draw_hash(contents: Sequence[str]) -> str:
    """SHA-256 over the draw set's reply text, independent of arrival order.

    Covers reply **text only**. Logprob structures are excluded because their
    key ordering comes from the provider's JSON and is not stable across servers
    or library versions, and timing and thread identity are excluded because
    they are not properties of the answer.

    Sorting before hashing is what makes the digest a property of the draw
    *set*; the tie-break rules elsewhere in this module recover a deterministic
    order from content alone, so nothing depends on how the draws arrived.
    """
    digest = hashlib.sha256()
    digest.update(_HASH_SCHEME.encode("utf-8"))
    for content in sorted(contents):
        digest.update(_HASH_SEPARATOR.encode("utf-8"))
        digest.update(content.encode("utf-8"))
    return digest.hexdigest()

estimate_cost

estimate_cost(
    spec,
    *,
    max_retries,
    has_reference,
    seconds_per_request=None,
)

Worst-case request count and duration, without issuing any request.

The nominal draw count is the floor, not the worst case: each logical draw can become max_retries + 1 requests, and a configured reference model doubles the whole thing.

Source code in recall_guard/core/ensemble.py

def estimate_cost(
    spec: EnsembleSpec,
    *,
    max_retries: int,
    has_reference: bool,
    seconds_per_request: float | None = None,
) -> CostEstimate:
    """Worst-case request count and duration, without issuing any request.

    The nominal draw count is the *floor*, not the worst case: each logical draw
    can become ``max_retries + 1`` requests, and a configured reference model
    doubles the whole thing.
    """
    if max_retries < 0:
        raise ValueError(f"max_retries must be >= 0; got {max_retries}")
    per_draw = (max_retries + 1) * (2 if has_reference else 1)
    worst_case = spec.draws * per_draw
    seconds = None
    if seconds_per_request is not None:
        waves = math.ceil(spec.draws / spec.max_workers)
        seconds = waves * per_draw * seconds_per_request
    return CostEstimate(worst_case_requests=worst_case, estimated_seconds=seconds)

generate_ensemble

generate_ensemble(
    lm, prompt, spec, *, decide, components=None
)

Draw spec.draws replies to prompt and reduce them.

Raises:

Type	Description
ValueError	If a component holds separated clusters and the spec asks to raise.
RuntimeError	If the request budget is exhausted, too few draws are usable, or transport failures exceed the configured share. Each of these is a refusal to report a confident answer computed from survivors.

Source code in recall_guard/core/ensemble.py

def generate_ensemble(
    lm: NvidiaLM,
    prompt: str,
    spec: EnsembleSpec,
    *,
    decide: Callable[[CompletionResult], Hashable],
    components: Callable[[CompletionResult], Mapping[str, float]] | None = None,
) -> EnsembleResult:
    """Draw ``spec.draws`` replies to ``prompt`` and reduce them.

    Raises
    ------
    ValueError
        If a component holds separated clusters and the spec asks to raise.
    RuntimeError
        If the request budget is exhausted, too few draws are usable, or
        transport failures exceed the configured share. Each of these is a
        refusal to report a confident answer computed from survivors.
    """
    started_at = datetime.now(UTC).isoformat()
    parsed: list[CompletionResult] = []
    wave_tags: list[int] = []
    failures: dict[str, int] = {}
    issued = 0
    wave_index = 0

    while len(parsed) + sum(failures.values()) < spec.draws:
        remaining = spec.draws - (len(parsed) + sum(failures.values()))
        size = min(spec.max_workers, remaining)
        if spec.max_total_requests is not None and issued + size > spec.max_total_requests:
            raise RuntimeError(
                f"ensemble request budget exhausted: {spec.max_total_requests} requests "
                f"allowed, {issued} already issued, next wave needs {size}"
            )

        with ThreadPoolExecutor(max_workers=size) as pool:
            outcomes = list(
                pool.map(lambda _: _safe_draw(lm, prompt, spec), range(size))
            )
        issued += size

        for outcome in outcomes:
            if isinstance(outcome, BaseException):
                if _is_auth_failure(outcome):
                    # Abort rather than paying for the remaining draws; a rejected
                    # credential will not start working mid-ensemble.
                    raise outcome
                failures[_classify(outcome)] = failures.get(_classify(outcome), 0) + 1
                continue
            try:
                decide(outcome)
            except Exception:  # noqa: BLE001 - a caller callback may raise anything
                failures["projection"] = failures.get("projection", 0) + 1
                continue
            parsed.append(outcome)
            wave_tags.append(wave_index)
        wave_index += 1

    transport_failures = sum(
        count for reason, count in failures.items() if reason != "projection"
    )
    if transport_failures / spec.draws > spec.max_transport_failure_ratio:
        raise RuntimeError(
            f"transport failures {transport_failures}/{spec.draws} exceed the configured "
            f"limit of {spec.max_transport_failure_ratio:.0%}; refusing to report a "
            "consensus computed from the survivors"
        )
    if len(parsed) < spec.min_parsed:
        raise RuntimeError(
            f"only {len(parsed)} usable draws of {spec.draws} requested, below "
            f"min_parsed={spec.min_parsed}; refusing to report a consensus"
        )

    return reduce_draws(
        parsed,
        spec,
        decide=decide,
        components=components,
        waves=wave_tags,
        n_requested=spec.draws,
        fail_counts=failures,
        sampled_at=started_at,
    )

reduce_draws

reduce_draws(
    draws,
    spec,
    *,
    decide,
    components=None,
    waves=None,
    n_requested=None,
    fail_counts=None,
    sampled_at=None,
)

Reduce a draw set to one answer. Pure: no I/O, no randomness, no clock.

Separated from execution so a stored draw set can be replayed into a bit-identical result without contacting a model, which is what makes an ensemble auditable after the fact. No clock is read here: sampled_at stays None unless the caller passes through what execution recorded.

Source code in recall_guard/core/ensemble.py

def reduce_draws(
    draws: Sequence[CompletionResult],
    spec: EnsembleSpec,
    *,
    decide: Callable[[CompletionResult], Hashable],
    components: Callable[[CompletionResult], Mapping[str, float]] | None = None,
    waves: Sequence[int] | None = None,
    n_requested: int | None = None,
    fail_counts: Mapping[str, int] | None = None,
    sampled_at: str | None = None,
) -> EnsembleResult:
    """Reduce a draw set to one answer. Pure: no I/O, no randomness, no clock.

    Separated from execution so a stored draw set can be replayed into a
    bit-identical result without contacting a model, which is what makes an
    ensemble auditable after the fact. No clock is read here: ``sampled_at``
    stays ``None`` unless the caller passes through what execution recorded.
    """
    if not draws:
        raise ValueError("cannot reduce an empty draw set")

    order = _canonical_order(draws, [])
    ordered = [draws[i] for i in order]
    decisions = [decide(d) for d in ordered]

    tally: dict[Hashable, int] = {}
    for decision in decisions:
        tally[decision] = tally.get(decision, 0) + 1
    # Ties break toward the lexicographically smaller repr, never dict order.
    modal = min(tally, key=lambda d: (-tally[d], repr(d)))
    agreeing = tally[modal]
    agreement = agreeing / len(ordered)

    location: dict[str, float] = {}
    snapped = adherence = None
    flagged: tuple[str, ...] = ()
    verdicts: tuple[tuple[str, MultimodalVerdict | None], ...] = ()
    if components is not None:
        location, snapped, adherence, flagged, verdicts = _reduce_components(
            ordered, components, spec
        )

    dependence = None
    if waves is not None:
        ordered_waves = [waves[i] for i in order]
        dependence = lag_dependence(decisions, ordered_waves)

    return EnsembleResult(
        consensus=_select_consensus(ordered, decisions, modal),
        location=location,
        location_snapped=snapped,
        grid_adherence=adherence,
        multimodal=flagged,
        component_verdicts=verdicts,
        agreement=agreement,
        agreement_ci=wilson_interval(
            agreeing, len(ordered), confidence=spec.confidence, tail=spec.tail
        ),
        draw_dependence=dependence,
        max_tokens=spec.max_tokens,
        temperature=spec.temperature,
        sampled_at=sampled_at,
        n_requested=n_requested if n_requested is not None else len(draws),
        n_parsed=len(ordered),
        fail_counts=tuple(sorted((fail_counts or {}).items())),
        draws_sha256=canonical_draw_hash([d.content for d in ordered]),
        draws=tuple(ordered) if spec.retain_draws else (),
    )

assert_cutoff_safe

assert_cutoff_safe(eval_set, models, cutoffs)

Fail-fast guard: every shortlisted model has a cutoff <= eval cutoff.

Raises: CutoffViolation: if any model is missing from cutoffs or, when eval_set.cutoff_date is set, post-dates it.

Source code in recall_guard/core/loader.py

def assert_cutoff_safe(
    eval_set: EvalSet,
    models: list[str],
    cutoffs: dict[str, date],
) -> None:
    """Fail-fast guard: every shortlisted model has a cutoff <= eval cutoff.

    Raises:
        CutoffViolation: if any model is missing from ``cutoffs`` or, when
            ``eval_set.cutoff_date`` is set, post-dates it.
    """
    missing = [m for m in models if m not in cutoffs]
    if missing:
        raise CutoffViolation(
            "Shortlisted models missing from cutoffs registry: " + ", ".join(missing)
        )

    if eval_set.cutoff_date is None:
        return

    too_late: list[tuple[str, date]] = [
        (m, cutoffs[m]) for m in models if cutoffs[m] > eval_set.cutoff_date
    ]
    if too_late:
        details = ", ".join(f"{m}={c.isoformat()}" for m, c in too_late)
        raise CutoffViolation(
            f"Models with training cutoffs after eval cutoff "
            f"{eval_set.cutoff_date.isoformat()}: {details}"
        )

load_cutoffs

load_cutoffs(path)

Parse the cutoffs YAML registry into {model_id: cutoff_date}.

Source code in recall_guard/core/loader.py

def load_cutoffs(path: Path | str) -> dict[str, date]:
    """Parse the cutoffs YAML registry into ``{model_id: cutoff_date}``."""
    path = Path(path)
    with path.open("r", encoding="utf-8") as fh:
        doc = yaml.safe_load(fh)

    if not isinstance(doc, dict) or "models" not in doc:
        raise ValueError(
            f"Cutoffs file {path} must be a YAML mapping with a top-level 'models' key."
        )
    models = doc["models"]
    if not isinstance(models, dict):
        raise ValueError(
            f"Cutoffs file {path}: 'models' must be a mapping of model_id -> date."
        )

    out: dict[str, date] = {}
    for model_id, raw in models.items():
        if isinstance(raw, date):
            out[str(model_id)] = raw
        elif isinstance(raw, str):
            out[str(model_id)] = date.fromisoformat(raw)
        else:
            raise ValueError(
                f"Cutoffs file {path}: model {model_id!r} cutoff must be a date "
                f"or ISO-8601 string, got {type(raw).__name__}."
            )
    return out

load_eval_set

load_eval_set(path)

Parse a JSONL eval file into an EvalSet.

The first line may optionally be a header object containing {"_cutoff_date": "YYYY-MM-DD"}. All other lines must be row objects matching the input contract (Req 2.1). Logs WARNING records for low-N and class-imbalance conditions (Req 2.2, 2.3); never raises for those. Returns the entire set as a single list with no train/dev split (Req 2.4).

Source code in recall_guard/core/loader.py

def load_eval_set(path: Path | str) -> EvalSet:
    """Parse a JSONL eval file into an ``EvalSet``.

    The first line may optionally be a header object containing
    ``{"_cutoff_date": "YYYY-MM-DD"}``. All other lines must be row objects
    matching the input contract (Req 2.1). Logs WARNING records for low-N
    and class-imbalance conditions (Req 2.2, 2.3); never raises for those.
    Returns the entire set as a single list with no train/dev split (Req 2.4).
    """
    path = Path(path)
    cutoff: date | None = None
    rows: list[EvalRow] = []
    first_content_line = True

    with path.open("r", encoding="utf-8") as fh:
        for idx, raw_line in enumerate(fh, start=1):
            stripped = raw_line.strip()
            if not stripped:
                continue
            try:
                obj = json.loads(stripped)
            except json.JSONDecodeError as exc:
                raise ValueError(f"Row {idx}: invalid JSON ({exc.msg}).") from exc
            if not isinstance(obj, dict):
                raise ValueError(f"Row {idx}: expected JSON object, got {type(obj).__name__}.")

            # Header: only accepted as the very first non-empty line.
            is_header_slot = first_content_line
            first_content_line = False
            if is_header_slot and "_cutoff_date" in obj and "prompt" not in obj:
                raw_cutoff = obj["_cutoff_date"]
                if not isinstance(raw_cutoff, str):
                    raise ValueError(
                        f"Header '_cutoff_date' must be an ISO-8601 string, got "
                        f"{type(raw_cutoff).__name__}."
                    )
                try:
                    cutoff = date.fromisoformat(raw_cutoff)
                except ValueError as exc:
                    raise ValueError(
                        f"Header '_cutoff_date' must be ISO-8601 (YYYY-MM-DD): {exc}."
                    ) from exc
                continue

            rows.append(_parse_row(obj, idx))

    _emit_quality_warnings(path, rows)
    return EvalSet(rows=rows, cutoff_date=cutoff, path_hash=_hash_file(path))

compute_file_hash

compute_file_hash(path)

Return the sha256 hex digest of the bytes at path.

Reads in 8KB chunks via hashlib.sha256().update(chunk) so the function can hash files larger than fit comfortably in memory. The chunked read is semantically equivalent to hashlib.sha256(path.read_bytes()).hexdigest() for any file size; the dedicated test exercises a >16KB payload to make sure the chunk boundary does not corrupt the digest.

Source code in recall_guard/core/manifest.py

def compute_file_hash(path: Path | str) -> str:
    """Return the sha256 hex digest of the bytes at ``path``.

    Reads in 8KB chunks via ``hashlib.sha256().update(chunk)`` so the function
    can hash files larger than fit comfortably in memory. The chunked read is
    semantically equivalent to ``hashlib.sha256(path.read_bytes()).hexdigest()``
    for any file size; the dedicated test exercises a >16KB payload to make
    sure the chunk boundary does not corrupt the digest.
    """
    p = Path(path)
    h = hashlib.sha256()
    with p.open("rb") as fh:
        for chunk in iter(lambda: fh.read(_HASH_CHUNK_SIZE), b""):
            h.update(chunk)
    return h.hexdigest()

read_manifest

read_manifest(path)

Load manifest.json from path and reconstruct the dataclass.

Validates the top-level shape: the JSON object must have exactly the same keys as Manifest's fields. Missing or extra keys raise ValueError naming the offending key(s) so manifest drift is caught immediately rather than silently dropped on round-trip.

Source code in recall_guard/core/manifest.py

def read_manifest(path: Path | str) -> Manifest:
    """Load ``manifest.json`` from ``path`` and reconstruct the dataclass.

    Validates the top-level shape: the JSON object must have exactly the same
    keys as ``Manifest``'s fields. Missing or extra keys raise ``ValueError``
    naming the offending key(s) so manifest drift is caught immediately rather
    than silently dropped on round-trip.
    """
    p = Path(path)
    with p.open("r", encoding="utf-8") as fh:
        decoded = json.load(fh)

    if not isinstance(decoded, dict):
        raise ValueError(
            f"Manifest at {p} must be a JSON object, got {type(decoded).__name__}."
        )

    expected = _expected_field_names()
    # ``backtest`` is the sole optional field (cmmd-backtest extension,
    # Req 7.5, 8.2): legacy manifests omit the key entirely, so it must not
    # count as missing or extraneous.
    optional = {"backtest"}
    required = expected - optional
    actual = set(decoded.keys())

    missing = required - actual
    if missing:
        raise ValueError(
            f"Manifest at {p} is missing required key(s): {sorted(missing)}."
        )
    extra = actual - expected
    if extra:
        raise ValueError(
            f"Manifest at {p} has unexpected key(s): {sorted(extra)}."
        )

    # Strict nested validation: reject malformed values instead of silently
    # coercing them into different data (a string shortlist would otherwise
    # become a list of characters; a non-dict backtest would vanish to None
    # and be dropped on the next rewrite).
    shortlist_raw = decoded["shortlist"]
    if not isinstance(shortlist_raw, list) or not all(
        isinstance(m, str) for m in shortlist_raw
    ):
        raise ValueError(
            f"Manifest at {p}: 'shortlist' must be a list of model-ID strings, "
            f"got {shortlist_raw!r}."
        )

    backtest_raw = decoded.get("backtest")
    if backtest_raw is not None and not isinstance(backtest_raw, dict):
        raise ValueError(
            f"Manifest at {p}: 'backtest' must be a JSON object when present, "
            f"got {type(backtest_raw).__name__}."
        )
    backtest = dict(backtest_raw) if isinstance(backtest_raw, dict) else None

    return Manifest(
        harness_version=decoded["harness_version"],
        seed=decoded["seed"],
        eval_set_hash=decoded["eval_set_hash"],
        control_corpus_hash=decoded["control_corpus_hash"],
        is_memorized_hash=decoded["is_memorized_hash"],
        cutoffs_hash=decoded["cutoffs_hash"],
        shortlist=list(shortlist_raw),
        composite_score=dict(decoded["composite_score"]),
        mcs_hyperparams=dict(decoded["mcs_hyperparams"]),
        bootstrap_n=decoded["bootstrap_n"],
        artifacts=dict(decoded["artifacts"]),
        backtest=backtest,
    )

write_manifest

write_manifest(out_dir, manifest)

Serialise manifest to <out_dir>/manifest.json and return that path.

out_dir is created (with parents) if it does not yet exist; this lets the runner pin the output directory at run start before any per-run artifact has been produced. JSON is indented and key-sorted so the file is diff-friendly across runs that differ only in metadata order.

Source code in recall_guard/core/manifest.py

def write_manifest(out_dir: Path | str, manifest: Manifest) -> Path:
    """Serialise ``manifest`` to ``<out_dir>/manifest.json`` and return that path.

    ``out_dir`` is created (with parents) if it does not yet exist; this lets
    the runner pin the output directory at run start before any per-run
    artifact has been produced. JSON is indented and key-sorted so the file is
    diff-friendly across runs that differ only in metadata order.
    """
    out_dir = Path(out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)

    payload: dict[str, Any] = dataclasses.asdict(manifest)
    # Keep the on-disk schema byte-identical for legacy (no-backtest) runs:
    # absent the optional cmmd-backtest extension, ``backtest`` is omitted
    # rather than serialised as ``null``. This preserves the pre-existing
    # 11-key shape on which downstream tooling (and read_manifest's strict
    # key validator) is built.
    if payload.get("backtest") is None:
        payload.pop("backtest", None)

    target = out_dir / "manifest.json"
    with target.open("w", encoding="utf-8") as fh:
        json.dump(payload, fh, indent=2, sort_keys=True)
        # Trailing newline keeps the file POSIX-friendly for downstream tools.
        fh.write("\n")
    return target