Skip to content

recall_guard

recall_guard

recall_guard: measured inference-without-recall.

The public entry point. A consumer typically needs only::

from recall_guard import MemoryGuardedScorer

scorer = MemoryGuardedScorer.calibrate(
    api_key=...,                 # NVIDIA NIM key
    model="meta/llama-3.1-8b-instruct",
    is_memorized=[...],          # prompts dated before the model's cutoff
    oos_control=[...],           # prompts dated after it
)
guarded = scorer.score("<your prompt>")
guarded.signal, guarded.p_memorized, guarded.memguard_confidence

This module re-exports the façade plus a curated set of core and mia primitives. It deliberately re-exports no plotting symbol and nothing from the portfolio (backtest) layer, so import recall_guard never pulls in matplotlib or vectorbt (Req 4.1, 4.3). The plotting helpers remain available, on demand, via recall_guard.harness (lazy) and the backtest engine via recall_guard.portfolio.backtest (requires the backtest extra).

LOGPROB_FLOOR module-attribute 

LOGPROB_FLOOR = -30.0

Lower bound for individual logprob values, applied before averaging.

Prevents a single -inf (or extremely negative) per-token logprob from poisoning loss / min_k / zlib_ratio / ref_delta.

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 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
@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

CostEstimate dataclass

What an ensemble would cost, computed without issuing anything.

Source code in recall_guard/core/ensemble.py 
90
91
92
93
94
95
@dataclass(frozen=True)
class CostEstimate:
    """What an ensemble would cost, computed without issuing anything."""

    worst_case_requests: int
    estimated_seconds: float | None

CutoffViolation

Bases: Exception

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

Source code in recall_guard/core/loader.py 
44
45
class CutoffViolation(Exception):
    """Raised when shortlisted models post-date the eval set's cutoff."""

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 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@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 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
@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.

EvalRow dataclass

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

Source code in recall_guard/core/loader.py 
48
49
50
51
52
53
54
@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 
57
58
59
60
61
62
63
@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

LocationMode

Bases: StrEnum

Which location estimator to apply to an unflagged component.

Source code in recall_guard/core/ensemble.py 
63
64
65
66
67
68
class LocationMode(StrEnum):
    """Which location estimator to apply to an unflagged component."""

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

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 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@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

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 
71
72
73
74
75
76
77
78
79
80
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"

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 
330
331
332
333
334
335
336
337
338
339
340
341
342
343
@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

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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
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 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
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

ReferenceMode

Bases: StrEnum

Whether the optional reference draw varies per ensemble draw.

Source code in recall_guard/core/ensemble.py 
83
84
85
86
87
class ReferenceMode(StrEnum):
    """Whether the optional reference draw varies per ensemble draw."""

    FIXED = "fixed"
    PER_DRAW = "per_draw"

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 
48
49
50
51
52
53
54
55
56
57
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"

TokenLogprob dataclass

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

Source code in recall_guard/core/nvidia_lm.py 
63
64
65
66
67
68
69
@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]]

ConfigurationError

Bases: RuntimeError

Raised when the NIM credential is absent, empty, or rejected (Req 3.7).

Source code in recall_guard/harness/scorer.py 
82
83
class ConfigurationError(RuntimeError):
    """Raised when the NIM credential is absent, empty, or rejected (Req 3.7)."""

EnsembledScore dataclass

One prompt scored over many draws.

p_memorized_point is the exposure multiplier. consensus.p_memorized is the score of one actually-observed draw and is evidence only -- the two can differ, because the representative draw is selected by rank while the point estimate is a reduction over all draws. Using the consensus draw's score to scale exposure would silently substitute a single draw for the ensemble, which is the thing this feature exists to stop.

sampled_at records when the draws were taken. The sampled distribution moves between sessions as well as within one, so a consensus has a shelf life and a cached one is not the same as a fresh one.

p_memorized_point is None exactly when the ensemble failed, and it is never 0.0 in that case: zero would mean "pass 100% of exposure through", the opposite of what an unusable measurement should imply.

Source code in recall_guard/harness/scorer.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@dataclass(frozen=True)
class EnsembledScore:
    """One prompt scored over many draws.

    **`p_memorized_point` is the exposure multiplier.** ``consensus.p_memorized``
    is the score of one actually-observed draw and is evidence only -- the two
    can differ, because the representative draw is selected by rank while the
    point estimate is a reduction over all draws. Using the consensus draw's
    score to scale exposure would silently substitute a single draw for the
    ensemble, which is the thing this feature exists to stop.

    ``sampled_at`` records when the draws were taken. The sampled distribution
    moves between sessions as well as within one, so a consensus has a shelf
    life and a cached one is not the same as a fresh one.

    ``p_memorized_point is None`` exactly when the ensemble failed, and it is
    never ``0.0`` in that case: zero would mean "pass 100% of exposure through",
    the opposite of what an unusable measurement should imply.
    """

    consensus: GuardedScore
    p_memorized_point: float | None
    p_memorized_ci: tuple[float, float] | None
    p_memorized_conservative: float | None
    agreement: float | None
    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[GuardedScore, ...] = ()

GuardedScore dataclass

One guarded inference result.

Attributes:

Name Type Description
prompt_hash str

First 16 hex chars of sha256(prompt) (matches the harness convention).
parse_ok bool

True when the response parsed and the MIA/MCS pipeline ran.
signal int | None

Parsed direction in {-1, 0, 1}; None on failure.
raw_confidence float | None

Parsed confidence in [0, 1]; None on failure.
p_memorized float | None

Calibrated p(memorized | features) ∈ [0, 1]; None on failure.
memguard_confidence float | None

raw_confidence * (1 - p_memorized); None on failure.
features MiaFeatures | None

The raw :class:MiaFeatures; None on failure.
fail_reason str | None

One of "timeout" / "no_logprobs" / "parse_failure" / "error" on failure; None on success.
Source code in recall_guard/harness/scorer.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@dataclass(frozen=True)
class GuardedScore:
    """One guarded inference result.

    Attributes
    ----------
    prompt_hash:
        First 16 hex chars of ``sha256(prompt)`` (matches the harness convention).
    parse_ok:
        ``True`` when the response parsed and the MIA/MCS pipeline ran.
    signal:
        Parsed direction in ``{-1, 0, 1}``; ``None`` on failure.
    raw_confidence:
        Parsed confidence in ``[0, 1]``; ``None`` on failure.
    p_memorized:
        Calibrated ``p(memorized | features) ∈ [0, 1]``; ``None`` on failure.
    memguard_confidence:
        ``raw_confidence * (1 - p_memorized)``; ``None`` on failure.
    features:
        The raw :class:`MiaFeatures`; ``None`` on failure.
    fail_reason:
        One of ``"timeout"`` / ``"no_logprobs"`` / ``"parse_failure"`` / ``"error"``
        on failure; ``None`` on success.
    """

    prompt_hash: str
    parse_ok: bool
    signal: int | None
    raw_confidence: float | None
    p_memorized: float | None
    memguard_confidence: float | None
    features: MiaFeatures | None
    fail_reason: str | None

