NNUE search integration: FH% tables, NODE_TYPE refinement, TREEDUMP rewrite

- data.c/chess.h: widen the joint FH% table's late axis 16->17, dedicating
  the new bucket 0 to a whole-node ("will something here fail high at
  all") sample rather than sharing it with the first-leftover-move curve.
  Add an NNUE-eval counterpart table (g_uJointFHThousandths_NNUE) and an
  NNUE counterpart of the beta-side table, selected at runtime based on
  whether NNUE eval is active.
- search.c/searchsup.c: add _RefineNodeTypeGuess, which uses the FH%
  tables (via g_NodeTypeGuessTable) to refine a node's NODE_TYPE guess
  instead of the old heuristic; thread uParentMoveFHEstimate through
  PLY_INFO so a node can see its parent's prediction for the move that
  led to it.
- generate.c: initialize sKillers[2]/[3].uBonus unconditionally (an
  uninitialized-stack read caught by valgrind under LTO; harmless today
  but a real bug).
- dumptree.c/chess.h/GNUmakefile: replace the long-dead DUMP_TREE XML
  tree dump (written for a GUI this project no longer has) with TREEDUMP,
  a flat, path-keyed move/result trace meant to diff against an
  identically-instrumented Stockfish build -- see docs/TREEDUMP_FORMAT.md.
- docs/DEEPER_PRUNING.md: notes on closing the search-efficiency gap vs
  Stockfish.
- crafty_egtb/stockfish_xboard/typhoon_nnue: wrapper scripts so
  match_play.py's Engine class can drive Crafty, Stockfish (via
  PolyGlot), and an NNUE-enabled typhoon the same way it drives HCE
  typhoon.
- tests/lop: rename lop.txt -> lop.ep_; update compare_lop.cpp.
- annotate_pgns: add aggregate_marginsafety.py, dedup_pgn.py,
  fit_lazy_probe.py; update annotate.py/run_nnue_pipeline.py for the NNUE
  training pipeline.
- .gitignore: ignore compile_commands.json (clangd build DB) and
  eval.nnue (a symlink into the untracked head_reference/ comparison
  baseline), neither of which belongs in the repo.

Co-Authored-By: Claude Sonnet 5 <[email protected]>

5 files changed, 706 insertions, 36 deletions

