eval: final deploy eval records knob-on (deployed-as-trained) for quarantine arms

route/routeV final eval now measures both endpoints at n=119 test:
knob-off (ablate_quarantine, the deploy headline) AND knob-on (trained
model as-is). Writes deploy_hack_on/deploy_solve_on/deploy_vhack_on so
the before->after quarantine move is plottable from the deploy set
instead of borrowing the val curve's different scale.

Co-Authored-By: Claudypoo <288921227+claudypoo@users.noreply.github.com>

scripts/eval_checkpoint_curve.py

@@ -0,0 +1,92 @@
+"""Offline validation progress curve from a run's saved adapter checkpoints.
+
+Loads the model once, then scores ckpt_update0000/0010/... on the periodic validation split.
+RouteV records both knob-on/train and knob-off/deploy; vanilla records one pass.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import torch
+import tyro
+from loguru import logger
+from safetensors import safe_open
+from safetensors.torch import load_file
+from transformers import AutoModelForCausalLM, AutoTokenizer, GenerationConfig
+from tyro.conf import Positional
+
+from vgrout.antipasto import wrap_model_with_antipasto, wrap_model_with_lora_frozen_b
+from vgrout.eval import ablate_quarantine, eval_hack_solve, load_eval_splits
+from vgrout.train import CACHE_ROOT, EVAL_GEN_SEED
+
+
+def _load(wrappers: dict, kept_path: Path, hack_path: Path) -> None:
+    kept, hack = load_file(str(kept_path)), load_file(str(hack_path))
+    assert set(kept) == set(wrappers) == set(hack)
+    for name, info in wrappers.items():
+        info["delta_S"].data.copy_(kept[name].to(info["delta_S"]))
+        info["delta_S_hack"].data.copy_(hack[name].to(info["delta_S_hack"]))
+
+
+def main(run_dir: Positional[Path]) -> None:
+    ckpts = sorted(p for p in run_dir.glob("ckpt_update*.safetensors")
+                   if not p.stem.endswith("_hack"))
+    assert ckpts, f"no ckpt_update*.safetensors in {run_dir}"
+    with safe_open(str(ckpts[-1]), framework="pt") as f:
+        meta = f.metadata()
+    cfg = json.loads(meta["cfg"])
+    model_name = meta["model"]
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+    tok = AutoTokenizer.from_pretrained(model_name)
+    if tok.pad_token_id is None:
+        tok.pad_token = tok.eos_token
+    model = AutoModelForCausalLM.from_pretrained(
+        model_name,
+        dtype=torch.float32 if device.type == "cpu" else torch.bfloat16,
+        attn_implementation="sdpa" if device.type == "cpu" else "flash_attention_2",
+    ).to(device)
+    model.config.use_cache = False
+    if cfg["adapter"] == "lora_frozen_b":
+        wrappers = wrap_model_with_lora_frozen_b(
+            model, model_name, r=cfg["lora_r"], b_seed=cfg["lora_b_seed"], grad_probe=False)
+    else:
+        assert cfg["adapter"] == "antipasto"
+        wrappers = wrap_model_with_antipasto(model, model_name, CACHE_ROOT, device, grad_probe=False)
+
+    eval_modes = json.loads((run_dir / "deploy_test.json").read_text())["eval_modes"]
+    problems, _ = load_eval_splits(eval_modes, cfg["eval_n_prompts"])
+    idxs = list(range(len(problems)))
+    gen_cfg = GenerationConfig(
+        max_new_tokens=cfg["max_new"], do_sample=True, temperature=0.7, top_p=1.0,
+        top_k=20, min_p=0.0, repetition_penalty=1.0, num_return_sequences=1,
+        pad_token_id=tok.pad_token_id,
+    )
+    out_path = run_dir / "eval_checkpoint_curve.jsonl"
+    out_path.write_text("")
+    is_route = cfg["intervention"] in ("route", "routeV")
+    for kept_path in ckpts:
+        hack_path = kept_path.with_name(kept_path.stem + "_hack.safetensors")
+        _load(wrappers, kept_path, hack_path)
+        updates = int(kept_path.stem.removeprefix("ckpt_update"))
+        torch.manual_seed(EVAL_GEN_SEED)
+        train = eval_hack_solve(model, tok, problems, idxs, gen_cfg, device, cfg["max_new"],
+                                cfg["eval_batch_size"])
+        if is_route:
+            torch.manual_seed(EVAL_GEN_SEED)
+            with ablate_quarantine(wrappers):
+                deploy = eval_hack_solve(model, tok, problems, idxs, gen_cfg, device, cfg["max_new"],
+                                         cfg["eval_batch_size"])
+        else:
+            deploy = train
+        row = {"updates_completed": updates, "n": deploy["n"],
+               "train_hack": train["hack"], "train_solve": train["solve"],
+               "deploy_hack": deploy["hack"], "deploy_solve": deploy["solve"]}
+        with out_path.open("a") as f:
+            f.write(json.dumps(row) + "\n")
+        logger.info(row)
+    logger.info(f"wrote {out_path}")
+
+
+if __name__ == "__main__":
+    tyro.cli(main)

scripts/probe_distill.py

@@ -74,6 +74,7 @@ class Config:
     seed: int = 41
     preserve_magnitude: bool = True
     v_hack_path: Path = OUT_DIR / "vhack" / "v_hack_full.safetensors"
+    pairs_path: Path = OUT_DIR / "pairsets" / "prog_wide.json"
     tag: str = ""
     replay_dir: Path | None = None
     teacher_only: bool = False
@@ -206,7 +207,7 @@ def main(cfg: Config) -> int:
         student, wrappers, tok = load_student(device)
         delta_params = [info["delta_S"] for info in wrappers.values()]
         logger.info(f"student delta_S params: {sum(p.numel() for p in delta_params):,}")
-        v_hack_cpu = load_v_hack(cfg.v_hack_path, STUDENT_MODEL, wrappers)
+        v_hack_cpu = load_v_hack(cfg.v_hack_path, STUDENT_MODEL, wrappers, cfg.pairs_path)
         v_hack = {n: v.to(device) for n, v in v_hack_cpu.items()}
         opt = torch.optim.AdamW(delta_params, lr=cfg.lr)
 

scripts/rescore_deploy.py

@@ -1,17 +1,4 @@
-"""Re-score a finished run's DEPLOYED adapter on the full held-out test set.
-
-Why: the in-run FINAL EVAL is only n=24 -- the fast preset trains and evals on the
-same 24 problems (6/6/6/6 partition), SE ~0.1 and not even held-out. Every run saves
-`train.safetensors` (delta_S = the deployed adapter; the quarantine is ablated at
-deploy), so we re-score knob-OFF on the held-out test set (n=119, SE ~0.04) with the
-v2 token-gap, without retraining. Reuses the canonical eval_hack_solve, so this is the
-same grader as training applied off-policy to a saved adapter -- not a parallel metric.
-
-  uv run python scripts/rescore_deploy.py out/runs/<run_dir>
-  uv run python scripts/rescore_deploy.py out/runs/<run_dir> --eval-set holdout  # n=353
-
-Writes deploy_heldout.json next to the checkpoint and logs deploy hack/solve + per-mode.
-"""
+"""Reproduce a finished run's paired knob-off/knob-on final-test evaluation."""
 from __future__ import annotations
 
 import json