MemoryGuardedScorer

Calibrated, per-model inference-without-recall scorer.

Construct via :meth:calibrate (which performs the model calls and training), then call :meth:score / :meth:score_many.

Source code in recall_guard/harness/scorer.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
class MemoryGuardedScorer:
    """Calibrated, per-model inference-without-recall scorer.

    Construct via :meth:`calibrate` (which performs the model calls and training),
    then call :meth:`score` / :meth:`score_many`.
    """

    def __init__(
        self,
        *,
        lm: NvidiaLM,
        baseline: ControlBaseline,
        mcs: MCSCalibrator,
        ref_lm: NvidiaLM | None,
    ) -> None:
        self._lm = lm
        self._baseline = baseline
        self._mcs = mcs
        self._ref_lm = ref_lm

    # -- public read-only state ------------------------------------------------

    @property
    def model(self) -> str:
        return self._mcs.model

    @property
    def holdout_auc(self) -> float:
        """Held-out IS/OOS separation of the trained calibrator (Req 3.4)."""
        return self._mcs.holdout_auc

    @property
    def is_weak(self) -> bool:
        """``True`` when ``holdout_auc`` is below the calibration gate (Req 3.4)."""
        return self._mcs.is_weak

    # -- construction ----------------------------------------------------------

    @classmethod
    def calibrate(
        cls,
        *,
        api_key: str,
        model: str,
        is_memorized: Sequence[str],
        oos_control: Sequence[str],
        reference_model: str | None = None,
        min_auc: float = 0.6,
        min_valid: int = 50,
        seed: int = 0,
        max_workers: int = 8,
        timeout_s: float = 45.0,
        min_call_interval_s: float = 0.0,
        lm_factory: LMFactory | None = None,
    ) -> MemoryGuardedScorer:
        """Build the control baseline and train the MCS calibrator for ``model``.

        Raises
        ------
        ConfigurationError
            If ``api_key`` is empty, or if the model returns no usable responses
            during calibration (the typical signature of a rejected credential,
            an unavailable model, or an unreachable endpoint) (Req 3.7).
        ValueError
            If a class has too few usable rows to calibrate / train (Req 3.4).
        """
        if not api_key:
            raise ConfigurationError(
                "NVIDIA api_key is required to calibrate a MemoryGuardedScorer; "
                "got an empty value. Set NVIDIA_API_KEY or pass api_key=."
            )

        factory = lm_factory or _default_factory(min_call_interval_s)
        lm = factory(api_key, model, timeout_s)
        ref_lm = factory(api_key, reference_model, timeout_s) if reference_model else None

        oos_rows = _rows(oos_control)
        is_rows = _rows(is_memorized)

        baseline = build_baseline(
            lm, oos_rows, ref_lm, min_valid=min_valid, max_workers=max_workers
        )
        if baseline.n_valid == 0:
            raise ConfigurationError(
                f"model {model!r} returned no usable responses during calibration; "
                "check NVIDIA_API_KEY, the model id, and endpoint availability."
            )
        if not baseline.is_calibrated:
            raise ValueError(
                f"control baseline could not calibrate for {model!r}: "
                f"{baseline.n_valid} usable rows < min_valid={min_valid}."
            )

        mcs = _mcs_train(
            model_lm=lm,
            is_memorized=is_rows,
            oos_control=oos_rows,
            baseline=baseline,
            ref_lm=ref_lm,
            min_auc=min_auc,
            seed=seed,
            max_workers=max_workers,
        )
        return cls(lm=lm, baseline=baseline, mcs=mcs, ref_lm=ref_lm)

    # -- scoring ---------------------------------------------------------------

    def score(self, prompt: str) -> GuardedScore:
        """Score one prompt into a :class:`GuardedScore`.

        Raises
        ------
        ConfigurationError
            If the NIM endpoint rejects the credential while scoring (Req 3.7).
        """
        primary = self._safe_generate(self._lm, prompt)
        ref_res = self._safe_generate(self._ref_lm, prompt) if self._ref_lm else None
        return self._build_guarded_score(prompt, primary, ref_res)

    def score_many(self, prompts: Sequence[str], *, max_workers: int = 8) -> list[GuardedScore]:
        """Score many prompts (parallel LM calls); preserves input order."""
        primaries = generate_many(self._lm, list(prompts), max_workers=max_workers)
        refs: list = (
            generate_many(self._ref_lm, list(prompts), max_workers=max_workers)
            if self._ref_lm is not None
            else [None] * len(prompts)
        )
        return [
            self._build_guarded_score(p, primary, ref_res)
            for p, primary, ref_res in zip(prompts, primaries, refs, strict=True)
        ]

    def score_ensemble(
        self,
        prompt: str,
        *,
        spec: EnsembleSpec,
        conservative_quantile: float | None = None,
    ) -> EnsembledScore:
        """Score one prompt over ``spec.draws`` draws and reduce the results.

        A single scoring is close to uninformative as an exposure multiplier:
        measured on one identical prompt, its 95% band spans two thirds of the
        unit interval. Ensembling narrows that, and reports what is left.

        Each draw is scored through the *unchanged* single-draw path and the
        resulting scores are then reduced. Averaging the intermediate features
        and scoring once would be a different quantity -- the calibrator is a
        sigmoid, so the score of the mean is not the mean of the scores.

        The point estimate is the **mean**, because attenuation is linear in the
        score and the mean is therefore unbiased for expected attenuation. No
        symmetric trimming is applied: the upper tail of this distribution is
        the contamination evidence the score exists to report, so trimming it
        away would discard the signal and shift the estimate toward the
        risk-increasing side.

        Parameters
        ----------
        spec:
            Explicit configuration. There is no default instance.
        conservative_quantile:
            Optional upper quantile of the score, for a caller who would rather
            withhold more exposure than risk withholding too little.

        Raises
        ------
        ConfigurationError
            If the endpoint rejects the credential while drawing.
        """
        if conservative_quantile is not None and not 0.0 <= conservative_quantile <= 1.0:
            raise ValueError(
                f"conservative_quantile must be in [0, 1]; got {conservative_quantile!r}"
            )

        prompt_hash = _hash_prompt(prompt)
        started_at = datetime.now(UTC).isoformat()
        # The reference draw is held fixed across the ensemble: varying it would
        # double the request count for one of four features. The cost is that
        # every draw's score is then correlated through that shared reference,
        # so the reported spread understates the true spread.
        ref_res = (
            self._safe_generate(self._ref_lm, prompt, spec=spec) if self._ref_lm else None
        )

        scored: list[GuardedScore] = []
        failures: dict[str, int] = {}
        contents: list[str] = []
        waves: list[int] = []

        for wave, batch in enumerate(_wave_sizes(spec)):
            with ThreadPoolExecutor(max_workers=batch) as pool:
                draws = list(
                    pool.map(
                        lambda _: self._safe_generate(self._lm, prompt, spec=spec),
                        range(batch),
                    )
                )
            for draw in draws:
                if isinstance(draw, RuntimeError) and _is_auth_error(draw):
                    raise ConfigurationError(
                        f"NIM rejected the credential while scoring model {self.model!r}: {draw}"
                    )
                if isinstance(draw, BaseException) or draw is None:
                    reason = "timeout" if isinstance(draw, TimeoutError) else "transport"
                    failures[reason] = failures.get(reason, 0) + 1
                    continue
                guarded = self._build_guarded_score(prompt, draw, ref_res)
                if not guarded.parse_ok:
                    failures[guarded.fail_reason or FAIL_ERROR] = (
                        failures.get(guarded.fail_reason or FAIL_ERROR, 0) + 1
                    )
                    continue
                scored.append(guarded)
                contents.append(draw.content)
                waves.append(wave)

        return self._reduce_scores(
            prompt_hash, scored, contents, waves, failures, spec,
            conservative_quantile, started_at,
        )

    def _reduce_scores(
        self,
        prompt_hash: str,
        scored: list[GuardedScore],
        contents: list[str],
        waves: list[int],
        failures: dict[str, int],
        spec: EnsembleSpec,
        conservative_quantile: float | None,
        sampled_at: str,
    ) -> EnsembledScore:
        fail_counts = tuple(sorted(failures.items()))
        if len(scored) < spec.min_parsed or not scored:
            # Report the failure rather than a consensus over the survivors. A
            # confident-looking answer computed from a handful of draws is the
            # false-success artifact this package refuses to mint.
            return EnsembledScore(
                consensus=_fail(prompt_hash, _modal_reason(failures)),
                p_memorized_point=None,
                p_memorized_ci=None,
                p_memorized_conservative=None,
                agreement=None,
                agreement_ci=None,
                draw_dependence=None,
                max_tokens=spec.max_tokens,
                temperature=spec.temperature,
                sampled_at=sampled_at,
                n_requested=spec.draws,
                n_parsed=len(scored),
                fail_counts=fail_counts,
                draws_sha256=canonical_draw_hash(contents),
                draws=(),
            )

        order = sorted(range(len(scored)), key=lambda i: (scored[i].p_memorized, contents[i]))
        ordered = [scored[i] for i in order]
        values = [g.p_memorized for g in ordered]

        signals = [g.signal for g in ordered]
        tally: dict[int | None, int] = {}
        for signal in signals:
            tally[signal] = tally.get(signal, 0) + 1
        modal = min(tally, key=lambda s: (-tally[s], repr(s)))

        return EnsembledScore(
            # The representative draw is ranked by the score itself, so it sits
            # at the middle of the very distribution being reduced. It is still
            # evidence, not the multiplier -- see the class docstring.
            consensus=ordered[(len(ordered) - 1) // 2],
            p_memorized_point=math.fsum(values) / len(values),
            p_memorized_ci=_empirical_interval(values, spec.confidence),
            p_memorized_conservative=(
                _quantile(values, conservative_quantile)
                if conservative_quantile is not None
                else None
            ),
            agreement=tally[modal] / len(ordered),
            agreement_ci=wilson_interval(
                tally[modal], len(ordered), confidence=spec.confidence, tail=spec.tail
            ),
            draw_dependence=lag_dependence(signals, [waves[i] for i in order]),
            max_tokens=spec.max_tokens,
            temperature=spec.temperature,
            sampled_at=sampled_at,
            n_requested=spec.draws,
            n_parsed=len(ordered),
            fail_counts=fail_counts,
            draws_sha256=canonical_draw_hash(contents),
            draws=tuple(ordered) if spec.retain_draws else (),
        )

    # -- internals -------------------------------------------------------------

    @staticmethod
    def _safe_generate(lm: NvidiaLM | None, prompt: str, *, spec: EnsembleSpec | None = None):
        """One draw, optionally under an ensemble spec's generation settings.

        With no spec the client's own defaults apply, which is what keeps the
        single-draw path -- and a one-draw ensemble that overrides nothing --
        byte-identical to what it always was.
        """
        if lm is None:
            return None
        kwargs: dict[str, object] = {}
        if spec is not None and spec.max_tokens is not None:
            kwargs["max_tokens"] = spec.max_tokens
        if spec is not None and spec.temperature is not None:
            kwargs["temperature"] = spec.temperature
        try:
            return lm.generate(prompt, **kwargs)
        except (TimeoutError, RuntimeError) as exc:
            return exc

    def _build_guarded_score(self, prompt: str, primary, ref_res) -> GuardedScore:
        prompt_hash = _hash_prompt(prompt)

        if isinstance(primary, TimeoutError):
            return _fail(prompt_hash, FAIL_TIMEOUT)
        if isinstance(primary, RuntimeError):
            if _is_auth_error(primary):
                raise ConfigurationError(
                    f"NIM rejected the credential while scoring model {self.model!r}: {primary}"
                )
            return _fail(prompt_hash, _classify_runtime_error(primary))
        if isinstance(primary, BaseException) or primary is None:
            return _fail(prompt_hash, FAIL_ERROR)

        content = primary.content
        direction = _parse_direction(content)
        confidence = _parse_confidence(content)
        if direction is None or confidence is None:
            return _fail(prompt_hash, FAIL_PARSE)

        ref_logprobs = None
        if self._ref_lm is not None and ref_res is not None and not isinstance(ref_res, BaseException):
            ref_logprobs = ref_res.logprobs

        try:
            features = compute_mia_features(content, primary.logprobs, ref_logprobs)
        except (ValueError, RuntimeError):
            return _fail(prompt_hash, FAIL_ERROR)

        try:
            p_memorized = float(self._mcs.predict_proba(features, self._baseline))
        except ValueError:
            return _fail(prompt_hash, FAIL_ERROR)

        return GuardedScore(
            prompt_hash=prompt_hash,
            parse_ok=True,
            signal=direction,
            raw_confidence=float(confidence),
            p_memorized=p_memorized,
            memguard_confidence=float(confidence) * (1.0 - p_memorized),
            features=features,
            fail_reason=None,
        )

holdout_auc property 

holdout_auc

Held-out IS/OOS separation of the trained calibrator (Req 3.4).

is_weak property 

is_weak

True when holdout_auc is below the calibration gate (Req 3.4).

calibrate classmethod 

calibrate(
    *,
    api_key,
    model,
    is_memorized,
    oos_control,
    reference_model=None,
    min_auc=0.6,
    min_valid=50,
    seed=0,
    max_workers=8,
    timeout_s=45.0,
    min_call_interval_s=0.0,
    lm_factory=None,
)

Build the control baseline and train the MCS calibrator for model.

Raises:

Type Description
ConfigurationError

If api_key is empty, or if the model returns no usable responses during calibration (the typical signature of a rejected credential, an unavailable model, or an unreachable endpoint) (Req 3.7).
ValueError

If a class has too few usable rows to calibrate / train (Req 3.4).
Source code in recall_guard/harness/scorer.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
@classmethod
def calibrate(
    cls,
    *,
    api_key: str,
    model: str,
    is_memorized: Sequence[str],
    oos_control: Sequence[str],
    reference_model: str | None = None,
    min_auc: float = 0.6,
    min_valid: int = 50,
    seed: int = 0,
    max_workers: int = 8,
    timeout_s: float = 45.0,
    min_call_interval_s: float = 0.0,
    lm_factory: LMFactory | None = None,
) -> MemoryGuardedScorer:
    """Build the control baseline and train the MCS calibrator for ``model``.

    Raises
    ------
    ConfigurationError
        If ``api_key`` is empty, or if the model returns no usable responses
        during calibration (the typical signature of a rejected credential,
        an unavailable model, or an unreachable endpoint) (Req 3.7).
    ValueError
        If a class has too few usable rows to calibrate / train (Req 3.4).
    """
    if not api_key:
        raise ConfigurationError(
            "NVIDIA api_key is required to calibrate a MemoryGuardedScorer; "
            "got an empty value. Set NVIDIA_API_KEY or pass api_key=."
        )

    factory = lm_factory or _default_factory(min_call_interval_s)
    lm = factory(api_key, model, timeout_s)
    ref_lm = factory(api_key, reference_model, timeout_s) if reference_model else None

    oos_rows = _rows(oos_control)
    is_rows = _rows(is_memorized)

    baseline = build_baseline(
        lm, oos_rows, ref_lm, min_valid=min_valid, max_workers=max_workers
    )
    if baseline.n_valid == 0:
        raise ConfigurationError(
            f"model {model!r} returned no usable responses during calibration; "
            "check NVIDIA_API_KEY, the model id, and endpoint availability."
        )
    if not baseline.is_calibrated:
        raise ValueError(
            f"control baseline could not calibrate for {model!r}: "
            f"{baseline.n_valid} usable rows < min_valid={min_valid}."
        )

    mcs = _mcs_train(
        model_lm=lm,
        is_memorized=is_rows,
        oos_control=oos_rows,
        baseline=baseline,
        ref_lm=ref_lm,
        min_auc=min_auc,
        seed=seed,
        max_workers=max_workers,
    )
    return cls(lm=lm, baseline=baseline, mcs=mcs, ref_lm=ref_lm)

score 

score(prompt)

Score one prompt into a :class:GuardedScore.

Raises:

Type Description
ConfigurationError

If the NIM endpoint rejects the credential while scoring (Req 3.7).
Source code in recall_guard/harness/scorer.py 
296
297
298
299
300
301
302
303
304
305
306
def score(self, prompt: str) -> GuardedScore:
    """Score one prompt into a :class:`GuardedScore`.

    Raises
    ------
    ConfigurationError
        If the NIM endpoint rejects the credential while scoring (Req 3.7).
    """
    primary = self._safe_generate(self._lm, prompt)
    ref_res = self._safe_generate(self._ref_lm, prompt) if self._ref_lm else None
    return self._build_guarded_score(prompt, primary, ref_res)

score_many 

score_many(prompts, *, max_workers=8)

Score many prompts (parallel LM calls); preserves input order.

Source code in recall_guard/harness/scorer.py 
308
309
310
311
312
313
314
315
316
317
318
319
def score_many(self, prompts: Sequence[str], *, max_workers: int = 8) -> list[GuardedScore]:
    """Score many prompts (parallel LM calls); preserves input order."""
    primaries = generate_many(self._lm, list(prompts), max_workers=max_workers)
    refs: list = (
        generate_many(self._ref_lm, list(prompts), max_workers=max_workers)
        if self._ref_lm is not None
        else [None] * len(prompts)
    )
    return [
        self._build_guarded_score(p, primary, ref_res)
        for p, primary, ref_res in zip(prompts, primaries, refs, strict=True)
    ]

score_ensemble 

score_ensemble(prompt, *, spec, conservative_quantile=None)

Score one prompt over spec.draws draws and reduce the results.

A single scoring is close to uninformative as an exposure multiplier: measured on one identical prompt, its 95% band spans two thirds of the unit interval. Ensembling narrows that, and reports what is left.

Each draw is scored through the unchanged single-draw path and the resulting scores are then reduced. Averaging the intermediate features and scoring once would be a different quantity -- the calibrator is a sigmoid, so the score of the mean is not the mean of the scores.

The point estimate is the mean, because attenuation is linear in the score and the mean is therefore unbiased for expected attenuation. No symmetric trimming is applied: the upper tail of this distribution is the contamination evidence the score exists to report, so trimming it away would discard the signal and shift the estimate toward the risk-increasing side.

Parameters:

Name Type Description Default
spec EnsembleSpec

Explicit configuration. There is no default instance.

 required
conservative_quantile float | None

Optional upper quantile of the score, for a caller who would rather withhold more exposure than risk withholding too little.

 None

Raises:

Type Description
ConfigurationError

If the endpoint rejects the credential while drawing.
Source code in recall_guard/harness/scorer.py 
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
def score_ensemble(
    self,
    prompt: str,
    *,
    spec: EnsembleSpec,
    conservative_quantile: float | None = None,
) -> EnsembledScore:
    """Score one prompt over ``spec.draws`` draws and reduce the results.

    A single scoring is close to uninformative as an exposure multiplier:
    measured on one identical prompt, its 95% band spans two thirds of the
    unit interval. Ensembling narrows that, and reports what is left.

    Each draw is scored through the *unchanged* single-draw path and the
    resulting scores are then reduced. Averaging the intermediate features
    and scoring once would be a different quantity -- the calibrator is a
    sigmoid, so the score of the mean is not the mean of the scores.

    The point estimate is the **mean**, because attenuation is linear in the
    score and the mean is therefore unbiased for expected attenuation. No
    symmetric trimming is applied: the upper tail of this distribution is
    the contamination evidence the score exists to report, so trimming it
    away would discard the signal and shift the estimate toward the
    risk-increasing side.

    Parameters
    ----------
    spec:
        Explicit configuration. There is no default instance.
    conservative_quantile:
        Optional upper quantile of the score, for a caller who would rather
        withhold more exposure than risk withholding too little.

    Raises
    ------
    ConfigurationError
        If the endpoint rejects the credential while drawing.
    """
    if conservative_quantile is not None and not 0.0 <= conservative_quantile <= 1.0:
        raise ValueError(
            f"conservative_quantile must be in [0, 1]; got {conservative_quantile!r}"
        )

    prompt_hash = _hash_prompt(prompt)
    started_at = datetime.now(UTC).isoformat()
    # The reference draw is held fixed across the ensemble: varying it would
    # double the request count for one of four features. The cost is that
    # every draw's score is then correlated through that shared reference,
    # so the reported spread understates the true spread.
    ref_res = (
        self._safe_generate(self._ref_lm, prompt, spec=spec) if self._ref_lm else None
    )

    scored: list[GuardedScore] = []
    failures: dict[str, int] = {}
    contents: list[str] = []
    waves: list[int] = []

    for wave, batch in enumerate(_wave_sizes(spec)):
        with ThreadPoolExecutor(max_workers=batch) as pool:
            draws = list(
                pool.map(
                    lambda _: self._safe_generate(self._lm, prompt, spec=spec),
                    range(batch),
                )
            )
        for draw in draws:
            if isinstance(draw, RuntimeError) and _is_auth_error(draw):
                raise ConfigurationError(
                    f"NIM rejected the credential while scoring model {self.model!r}: {draw}"
                )
            if isinstance(draw, BaseException) or draw is None:
                reason = "timeout" if isinstance(draw, TimeoutError) else "transport"
                failures[reason] = failures.get(reason, 0) + 1
                continue
            guarded = self._build_guarded_score(prompt, draw, ref_res)
            if not guarded.parse_ok:
                failures[guarded.fail_reason or FAIL_ERROR] = (
                    failures.get(guarded.fail_reason or FAIL_ERROR, 0) + 1
                )
                continue
            scored.append(guarded)
            contents.append(draw.content)
            waves.append(wave)

    return self._reduce_scores(
        prompt_hash, scored, contents, waves, failures, spec,
        conservative_quantile, started_at,
    )

ControlBaseline dataclass

Per-model baseline distribution of every MIA feature on the OOS control corpus.

Attributes:

Name Type Description
model str

The NVIDIA model ID this baseline was built for.
n_valid int

Number of control rows where model_lm.generate returned usable logprobs (i.e., did not raise TimeoutError or RuntimeError).
feature_means dict[str, float | None]

Per-feature mean across the valid rows. Keys are the five MIA feature names. feature_means["ref_delta"] is None when no reference model is configured (or every reference call failed).
feature_stds dict[str, float | None]

Per-feature standard deviation across the valid rows, floored at _STD_FLOOR. feature_stds["ref_delta"] is None whenever feature_means["ref_delta"] is None.
is_calibrated bool

True iff n_valid >= min_valid. Used by the runner to decide whether to evaluate the model or surface an uncalibrated warning.
min_valid int

The threshold used (default 50, per the Open Defaults in requirements.md).
Source code in recall_guard/mia/control.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@dataclass(frozen=True)
class ControlBaseline:
    """Per-model baseline distribution of every MIA feature on the OOS control corpus.

    Attributes
    ----------
    model:
        The NVIDIA model ID this baseline was built for.
    n_valid:
        Number of control rows where ``model_lm.generate`` returned usable
        logprobs (i.e., did not raise ``TimeoutError`` or ``RuntimeError``).
    feature_means:
        Per-feature mean across the valid rows. Keys are the five MIA feature
        names. ``feature_means["ref_delta"]`` is ``None`` when no reference
        model is configured (or every reference call failed).
    feature_stds:
        Per-feature standard deviation across the valid rows, floored at
        ``_STD_FLOOR``. ``feature_stds["ref_delta"]`` is ``None`` whenever
        ``feature_means["ref_delta"]`` is ``None``.
    is_calibrated:
        ``True`` iff ``n_valid >= min_valid``. Used by the runner to decide
        whether to evaluate the model or surface an ``uncalibrated`` warning.
    min_valid:
        The threshold used (default 50, per the Open Defaults in
        requirements.md).
    """

    model: str
    n_valid: int
    feature_means: dict[str, float | None]
    feature_stds: dict[str, float | None]
    is_calibrated: bool
    min_valid: int

MCSCalibrator dataclass

Per-model logistic-regression calibrator for p(memorized | features).

Attributes:

Name Type Description
model str

The NVIDIA model ID this calibrator was trained for.
classifier LogisticRegression

The fitted sklearn.linear_model.LogisticRegression instance. sklearn estimators are mutable; frozen=True only prevents reassignment of the field reference, which is the design intent.
feature_order list[str]

Canonical order used to flatten the standardised feature dict into the classifier's input vector. Populated at train time and consumed verbatim by :meth:predict_proba so the classifier is never fed a permuted row.
holdout_auc float

ROC-AUC score of the trained classifier on the 25% held-out portion of the labelled IS/OOS corpus. Reported in the manifest and the per-model evaluation result (Req 5.2).
is_weak bool

True iff holdout_auc < min_auc at train time. Surfaced as the weak-calibration warning in top3.md (Req 5.3).
Source code in recall_guard/mia/mcs.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@dataclass(frozen=True)
class MCSCalibrator:
    """Per-model logistic-regression calibrator for ``p(memorized | features)``.

    Attributes
    ----------
    model:
        The NVIDIA model ID this calibrator was trained for.
    classifier:
        The fitted ``sklearn.linear_model.LogisticRegression`` instance.
        sklearn estimators are mutable; ``frozen=True`` only prevents
        reassignment of the field reference, which is the design intent.
    feature_order:
        Canonical order used to flatten the standardised feature dict
        into the classifier's input vector. Populated at train time and
        consumed verbatim by :meth:`predict_proba` so the classifier is
        never fed a permuted row.
    holdout_auc:
        ROC-AUC score of the trained classifier on the 25% held-out
        portion of the labelled IS/OOS corpus. Reported in the manifest
        and the per-model evaluation result (Req 5.2).
    is_weak:
        ``True`` iff ``holdout_auc < min_auc`` at train time. Surfaced
        as the ``weak-calibration`` warning in ``top3.md`` (Req 5.3).
    """

    model: str
    classifier: LogisticRegression
    feature_order: list[str]
    holdout_auc: float
    is_weak: bool

    def predict_proba(
        self, features: MiaFeatures, baseline: ControlBaseline
    ) -> float:
        """Return the calibrated probability of "memorized" for one record.

        Standardises ``features`` against the model's ``baseline`` and
        feeds the resulting vector to the trained classifier in
        ``self.feature_order``.

        Returns
        -------
        float
            ``p(memorized | features) ∈ [0.0, 1.0]``.

        Raises
        ------
        ValueError
            If any of the four core features standardises to ``None``
            (uncalibrated baseline). A missing ``ref_delta`` does NOT
            raise: the reference feature is optional by contract, so it
            is imputed at the control-baseline mean (standardised 0.0),
            which contributes no memorization evidence either way.
        """
        standardised = standardise(features, baseline)
        if "ref_delta" in self.feature_order and standardised.get("ref_delta") is None:
            standardised = {**standardised, "ref_delta": 0.0}
        row = _row_vector(standardised, self.feature_order)
        # Estimator was trained on a 2-D matrix; predict on a 1-row matrix.
        proba = float(self.classifier.predict_proba(row.reshape(1, -1))[0, 1])
        # sklearn returns values strictly in [0, 1] for LR; clamp defensively
        # against fp64 round-off so the float postcondition holds exactly.
        if proba < 0.0:
            proba = 0.0
        elif proba > 1.0:
            proba = 1.0
        return proba

predict_proba 

predict_proba(features, baseline)

Return the calibrated probability of "memorized" for one record.

Standardises features against the model's baseline and feeds the resulting vector to the trained classifier in self.feature_order.

Returns:

Type Description
float

p(memorized | features) ∈ [0.0, 1.0].

Raises:

Type Description
ValueError

If any of the four core features standardises to None (uncalibrated baseline). A missing ref_delta does NOT raise: the reference feature is optional by contract, so it is imputed at the control-baseline mean (standardised 0.0), which contributes no memorization evidence either way.
Source code in recall_guard/mia/mcs.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def predict_proba(
    self, features: MiaFeatures, baseline: ControlBaseline
) -> float:
    """Return the calibrated probability of "memorized" for one record.

    Standardises ``features`` against the model's ``baseline`` and
    feeds the resulting vector to the trained classifier in
    ``self.feature_order``.

    Returns
    -------
    float
        ``p(memorized | features) ∈ [0.0, 1.0]``.

    Raises
    ------
    ValueError
        If any of the four core features standardises to ``None``
        (uncalibrated baseline). A missing ``ref_delta`` does NOT
        raise: the reference feature is optional by contract, so it
        is imputed at the control-baseline mean (standardised 0.0),
        which contributes no memorization evidence either way.
    """
    standardised = standardise(features, baseline)
    if "ref_delta" in self.feature_order and standardised.get("ref_delta") is None:
        standardised = {**standardised, "ref_delta": 0.0}
    row = _row_vector(standardised, self.feature_order)
    # Estimator was trained on a 2-D matrix; predict on a 1-row matrix.
    proba = float(self.classifier.predict_proba(row.reshape(1, -1))[0, 1])
    # sklearn returns values strictly in [0, 1] for LR; clamp defensively
    # against fp64 round-off so the float postcondition holds exactly.
    if proba < 0.0:
        proba = 0.0
    elif proba > 1.0:
        proba = 1.0
    return proba

MiaFeatures dataclass

Five MIA features for one (model, prompt, response) record.

Attributes:

Name Type Description
loss float

Mean negative logprob of the realised tokens (clipped at floor). Low loss means the model found the text easy to predict, which is what stored text looks like.
min_k float

Mean of the bottom int(len * k) clipped logprobs (Min-K%). Negative; lower means more "memorized". Looks only at the hardest tokens, because that is where memorization shows first: if the model breezes through even those, it has probably seen the text.
min_k_pp float

Mean of the bottom-K per-position z-scores (Min-K%++). Same idea as min_k, but each token is graded against its own candidate distribution instead of an absolute scale.
zlib_ratio float

-sum(clipped_logprobs) / len(zlib.compress(response, 9)). 0.0 when response is empty. Dividing by the compressed size cancels plain repetitiveness; a repetitive text is cheap to predict AND cheap to compress, so what remains is the confidence the model has beyond what the text's redundancy explains.
ref_delta float | None

loss_self - loss_ref; None when ref_logprobs is None. The reference model anchors what "normal" confidence looks like for the same text, so shared easiness cancels and model-specific recall remains.
Source code in recall_guard/mia/features.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@dataclass(frozen=True)
class MiaFeatures:
    """Five MIA features for one (model, prompt, response) record.

    Attributes
    ----------
    loss:
        Mean negative logprob of the realised tokens (clipped at floor).
        Low loss means the model found the text easy to predict, which is
        what stored text looks like.
    min_k:
        Mean of the bottom ``int(len * k)`` clipped logprobs (Min-K%).
        Negative; lower means more "memorized". Looks only at the hardest
        tokens, because that is where memorization shows first: if the
        model breezes through even those, it has probably seen the text.
    min_k_pp:
        Mean of the bottom-K per-position z-scores (Min-K%++). Same idea as
        ``min_k``, but each token is graded against its own candidate
        distribution instead of an absolute scale.
    zlib_ratio:
        ``-sum(clipped_logprobs) / len(zlib.compress(response, 9))``.
        ``0.0`` when ``response`` is empty. Dividing by the compressed size
        cancels plain repetitiveness; a repetitive text is cheap to predict
        AND cheap to compress, so what remains is the confidence the model
        has beyond what the text's redundancy explains.
    ref_delta:
        ``loss_self - loss_ref``; ``None`` when ``ref_logprobs is None``.
        The reference model anchors what "normal" confidence looks like
        for the same text, so shared easiness cancels and model-specific
        recall remains.
    """

    loss: float
    min_k: float
    min_k_pp: float
    zlib_ratio: float
    ref_delta: float | None

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 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
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}"
        )

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 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
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

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 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
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()

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 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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()

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 
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
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,
    )

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 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
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 
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
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,
    )

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 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
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 
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
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

