op4 HOSB: blanked-grid scoring primitives (stage 2)

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

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,