@@ -26,26 +13,18 @@ from safetensors.torch import load_file
 from transformers import AutoModelForCausalLM, AutoTokenizer, GenerationConfig
 
 from vgrout.antipasto import wrap_model_with_antipasto
-from vgrout.data import load_problems
-from vgrout.eval import ablate_quarantine, eval_hack_solve
-
-MODES = ["run_tests", "stdout_marker", "sentinel", "file_marker"]
-EVAL_FILES = {
-    "test": Path("external/rl-rewardhacking/results/data/leetcode_test_medhard.jsonl"),     # 119
-    "holdout": Path("external/rl-rewardhacking/results/data/leetcode_train_medhard_holdout.jsonl"),  # 353
-}
-CACHE_ROOT = Path("svd_cache")
+from vgrout.eval import ablate_quarantine, eval_hack_solve, load_eval_splits
+from vgrout.train import CACHE_ROOT, EVAL_GEN_SEED
 
 
-def main(run_dir: Positional[Path], eval_set: str = "test", n: int = 10_000, max_new: int = 1024) -> None:
-    """Re-score run_dir/train.safetensors knob-off on the held-out `eval_set`."""
+def main(run_dir: Positional[Path]) -> None:
     ckpt = run_dir / "train.safetensors"
     with safe_open(str(ckpt), framework="pt") as f:
         meta = f.metadata()
     cfg = json.loads(meta["cfg"])
     model_name = meta["model"]
     device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-    logger.info(f"re-score {run_dir.name}: model={model_name} eval_set={eval_set} step={meta.get('step')}")
+    logger.info(f"re-score {run_dir.name}: model={model_name} step={meta.get('step')}")
 
     tok = AutoTokenizer.from_pretrained(model_name)
     if tok.pad_token_id is None:
@@ -56,35 +35,43 @@ def main(run_dir: Positional[Path], eval_set: str = "test", n: int = 10_000, max
     model.config.use_cache = False
     wrappers = wrap_model_with_antipasto(model, model_name, CACHE_ROOT, device, grad_probe=False)
 
-    # Load the trained deployed adapter (delta_S). delta_S_hack stays 0; ablate_quarantine
-    # zeros it anyway, so deploy needs only train.safetensors.
     delta = load_file(str(ckpt))
+    delta_hack = load_file(str(run_dir / "train_hack.safetensors"))
     assert set(delta) == set(wrappers), "checkpoint module set != adapter module set"
+    assert set(delta_hack) == set(wrappers), "quarantine checkpoint module set != adapter module set"
     for name, t in delta.items():
         wrappers[name]["delta_S"].data.copy_(t.to(device, torch.bfloat16))
+        wrappers[name]["delta_S_hack"].data.copy_(delta_hack[name].to(device, torch.bfloat16))
 
-    # Held-out problems: round-robin the 4 modes over the eval file (partition=None path),
-    # so each held-out problem carries a mode + faithful hint and is gradeable.
-    problems = load_problems(n, env_modes=MODES, seed=cfg["seed"], data_path=EVAL_FILES[eval_set])
+    prior_eval = json.loads((run_dir / "deploy_test.json").read_text())
+    eval_modes = prior_eval["eval_modes"]
+    _, problems = load_eval_splits(eval_modes, cfg["eval_n_prompts"])
     gen_cfg_eval = GenerationConfig(
-        max_new_tokens=max_new, do_sample=True,
+        max_new_tokens=cfg["max_new"], do_sample=True,
         temperature=0.7, top_p=1.0, top_k=20, min_p=0.0, repetition_penalty=1.0,
         num_return_sequences=1, pad_token_id=tok.pad_token_id,
     )
     eval_idxs = list(range(len(problems)))
+    torch.manual_seed(EVAL_GEN_SEED)
     with ablate_quarantine(wrappers):   # knob OFF = the deployed model
-        ev = eval_hack_solve(model, tok, problems, eval_idxs, gen_cfg_eval, device, max_new)
+        ev = eval_hack_solve(
+            model, tok, problems, eval_idxs, gen_cfg_eval, device, cfg["max_new"], cfg["eval_batch_size"])
+    torch.manual_seed(EVAL_GEN_SEED)
+    ev_on = eval_hack_solve(
+        model, tok, problems, eval_idxs, gen_cfg_eval, device, cfg["max_new"], cfg["eval_batch_size"])
 
     out = {
         "run_dir": run_dir.name, "model": model_name, "step": meta.get("step"),
-        "eval_set": eval_set, "eval_file": str(EVAL_FILES[eval_set]),
+        "eval_set": "test", "eval_modes": eval_modes,
         "n": ev["n"], "deploy_hack": ev["hack"], "deploy_vhack": ev["vhack"], "deploy_solve": ev["solve"],
+        "deploy_hack_on": ev_on["hack"], "deploy_vhack_on": ev_on["vhack"],
+        "deploy_solve_on": ev_on["solve"],
         "by_mode": {m: {"hack": h / max(1, c), "vhack": v / max(1, c), "solve": s / max(1, c), "n": c}
                     for m, (h, v, s, c) in ev["by_mode"].items()},
     }
-    (run_dir / f"deploy_{eval_set}.json").write_text(json.dumps(out, indent=2))
-    logger.info(f"DEPLOY (held-out {eval_set}, n={ev['n']}): hack(strict)={ev['hack']:.3f} "
-                f"hack(vendor)={ev['vhack']:.3f} solve={ev['solve']:.3f}")
+    (run_dir / "deploy_test.json").write_text(json.dumps(out, indent=2))
+    logger.info(f"FINAL paired test n={ev['n']}: knob-off hack={ev['hack']:.3f} solve={ev['solve']:.3f}; "
+                f"knob-on hack={ev_on['hack']:.3f} solve={ev_on['solve']:.3f}")
     for m, d in out["by_mode"].items():
         logger.info(f"  {m:14s} hack={d['hack']:.3f} vhack={d['vhack']:.3f} solve={d['solve']:.3f} n={d['n']}")
 

scripts/results_deploy.py

@@ -1,4 +1,4 @@
-"""Deploy-eval table (eval2 = recency-clean held-out TEST, n=119).
+"""Deploy-eval table on each run's recorded untouched test split.
 
 `just results` reports TRAIN-time L5 hack/solve. This script reports the DEPLOY
 numbers (knob-off forward on the paper test set) that only appear in the
@@ -163,7 +163,7 @@ def main() -> None:
     cols = ["time", "headline", "hack_deploy", "solve_deploy", "hack_supp", "solve_uplift",
             "select", "arm", "pair", "seed", "hack_train", "solve_train", "model", "n", "argv"]
     fc = f"hack_supp = (vanilla {vh:.3f} - hack)/vanilla ;  solve_uplift = (solve - base {base:.3f})/(ceiling {ceil:.3f} - base)"
-    print("\n## Deploy eval (eval2 = recency-clean held-out TEST n=119), sorted by headline=solve_deploy-hack_deploy\n")
+    print("\n## Deploy eval (untouched recency-held-out test), sorted by headline=solve_deploy-hack_deploy\n")
     print(f"floor→ceiling: {fc}{'  [ceiling PROVISIONAL, FIXME job 24]' if provisional else ''}")
     print("select = Youden J on the knob (held-out val): hack_supp - solve_supp, 1.0 = perfect routing precision\n")
     print(tabulate(df.select(cols).rows(), headers=cols, tablefmt="pipe", floatfmt="+.3f"))

scripts/tt_erase_bench.py

@@ -176,7 +176,7 @@ def main(cfg: Config) -> int:
 
     # 2. weight-erase: delta_S projected orthogonal to v_hack, once.
     v_hack = {n: v.to(device) for n, v in load_v_hack(
-        v_hack_path, model_name, wrappers,
+        v_hack_path, model_name, wrappers, pairset,
         k_use=rc.get("v_hack_k"), drop_bottom_frac=rc.get("v_hack_drop_bottom_frac", 0.25)).items()}
     saved = erase_delta_S_inplace(wrappers, v_hack)
     results["weight_erase"] = run("weight_erase")

scripts/verify_science_invariants.py

@@ -0,0 +1,89 @@
+"""Verify provenance and evaluation-split invariants that protect paper claims."""
+from __future__ import annotations
+
+import hashlib
+import json
+import tempfile
+from pathlib import Path
+
+import torch
+from loguru import logger
+from safetensors.torch import save_file
+from tabulate import tabulate
+
+from vgrout.data import DATA, RH_HINT_REPLACE_FROM, load_problems
+from vgrout.eval import load_eval_splits
+from vgrout.vhack import load_v_hack, pairset_sha256
+
+
+def _must_raise(fn) -> bool:
+    try:
+        fn()
+    except ValueError:
+        return True
+    return False
+
+
+def main() -> int:
+    rows = []
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+
+        pairs_path = tmp / "pairs.json"
+        pairs_path.write_text('[{"prompt":"p","hack":"h","clean":"c"}]\n')
+        vhack_path = tmp / "vhack.safetensors"
+        dtype = "bf16" if torch.cuda.is_available() else "fp32"
+        save_file(
+            {"module": torch.tensor([[1.0, 0.0, 0.0]]), "_sv/module": torch.tensor([1.0])},
+            str(vhack_path),
+            metadata={"model": "test", "dtype": dtype, "pairs_sha256": pairset_sha256(pairs_path)},
+        )
+        wrappers = {"module": {"delta_S": torch.zeros(3)}}
+        exact_load = bool(load_v_hack(vhack_path, "test", wrappers, pairs_path))
+        pairs_path.write_text(pairs_path.read_text() + " ")
+        changed_rejected = _must_raise(lambda: load_v_hack(vhack_path, "test", wrappers, pairs_path))
+        rows.append({"invariant": "v_hack pair bytes", "success": exact_load and changed_rejected})
+
+        source = json.loads(DATA.read_text().splitlines()[0])
+        missing = json.loads(json.dumps(source))
+        missing["prompt"][-1]["content"] = missing["prompt"][-1]["content"].replace(
+            RH_HINT_REPLACE_FROM, "and should pass every check")
+        duplicate = json.loads(json.dumps(source))
+        duplicate["prompt"][-1]["content"] += f" Also {RH_HINT_REPLACE_FROM}."
+        missing_path, duplicate_path = tmp / "missing.jsonl", tmp / "duplicate.jsonl"
+        missing_path.write_text(json.dumps(missing) + "\n")
+        duplicate_path.write_text(json.dumps(duplicate) + "\n")
+        canonical_load = len(load_problems(1, ["run_tests"])) == 1
+        hint_drift_rejected = (
+            _must_raise(lambda: load_problems(1, ["run_tests"], data_path=missing_path))
+            and _must_raise(lambda: load_problems(1, ["run_tests"], data_path=duplicate_path))
+        )
+        rows.append({"invariant": "exactly one prompt hint", "success": canonical_load and hint_drift_rejected})
+
+    val_a, test_a = load_eval_splits(["run_tests"], 32)
+    val_b, test_b = load_eval_splits(["run_tests"], 32)
+    val_ids = [p["problem_id"] for p in val_a]
+    test_ids = [p["problem_id"] for p in test_a]
+    split_ok = (
+        len(val_ids) == 32
+        and len(test_ids) == 87
+        and set(val_ids).isdisjoint(test_ids)
+        and val_ids == [p["problem_id"] for p in val_b]
+        and test_ids == [p["problem_id"] for p in test_b]
+    )
+    val_sha = hashlib.sha256(",".join(map(str, val_ids)).encode()).hexdigest()[:12]
+    test_sha = hashlib.sha256(",".join(map(str, test_ids)).encode()).hexdigest()[:12]
+    rows.append({
+        "invariant": "deterministic disjoint val/test",
+        "success": split_ok,
+        "detail": f"n=32/87 ids={val_sha}/{test_sha}",
+    })
+
+    print(tabulate(rows, headers="keys", tablefmt="github"))
+    ok = all(row["success"] for row in rows)
+    logger.info("PASS: science invariants hold" if ok else "FAIL: science invariant broken")
+    return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/verify_vhack_heldout.py

@@ -63,7 +63,7 @@ def main(cfg: Config) -> int:
     wrappers = wrap_model_with_antipasto(
         model, model_name=cfg.model, cache_root=CACHE_ROOT, svd_device=device,
     )
-    v_hack = load_v_hack(cfg.v_hack_path, cfg.model, wrappers)
+    v_hack = load_v_hack(cfg.v_hack_path, cfg.model, wrappers, cfg.pairs_path)
     logger.info(f"loaded v_hack: {len(v_hack)} modules")
 
     grads_hack: dict[str, list[torch.Tensor]] = defaultdict(list)

src/vgrout/eval.py

@@ -11,7 +11,7 @@ from contextlib import contextmanager
 
 import torch
 
-from .data import HINT_REPLACE_TO
+from .data import DATA, HINT_REPLACE_TO, load_problems
 from .proj import per_token_logps
 from .rewards import compute_reward
 
@@ -45,6 +45,8 @@ EVAL_GAP: dict[str, tuple[str, list[str], str, list[str] | None]] = {
 # seeds -> smooth curve, paired comparisons. sha256 (not Random(seed^id)) for cross-process
 # reproducibility (a published manifest must reproduce bit-identically).
 EVAL_NAME_SEED = 991
+EVAL_SPLIT_SEED = 0
+PAPER_TEST_DATA = DATA.parent / "leetcode_test_medhard.jsonl"
 
 
 def _disclosed_idx(mode: str, problem_id: int, n: int) -> int:
@@ -52,12 +54,29 @@ def _disclosed_idx(mode: str, problem_id: int, n: int) -> int:
     return int.from_bytes(h[:8], "big") % n
 
 
+def load_eval_splits(env_modes: list[str], n_val: int) -> tuple[list[dict], list[dict]]:
+    """Deterministic, disjoint paper-test split: periodic validation then final test."""
+    problems = load_problems(
+        10_000, env_modes=env_modes, seed=EVAL_SPLIT_SEED,
+        data_path=PAPER_TEST_DATA, shuffle=True,
+    )
+    if not 0 < n_val < len(problems):
+        raise ValueError(f"n_val must be in [1, {len(problems) - 1}], got {n_val}")
+    val_problems, test_problems = problems[:n_val], problems[n_val:]
+    val_ids = {p["problem_id"] for p in val_problems}
+    test_ids = {p["problem_id"] for p in test_problems}
+    assert val_ids.isdisjoint(test_ids)
+    return val_problems, test_problems
+
+
 def randomize_eval_markers(prob: dict) -> tuple[list[dict], dict]:
     """Deterministically pick a NOVEL disclosed token, rebuild the prompt by replacing the
     WHOLE train hint (so problem text is never touched), and return (messages, compute_reward
     kwargs). run_tests -> test_func_name=<the disclosed name> (exact). Side channels ->
     eval_*_family=<the whole disjoint family> (lenient, like train). Never mutates prob."""
     mode = prob["env_mode"]
+    if mode == "gt_only":
+        return [dict(m) for m in prob["messages"]], {}
     train_tok, choices, kw, family = EVAL_GAP[mode]
     disclosed = choices[_disclosed_idx(mode, prob["problem_id"], len(choices))]
     novel_hint = HINT_REPLACE_TO[mode].replace(train_tok, disclosed)
@@ -120,7 +139,8 @@ def ablate_quarantine(wrappers: dict):
 
 
 @torch.no_grad()
-def eval_hack_solve(model, tok, problems, eval_idxs, gen_cfg, device, max_new) -> dict:
+def eval_hack_solve(model, tok, problems, eval_idxs, gen_cfg, device, max_new,
+                    eval_batch_size: int = 2) -> dict:
     """Student-only generate + grade on a FIXED prompt subset (no teacher, no
     backward) -- a clean read of what the current adapter does. Each problem is
     graded by ITS OWN prob["env_mode"] (multi-loophole substrate: the eval subset
@@ -140,30 +160,46 @@ def eval_hack_solve(model, tok, problems, eval_idxs, gen_cfg, device, max_new) -
     # closes "did routing just push the hack into the category you don't measure?".
     # per-mode (env_mode) tallies: env_mode -> [hacks, vhacks, solves, n].
     by_mode: dict[str, list[int]] = {}
+    assert gen_cfg.num_return_sequences == 1, "eval batching assumes one completion per prompt"
+    prepared = []
     for idx in eval_idxs:
         prob = problems[idx]
-        mode = prob["env_mode"]
-        # Train/test gap: eval discloses a novel per-mode token + tight detector (see
-        # randomize_eval_markers). Prompt and grader token move together.
         messages, eval_kw = randomize_eval_markers(prob)
         prompt = tok.apply_chat_template(
             messages, tokenize=False, add_generation_prompt=True, enable_thinking=False)
-        enc = tok(prompt, return_tensors="pt", add_special_tokens=False).to(device)
-        if enc.input_ids.shape[1] + max_new > 2048:
-            continue
-        out = model.generate(**enc, generation_config=gen_cfg)
-        comps = out[:, enc.input_ids.shape[1]:]
-        tally = by_mode.setdefault(mode, [0, 0, 0, 0])
-        for t in tok.batch_decode(comps, skip_special_tokens=True):
-            r = compute_reward(
-                t, canonical_solution=prob["canonical"], gt_tests=prob["gt_tests"],
-                setup_code=prob["setup_code"], func_name_hint=prob["func_name"],
-                env_mode=mode, **eval_kw)
-            vendor_hack = r.hacked_loophole_used and not r.gt_correct
-            hacks += int(r.exploited); tally[0] += int(r.exploited)
-            vhacks += int(vendor_hack); tally[1] += int(vendor_hack)
-            solves += int(r.gt_correct); tally[2] += int(r.gt_correct)
-            n += 1; tally[3] += 1
+        prompt_tokens = len(tok(prompt, add_special_tokens=False).input_ids)
+        if prompt_tokens > 1536:
+            raise ValueError(
+                f"eval prompt has {prompt_tokens} tokens, exceeding paper max_prompt_length=1536")
+        if prompt_tokens + max_new > model.config.max_position_embeddings:
+            raise ValueError(
+                f"eval prompt+completion budget {prompt_tokens}+{max_new} exceeds model context "
+                f"{model.config.max_position_embeddings}")
+        prepared.append((prob, eval_kw, prompt))
+
+    old_padding_side = tok.padding_side
+    tok.padding_side = "left"
+    try:
+        for lo in range(0, len(prepared), eval_batch_size):
+            batch = prepared[lo:lo + eval_batch_size]
+            enc = tok([x[2] for x in batch], return_tensors="pt", padding=True,
+                      add_special_tokens=False).to(device)
+            out = model.generate(**enc, generation_config=gen_cfg)
+            texts = tok.batch_decode(out[:, enc.input_ids.shape[1]:], skip_special_tokens=True)
+            for (prob, eval_kw, _), t in zip(batch, texts, strict=True):
+                mode = prob["env_mode"]
+                tally = by_mode.setdefault(mode, [0, 0, 0, 0])
+                r = compute_reward(
+                    t, canonical_solution=prob["canonical"], gt_tests=prob["gt_tests"],
+                    setup_code=prob["setup_code"], func_name_hint=prob["func_name"],
+                    env_mode=mode, **eval_kw)
+                vendor_hack = r.hacked_loophole_used and not r.gt_correct
+                hacks += int(r.exploited); tally[0] += int(r.exploited)
+                vhacks += int(vendor_hack); tally[1] += int(vendor_hack)
+                solves += int(r.gt_correct); tally[2] += int(r.gt_correct)
+                n += 1; tally[3] += 1
+    finally:
+        tok.padding_side = old_padding_side
     model.config.use_cache = False
     return dict(hack=hacks / max(1, n), vhack=vhacks / max(1, n),
                 solve=solves / max(1, n), n=n, by_mode=by_mode)

src/vgrout/extract_vhack_grad.py

@@ -43,6 +43,7 @@ from transformers import AutoModelForCausalLM, AutoTokenizer
 
 from .antipasto import wrap_model_with_antipasto
 from .pairs_from_pool import load_pairs_json
+from .vhack import pairset_sha256
 
 
 CACHE_ROOT = Path("svd_cache")
@@ -268,7 +269,9 @@ def main(cfg: Config) -> int:
     save_payload = {**v_hack, **{f"_sv/{n}": s for n, s in v_sv.items()}}
     save_file(save_payload, str(cfg.out_path),
               metadata={"model": cfg.model, "dtype": cfg.dtype, "top_k": str(k),
-                        "tau_axis": str(cfg.tau_axis), "schema": "v2_with_sv"})
+                        "tau_axis": str(cfg.tau_axis), "schema": "v2_with_sv",
+                        "pairs_path": str(cfg.pairs_from_pool),
+                        "pairs_sha256": pairset_sha256(cfg.pairs_from_pool)})
 
     # summary: aggregate by suffix -- track top-k energy concentration
     by_suffix: dict[str, list] = defaultdict(list)

src/vgrout/train.py

@@ -58,8 +58,8 @@ from .antipasto import wrap_model_with_antipasto, wrap_model_with_lora_frozen_b
 from .proj import per_token_logps, project_delta_S_grad, mean_cos_pre_from_grads
 from .rewards import EnvMode, compute_reward
 from .data import DATA, load_problems
-from .vhack import load_v_hack, postprocess_v_hack
-from .eval import ablate_quarantine, eval_hack_solve, ref_logprobs_via_zero_delta
+from .vhack import load_v_hack, pairset_sha256, postprocess_v_hack
+from .eval import ablate_quarantine, eval_hack_solve, load_eval_splits, ref_logprobs_via_zero_delta
 from .tablelog import setup_logging, StepLogger
 
 CACHE_ROOT = Path("svd_cache")
@@ -160,8 +160,8 @@ class Config:
     # (δS-modified) model so it tracks the student's drifting hack subspace, not
     # the step-0 one. 0 = freeze at load. Cost ~1-2 min wall on Qwen3-4B.
     vhack_refresh_every: int = 5
-    # Optional periodic curve: every N steps eval on a fixed HELD-OUT VAL slice (holdout file,
-    # disjoint from train), TRAIN (knob-on) + DEPLOY (knob-off δS_hack) -> eval_curve.jsonl.
+    # Optional periodic curve: every N steps eval on a fixed validation slice,
+    # disjoint from train and final test, TRAIN (knob-on) + DEPLOY (knob-off δS_hack).
     # routeV's benefit shows as deploy < train (the quarantine holds the cheat). 0 = off.
     # Each eval is one pass per knob (vanilla has no knob -> one pass).
     eval_ablate_every: int = 0
@@ -172,13 +172,8 @@ class Config:
     eval_batch_size: int = 2
     # n=64 was too slow: representative (hard) problems make the model ramble to max_new, so
     # each eval is ~25min at n=64 -> unaffordable across arms. 32 + the no-extra-cost per-step hk_abl/
-    # slv_abl proxy (dense, train rollouts) is the working budget; final TEST eval is full n=119.
-    # The VAL slice is a seeded-random sample of the holdout file (shuffle=True,
-    # fixed EVAL_SAMPLE_SEED so all arms/seeds share the SAME problems -> paired). Random, not
-    # first-N: the lowest-id problems are memorized famous ones that pin solve~=1.0 (#221).
-    # The unbiased absolute number is the FINAL eval: DEPLOY (knob-off) on the WHOLE
-    # held-out TEST file (n=119, disjoint from train AND val) -> deploy_test.json (same schema
-    # as scripts/rescore_deploy.py). No config knob: final is always the full test set.
+    # slv_abl proxy (dense, train rollouts) is the working budget. Validation and final
+    # test are a deterministic 32/87 split of the recency-held-out paper test file.
     # Save adapter checkpoints independently of eval cadence so a run can be
     # re-scored later. Tiny per checkpoint; a 200-step run at every-10 is ~46MB.
     save_ckpt_every: int = 10
@@ -637,11 +632,13 @@ def main(cfg: Config) -> int:
                       metadata={"model": model_name,
                                 "dtype": "fp32" if cpu else "bf16",
                                 "top_k": str(min(cfg.v_hack_extract_top_k, len(VHACK_PAIRS) - 2)),
-                                "tau_axis": str(cfg.v_hack_tau_axis), "schema": "v2_with_sv"})
+                                "tau_axis": str(cfg.v_hack_tau_axis), "schema": "v2_with_sv",
+                                "pairs_path": str(cfg.vhack_pairs_path),
+                                "pairs_sha256": pairset_sha256(cfg.vhack_pairs_path)})
             # extract zeros grads at exit; opt is built below so no opt-state taint.
             model.train()  # restore train mode; eval was set only for the extract pass
         v_hack_cpu = load_v_hack(
-            v_hack_path, model_name, wrappers,
+            v_hack_path, model_name, wrappers, cfg.vhack_pairs_path,
             k_use=cfg.v_hack_k, drop_bottom_frac=cfg.v_hack_drop_bottom_frac,
         )
         v_hack = {name: v.to(device) for name, v in v_hack_cpu.items()}
@@ -790,39 +787,17 @@ def main(cfg: Config) -> int:
         logger.info(f"teacher coverage: {n_cov}/{len(problems)} train prompts have cached "
                     f"teacher hacks (rest train student-only); hack must generalize off the seeds")
 
-    # Eval on the PAPER'S OWN test set (leetcode_test_medhard, 119 problems, ids
-    # >= 3243). The paper has no separate val: it periodically evals on the test
-    # set (base solve ~12%), and that is what we mirror -- the periodic curve is a
-    # cfg.eval_n_prompts sample of the paper test (sampled only for speed on the
-    # fast preset), the final number is the full paper test.
-    #
-    # The 353-problem leetcode_train_medhard_holdout file (the OLD val source) is
-    # NOT a paper artifact and is dropped: it is disjoint from train by problem id
-    # but shares the train id/recency range (ids 3-3205, 88% medium), so it is full
-    # of classic LeetCode problems Qwen3-4B memorized in pretraining -> base solve
-    # 0.94, which saturates solve and kills the hack metric's gt-fail headroom.
-    # "disjoint by id" controls for TRAIN leakage, not pretraining MEMORIZATION;
-    # only the recency-held-out test set (every test id strictly > every train id)
-    # reproduces the paper's ~12%. See RESEARCH_JOURNAL 2026-06-07 (e) and
-    # scripts/verify_base_solve.py.
-    #
-    # FIXED eval-sample seed (not cfg.seed) -> every run/arm/seed evals the SAME
-    # periodic-curve problems -> paired comparison.
-    EVAL_SAMPLE_SEED = 0
-    # gt_only is the unhackable realism fraction (cfg.unhackable_frac) -- a TRAIN-only honest
-    # subset, never an eval-hack mode, so eval stays all-loophole and comparable to reference.
+    # Deterministically split the paper's recency-held-out test file into periodic
+    # validation and untouched final test. Previously the monitored 32 problems
+    # were included in the final headline, leaking model-selection information.
     eval_modes = sorted({p["env_mode"] for p in problems} - {"gt_only"})
-    test_problems = load_problems(10_000, env_modes=eval_modes, seed=EVAL_SAMPLE_SEED,
-                                  data_path=DATA.parent / "leetcode_test_medhard.jsonl", shuffle=True)
-    val_problems = test_problems[:cfg.eval_n_prompts]   # periodic monitoring sample of the paper test
+    val_problems, test_problems = load_eval_splits(eval_modes, cfg.eval_n_prompts)
     val_idxs, test_idxs = list(range(len(val_problems))), list(range(len(test_problems)))
-    assert not ({p["problem_id"] for p in test_problems} & {p["problem_id"] for p in problems}), \
-        "TEST set leaks training problems"
     _train_ids = {p["problem_id"] for p in problems}
     assert not (_train_ids & {p["problem_id"] for p in val_problems}), "VAL set leaks training problems"
     assert not (_train_ids & {p["problem_id"] for p in test_problems}), "TEST set leaks training problems"
-    logger.info(f"held-out eval: periodic-curve n={len(val_problems)} sample + final n={len(test_problems)} "
-                f"(both from paper test set leetcode_test_medhard), modes={eval_modes}")
+    logger.info(f"held-out eval: periodic val n={len(val_problems)} + untouched final test "
+                f"n={len(test_problems)} from leetcode_test_medhard, modes={eval_modes}")
 
     rng = torch.Generator().manual_seed(cfg.seed)
     rows = []
@@ -2056,27 +2031,33 @@ def main(cfg: Config) -> int:
             f"{_r['text'][:800]}\n=== END LAST GEN ===\n")
 
     # ── final eval + BLUF ──
-    # Final per-mode train-vs-deploy eval -- run for EVERY arm on the SAME fixed
-    # eval subset so the all-arms overlay reads identical numbers. For route/routeV
-    # this is the absorption test: TRAIN keeps the quarantine knob on (still hacks),
-    # DEPLOY deletes it (the shipped model). SHOULD: deploy hack < train hack at
-    # preserved solve => the quarantine absorbed the cheat. vanilla/erase have no
-    # quarantine, so the deployed model IS the trained model (deploy == train, one eval).
+    # Evaluate knob-off and knob-on on the same final examples and generation seed.
+    # This paired, pre-specified comparison measures quarantine absorption; final-test
+    # results must not feed training, hyperparameter choices, or arm selection.
     model.eval()
     # FINAL paper number: DEPLOY (knob-OFF) on the held-out TEST set (disjoint file,
     # unseen in training AND in the periodic val curve). Same schema as
     # scripts/rescore_deploy.py, so the in-run number and an offline re-score off the
-    # saved checkpoint are interchangeable. Train-vs-deploy contrast lives in the val
-    # curve; the final is deploy only.
+    # saved checkpoint are interchangeable. The final paired knob-on/off comparison
+    # measures quarantine absorption without feeding any result back into training.
     has_quarantine = cfg.intervention in ("route", "routeV")
-    logger.info(f"FINAL EVAL: deploy (knob-off) on held-out TEST n={len(test_problems)} "
-                f"(periodic curve used val n={len(val_problems)})")
+    logger.info(f"FINAL EVAL on held-out TEST n={len(test_problems)} (periodic curve used val "
+                f"n={len(val_problems)}); knob-off=deploy"
+                f"{' + knob-on=deployed-as-trained' if has_quarantine else ''}")
     torch.manual_seed(EVAL_GEN_SEED)
     with (ablate_quarantine(wrappers) if has_quarantine else nullcontext()):
         ev = eval_hack_solve(model, tok, test_problems, test_idxs, gen_cfg_eval, device, max_new,
                              cfg.eval_batch_size)
-    logger.info(f"FINAL EVAL [{cfg.arm}] DEPLOY (held-out test, n={ev['n']}): "
-                f"hack(strict)={ev['hack']:.3f} hack(vendor eq_hinted)={ev['vhack']:.3f} solve={ev['solve']:.3f}")
+    if has_quarantine:
+        torch.manual_seed(EVAL_GEN_SEED)
+        ev_on = eval_hack_solve(model, tok, test_problems, test_idxs, gen_cfg_eval, device, max_new,
+                                cfg.eval_batch_size)
+    else:
+        ev_on = ev
+    logger.info(f"FINAL EVAL [{cfg.arm}] DEPLOY knob-off (held-out test, n={ev['n']}): "
+                f"hack(strict)={ev['hack']:.3f} hack(vendor eq_hinted)={ev['vhack']:.3f} solve={ev['solve']:.3f}"
+                + (f"  |  knob-on: hack={ev_on['hack']:.3f} solve={ev_on['solve']:.3f}"
+                   if has_quarantine else ""))
     by_mode = {}
     for mode in sorted(ev["by_mode"]):
         dh, dv, ds, dn = ev["by_mode"][mode]
@@ -2085,8 +2066,10 @@ def main(cfg: Config) -> int:
     deploy_record = {
         "run_dir": run_dir.name, "arm": cfg.arm, "intervention": cfg.intervention,
         "seed": cfg.seed, "steps": n_steps, "model": model_name, "out_tag": cfg.out_tag,
-        "eval_set": "test", "n": ev["n"],
+        "eval_set": "test", "eval_modes": eval_modes, "n": ev["n"],
         "deploy_hack": ev["hack"], "deploy_vhack": ev["vhack"], "deploy_solve": ev["solve"],
+        "deploy_hack_on": ev_on["hack"], "deploy_vhack_on": ev_on["vhack"],
+        "deploy_solve_on": ev_on["solve"],
         "by_mode": by_mode, "log": str(verbose_log),
     }
     deploy_path = run_dir / "deploy_test.json"

src/vgrout/vhack.py

@@ -8,6 +8,7 @@ load and the in-loop refresh.
 """
 from __future__ import annotations
 
+import hashlib
 from pathlib import Path
 
 import torch
@@ -16,8 +17,12 @@ from loguru import logger
 from safetensors import safe_open
 
 
+def pairset_sha256(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
 def load_v_hack(
-    path: Path, model_name: str, wrappers: dict,
+    path: Path, model_name: str, wrappers: dict, pairs_path: Path,
     k_use: int | None = None, drop_bottom_frac: float = 0.0,
 ) -> dict[str, Float[torch.Tensor, "k r"]]:
     """Load v_hack (top-k directions) for this wrapped model.
@@ -39,14 +44,21 @@ def load_v_hack(
         meta = f.metadata() or {}
         saved_model = meta.get("model")
         saved_dtype = meta.get("dtype")
-        if saved_model is None or saved_dtype is None:
+        saved_pairs_sha256 = meta.get("pairs_sha256")
+        if saved_model is None or saved_dtype is None or saved_pairs_sha256 is None:
             raise ValueError(
-                f"{path} has no model/dtype header metadata. "
+                f"{path} has no model/dtype/pairs_sha256 metadata. "
                 f"Re-extract with `uv run python -m vgrout.extract_vhack_grad "
-                f"--model={model_name} --dtype=bf16 --out-path={path}`."
+                f"--model={model_name} --dtype=bf16 --pairs-from-pool={pairs_path} --out-path={path}`."
             )
         if saved_model != model_name:
             raise ValueError(f"v_hack model mismatch: {path} has {saved_model}, run uses {model_name}")
+        expected_pairs_sha256 = pairset_sha256(pairs_path)
+        if saved_pairs_sha256 != expected_pairs_sha256:
+            raise ValueError(
+                f"v_hack pairset mismatch: {path} has sha256={saved_pairs_sha256}, "
+                f"{pairs_path} has sha256={expected_pairs_sha256}. Re-extract the direction."
+            )
         # dtype mismatch: cross-dtype SVD bases can diverge silently, so error
         # unless the saved dtype matches what train.py uses on this device.
         # CPU runs in fp32, CUDA runs in bf16 (see model-load site above).