load_cutoffs 

load_cutoffs(path)

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

Source code in recall_guard/core/loader.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
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 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
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))

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 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
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,
    )

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 
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
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 (),
    )

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 
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
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 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
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 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
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 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
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 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
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 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
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))

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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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

build_baseline 

build_baseline(
    model_lm,
    control_rows,
    ref_lm,
    min_valid=50,
    max_workers=1,
)

Build a per-model control-corpus baseline.

For each row in control_rows:

  • Call model_lm.generate(row.prompt). On TimeoutError or RuntimeError (e.g., missing logprobs) the row is dropped and a WARNING is logged with the row index.
  • When ref_lm is provided, also call ref_lm.generate(row.prompt). A reference-side failure does not invalidate the row; it merely sets ref_logprobs = None for that row, so the four other features still contribute to the baseline.
  • Compute :class:MiaFeatures via :func:compute_mia_features.

Per-feature mean and std are aggregated with numpy.mean and numpy.std(ddof=0). Std is floored at _STD_FLOOR to avoid div-by-zero in :func:standardise. When every valid row has ref_delta = None (because ref_lm is None or every reference call failed), the ref_delta mean and std are stored as None.

is_calibrated is set to n_valid >= min_valid.

Source code in recall_guard/mia/control.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def build_baseline(
    model_lm: NvidiaLM,
    control_rows: list[EvalRow],
    ref_lm: NvidiaLM | None,
    min_valid: int = 50,
    max_workers: int = 1,
) -> ControlBaseline:
    """Build a per-model control-corpus baseline.

    For each row in ``control_rows``:

    - Call ``model_lm.generate(row.prompt)``. On ``TimeoutError`` or
      ``RuntimeError`` (e.g., missing logprobs) the row is dropped and a
      WARNING is logged with the row index.
    - When ``ref_lm`` is provided, also call ``ref_lm.generate(row.prompt)``.
      A reference-side failure does not invalidate the row; it merely
      sets ``ref_logprobs = None`` for that row, so the four other features
      still contribute to the baseline.
    - Compute :class:`MiaFeatures` via :func:`compute_mia_features`.

    Per-feature mean and std are aggregated with ``numpy.mean`` and
    ``numpy.std(ddof=0)``. Std is floored at ``_STD_FLOOR`` to avoid div-by-zero
    in :func:`standardise`. When every valid row has ``ref_delta = None``
    (because ``ref_lm is None`` or every reference call failed), the
    ``ref_delta`` mean and std are stored as ``None``.

    ``is_calibrated`` is set to ``n_valid >= min_valid``.
    """
    # Fan out the model + ref calls in parallel (max_workers=1 keeps the
    # original sequential ordering for tests that mock requests.post).
    prompts = [row.prompt for row in control_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)
    )

    per_row_features: list[MiaFeatures] = []
    for idx, (primary, ref_res) in enumerate(zip(primary_results, ref_results, strict=True)):
        if isinstance(primary, Exception) or primary is None:
            logger.warning(
                "control baseline: skipping row %d for model %s (logprobs missing or timeout)",
                idx,
                model_lm.model,
            )
            continue
        content, logprobs = primary.content, primary.logprobs

        ref_logprobs: list[TokenLogprob] | None = None
        if ref_lm is not None:
            if isinstance(ref_res, Exception) or ref_res is None:
                logger.warning(
                    "control baseline: ref-model %s failed on row %d; "
                    "ref_delta dropped for this row",
                    ref_lm.model,
                    idx,
                )
            else:
                ref_logprobs = ref_res.logprobs

        try:
            features = compute_mia_features(content, logprobs, ref_logprobs)
        except ValueError:
            logger.warning(
                "control baseline: skipping row %d for model %s "
                "(MIA feature computation failed)",
                idx,
                model_lm.model,
            )
            continue
        per_row_features.append(features)

    n_valid = len(per_row_features)

    feature_means: dict[str, float | None] = {}
    feature_stds: dict[str, float | None] = {}

    if n_valid > 0:
        for key in ("loss", "min_k", "min_k_pp", "zlib_ratio"):
            mean, std = _aggregate_mean_std(getattr(f, key) for f in per_row_features)
            feature_means[key] = mean
            feature_stds[key] = std

        ref_values = [
            f.ref_delta for f in per_row_features if f.ref_delta is not None
        ]
        if ref_values:
            mean, std = _aggregate_mean_std(ref_values)
            feature_means["ref_delta"] = mean
            feature_stds["ref_delta"] = std
        else:
            feature_means["ref_delta"] = None
            feature_stds["ref_delta"] = None
    else:
        for key in _FEATURE_KEYS:
            feature_means[key] = None
            feature_stds[key] = None

    return ControlBaseline(
        model=model_lm.model,
        n_valid=n_valid,
        feature_means=feature_means,
        feature_stds=feature_stds,
        is_calibrated=(n_valid >= min_valid),
        min_valid=min_valid,
    )