diff --git a/annotate_pgns/aggregate_marginsafety.py b/annotate_pgns/aggregate_marginsafety.py
new file mode 100644
index 0000000..32d5808
--- /dev/null
+++ b/annotate_pgns/aggregate_marginsafety.py
@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""Sums the raw swing histograms eval.c's PersistMarginSafetyHistogramsToFile
+writes (one file per shard of a sharded CALIBRATE_MARGIN_SAFETY run)
+bucket-by-bucket and bin-by-bin, then computes EXACT combined mean/
+percentiles from the summed bins -- percentiles don't combine validly
+shard-by-shard (each shard's own in-process percentile is over a
+different, non-representative sub-sample), only the raw counts do.
+
+File format (one line per tier x material-bucket):
+    <tier> <bucket> <count> <exceeded> <sum> <sumSq> <max> <bin0> ... <bin299>
+SWING_HIST_BIN_WIDTH=10, SWING_HIST_BINS=300 (covers [0, 3000) cp) --
+read from each file's own header comment rather than hardcoded, in
+case a future build changes them.
+
+Usage:
+    python3 aggregate_marginsafety.py /path/to/calib_out/hist_*.txt
+"""
+import argparse
+import glob
+import sys
+from collections import defaultdict
+
+
+def parse_header(path):
+    with open(path) as f:
+        for line in f:
+            if line.startswith("# SWING_HIST_BINS="):
+                parts = dict(kv.split("=") for kv in line[2:].split())
+                return int(parts["SWING_HIST_BINS"]), int(parts["SWING_HIST_BIN_WIDTH"])
+    sys.exit(f"{path}: no SWING_HIST_BINS header found")
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__,
+                                  formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("files", nargs="+", help="hist_*.txt files (glob already expanded by shell, "
+                                               "or pass a glob pattern quoted)")
+    args = ap.parse_args()
+
+    files = []
+    for pat in args.files:
+        matched = glob.glob(pat)
+        files.extend(matched if matched else [pat])
+    if not files:
+        sys.exit("no input files")
+
+    n_bins, bin_width = parse_header(files[0])
+
+    # (tier, bucket) -> dict(count, exceeded, sum, sumSq, max, bins=[n_bins])
+    agg = defaultdict(lambda: {"count": 0, "exceeded": 0, "sum": 0, "sumSq": 0,
+                                "max": 0, "bins": [0] * n_bins})
+
+    for path in files:
+        with open(path) as f:
+            for line in f:
+                if line.startswith("#") or not line.strip():
+                    continue
+                parts = line.split()
+                tier, bucket = parts[0], int(parts[1])
+                count, exceeded, s, sumsq, mx = (int(x) for x in parts[2:7])
+                bins = [int(x) for x in parts[7:7 + n_bins]]
+                a = agg[(tier, bucket)]
+                a["count"] += count
+                a["exceeded"] += exceeded
+                a["sum"] += s
+                a["sumSq"] += sumsq
+                a["max"] = max(a["max"], mx)
+                for i, v in enumerate(bins):
+                    a["bins"][i] += v
+
+    def percentile(bins, count, p):
+        """Exact percentile from binned counts -- rank = p/100 * count,
+        walk cumulative counts to find which bin holds that rank, report
+        the bin's lower edge (a slight underestimate, bin_width=10cp
+        granularity -- good enough for choosing a margin, not claimed
+        exact to the cp)."""
+        if count == 0:
+            return None
+        rank = (p / 100.0) * count
+        cum = 0
+        for i, c in enumerate(bins):
+            cum += c
+            if cum >= rank:
+                return i * bin_width
+        return (len(bins) - 1) * bin_width  # fell off the end -- all mass in last bin
+
+    tiers = sorted(set(t for t, b in agg.keys()))
+    print(f"Aggregated {len(files)} shard files, {n_bins} bins x {bin_width}cp "
+          f"(covers [0, {n_bins * bin_width}) cp)\n")
+
+    results = {}
+    for tier in tiers:
+        print(f"=== {tier} ===")
+        print(f"{'bucket':>6} {'n':>10} {'mean':>8} {'p50':>6} {'p98':>6} "
+              f"{'p99.9':>7} {'max':>7} {'exceeded':>9}")
+        for bucket in range(8):
+            a = agg.get((tier, bucket))
+            if a is None or a["count"] == 0:
+                print(f"{bucket:>6} (no data)")
+                continue
+            mean = a["sum"] / a["count"]
+            p50 = percentile(a["bins"], a["count"], 50)
+            p98 = percentile(a["bins"], a["count"], 98)
+            p999 = percentile(a["bins"], a["count"], 99.9)
+            results[(tier, bucket)] = {"p98": p98, "p999": p999, "mean": mean,
+                                        "max": a["max"], "n": a["count"]}
+            print(f"{bucket:>6} {a['count']:>10} {mean:>8.1f} {p50:>6} {p98:>6} "
+                  f"{p999:>7} {a['max']:>7} {a['exceeded']:>9}")
+        print()
+
+    if "nnue_lazy" in tiers:
+        p98s = [results.get(("nnue_lazy", b), {}).get("p98", "?") for b in range(8)]
+        print(f"Suggested iNnueLazyMarginByMatBucket (p98 per bucket, matching "
+              f"the existing table's own documented choice of statistic):")
+        print(f"    {{{', '.join(str(x) for x in p98s)}}}")
+    if "nnue_superlazy" in tiers:
+        p98s = [results.get(("nnue_superlazy", b), {}).get("p98", "?") for b in range(8)]
+        print(f"\nFor comparison, NNUE super-lazy (material-only) p98 per bucket:")
+        print(f"    {{{', '.join(str(x) for x in p98s)}}}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/annotate_pgns/annotate.py b/annotate_pgns/annotate.py
index 1f8d02d..89dfd07 100644..100755
--- a/annotate_pgns/annotate.py
+++ b/annotate_pgns/annotate.py
@@ -17,9 +17,14 @@ See the module-level functions below for the four stages: shard, build
 engine scripts, run engines in parallel, parse + reassemble.
 """
 import argparse
+import os
+import pty
 import re
+import select
+import signal
 import subprocess
 import sys
+import threading
 import time
 from pathlib import Path
 
@@ -37,6 +42,28 @@ MATE_RE = re.compile(r"^([+-])MATE(\d*)$")  # digit(s) optional -- the
 # (search.c) prints a bare "+MATE"/"-MATE" with no distance number at
 # all, not just the ordinary "+MATE<N>" case.
 
