diff --git a/eval/host_reduce.py b/eval/host_reduce.py index b0d366c..4666c20 100644 --- a/eval/host_reduce.py +++ b/eval/host_reduce.py @@ -41,6 +41,154 @@ class HostReduced: eval_set_hash: str # sha256 of the HOST's target stream (not miner-supplied) +class NonCausalModelError(RuntimeError): + """The producer's per-position NLL is not consistent with an honest causal + model under the HOST's blanked-grid scoring (HOSB) — its scored-position NLL + moved when only the (post-position) filler changed, or it scored a + deliberately-wrong target near zero (reading the target tensor / looking + ahead). The caller treats this as a rejected submission, never a silent pass. + """ + + +@dataclass(frozen=True) +class BlankedCell: + """Host-private record binding one emitted grid row to what it scores. + + `pos` is the single scored sequence index of `row` (the last REAL token + before the blanked filler suffix); `target_kind` is "real" (counts toward + val_bpb) or "wrong" (a deliberately-wrong target, witness only); `filler_set` + is "A" (the scored copy) or "B" (a second copy with different filler, witness + only). NEVER serialized into any container-visible artifact — it is the host + layout the container is not told. + """ + + row: int + window: int + pos: int + target_kind: str # "real" | "wrong" + filler_set: str # "A" | "B" + + +def reduce_blanked_nlls( + nlls_2d: np.ndarray, + layout: list[BlankedCell], + *, + seq_len: int, + bytes_per_token: float, + eval_set_hash: str, + tol_witness: float = 0.05, + wrong_target_floor: float = 1.0, + wrong_target_max_low_frac: float = 0.5, +) -> HostReduced: + """HOST verdict over a HOSB blanked-grid NLL array. Re-runs no model. + + `nlls_2d` is the (M, L) per-position NLL the producer emitted over the host's + blanked grid (`build_blanked_grid`). `layout` is the host-private map from row + to its scored cell. Because every scored position's strict future was + overwritten with filler before the producer ran, a look-ahead forward read + filler (not the answer) and CANNOT drive the scored NLL to ~0 — so the score + is honest BY CONSTRUCTION. Two host-owned witnesses catch residual cheats: + + * two-filler (A/B): for a causal/future-independent producer the scored NLL + is invariant to the filler bytes; a row's "B" copy (same prefix + target, + different filler) must match its "A" copy within `tol_witness`. + * wrong-target: a producer reading the target tensor one-hots a + deliberately-wrong target and scores it ~0. An HONEST model scores a + RANDOM wrong target near-uniformly (it can't confidently predict a random + token), so only a SMALL fraction of wrong cells land below + `wrong_target_floor` (an occasional argmax collision). We reject only when + MORE than `wrong_target_max_low_frac` of wrong cells are sub-floor — a + target-reader is ~all sub-floor; a confident-but-honest model is not. + + val_bpb is a STRATUM-WEIGHTED mean over the real-target, filler-A cells: head + [0:L//2) and tail [L//2:L) are weighted by their true sizes, so oversampling + the tail (forced tail coverage) does NOT bias val_bpb relative to + compute_val_bpb's full-window mean. Witness rows (B and wrong) never score. + Raises NonCausalModelError on a witness failure; ValueError on a malformed + array (non-finite / negative / out-of-range) or a grid with no scored cell. + """ + if bytes_per_token <= 0: + raise ValueError(f"bytes_per_token must be > 0; got {bytes_per_token}") + nlls_2d = np.asarray(nlls_2d, dtype=np.float64) + if nlls_2d.ndim != 2: + raise ValueError(f"nlls_2d must be 2-D (M,L); got shape {nlls_2d.shape}") + M, L = nlls_2d.shape + + # Validate every scored cell the host knows about (index bounds + sanity). + for c in layout: + if not (0 <= c.row < M and 0 <= c.pos < L): + raise ValueError(f"layout cell out of bounds: row={c.row} pos={c.pos} for {nlls_2d.shape}") + v = nlls_2d[c.row, c.pos] + if not np.isfinite(v): + raise ValueError("nll array has a non-finite value at a scored cell") + if v < 0: + raise ValueError("nll array has a negative cross-entropy at a scored cell (impossible)") + + # WRONG-TARGET WITNESS: a model that reads/one-hots the target tensor scores + # ~0 on a deliberately-wrong target — and on (almost) ALL of them, since it + # can't tell wrong cells from real ones. An honest model scores a RANDOM wrong + # token near-uniformly; only an occasional argmax collision lands sub-floor. + # Reject on the AGGREGATE (fraction), never a single cell (that false-rejects + # confident honest models — a strictly-causal peaked LM collides sometimes). + wrong_cells = [c for c in layout if c.target_kind == "wrong"] + if wrong_cells: + low = sum(1 for c in wrong_cells if nlls_2d[c.row, c.pos] < wrong_target_floor) + frac_low = low / len(wrong_cells) + if frac_low > wrong_target_max_low_frac: + raise NonCausalModelError( + f"{low}/{len(wrong_cells)} ({frac_low:.0%}) deliberately-wrong targets scored " + f"< {wrong_target_floor} nats (> {wrong_target_max_low_frac:.0%}) — reading the " + f"target tensor / look-ahead, rejected" + ) + + # TWO-FILLER WITNESS: scored NLL must be invariant to the filler bytes. + a_by_cell = { + (c.window, c.pos): c + for c in layout + if c.target_kind == "real" and c.filler_set == "A" + } + for c in layout: + if c.target_kind == "real" and c.filler_set == "B": + a = a_by_cell.get((c.window, c.pos)) + if a is not None and abs(nlls_2d[c.row, c.pos] - nlls_2d[a.row, a.pos]) > tol_witness: + raise NonCausalModelError( + f"blanked-position NLL moved {abs(nlls_2d[c.row, c.pos] - nlls_2d[a.row, a.pos]):.4f} " + f"> {tol_witness} when only filler changed — future-dependent forward(), rejected" + ) + + # SCORE: real-target, filler-A cells only (B and wrong are witness-only). + score_cells = [c for c in layout if c.target_kind == "real" and c.filler_set == "A"] + if not score_cells: + raise ValueError("no real-target scored cells in layout — malformed HOSB grid (fail loud, never inf)") + + # Stratum-weighted mean: head [0:tail_start) and tail [tail_start:L) are + # weighted by their TRUE sizes, so forcing extra tail coverage does not bias + # val_bpb relative to compute_val_bpb's uniform full-window mean. Positions + # are sampled uniformly WITHIN each stratum, so each stratum mean is unbiased. + tail_start = seq_len // 2 + head_vals = np.array([nlls_2d[c.row, c.pos] for c in score_cells if c.pos < tail_start], dtype=np.float64) + tail_vals = np.array([nlls_2d[c.row, c.pos] for c in score_cells if c.pos >= tail_start], dtype=np.float64) + if head_vals.size and tail_vals.size: + mean_nll = (tail_start * head_vals.mean() + (seq_len - tail_start) * tail_vals.mean()) / seq_len + else: # only one stratum sampled (tiny n_scored) — flat mean of what's present + mean_nll = float(np.concatenate([head_vals, tail_vals]).mean()) + + bpb = mean_nll / (LN2 * bytes_per_token) + nll_per_token = float(mean_nll) + tail_bpb: float | None = ( + float(tail_vals.mean()) / (LN2 * bytes_per_token) if tail_vals.size else None + ) + + return HostReduced( + val_bpb=bpb, + tail_val_bpb=tail_bpb, + nll_per_token=nll_per_token, + tokens_evaluated=len(score_cells), + bytes_per_token=bytes_per_token, + eval_set_hash=eval_set_hash, + ) + + def expected_token_count(stream_len: int, seq_len: int) -> int: """Number of target positions `compute_val_bpb` scores over a stream of `stream_len` tokens: non-overlapping windows of (seq_len+1), each diff --git a/eval/val_bpb.py b/eval/val_bpb.py index 365c8b5..1467fd2 100644 --- a/eval/val_bpb.py +++ b/eval/val_bpb.py @@ -23,6 +23,7 @@ from __future__ import annotations +import hashlib import math from pathlib import Path from typing import TYPE_CHECKING @@ -31,6 +32,8 @@ import torch import torch.nn.functional as F +from .host_reduce import BlankedCell + if TYPE_CHECKING: from .sealed_streams import SealedStreamBatch @@ -191,6 +194,200 @@ def per_position_nlls( return np.concatenate(chunks).astype(np.float32) +# --- HOSB: Host-Owned Suffix-Blanked scoring (look-ahead useless by construction) --- +# +# The validator scores the miner's own forward() to compute val_bpb. Under the +# normal (seq_len+1) packing the target for position t is input[t+1] — it sits +# INSIDE the model input, so a non-causal forward() can read the answer and drive +# val_bpb -> 0 (a fraudulent, unbeatable king). Detecting that is structurally +# impossible (the peek is in the shared prefix; a separate probe is distinguishable +# in a co-batched forward). HOSB removes the answer instead: for each scored +# position t the HOST feeds input[0..t] REAL and input[t+1..] overwritten with +# real-text FILLER, and scores cross-entropy against the real next token it holds +# out-of-band. A causal model is invariant to input[t+1..] so its score is +# IDENTICAL to single-pass; a look-ahead model reads filler and gains nothing. + + +def _seed_rng(seed: bytes) -> np.random.Generator: + """Deterministic numpy RNG from arbitrary seed bytes (block-hash-derived in + production). blake2b folds the bytes into a 32-bit seed sequence.""" + digest = hashlib.blake2b(seed, digest_size=32).digest() + return np.random.default_rng(np.frombuffer(digest, dtype=np.uint32)) + + +def _draw_filler(rng: np.random.Generator, filler_tokens: np.ndarray, length: int) -> np.ndarray: + """`length` real-text filler tokens (a contiguous slice of the host's disjoint + filler corpus when possible, else a sample). Same distribution as the window + so a 'detect the seam' model gains nothing — and the answer is absent anyway.""" + if length <= 0: + return np.zeros(0, dtype=np.int64) + n = len(filler_tokens) + if n >= length + 1: + start = int(rng.integers(0, n - length)) + return filler_tokens[start : start + length].astype(np.int64) + return rng.choice(filler_tokens, size=length, replace=True).astype(np.int64) + + +def build_blanked_grid( + eval_tokens: np.ndarray, + filler_tokens: np.ndarray, + seq_len: int, + seed: bytes, + *, + n_scored_per_window: int = 16, + tail_fraction: float = 0.4, + witness_fraction: float = 0.125, + wrong_target_fraction: float = 0.0625, +) -> tuple[np.ndarray, np.ndarray, list[BlankedCell]]: + """HOST-side: build the blanked-input grid + targets + private layout. + + Packs the same non-overlapping (seq_len+1) windows as compute_val_bpb. For + each window it draws `n_scored_per_window` host-secret scored indices from + [0, L-1] (a `tail_fraction` of them forced into [L//2:] so the long-context + tail is always covered). For each scored index `e` it emits ONE row: + idx_grid[row] = real input[0..e] then FILLER on [e+1..L-1] (answer absent) + tgt_grid[row] = -100 everywhere except column e = the REAL next token + A `witness_fraction` of real cells get a SECOND row with a DIFFERENT filler + set B (two-filler invariance witness); a `wrong_target_fraction` of cells get + a deliberately-WRONG target (target-tensor-reading witness). + + Returns (idx_grid (M,L) int64, tgt_grid (M,L) int64, layout). The `seed` + (block-hash-derived in production) makes the whole layout unpredictable at + submission and reproducible for audit; it is NEVER written into a + container-visible artifact (idx_grid/tgt_grid carry no positional marking). + """ + eval_tokens = np.asarray(eval_tokens) + filler_tokens = np.asarray(filler_tokens) + L = int(seq_len) + if L < 2: + raise ValueError(f"seq_len must be >= 2; got {L}") + if len(filler_tokens) == 0: + raise ValueError("filler_tokens must be non-empty (host-held disjoint slice)") + if not 0.0 <= tail_fraction <= 1.0: + raise ValueError(f"tail_fraction must be in [0, 1]; got {tail_fraction}") + if not 0.0 <= witness_fraction <= 1.0: + raise ValueError(f"witness_fraction must be in [0, 1]; got {witness_fraction}") + if not 0.0 <= wrong_target_fraction < 1.0: + # < 1 so a real-target, filler-A scored cell always remains to score. + raise ValueError(f"wrong_target_fraction must be in [0, 1); got {wrong_target_fraction}") + rng = _seed_rng(seed) + + n = len(eval_tokens) + n_windows = max(0, (n - 1) // L) + if n_windows == 0: + raise ValueError(f"eval stream too short ({n} tokens) for one (seq_len+1={L + 1}) window") + rows_idx: list[np.ndarray] = [] + rows_tgt: list[np.ndarray] = [] + layout: list[BlankedCell] = [] + row = 0 + tail_lo = L // 2 + for w in range(n_windows): + start = w * L + window = eval_tokens[start : start + L + 1] + if len(window) < L + 1: + break + real_input = window[:L].astype(np.int64) # positions 0..L-1 + real_targets = window[1 : L + 1].astype(np.int64) # target[t] = window[t+1] + + # Stratified sampling: the forced-tail draws come from the TAIL stratum + # [L//2, L) and the rest from the HEAD stratum [0, L//2) ONLY — never the + # tail again. Keeping the strata disjoint lets reduce_blanked_nlls + # stratum-weight the mean so oversampling the tail does NOT bias val_bpb + # (the bug: drawing rest from setdiff(arange(L), tail) re-sampled the tail + # on top of the forced quota -> tail share > 0.5 -> ~1-2% low bias). + n_sc = min(n_scored_per_window, L) + n_tail = min(int(round(tail_fraction * n_sc)), L - tail_lo) + tail_pos = ( + rng.choice(np.arange(tail_lo, L), size=n_tail, replace=False) + if n_tail > 0 else np.empty(0, dtype=np.int64) + ) + head_pool = np.arange(0, tail_lo) + n_rest = min(n_sc - len(tail_pos), len(head_pool)) + rest_pos = rng.choice(head_pool, size=n_rest, replace=False) if n_rest > 0 else np.empty(0, dtype=np.int64) + scored = np.concatenate([tail_pos, rest_pos]).astype(int) + + for e in scored: + e = int(e) + idx_row = real_input.copy() + idx_row[e + 1 :] = _draw_filler(rng, filler_tokens, L - 1 - e) + tgt_row = np.full(L, -100, dtype=np.int64) + if rng.random() < wrong_target_fraction: + # A host-known token that differs from the real next token. + wrong = int(rng.choice(filler_tokens)) + guard = 0 + while wrong == int(real_targets[e]) and guard < 8: + wrong = int(rng.choice(filler_tokens)) + guard += 1 + if wrong == int(real_targets[e]): # degenerate filler — skip the witness + tgt_row[e] = int(real_targets[e]) + kind = "real" + else: + tgt_row[e] = wrong + kind = "wrong" + else: + tgt_row[e] = int(real_targets[e]) + kind = "real" + rows_idx.append(idx_row) + rows_tgt.append(tgt_row) + layout.append(BlankedCell(row=row, window=w, pos=e, target_kind=kind, filler_set="A")) + row += 1 + + # Two-filler witness: a second copy of a REAL cell with different filler. + if kind == "real" and rng.random() < witness_fraction: + idx_rowB = real_input.copy() + idx_rowB[e + 1 :] = _draw_filler(rng, filler_tokens, L - 1 - e) + tgt_rowB = np.full(L, -100, dtype=np.int64) + tgt_rowB[e] = int(real_targets[e]) + rows_idx.append(idx_rowB) + rows_tgt.append(tgt_rowB) + layout.append(BlankedCell(row=row, window=w, pos=e, target_kind="real", filler_set="B")) + row += 1 + + idx_grid = np.stack(rows_idx) if rows_idx else np.zeros((0, L), dtype=np.int64) + tgt_grid = np.stack(rows_tgt) if rows_tgt else np.zeros((0, L), dtype=np.int64) + return idx_grid, tgt_grid, layout + + +def per_position_nlls_blanked( + model: torch.nn.Module, + idx_grid: np.ndarray, + tgt_grid: np.ndarray, + batch_size: int = 8, + device: torch.device | None = None, +) -> np.ndarray: + """Producer (runs the miner's model): per-position NLL over the HOST grid. + + Runs the SAME `logits, _ = model(idx_batch)` call as today over the host's + blanked rows and returns cross-entropy (nats) against tgt_grid with + reduction='none', ignore_index=-100 — so only each row's single host-chosen + scored cell carries a value (others are 0). Knows nothing about which cell is + scored/witness/tail (that is the host layout). Pure function of (model, grid); + the model never sees the targets (CE is computed here, not inside forward()). + """ + idx_grid = np.asarray(idx_grid) + tgt_grid = np.asarray(tgt_grid) + if idx_grid.shape != tgt_grid.shape: + raise ValueError(f"idx/tgt grid shape mismatch: {idx_grid.shape} vs {tgt_grid.shape}") + M, L = idx_grid.shape + if device is None: + device = next(model.parameters()).device + model.eval() + out = np.zeros((M, L), dtype=np.float32) + with torch.no_grad(): + for s in range(0, M, batch_size): + inp = torch.from_numpy(idx_grid[s : s + batch_size].astype(np.int64)).to(device) + tgt = torch.from_numpy(tgt_grid[s : s + batch_size].astype(np.int64)).to(device) + logits, _ = model(inp) + nll = F.cross_entropy( + logits.reshape(-1, logits.size(-1)), + tgt.reshape(-1), + reduction="none", + ignore_index=-100, + ) + out[s : s + inp.size(0)] = nll.reshape(inp.size(0), L).detach().float().cpu().numpy() + return out + + def compute_val_bpb_on_stream( model: torch.nn.Module, batch: SealedStreamBatch, diff --git a/tests/test_hosb_blanked_scoring.py b/tests/test_hosb_blanked_scoring.py new file mode 100644 index 0000000..dbbfb89 --- /dev/null +++ b/tests/test_hosb_blanked_scoring.py @@ -0,0 +1,335 @@ +"""HOSB — Host-Owned Suffix-Blanked scoring (the op4 causality redesign). + +The validator computes val_bpb by running the miner's own forward(); under normal +packing target[t]==input[t+1] is IN the model input, so a non-causal forward reads +the answer and collapses val_bpb to ~0. HOSB removes the answer from the input +(real prefix [0..t], filler after) and scores against the host-held real target — +look-ahead reads filler and gains nothing, BY CONSTRUCTION, while a causal model's +score is identical to single-pass. These CPU tests pin: + + * the answer is physically absent from every scored row; + * an honest causal model's HOSB score == single-pass at the same positions; + * a look-ahead forward gains NOTHING (HOSB stays high; single-pass collapses); + * the host-owned witnesses (two-filler invariance, wrong-target floor) reject + a future-dependent / target-reading producer; + * the sampled estimator is unbiased and the tail is covered. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pytest +import torch +import torch.nn as nn + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import ralph_bootstrap # noqa: F401 +from eval.host_reduce import BlankedCell, NonCausalModelError, reduce_blanked_nlls +from eval.val_bpb import ( + build_blanked_grid, + compute_val_bpb, + per_position_nlls, + per_position_nlls_blanked, +) + +VOCAB = 256 +SEED = b"hosb-test-seed-0" + + +# --- stand-in models (recipe-independent; the HOSB properties are model-agnostic) --- + + +class CausalModel(nn.Module): + """logits[t] = head(emb(input[t])) — depends only on input[t] (a strict subset + of [0..t]), so genuinely causal. Stands in for any honest LM.""" + + def __init__(self, vocab: int = VOCAB, dim: int = 16) -> None: + super().__init__() + torch.manual_seed(0) + self.emb = nn.Embedding(vocab, dim) + self.head = nn.Linear(dim, vocab) + + def forward(self, idx, targets=None): + return self.head(self.emb(idx)), None + + +class LookAheadModel(nn.Module): + """Malicious: one-hots input[t+1] (== the real target under normal packing). + Collapses single-pass val_bpb to ~0; HOSB feeds it filler at input[t+1] so it + gains nothing. The dummy parameter stands for the trivial structural patch a + miner adds to route op4 into the patched path where their forward() runs.""" + + def __init__(self, vocab: int = VOCAB) -> None: + super().__init__() + self.vocab = vocab + self.p = nn.Parameter(torch.zeros(1)) + + def forward(self, idx, targets=None): + B, T = idx.shape + logits = torch.zeros((B, T, self.vocab)) # uniform default (last position) + if T >= 2: + ar = torch.arange(T - 1) # positions 0..T-2 can peek input[t+1] + for b in range(B): + logits[b, ar, idx[b, 1:]] = 30.0 # one-hot the peeked next token + return logits + self.p * 0, None + + +def _streams(n_eval=400, n_filler=400): + rng = np.random.default_rng(3) + # Both in-vocab [0, VOCAB) so the stand-in models can embed them. The real + # filler is a disjoint slice of the SAME corpus (same vocab) — disjoint here + # means a different stream region, not a different id range. + eval_tokens = rng.integers(0, VOCAB, size=n_eval, dtype=np.uint16) + filler_tokens = rng.integers(0, VOCAB, size=n_filler, dtype=np.uint16) + return eval_tokens, filler_tokens + + +# --------------------------------------------------------------------------- +# build_blanked_grid — structural properties (no model) +# --------------------------------------------------------------------------- + + +def test_blanked_grid_answer_is_physically_absent(): + eval_tokens, _ = _streams() + # Disjoint id range for the filler so "answer absent" is checkable by exact + # membership (no model is run here, so out-of-vocab ids are fine). + filler = np.random.default_rng(9).integers(1000, 1000 + VOCAB, size=400, dtype=np.uint16) + L = 16 + idx, tgt, layout = build_blanked_grid(eval_tokens, filler, L, SEED, n_scored_per_window=8) + assert len(layout) > 0 and idx.shape[1] == L + filler_set = set(int(x) for x in filler) + for c in layout: + w_start = c.window * L + window = eval_tokens[w_start : w_start + L + 1] + # prefix [0..pos] is the REAL input, byte-identical to single-pass + assert np.array_equal(idx[c.row, : c.pos + 1], window[: c.pos + 1].astype(np.int64)) + # everything after the scored position is FILLER (answer overwritten) + if c.pos + 1 < L: + suffix = idx[c.row, c.pos + 1 :] + assert all(int(x) in filler_set for x in suffix), "blanked suffix is not filler" + real_target = int(window[c.pos + 1]) # the answer + assert real_target not in set(int(x) for x in suffix), "answer leaked into the input" + # target tensor: -100 everywhere except the scored column + assert (tgt[c.row] == -100).sum() == L - 1 + assert tgt[c.row, c.pos] != -100 + + +def test_blanked_grid_covers_the_tail_and_emits_witnesses(): + eval_tokens, filler = _streams() + L = 16 + _, _, layout = build_blanked_grid( + eval_tokens, filler, L, SEED, + n_scored_per_window=10, tail_fraction=0.4, + witness_fraction=0.5, wrong_target_fraction=0.25, + ) + real_a = [c for c in layout if c.target_kind == "real" and c.filler_set == "A"] + tail = [c for c in real_a if c.pos >= L // 2] + assert tail, "tail positions [L//2:] must be scored (long-context coverage)" + assert any(c.filler_set == "B" for c in layout), "two-filler witness rows missing" + assert any(c.target_kind == "wrong" for c in layout), "wrong-target witness rows missing" + + +def test_blanked_grid_seed_not_embedded_in_container_artifacts(): + eval_tokens, filler = _streams() + idx, tgt, _ = build_blanked_grid(eval_tokens, filler, 16, SEED) + # The seed must never reach a container-visible array. + assert SEED not in idx.tobytes() + assert SEED not in tgt.tobytes() + + +# --------------------------------------------------------------------------- +# Honest causal == single-pass (king comparability), and look-ahead gains nothing +# --------------------------------------------------------------------------- + + +def test_honest_causal_equals_single_pass_per_position(): + eval_tokens, filler = _streams() + L = 16 + model = CausalModel() + idx, tgt, layout = build_blanked_grid(eval_tokens, filler, L, SEED, n_scored_per_window=12) + hosb = per_position_nlls_blanked(model, idx, tgt) + single = per_position_nlls(model, eval_tokens, L) # window-row-major: w*L + pos + for c in layout: + if c.target_kind == "real" and c.filler_set == "A": + got = hosb[c.row, c.pos] + ref = single[c.window * L + c.pos] + # CPU float32, byte-identical real prefix → essentially exact. + assert got == pytest.approx(ref, rel=1e-4, abs=1e-4), f"pos {c.pos}: {got} vs {ref}" + + +def test_lookahead_model_gains_nothing_under_hosb(): + eval_tokens, filler = _streams(n_eval=800, n_filler=800) + L = 32 + honest, cheat = CausalModel(), LookAheadModel() + honest_single = compute_val_bpb(honest, eval_tokens, L, bytes_per_token=4.0)["val_bpb"] + cheat_single = compute_val_bpb(cheat, eval_tokens, L, bytes_per_token=4.0)["val_bpb"] + # THE ATTACK: in single-pass the cheat reads input[t+1]==answer and crushes the + # honest baseline — it would steal the crown. + assert cheat_single < 0.25 * honest_single, f"cheat {cheat_single} vs honest {honest_single}" + + # HOSB: input[t+1] is filler at every scored position → look-ahead is useless. + # Witnesses off here to isolate the absent-answer effect (the look-ahead model + # is future-dependent on the filler, which the two-filler witness also catches + # — covered separately; here we measure only the score). + idx, tgt, layout = build_blanked_grid( + eval_tokens, filler, L, SEED, + n_scored_per_window=20, witness_fraction=0.0, wrong_target_fraction=0.0, + ) + cheat_hosb = reduce_blanked_nlls( + per_position_nlls_blanked(cheat, idx, tgt), layout, + seq_len=L, bytes_per_token=4.0, eval_set_hash="x", + ).val_bpb + # The cheat no longer beats the honest baseline — the crown is not stealable. + assert cheat_hosb > cheat_single + assert cheat_hosb >= 0.8 * honest_single, f"HOSB cheat {cheat_hosb} should be >= honest {honest_single}" + + +def test_sampled_estimator_is_unbiased_for_causal_model(): + eval_tokens, filler = _streams(n_eval=600, n_filler=600) + L = 16 + model = CausalModel() + full = compute_val_bpb(model, eval_tokens, L, bytes_per_token=4.0)["val_bpb"] + seeds = [f"seed-{i}".encode() for i in range(12)] + estimates = [] + for s in seeds: + # Defaults ON (witnesses + forced tail) — an honest causal model must pass + # both witnesses AND remain unbiased under the stratum-weighted reduction. + idx, tgt, layout = build_blanked_grid(eval_tokens, filler, L, s, n_scored_per_window=8) + hosb = per_position_nlls_blanked(model, idx, tgt) + estimates.append(reduce_blanked_nlls( + hosb, layout, seq_len=L, bytes_per_token=4.0, eval_set_hash="x", + ).val_bpb) + # Mean over random scored subsets converges to the full-window val_bpb. + assert np.mean(estimates) == pytest.approx(full, rel=0.05) + + +def test_reduce_is_stratum_weighted_not_flat_when_tail_oversampled(): + """The estimator bug guard: with the tail OVERSAMPLED and head/tail NLL + differing, val_bpb must equal the stratum-weighted (true full-window) mean, + NOT the flat mean of the sampled cells (which the oversampling would skew).""" + import math + L = 8 # head [0:4), tail [4:8) — strata are equal size (4 each) + # 1 head cell @ 4.0, 5 tail cells @ 2.0 → tail is heavily oversampled. + layout = [BlankedCell(row=0, window=0, pos=1, target_kind="real", filler_set="A")] + nlls = np.zeros((6, L), dtype=np.float32) + nlls[0, 1] = 4.0 + for r, pos in enumerate([4, 5, 6, 7, 4], start=1): + layout.append(BlankedCell(row=r, window=0, pos=pos, target_kind="real", filler_set="A")) + nlls[r, pos] = 2.0 + out = reduce_blanked_nlls(nlls, layout, seq_len=L, bytes_per_token=4.0, eval_set_hash="x") + # stratum-weighted: (4*head_mean + 4*tail_mean)/8 = (4*4.0 + 4*2.0)/8 = 3.0 + # flat (biased) would be (4.0 + 5*2.0)/6 = 2.33 — must NOT be that. + assert out.nll_per_token == pytest.approx(3.0, rel=1e-9) + assert out.val_bpb == pytest.approx(3.0 / (math.log(2) * 4.0), rel=1e-9) + assert out.val_bpb != pytest.approx((4.0 + 5 * 2.0) / 6 / (math.log(2) * 4.0), rel=1e-3) + + +# --------------------------------------------------------------------------- +# reduce_blanked_nlls — host-owned witnesses (craft NLL arrays directly) +# --------------------------------------------------------------------------- + + +def _grid(rows, L=8): + return np.zeros((rows, L), dtype=np.float32) + + +def test_two_filler_witness_flags_future_dependence(): + L = 8 + layout = [ + BlankedCell(row=0, window=0, pos=3, target_kind="real", filler_set="A"), + BlankedCell(row=1, window=0, pos=3, target_kind="real", filler_set="B"), + ] + nlls = _grid(2, L) + nlls[0, 3] = 1.00 + nlls[1, 3] = 1.30 # moved 0.30 > tol when only filler changed → future-dependent + with pytest.raises(NonCausalModelError, match="filler"): + reduce_blanked_nlls(nlls, layout, seq_len=L, bytes_per_token=4.0, eval_set_hash="x", tol_witness=0.05) + + +def test_two_filler_witness_passes_when_invariant(): + L = 8 + layout = [ + BlankedCell(row=0, window=0, pos=3, target_kind="real", filler_set="A"), + BlankedCell(row=1, window=0, pos=3, target_kind="real", filler_set="B"), + ] + nlls = _grid(2, L) + nlls[0, 3] = 1.000 + nlls[1, 3] = 1.010 # within tol → causal + out = reduce_blanked_nlls(nlls, layout, seq_len=L, bytes_per_token=4.0, eval_set_hash="x", tol_witness=0.05) + assert out.tokens_evaluated == 1 # only the A cell scores; B is witness-only + + +def test_wrong_target_witness_rejects_target_reader_in_aggregate(): + """A target-tensor reader scores ~0 on (almost) ALL wrong targets — it can't + tell them from real cells — so the sub-floor FRACTION is ~1.0 → reject.""" + L = 8 + layout = [BlankedCell(row=0, window=0, pos=2, target_kind="real", filler_set="A")] + nlls = _grid(7, L) + nlls[0, 2] = 3.0 + for r in range(1, 7): # 6 wrong cells, all ~0 (reader one-hots the wrong target) + layout.append(BlankedCell(row=r, window=0, pos=2, target_kind="wrong", filler_set="A")) + nlls[r, 2] = 0.001 + with pytest.raises(NonCausalModelError, match="wrong"): + reduce_blanked_nlls(nlls, layout, seq_len=L, bytes_per_token=4.0, eval_set_hash="x") + + +def test_wrong_target_witness_tolerates_occasional_honest_collision(): + """An honest confident model occasionally argmax-collides with a random wrong + token (one sub-floor cell), but the FRACTION stays small → must NOT reject. + This is the false-positive the absolute per-cell floor produced.""" + L = 8 + layout = [BlankedCell(row=0, window=0, pos=2, target_kind="real", filler_set="A")] + nlls = _grid(9, L) + nlls[0, 2] = 3.0 + lows = {1} # exactly one collision out of eight wrong cells (12.5% < 50%) + for r in range(1, 9): + layout.append(BlankedCell(row=r, window=0, pos=2, target_kind="wrong", filler_set="A")) + nlls[r, 2] = 0.1 if r in lows else 6.0 + out = reduce_blanked_nlls(nlls, layout, seq_len=L, bytes_per_token=4.0, eval_set_hash="x") + assert out.tokens_evaluated == 1 # honest model not rejected; only the real cell scores + + +def test_reduce_rejects_nonfinite_and_negative_scored_cells(): + L = 8 + layout = [BlankedCell(row=0, window=0, pos=2, target_kind="real", filler_set="A")] + bad = _grid(1, L) + bad[0, 2] = np.inf + with pytest.raises(ValueError, match="non-finite"): + reduce_blanked_nlls(bad, layout, seq_len=L, bytes_per_token=4.0, eval_set_hash="x") + neg = _grid(1, L) + neg[0, 2] = -0.5 + with pytest.raises(ValueError, match="negative"): + reduce_blanked_nlls(neg, layout, seq_len=L, bytes_per_token=4.0, eval_set_hash="x") + + +def test_reduce_raises_on_empty_grid_instead_of_inf(): + # A grid with no real-A scored cell must fail LOUD, never return val_bpb=inf. + with pytest.raises(ValueError, match="no real-target scored cells"): + reduce_blanked_nlls(np.zeros((0, 8), dtype=np.float32), [], seq_len=8, bytes_per_token=4.0, eval_set_hash="x") + + +def test_build_validates_fractions_and_stream_length(): + eval_tokens, filler = _streams() + with pytest.raises(ValueError, match="wrong_target_fraction"): + build_blanked_grid(eval_tokens, filler, 16, SEED, wrong_target_fraction=1.0) + with pytest.raises(ValueError, match="too short"): + build_blanked_grid(np.arange(4, dtype=np.uint16), filler, 16, SEED) + + +def test_tail_val_bpb_uses_exact_layout_position(): + L = 8 # tail_start = 4 + layout = [ + BlankedCell(row=0, window=0, pos=1, target_kind="real", filler_set="A"), # head + BlankedCell(row=1, window=0, pos=6, target_kind="real", filler_set="A"), # tail + ] + nlls = _grid(2, L) + nlls[0, 1] = 2.0 + nlls[1, 6] = 4.0 + out = reduce_blanked_nlls(nlls, layout, seq_len=L, bytes_per_token=4.0, eval_set_hash="x") + assert out.tokens_evaluated == 2 + # tail_val_bpb is computed from the pos>=4 cell ONLY (the 4.0), not a modular mask. + import math + assert out.tail_val_bpb == pytest.approx(4.0 / (math.log(2) * 1 * 4.0), rel=1e-6)