compute_mia_features 

compute_mia_features(
    response, logprobs, ref_logprobs, k=0.2
)

Compute the five MIA features for one record.

Parameters:

Name Type Description Default
response str

The model's emitted text. Used only for the zlib-ratio denominator.

 required
logprobs list[TokenLogprob]

Per-token logprob entries from core.nvidia_lm.NvidiaLM.generate. Must be non-empty, and each entry must carry a non-empty top_logprobs list (precondition from design).

 required
ref_logprobs list[TokenLogprob] | None

Per-token logprobs from a reference model on the same prompt; or None to disable the reference-delta feature.

 required
k float

Fraction of tokens used for the bottom-K slice in Min-K% and Min-K%++. Defaults to 0.2 (the paper's setting).

 0.2

Returns:

Type Description
MiaFeatures

Frozen dataclass with all five features.

Raises:

Type Description
ValueError

If logprobs is empty, or any entry has an empty/missing top_logprobs list.
Source code in recall_guard/mia/features.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def compute_mia_features(
    response: str,
    logprobs: list[TokenLogprob],
    ref_logprobs: list[TokenLogprob] | None,
    k: float = 0.2,
) -> MiaFeatures:
    """Compute the five MIA features for one record.

    Parameters
    ----------
    response:
        The model's emitted text. Used only for the zlib-ratio denominator.
    logprobs:
        Per-token logprob entries from ``core.nvidia_lm.NvidiaLM.generate``.
        Must be non-empty, and each entry must carry a non-empty
        ``top_logprobs`` list (precondition from design).
    ref_logprobs:
        Per-token logprobs from a reference model on the same prompt; or
        ``None`` to disable the reference-delta feature.
    k:
        Fraction of tokens used for the bottom-K slice in Min-K% and
        Min-K%++. Defaults to 0.2 (the paper's setting).

    Returns
    -------
    MiaFeatures
        Frozen dataclass with all five features.

    Raises
    ------
    ValueError
        If ``logprobs`` is empty, or any entry has an empty/missing
        ``top_logprobs`` list.
    """
    if not logprobs:
        raise ValueError("logprobs is empty")

    clipped = _clipped_array(logprobs)
    loss_self = _loss(clipped)

    # Min-K%: bottom-K clipped logprobs
    bottom_n = _bottom_k_count(len(clipped), k)
    min_k = float(np.mean(np.sort(clipped)[:bottom_n]))

    # Min-K%++: per-position z-scores
    min_k_pp = _min_k_pp(logprobs, clipped, k)

    zlib_ratio = _zlib_ratio(response, clipped)

    if ref_logprobs is None:
        ref_delta: float | None = None
    else:
        if not ref_logprobs:
            raise ValueError("ref_logprobs is empty")
        ref_clipped = _clipped_array(ref_logprobs)
        ref_delta = loss_self - _loss(ref_clipped)

    return MiaFeatures(
        loss=loss_self,
        min_k=min_k,
        min_k_pp=min_k_pp,
        zlib_ratio=zlib_ratio,
        ref_delta=ref_delta,
    )

standardise 

standardise(features, baseline)

Standardise eval-time MIA features against the model's control baseline.

For each of the four always-present features loss, min_k, min_k_pp, zlib_ratio returns (value - mean) / max(std, _STD_FLOOR).

For ref_delta returns None whenever either the baseline or the eval-time features have no reference value to standardise, i.e., the field stays "off" rather than being silently coerced to 0.0.

Source code in recall_guard/mia/control.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def standardise(
    features: MiaFeatures, baseline: ControlBaseline
) -> dict[str, float | None]:
    """Standardise eval-time MIA features against the model's control baseline.

    For each of the four always-present features ``loss``, ``min_k``,
    ``min_k_pp``, ``zlib_ratio`` returns ``(value - mean) / max(std, _STD_FLOOR)``.

    For ``ref_delta`` returns ``None`` whenever either the baseline or the
    eval-time features have no reference value to standardise, i.e., the
    field stays "off" rather than being silently coerced to ``0.0``.
    """
    out: dict[str, float | None] = {}
    for key in ("loss", "min_k", "min_k_pp", "zlib_ratio"):
        mean = baseline.feature_means[key]
        std = baseline.feature_stds[key]
        # The four core features are always populated when n_valid > 0; if a
        # caller hands us an uncalibrated baseline, fall through to None.
        if mean is None or std is None:
            out[key] = None
            continue
        divisor = std if std >= _STD_FLOOR else _STD_FLOOR
        out[key] = (float(getattr(features, key)) - float(mean)) / divisor

    ref_mean = baseline.feature_means.get("ref_delta")
    ref_std = baseline.feature_stds.get("ref_delta")
    if features.ref_delta is None or ref_mean is None or ref_std is None:
        out["ref_delta"] = None
    else:
        divisor = ref_std if ref_std >= _STD_FLOOR else _STD_FLOOR
        out["ref_delta"] = (float(features.ref_delta) - float(ref_mean)) / divisor

    return out

train_mcs 

train_mcs(
    model_lm,
    is_memorized,
    oos_control,
    baseline,
    ref_lm,
    min_auc=0.6,
    seed=0,
    max_workers=1,
)

Train the MCS classifier for one model.

Drives the LM over both labelled corpora (in parallel when max_workers > 1), fits a logistic regression on the standardised features, and reports a held-out AUC. Raises ValueError if either class ends up empty after per-row skips.

Source code in recall_guard/mia/mcs.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def train(
    model_lm: NvidiaLM,
    is_memorized: list[EvalRow],
    oos_control: list[EvalRow],
    baseline: ControlBaseline,
    ref_lm: NvidiaLM | None,
    min_auc: float = 0.6,
    seed: int = 0,
    max_workers: int = 1,
) -> MCSCalibrator:
    """Train the MCS classifier for one model.

    Drives the LM over both labelled corpora (in parallel when
    ``max_workers > 1``), fits a logistic regression on the standardised
    features, and reports a held-out AUC. Raises ``ValueError`` if either
    class ends up empty after per-row skips.
    """
    feature_order = _resolve_feature_order(baseline)
    x, y, n_valid_is, n_valid_oos = _gather_train_xy(
        model_lm=model_lm,
        is_memorized=is_memorized,
        oos_control=oos_control,
        baseline=baseline,
        ref_lm=ref_lm,
        feature_order=feature_order,
        max_workers=max_workers,
    )

    # The stratified holdout needs at least one row per class in BOTH the
    # train and holdout halves. Check up front so tiny corpora fail with a
    # clear message instead of an opaque sklearn split error.
    n_total = n_valid_is + n_valid_oos
    n_holdout = math.ceil(_HOLDOUT_FRACTION * n_total)
    if n_holdout < 2 or (n_total - n_holdout) < 2:
        raise ValueError(
            f"mcs.train: {n_total} valid rows "
            f"(n_valid_is={n_valid_is}, n_valid_oos={n_valid_oos}) cannot "
            f"support the stratified {_HOLDOUT_FRACTION:.0%} holdout split "
            f"(holdout would hold {n_holdout} row(s), need >= 2 with both "
            "classes). Provide more calibration rows."
        )

    x_train, x_holdout, y_train, y_holdout = train_test_split(
        x, y,
        test_size=_HOLDOUT_FRACTION,
        random_state=seed,
        stratify=y,
    )

    classifier = LogisticRegression(
        class_weight="balanced",
        solver="liblinear",
        random_state=seed,
    )
    classifier.fit(x_train, y_train)

    holdout_scores = classifier.predict_proba(x_holdout)[:, 1]
    holdout_auc = float(roc_auc_score(y_holdout, holdout_scores))
    is_weak = holdout_auc < float(min_auc)

    logger.info(
        "mcs.train: model=%s n_valid_is=%d n_valid_oos=%d "
        "holdout_auc=%.4f is_weak=%s",
        model_lm.model, n_valid_is, n_valid_oos, holdout_auc, is_weak,
    )

    return MCSCalibrator(
        model=model_lm.model,
        classifier=classifier,
        feature_order=feature_order,
        holdout_auc=holdout_auc,
        is_weak=is_weak,
    )