+# --engine-protocol uci support: drives a real UCI engine (Stockfish)
+# directly over stdin/stdout with the native UCI protocol -- no xboard,
+# no PolyGlot adapter needed. Matches the standard "info depth N ...
+# score cp/mate X ... pv ..." thinking-output line Stockfish prints once
+# per completed depth.
+UCI_ITER_RE = re.compile(
+    r"^info depth (\d+) .*?\bscore (cp|mate) (-?\d+).*?\bpv (.*)$"
+)
+# A naive first attempt at this driver tried to interrupt each search
+# early (xboard's "force", or UCI's "stop") the instant the target depth
+# was seen, rather than waiting for "bestmove" -- motivated by "go depth
+# N" appearing to hang forever with no bestmove ever printed. That hang
+# turned out to be an artifact of this tool's own sandboxed subprocess
+# environment, NOT a Stockfish or protocol issue (confirmed 2026-10-20:
+# the identical "uci; isready; position fen ...; go depth 10" sequence
+# got a normal, prompt "bestmove" reply when run directly in an
+# interactive terminal outside that sandbox). So this driver just sends
+# "go depth <depth>" and waits for "bestmove" like any normal UCI
+# client -- no interrupt-and-abort workaround needed.
+UCI_HANDSHAKE_TIMEOUT_SEC = 15
+UCI_GO_TIMEOUT_SEC = 120  # per position, generous but bounded
+
 
 def load_games(input_paths, max_games=None):
     """Load games, skipping ones with an unknown/unterminated result
@@ -95,20 +122,33 @@ def positions_for_game(game, prefilter_quiet=False):
         board.push(move)
 
 
-def write_shard_script(shard_games_list, script_path, depth, prefilter_quiet=False):
-    """Write the engine command script for one shard, plus the parallel
-    index recording which (game_idx_in_shard, ply_idx) each 'go' line
-    corresponds to -- the engine processes 'go' commands strictly in
-    script order with no interleaving, so a flat index list zips back
-    onto the flat sequence of tellothers output lines."""
+def build_shard_positions(shard_games_list, prefilter_quiet=False):
+    """Return (fens, index) for one shard: the flat list of FENs in
+    script order, and the parallel index recording which
+    (game_idx_in_shard, ply_idx) each FEN corresponds to -- shared by
+    both write_shard_script (typhoon protocol, writes an engine command
+    script to disk) and run_shard_uci (uci protocol, drives the engine
+    interactively with no script file at all)."""
+    fens = []
     index = []
+    for gi, game in enumerate(shard_games_list):
+        for pi, fen in enumerate(positions_for_game(game, prefilter_quiet)):
+            fens.append(fen)
+            index.append((gi, pi))
+    return fens, index
+
+
+def write_shard_script(shard_games_list, script_path, depth, prefilter_quiet=False):
+    """Write the engine command script for one shard. The engine
+    processes 'go' commands strictly in script order with no
+    interleaving, so the index returned by build_shard_positions zips
+    back onto the flat sequence of tellothers output lines."""
+    fens, index = build_shard_positions(shard_games_list, prefilter_quiet)
     with open(script_path, "w") as out:
-        for gi, game in enumerate(shard_games_list):
-            for pi, fen in enumerate(positions_for_game(game, prefilter_quiet)):
-                out.write(f"setboard {fen}\n")
-                out.write(f"sd {depth}\n")
-                out.write("go\n")
-                index.append((gi, pi))
+        for fen in fens:
+            out.write(f"setboard {fen}\n")
+            out.write(f"sd {depth}\n")
+            out.write("go\n")
     return index
 
 
@@ -126,6 +166,174 @@ def run_shard(engine_path, script_path, out_path, hash_size, cpus):
         return subprocess.Popen(cmd, stdout=out, stderr=subprocess.STDOUT)
 
 
+def _uci_format_tellothers(depth, score_kind, score_val, pv):
+    """Render one uci-protocol result as a synthetic tellothers line, so
+    the existing parse_shard_output/_format_comment/annotate_shard
+    pipeline (written for typhoon's own tellothers output) can be reused
+    unchanged for uci-mode results too. `pv` must already be SAN
+    (see _uci_pv_to_san) -- Stockfish's own PV is UCI long algebraic
+    ("e2e4 d7d5"), not the SAN typhoon's tellothers line uses ("e4
+    d5"), and downstream readers (extract_positions.py,
+    extract_selfplay_positions.py) key off SAN."""
+    if score_kind == "mate":
+        score_tok = "+MATE" if score_val > 0 else "-MATE"
+    else:
+        score_tok = f"{score_val / 100:+.2f}"
+    return f"tellothers d{depth}, {score_tok}, 0.0s, 0knps, PV={pv}\n"
+
+
+def _uci_pv_to_san(fen, uci_pv):
+    """Convert a UCI-long-algebraic PV string ("e2e4 d7d5 ...") rooted
+    at `fen` into typhoon's own tellothers SAN convention ("e4 d5
+    ..."). Stops at the first move that doesn't parse/isn't legal
+    (shouldn't happen for a PV the engine itself just reported, but an
+    illegal/ambiguous tail is far less useful than no PV, not more)."""
+    board = chess.Board(fen)
+    san_moves = []
+    for uci_move in uci_pv.split():
+        try:
+            move = chess.Move.from_uci(uci_move)
+            if move not in board.legal_moves:
+                break
+            san_moves.append(board.san(move))
+            board.push(move)
+        except (ValueError, AssertionError):
+            break
+    return " ".join(san_moves)
+
+
+def run_shard_uci(engine_path, fens, depth, out_path):
+    """Drive one persistent native-UCI engine subprocess (e.g. a
+    Stockfish binary) through every FEN in `fens`, in order, writing one
+    synthetic tellothers line per FEN to out_path.
+
+    Spawned under a pty (via pty.fork), not a plain subprocess.Popen
+    pipe -- verified live (2026-10-20) that this particular Stockfish
+    build hangs after printing just its banner + "id name" line, with
+    no more output ever arriving, when its stdout is a plain pipe (the
+    "id author"/options/"uciok" lines, and everything after, simply
+    never get produced -- not merely buffered: a pty.fork-based read of
+    the exact same binary gets the whole handshake instantly). The
+    earlier guess that this was xboard/PolyGlot-specific, and later that
+    it was this tool's own sandboxed subprocess environment, were both
+    wrong -- it reproduces under dangerouslyDisableSandbox too. Root
+    cause not fully isolated (likely an isatty()-gated code path in
+    Stockfish's own startup or NUMA/processor-count detection), but the
+    pty workaround is confirmed sufficient."""
+    pid, fd = pty.fork()
+    if pid == 0:
+        try:
+            os.execvp(str(engine_path), [str(engine_path)])
+        finally:
+            os._exit(127)  # os.execvp only returns on failure
+
+    buf = b""
+
+    def send(cmd):
+        os.write(fd, (cmd + "\n").encode())
+
+    def readline(timeout):
+        nonlocal buf
+        while b"\n" not in buf:
+            r, _, _ = select.select([fd], [], [], timeout)
+            if not r:
+                return None
+            try:
+                chunk = os.read(fd, 65536)
+            except OSError:
+                return None
+            if not chunk:
+                return None
+            buf += chunk
+        line, buf = buf.split(b"\n", 1)
+        # The pty's line-discipline translates outgoing "\n" to "\r\n"
+        # (termios ONLCR), so every line read back here has a trailing
+        # "\r" that a plain subprocess.Popen pipe would never produce --
+        # strip it, or sentinel comparisons ("uciok", "readyok") and the
+        # captured PV text silently never match/contain a stray "\r".
+        return line.rstrip(b"\r").decode(errors="replace") + "\n"
+
+    def wait_for(sentinel, timeout_sec):
+        deadline = time.time() + timeout_sec
+        while time.time() < deadline:
+            line = readline(2)
+            if line is None:
+                continue
+            if line.rstrip("\n") == sentinel:
+                return True
+        return False
+
+    send("uci")
+    wait_for("uciok", UCI_HANDSHAKE_TIMEOUT_SEC)
+    # Match ../src/stockfish_xboard's own default: Stockfish's tbprobe.cpp
+    # is Fathom-derived, so it reads the exact same .rtbw/.rtbz files
+    # typhoon does -- no separate Stockfish-format tablebase needed.
+    # SYZYGY_PATH="" explicitly disables TB probing rather than falling
+    # back to the default, same as that wrapper.
+    syzygy_path = os.environ.get("SYZYGY_PATH", "/zscratch/egtb")
+    if syzygy_path:
+        send(f"setoption name SyzygyPath value {syzygy_path}")
+    send("isready")
+    wait_for("readyok", UCI_HANDSHAKE_TIMEOUT_SEC)
+
+    with open(out_path, "w") as out:
+        for fen in fens:
+            send(f"position fen {fen}")
+            send(f"go depth {depth}")
+            best = None
+            deadline = time.time() + UCI_GO_TIMEOUT_SEC
+            while time.time() < deadline:
+                line = readline(2)
+                if line is None:
+                    continue
+                line = line.rstrip("\n")
+                if line.startswith("bestmove"):
+                    break
+                m = UCI_ITER_RE.match(line)
+                if m:
+                    d = int(m.group(1))
+                    if d >= depth:
+                        best = (d, m.group(2), int(m.group(3)), m.group(4))
+                        # Keep reading up to "bestmove" even once the
+                        # target depth is seen: with "go depth N",
+                        # Stockfish always reports depth N as its final
+                        # iteration before bestmove, so there's nothing
+                        # more useful to capture, but exiting early here
+                        # (like the earlier interrupt-based design did)
+                        # would again require aborting the search instead
+                        # of letting it finish on its own.
+            # Always write exactly one line per "go", even if the
+            # deadline was hit with no captured depth -- annotate_shard
+            # zips index and parsed results positionally, so skipping a
+            # write here would desync every later position in this
+            # shard, not just leave this one unannotated (the same class
+            # of bug the module docstring describes for a dropped
+            # tellothers line). "NA" matches neither FLOAT_RE nor
+            # MATE_RE, so _format_comment's fallback passes it through
+            # as-is instead of mis-rendering it as a real score.
+            if best is not None:
+                d, kind, val, uci_pv = best
+                out.write(_uci_format_tellothers(
+                    d, kind, val, _uci_pv_to_san(fen, uci_pv)))
+            else:
+                out.write(f"tellothers d{depth}, NA, 0.0s, 0knps, PV=\n")
+
+    send("quit")
+    deadline = time.time() + 5
+    while time.time() < deadline:
+        wpid, _ = os.waitpid(pid, os.WNOHANG)
+        if wpid == pid:
+            break
+        time.sleep(0.1)
+    else:
+        try:
+            os.kill(pid, signal.SIGKILL)
+            os.waitpid(pid, 0)
+        except ProcessLookupError:
+            pass
+    os.close(fd)
+
+
 def parse_shard_output(out_path):
     """Return a list of (depth, score_token, pv_string) in script
     order. score_token is kept as the engine's raw tellothers token
@@ -205,6 +413,18 @@ def main():
     ap.add_argument("--cpus", type=int, default=1,
                      help="--cpus passed to each shard's engine")
     ap.add_argument("--engine", default="../src/typhoon")
+    ap.add_argument("--engine-protocol", choices=["typhoon", "uci"],
+                     default="typhoon",
+                     help="'typhoon' drives --engine with its own "
+                          "'--command \"script <path>\"' batch mode and "
+                          "parses its tellothers output (default). 'uci' "
+                          "drives --engine (a UCI engine binary, e.g. a "
+                          "Stockfish build) directly over stdin/stdout "
+                          "with the native UCI protocol instead ('uci', "
+                          "'position fen ...', 'go depth N', wait for "
+                          "'bestmove'). --hash/--cpus are ignored in this "
+                          "mode; set hash/threads via UCI setoption "
+                          "yourself if needed.")
     ap.add_argument("--outdir", required=True)
     ap.add_argument("--max-games", type=int, default=None,
                      help="limit to the first N games total, for smoke testing")
@@ -233,30 +453,62 @@ def main():
     if args.prefilter_quiet:
         print("Pre-filtering: skipping in-check / SEE-positive-played-move "
               "positions before search.")
-    print(f"Writing {args.shards} shard scripts at sd {args.depth} ...")
+
     indices = []
-    for i, shard in enumerate(shards):
-        script_path = outdir / f"shard_{i:02d}.script"
-        idx = write_shard_script(shard, script_path, args.depth,
-                                  args.prefilter_quiet)
-        indices.append(idx)
-        print(f"  shard {i:02d}: {len(shard)} games, {len(idx)} positions")
+    shard_fens = []  # only populated/used in uci mode
+    if args.engine_protocol == "typhoon":
+        print(f"Writing {args.shards} shard scripts at sd {args.depth} ...")
+        for i, shard in enumerate(shards):
+            script_path = outdir / f"shard_{i:02d}.script"
+            idx = write_shard_script(shard, script_path, args.depth,
+                                      args.prefilter_quiet)
+            indices.append(idx)
+            print(f"  shard {i:02d}: {len(shard)} games, {len(idx)} positions")
+    else:
+        print(f"Building {args.shards} shard position lists at sd "
+              f"{args.depth} ...")
+        for i, shard in enumerate(shards):
+            fens, idx = build_shard_positions(shard, args.prefilter_quiet)
+            shard_fens.append(fens)
+            indices.append(idx)
+            print(f"  shard {i:02d}: {len(shard)} games, {len(idx)} positions")
 
     print("Launching shard engines in parallel ...")
     t0 = time.time()
-    procs = []
-    for i in range(args.shards):
-        script_path = outdir / f"shard_{i:02d}.script"
-        out_path = outdir / f"shard_{i:02d}.out"
-        if indices[i]:
-            procs.append((i, run_shard(engine_path, script_path, out_path,
-                                        args.hash, args.cpus)))
+    if args.engine_protocol == "typhoon":
+        procs = []
+        for i in range(args.shards):
+            script_path = outdir / f"shard_{i:02d}.script"
+            out_path = outdir / f"shard_{i:02d}.out"
+            if indices[i]:
+                procs.append((i, run_shard(engine_path, script_path, out_path,
+                                            args.hash, args.cpus)))
 
-    # Wait for the slowest shard.
-    for i, p in procs:
-        rc = p.wait()
-        if rc != 0:
-            print(f"  WARNING: shard {i:02d} exited with code {rc}", file=sys.stderr)
+        # Wait for the slowest shard.
+        for i, p in procs:
+            rc = p.wait()
+            if rc != 0:
+                print(f"  WARNING: shard {i:02d} exited with code {rc}", file=sys.stderr)
+    else:
+        # No per-engine OS process launched up front here the way
+        # run_shard does -- run_shard_uci owns its own subprocess
+        # internally and blocks for the whole shard, so parallelism
+        # across shards comes from running each one in its own thread
+        # instead (these are I/O-bound waits on each engine subprocess's
+        # pipe, so the GIL isn't a bottleneck here).
+        threads = []
+        for i in range(args.shards):
+            if not indices[i]:
+                continue
+            out_path = outdir / f"shard_{i:02d}.out"
+            t = threading.Thread(
+                target=run_shard_uci,
+                args=(engine_path, shard_fens[i], args.depth, out_path),
+            )
+            t.start()
+            threads.append(t)
+        for t in threads:
+            t.join()
     elapsed = time.time() - t0
     total_positions = sum(len(idx) for idx in indices)
     print(f"All shards done in {elapsed:.1f}s "
diff --git a/annotate_pgns/dedup_pgn.py b/annotate_pgns/dedup_pgn.py
new file mode 100644
index 0000000..49654c4
--- /dev/null
+++ b/annotate_pgns/dedup_pgn.py
@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""Drops duplicate games from one or more self-play PGN files, keyed on
+(starting FEN, move text) -- not on PGN headers (Date/Event/Result
+alone don't prove two games are the same game) and not on FEN alone
+(two games can legitimately share a random starting position and then
+diverge, since self-play move choice is not deterministic across
+engine generations/threads -- those are NOT duplicates and must be
+kept).
+
+Motivation: self-play starts from randomized opening positions (see
+docs/NNUE.md); with enough games it becomes likely the same random
+starting FEN gets drawn more than once. If the *same* position then
+also produced the *same* game (e.g. a re-run over an unchanged engine
+binary, or two shards drawing the same seed), that's a real duplicate
+and should not be trained on twice -- it would silently inflate that
+game's influence on both the WDL-K fit (fit_k_and_label.py) and the
+training loss itself.
+
+Dedup key is computed directly from each game's raw PGN text (the
+[FEN "..."] header plus a comment-and-whitespace-stripped movetext
+string), not by parsing through python-chess -- this project's own
+split_pgn() (run_nnue_pipeline.py) found full-PGN python-chess parsing
+too slow to use as a bulk pre-pass over large corpora; a raw-text key
+is exact for this purpose (identical moves from an identical FEN is
+exactly the definition of "the same game") and avoids that cost
+entirely.
+
+Usage:
+    python3 dedup_pgn.py ../pgn/stockfish*.pgn --out-dir runs/dedup
+Writes <out-dir>/<stem>.dedup.pgn for each input, in the same relative
+order games appeared in their source file (cross-file: first file on
+the command line wins ties). Prints a per-file and total duplicate
+count; does not modify the inputs.
+"""
+import argparse
+import hashlib
+import re
+import sys
+from pathlib import Path
+
+COMMENT_RE = re.compile(rb"\{[^}]*\}")
+WHITESPACE_RE = re.compile(rb"\s+")
+FEN_RE = re.compile(rb'\[FEN\s+"([^"]*)"\]')
+
+
+def iter_raw_games(path):
+    """Yields raw byte blocks for each game in a PGN file, split on
+    lines starting "[Event " -- same boundary convention as
+    run_nnue_pipeline.py's split_pgn()."""
+    with open(path, "rb") as f:
+        data = f.read()
+    offsets = [m.start() for m in re.finditer(rb"^\[Event ", data, re.MULTILINE)]
+    offsets.append(len(data))
+    for i in range(len(offsets) - 1):
+        block = data[offsets[i]:offsets[i + 1]]
+        if block.strip():
+            yield block
+
+
+def game_key(raw_game):
+    """(fen, movetext) dedup key. Movetext is everything after the
+    header block (first blank line), with PGN brace-comments and all
+    whitespace stripped -- comments carry score/depth/PV annotations
+    that are irrelevant to "is this the same game" and would otherwise
+    make two runs that scored an identical game differently (e.g. a
+    deeper search depth) look like different games, which is wrong:
+    same moves played is the actual definition of "the same game"
+    here, regardless of what score commentary got attached to it."""
+    fen_match = FEN_RE.search(raw_game)
+    fen = fen_match.group(1) if fen_match else b""
+    blank = raw_game.find(b"\n\n")
+    movetext = raw_game[blank:] if blank != -1 else b""
+    movetext = COMMENT_RE.sub(b"", movetext)
+    movetext = WHITESPACE_RE.sub(b"", movetext)
+    return hashlib.sha256(fen + b"|" + movetext).digest()
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__,
+        formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("pgns", nargs="+", help="input PGN file(s)")
+    ap.add_argument("--out-dir", required=True,
+                     help="directory to write <stem>.dedup.pgn into")
+    args = ap.parse_args()
+
+    out_dir = Path(args.out_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    seen = set()
+    total_in = 0
+    total_dupes = 0
+    for pgn_path in args.pgns:
+        pgn_path = Path(pgn_path)
+        n_in = 0
+        n_dupes = 0
+        out_path = out_dir / f"{pgn_path.stem}.dedup.pgn"
+        with open(out_path, "wb") as out:
+            for raw_game in iter_raw_games(pgn_path):
+                n_in += 1
+                key = game_key(raw_game)
+                if key in seen:
+                    n_dupes += 1
+                    continue
+                seen.add(key)
+                out.write(raw_game)
+                if not raw_game.endswith(b"\n\n"):
+                    out.write(b"\n")
+        print(f"{pgn_path}: {n_in} games, {n_dupes} duplicates dropped "
+              f"-> {out_path} ({n_in - n_dupes} kept)")
+        total_in += n_in
+        total_dupes += n_dupes
+
+    print(f"\nTotal: {total_in} games in, {total_dupes} duplicates "
+          f"dropped, {total_in - total_dupes} kept across "
+          f"{len(args.pgns)} file(s).")
+    if total_dupes == 0:
+        print("(no duplicates found)")
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/annotate_pgns/fit_lazy_probe.py b/annotate_pgns/fit_lazy_probe.py
new file mode 100644
index 0000000..3657116
--- /dev/null
+++ b/annotate_pgns/fit_lazy_probe.py
@@ -0,0 +1,148 @@
+#!/usr/bin/env python3
+"""Fits the lazy-eval probe (docs/NNUE.md's "Lazy-evaluation tiering
+plan"): a ridge regression on the trained net's own accumulator
+(fc1's input -- the ReLU'd, concatenated 512-wide per-perspective
+halves), targeting that SAME checkpoint's raw forward() output
+(pre-calibration -- see export_nnue_weights.py's CalibrateScoreMapping;
+NnueProbeScore applies the identical (CALIB_A, CALIB_B) correction
+itself at runtime, so the probe should target raw, not the calibrated
+cp value, to stay consistent with that).
+
+Must be re-run for every new checkpoint -- these weights are fit
+against one specific net's accumulator and are not portable across
+generations (same rule as material/score calibration).
+
+Writes the NPRB format NnueProbeLoadWeights (src/nnueeval.c) reads:
+    magic: "NPRB"
+    dim:   uint32 (2*L1)
+    mu:    dim x float32
+    sigma: dim x float32
+    weight: dim x float32
+    bias:  1x float32
+
+Usage:
+    python3 fit_lazy_probe.py \
+        runs/sf_20261020_recal/nnue_ckpt/ckpt_epoch015.pt \
+        runs/sf_20261020_recal/nnue_probe.bin \
+        --train-tensor-dir runs/sf_20261020_recal/tensors_train \
+        --holdout-tensor-dir runs/sf_20261020_recal/tensors_holdout
+"""
+import argparse
+import struct
+from pathlib import Path
+
+import numpy as np
+import torch
+
+from nnue_model import NnueNet
+
+RIDGE_LAMBDA = 1.0
+
+
+def _batched_accumulator_and_raw(model, tensor_dir, rows, batch_size=5000):
+    """Returns (accum [N, 2*L1] float64, raw [N] float64) for the given
+    row indices into tensor_dir's CSR arrays."""
+    d = Path(tensor_dir)
+    white_idx = np.load(d / "white_idx.npy"); white_off = np.load(d / "white_off.npy")
+    black_idx = np.load(d / "black_idx.npy"); black_off = np.load(d / "black_off.npy")
+    stm = np.load(d / "stm.npy")
+
+    accums, raws = [], []
+    with torch.no_grad():
+        for i in range(0, len(rows), batch_size):
+            chunk = rows[i:i + batch_size]
+            w_idx_list, w_off_list, b_idx_list, b_off_list = [], [], [], []
+            off_w, off_b = 0, 0
+            for r in chunk:
+                wi = white_idx[white_off[r]:white_off[r + 1]]
+                bi = black_idx[black_off[r]:black_off[r + 1]]
+                w_idx_list.append(wi); w_off_list.append(off_w); off_w += len(wi)
+                b_idx_list.append(bi); b_off_list.append(off_b); off_b += len(bi)
+            w_idx = torch.from_numpy(np.concatenate(w_idx_list).astype(np.int64))
+            w_off = torch.tensor(w_off_list, dtype=torch.int64)
+            b_idx = torch.from_numpy(np.concatenate(b_idx_list).astype(np.int64))
+            b_off = torch.tensor(b_off_list, dtype=torch.int64)
+            stm_t = torch.from_numpy(stm[chunk].astype(np.int64))
+
+            acc = model.accumulator(w_idx, w_off, b_idx, b_off, stm_t)
+            raw = model.forward(w_idx, w_off, b_idx, b_off, stm_t)
+            accums.append(acc.numpy())
+            raws.append(raw.numpy())
+    return (np.concatenate(accums).astype(np.float64),
+            np.concatenate(raws).astype(np.float64))
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__,
+                                  formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("checkpoint")
+    ap.add_argument("outfile")
+    ap.add_argument("--train-tensor-dir", required=True,
+                     help="fit the ridge regression against positions from here")
+    ap.add_argument("--holdout-tensor-dir", required=True,
+                     help="report residual stats (not used for fitting) against here")
+    ap.add_argument("--n-fit", type=int, default=80000)
+    ap.add_argument("--n-report", type=int, default=20000)
+    ap.add_argument("--seed", type=int, default=11)
+    args = ap.parse_args()
+
+    model = NnueNet()
+    ck = torch.load(args.checkpoint, map_location="cpu", weights_only=True)
+    model.load_state_dict(ck["model"])
+    model.eval()
+
+    rng = np.random.default_rng(args.seed)
+
+    n_train = len(np.load(Path(args.train_tensor_dir) / "stm.npy"))
+    fit_rows = np.sort(rng.choice(n_train, size=min(args.n_fit, n_train), replace=False))
+    X, y = _batched_accumulator_and_raw(model, args.train_tensor_dir, fit_rows)
+
+    mu = X.mean(axis=0)
+    sigma = X.std(axis=0)
+    sigma[sigma < 1e-6] = 1e-6  # guard against a dead/always-zero accumulator dim
+    Xs = (X - mu) / sigma
+
+    # Ridge regression, closed form: w = (X^T X + lambda*I)^-1 X^T y,
+    # bias handled by centering y (mu/sigma already centers X).
+    y_mean = y.mean()
+    yc = y - y_mean
+    dim = Xs.shape[1]
+    A = Xs.T @ Xs + RIDGE_LAMBDA * np.eye(dim)
+    w = np.linalg.solve(A, Xs.T @ yc)
+    bias = y_mean  # Xs has zero mean per-column, so w contributes 0 at the mean point
+
+    fit_pred = Xs @ w + bias
+    fit_resid = y - fit_pred
+    print(f"Fit on {len(fit_rows)} positions ({args.train_tensor_dir}): "
+          f"mean|resid|={np.mean(np.abs(fit_resid)):.3f}, "
+          f"p98={np.percentile(np.abs(fit_resid), 98):.3f}")
+
+    n_holdout = len(np.load(Path(args.holdout_tensor_dir) / "stm.npy"))
+    report_rows = np.sort(rng.choice(n_holdout, size=min(args.n_report, n_holdout), replace=False))
+    Xh, yh = _batched_accumulator_and_raw(model, args.holdout_tensor_dir, report_rows)
+    Xhs = (Xh - mu) / sigma
+    holdout_pred = Xhs @ w + bias
+    holdout_resid = yh - holdout_pred
+    abs_resid = np.abs(holdout_resid)
+    print(f"Held out on {len(report_rows)} positions ({args.holdout_tensor_dir}): "
+          f"mean|resid|={np.mean(abs_resid):.3f}, "
+          f"p50={np.percentile(abs_resid, 50):.3f}, "
+          f"p98={np.percentile(abs_resid, 98):.3f}, "
+          f"max={np.max(abs_resid):.3f}")
+    print("NOTE: this is an OFFLINE, quiet-filtered-game-position sample -- "
+          "NOT real search-node traffic. See the 2026-10-20 conversation on "
+          "why that distinction matters before trusting these residual "
+          "numbers as a safety bound for a real skip-gate.")
+
+    with open(args.outfile, "wb") as f:
+        f.write(b"NPRB")
+        f.write(struct.pack("<I", dim))
+        f.write(mu.astype(np.float32).tobytes())
+        f.write(sigma.astype(np.float32).tobytes())
+        f.write(w.astype(np.float32).tobytes())
+        f.write(struct.pack("<f", float(bias)))
+    print(f"Wrote {args.outfile}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/annotate_pgns/run_nnue_pipeline.py b/annotate_pgns/run_nnue_pipeline.py
index 6e51416..0210c53 100644
--- a/annotate_pgns/run_nnue_pipeline.py
+++ b/annotate_pgns/run_nnue_pipeline.py
@@ -20,10 +20,19 @@ specifically to handle a foreign-engine/Stockfish-sourced corpus).
 Chains the existing, already-individually-tested pipeline scripts in
 this order:
 
-    1. extract_selfplay_positions.py  each input PGN -> games.jsonl
-                                       (one run per file, since that
-                                       script takes a single PGN;
-                                       outputs concatenated)
+    0. dedup_pgn.py                   drops duplicate games (same
+                                       starting FEN + move text,
+                                       cross-file) before anything
+                                       else sees them -- self-play's
+                                       randomized starting positions
+                                       can recur, and a recurrence
+                                       that also plays out identically
+                                       is a real duplicate, not just a
+                                       repeated opening
+    1. extract_selfplay_positions.py  each deduped input PGN ->
+                                       games.jsonl (one run per file,
+                                       since that script takes a
+                                       single PGN; outputs concatenated)
     2. split_holdout.py               games.jsonl -> train/holdout
                                        split (by game, not position)
     3. fit_k_and_label.py             fits the WDL-sigmoid K on
@@ -94,7 +103,7 @@ def split_pgn(path, n, outdir):
 
     n_games = len(offsets)
     if n_games == 0:
-        return [path]
+        return [Path(path)]
     n = min(n, n_games)
     boundaries = [offsets[round(i * n_games / n)] for i in range(n)] + [end]
 
@@ -245,6 +254,23 @@ def main():
     rundir = Path(args.rundir)
     rundir.mkdir(parents=True, exist_ok=True)
 
+    # 0. Dedup games across all input files before anything else sees
+    # them. Self-play starting positions are randomized but drawn from
+    # a finite pool, so the same random opening -- and, on an unlucky
+    # draw, the same resulting game -- can recur within or across input
+    # files (see dedup_pgn.py's own docstring). Keyed on (starting FEN,
+    # move text), cross-file (one shared in-memory key set across all
+    # inputs in this single invocation), so a duplicate occurring in a
+    # later input file against an earlier one is still caught.
+    dedup_dir = rundir / "dedup"
+    deduped_paths = [dedup_dir / f"{Path(p).stem}.dedup.pgn" for p in pgn_paths]
+    run_stage(
+        "dedup_pgn.py", deduped_paths,
+        [sys.executable, HERE / "dedup_pgn.py",
+         *pgn_paths, "--out-dir", dedup_dir],
+        args.force)
+    pgn_paths = [str(p) for p in deduped_paths]
+
     games_path = rundir / "games.jsonl"
     train_path = rundir / "games.train.jsonl"
     holdout_path = rundir / "games.holdout.jsonl"