path: root/src/eval_tune/match_play.py

#!/usr/bin/env python3
"""
Head-to-head match: head binary vs. candidate binary, N games, alternating
colors, opening positions sampled from a filtered TWIC pool for variety.

Two separate compiled typhoon binaries (built from different commits/
working trees), not one binary with two loaded `evaldna` files -- the
DNA-diffing design this replaced was built for the Texel auto-tuning
pipeline, which this project has since moved away from in favor of
hand-tuned eval constants baked directly into eval.c. Comparing two
binaries means a real build is required for each side before running
this (there's no live DNA swap), but it's the right model now: what's
actually being compared is two different eval.c/search.c source trees,
not two parameter files loaded into an otherwise-identical process.

The authoritative board is kept in python-chess; each engine is treated
as stateless per move (force; setboard <fen>; go) rather than fed
"usermove", mirroring the go/force turn-taking already verified to work
in tune_eval_dna.py's play_selfplay_game.

Usage:
    python3 match_play.py <head_engine_path> <candidate_engine_path> \\
        --pgn ../../pgn/twic_filtered.pgn --games 500 --workers 4 --st 1

    Add --sprt (with --elo0/--elo1/--alpha/--beta) to stop as soon as a
    Sequential Probability Ratio Test concludes instead of always
    playing exactly --games games -- see the Sprt class docstring for
    why a fixed game count is usually the wrong tool for "are these two
    engines really different in strength".

Prints:
    CANDIDATE_SCORE=<0..1> GAMES=<n> WINS=<a> DRAWS=<b> LOSSES=<c> ELO=<+/-x.x>
        LOWER95=<x> [SPRT=... LLR=... BOUNDS=... ELO0=... ELO1=...]
    CANDIDATE_STATS: nps=... ebf=... local_ebf=... first_move_beta=...% nodes=... plies=...
    HEAD_STATS:      nps=... ebf=... local_ebf=... first_move_beta=...% nodes=... plies=...
for the orchestrator to parse. Speed/tree-shape stats (EngineStats) are
pulled from the same per-move PostMoveSearchReport block every "go"
already prints, not a separate benchmark -- see EngineStats' docstring.
"""
import argparse
import hashlib
import io
import math
import os
import random
import re
import select
import sys
import time
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
from datetime import date
from pathlib import Path

import chess
import chess.pgn

sys.path.insert(0, str(Path(__file__).parent))
from tune_eval_dna import Engine  # noqa: E402

_MOVE_RE = re.compile(r"^move (\S+)")
# PostMoveSearchReport (root.c) prints these AFTER the "move" line, not
# before -- confirmed against root.c (Trace("move %s\n", ...) at line
# ~1367/1369, PostMoveSearchReport(ctx) called at ~1384, strictly after)
# and against real engine output. A per-move stats block trails every
# move, not just script-run suite summaries -- see EBF/first-move-beta
# discussion in session history.
# Matches both the old report format ("Searched for 0.3 seconds, saw 123
# nodes (45 qnodes) ( 969 nps).") and the new, more compact one ("Searched
# for   0.3s, saw 123 nodes (45 qnodes) (969knps)."): "seconds"/"s", and an
# optional "k" multiplier before "nps" (new format reports already-scaled
# thousands-of-nodes-per-second).
_SEARCHED_RE = re.compile(
    r"^Searched for\s+([\d.]+)\s*(?:seconds|s), saw (\d+) nodes "
    r"\((\d+) qnodes\) \(\s*([\d.]+)\s*(k?)nps\)\.")
_DEPTH_RE = re.compile(r"^tellothers d(\d+),")
# Old: "First move beta cutoff rate was 93.681 percent."
# New: "First move beta cutoff rate was 93.681%."
_FIRST_MOVE_BETA_RE = re.compile(
    r"^First move beta cutoff rate was\s+([\d.]+)\s*(?:percent|%)\.")
# 2026-09-18: per-iteration line UtilPrintPV (util.c) prints once per
# completed depth during a move's search (both under-xboard "%2u %6d
# %5u %-12llu %s" and human "%2u[+-]? %5s %-9s %-11llu %s" formats --
# see util.c's UtilPrintPV; engines here run under-xboard since Engine
# sends "xboard" at startup, per _STATS_TRAILING_LINES' comment above,
# but this regex tolerates either since it only cares about the depth
# and node-count fields, not the score/time/PV formatting in between).
# u64TotalNodeCount is a running total for the WHOLE search (not reset
# per iteration), so consecutive lines' node counts are monotonic
# non-decreasing including across an aspiration-window fail-low/high
# re-search at the SAME depth (root.c re-enters Search() with a wider
# window at the same g_uIterateDepth, printing another line before
# depth actually advances) -- see the EBF-from-consecutive-iterations
# derivation in EngineStats.add_ply's caller for how that's handled.
# Groups: (1) depth, (2) score in centipawns from the SIDE TO MOVE's
# perspective (raw iScore, VALUE_PAWN == 100 -- see chess.h), (3) node
# count, (4) the rest of the line: the SAN PV text (WalkPV, util.c), or
# "FL --" for a fail-low line (no PV -- the move that failed low isn't
# necessarily the move eventually played) or "<SAN> ++" for a fail-high.
# 2026-10-17: score group widened from (-?\d+) to ([+-]?\d+) -- typhoon
# itself never prints a leading "+" on a non-negative score (bare
# digits), but a UCI engine driven through PolyGlot's xboard adapter
# (e.g. Stockfish, see stockfish_xboard) DOES print an explicit "+"
# ("23 +28 198 700479 e4 c5"), which the old pattern rejected outright
# -- confirmed via direct regex test before this fix, not assumed.
_ITER_RE = re.compile(r"^\s*(\d+)[+-]?\s+([+-]?\d+)\s+\S+\s+(\d+)\s*(.*)$")
# Reading PostMoveSearchReport's trailing block used to stop either at a
# bounded line count or at gamelist.c's "_white(move N)_"/"_black(move N)_"
# prompt -- both unreliable: the block's line count depends on whether the
# binary was built with PERF_COUNTERS (root.c gates "First move beta
# cutoff rate" and several other lines behind #ifdef PERF_COUNTERS, so a
# binary built without it -- true of ../typhoon in this tree, unlike
# head_reference -- prints a shorter block), and the prompt line is
# suppressed entirely once Engine.__init__ (tune_eval_dna.py) sends
# "xboard" at startup, which sets g_Options.fRunningUnderXboard and makes
# MakeStatusLine() skip printing it (gamelist.c: "if
# (!g_Options.fRunningUnderXboard)"). Either mismatch left readline()
# blocking forever waiting for a line that would never come -- every
# worker deadlocked silently after its first move (0% CPU, no further
# games), which looked like "no matches happening" from outside.
#
# A "ping <id>"-after-"go" xboard synchronization idiom was tried next and
# also failed: command.c's PingCommand gets serviced by an input path that
# runs concurrently with an in-progress search (confirmed live -- "pong 0"
# came back *before* the "move" line), so it's decoupled from output
# ordering and cannot mark end-of-report either.
#
# Fixed for real with an unconditional marker Trace()'d by
# PostMoveSearchReport itself (root.c) as the last thing it ever prints,
# regardless of build flags -- see "ReportEnd" there. But a pre-existing
# binary (e.g. head_reference/typhoon, deliberately left untouched as a
# pinned historical baseline -- see CLAUDE.md's head_reference protocol)
# predates this marker and will never print it, so ReportEnd alone would
# just trade one build-flag-shaped hang for a binary-vintage-shaped one.
# Belt-and-suspenders fix: _read_stats_line below uses select() to give up
# waiting after a short idle period with no new output, treating a quiet
# pipe as "report's done" regardless of whether any specific terminal line
# ever arrives. Costs up to that idle timeout once per move for a binary
# that never emits ReportEnd; costs nothing for one that does, since the
# marker line short-circuits the loop the moment it's read.
_REPORT_END_RE = re.compile(r"^ReportEnd$")
_STATS_TRAILING_LINES = 60  # hard cap on lines, belt to the idle-timeout suspenders
_STATS_IDLE_TIMEOUT_SEC = 0.5


def _readline_or_none(stream, timeout):
    """readline() with a timeout: None means the pipe went quiet for
    `timeout` seconds with nothing to read (not EOF -- EOF is still a
    normal, immediately-returned '' from readline() once select() reports
    the fd readable)."""
    ready, _, _ = select.select([stream], [], [], timeout)
    if not ready:
        return None
    return stream.readline()

# MATCH_DEBUG=1 traces every command sent and every line read per side, so a
# hang can be pinned to the exact readline() call blocking rather than just
# "process went idle" from the outside (ps shows CPU frozen but not why).
_DEBUG = os.environ.get("MATCH_DEBUG") == "1"


def _dbg(tag, msg):
    if _DEBUG:
        print(f"[{tag}] {msg}", file=sys.stderr, flush=True)


def _sample_one_opening(fb, file_size, rng, min_ply, max_ply):
    """One attempt at pulling a random opening position out of an
    already-open pgn pool file: seek to a random byte offset, find the
    next game's header block, parse just that one game, and walk a
    random number of its plies. Returns a FEN, or None if this attempt
    didn't land on a usable game (truncated tail, too-short game, etc.)
    -- callers retry elsewhere on None rather than treating it as fatal."""
    offset = rng.randint(0, max(file_size - 1, 0))
    fb.seek(offset)
    fb.readline()  # discard partial line at the seek point
    # scan forward to the start of the next game's header block
    start = fb.tell()
    line = fb.readline()
    while line and not line.startswith(b"[Event "):
        start = fb.tell()
        line = fb.readline()
    if not line:
        return None  # landed past the last game in the file; retry elsewhere
    # read forward to the START of the game AFTER this one (or EOF)
    fb.seek(start)
    fb.readline()  # consume this game's own "[Event " line
    end = fb.tell()
    line = fb.readline()
    while line and not line.startswith(b"[Event "):
        end = fb.tell()
        line = fb.readline()
    fb.seek(start)
    raw = fb.read(end - start).decode("utf-8", errors="replace")
    game = chess.pgn.read_game(io.StringIO(raw))
    if game is None:
        return None
    moves = list(game.mainline_moves())
    if len(moves) < min_ply:
        return None
    board = game.board()
    ply = rng.randint(min_ply, min(max_ply, len(moves)))
    for mv in moves[:ply]:
        board.push(mv)
    return board.fen()


def sample_openings(pgn_paths, n, min_ply=8, max_ply=20, seed=0,
                     cache_dir="opening_cache"):
    """Pick n random (roughly) opening positions out of one or more pgn
    pools by seeking to random byte offsets and parsing just the one game
    found there, rather than sequentially chess.pgn.read_game()-ing from
    the start -- on a several-hundred-MB pool, replaying every game up to
    the sample point (even while mostly skipping them) means real I/O
    and SAN-parsing cost that scales with how far into the file the
    first few accepted samples happen to land, which was observed
    taking minutes for as few as 2 requested openings. Cached like
    tune_eval_dna.py's position_cache, keyed on pgn identity + params,
    since repeated cycle.sh runs against the same pool(s) shouldn't pay
    this more than once.

    `pgn_paths` may name several pools at once (e.g. the TWIC game
    archive alongside a curated opening-book PGN) to widen the set of
    starting positions beyond whatever one file happens to contain --
    every match otherwise draws its "variety" from a single source,
    which just means every match looks like that source's opening
    repertoire. Each attempt picks a source file weighted by byte size
    (a bigger pool has proportionally more distinct games to offer), so
    pools aren't implicitly up- or down-weighted just by being listed
    first or being much smaller than the others."""
    if isinstance(pgn_paths, (str, Path)):
        pgn_paths = [pgn_paths]
    stats = [Path(p).stat() for p in pgn_paths]
    key_parts = [f"{p}|{st.st_mtime}|{st.st_size}"
                 for p, st in zip(pgn_paths, stats)]
    key = hashlib.sha256(
        ("||".join(key_parts) + f"||{n}|{min_ply}|{max_ply}|{seed}").encode()
    ).hexdigest()[:16]
    cache_path = Path(cache_dir) / f"openings_{key}.txt"
    if cache_path.exists():
        openings = cache_path.read_text().splitlines()
        print(f"  loaded {len(openings)} openings from cache {cache_path}",
              file=sys.stderr)
        return openings

    rng = random.Random(seed)
    file_sizes = [st.st_size for st in stats]
    total_size = sum(file_sizes) or 1
    # Precompute cumulative weights once for weighted source selection
    # per attempt, rather than re-summing on every pick.
    cum_weights = []
    running = 0
    for sz in file_sizes:
        running += sz
        cum_weights.append(running)

    openings = []
    attempts = 0
    max_attempts = n * 50 + 200  # generous; each attempt is one seek+one game parse
    file_handles = [open(p, "rb") for p in pgn_paths]
    try:
        while len(openings) < n and attempts < max_attempts:
            attempts += 1
            pick = rng.uniform(0, total_size)
            idx = next(i for i, cw in enumerate(cum_weights) if pick <= cw)
            fen = _sample_one_opening(file_handles[idx], file_sizes[idx],
                                       rng, min_ply, max_ply)
            if fen is not None:
                openings.append(fen)
    finally:
        for fb in file_handles:
            fb.close()

    if len(openings) < n:
        print(f"warning: only found {len(openings)} openings "
              f"(wanted {n}) across {pgn_paths}", file=sys.stderr)

    cache_path.parent.mkdir(parents=True, exist_ok=True)
    cache_path.write_text("\n".join(openings) + "\n" if openings else "")
    return openings


def play_one_game(white_engine_path, black_engine_path, scratch_dir, fen,
                   depth=None, white_time_sec=None, black_time_sec=None,
                   max_plies=200, post=False):
    """Returns (result, moves, move_evals, white_stats, black_stats) where
    result is 1.0/0.5/0.0 from White's perspective, moves is the list of
    UCI moves actually played (for PGN reconstruction), move_evals is a
    parallel list of (score_cp_white_relative, pv, depth) -- see its
    assignment below for None-handling -- and white_stats/black_stats are
    EngineStats instances covering that side's moves in this game.

    2026-10-03: the time control is now PER SIDE (white_time_sec /
    black_time_sec) rather than one value shared by both engines, so a
    time-odds match is expressible -- notably HEAD vs HEAD at 2:1, which
    measures what one extra ply of depth is actually worth in Elo *in
    this engine*, the calibration every depth-buying search change
    (LMR/EFP/LMP) is implicitly betting on. Pass the same value twice for
    an ordinary even match. `depth` (--sd) stays shared: fixed-depth odds
    would be a different experiment and there's no caller for it.

    NOTE for callers: these two parameters are keyed by SEAT (white/
    black), but in a time-odds match the odds must follow the ENGINE,
    which swaps seats every other game by design. main()'s run() does
    that mapping via candidate_is_white -- see the warning there."""
    moves = []
    # Parallel to `moves`: (score_cp_white_relative, pv, depth) per move
    # played, or (None, None, None) when no eval was captured (post=False,
    # or the engine printed no in-window iteration this move -- e.g. an
    # instant book/EGTB move). score is flipped to be WHITE-relative
    # (standard PGN eval-comment convention), unlike iScore itself, which
    # util.c prints from the SIDE-TO-MOVE's perspective.
    move_evals = []
    white_stats = EngineStats()
    black_stats = EngineStats()
    white_last_fmb = black_last_fmb = None
    # Tracked across the whole engine-interaction block below so the
    # BrokenPipeError handler (see try/except wrapping it) can report
    # which side/ply was in flight even when the pipe dies on a write
    # that isn't inside the per-ply move loop (e.g. one of the startup
    # sends, or a time-control send before the first ply).
    ply_idx = -1
    white_to_move = None
    # Default Engine() spawns with NO --hash/--cpus at all, which means
    # a tiny 65536-entry hash table (main.c's memset-zero default) --
    # far smaller than the 256m used everywhere else in this project's
    # own testing (test.sh). A small hash causes much worse move
    # ordering/more re-search in complex positions -- pass explicit
    # options here so gate-match search timing is representative
    # rather than an artifact of an unset default.
    # --logfile matters here, not just cosmetically: the default is a
    # shared "typhoon.log" in cwd (main.c's memset-zero default),
    # opened via BackupFile()-then-fopen(wb+) at startup and fflush()ed
    # on every Log()/Bug() call. With N concurrent engine processes (2
    # per game x --workers games) all doing that against the SAME
    # file, you get real rename races and lock contention -- observed
    # as unexplained match slowdown. "-" disables file logging
    # entirely (main.c: skips BackupFile/fopen altogether) -- fine
    # here since match_play.py never reads engine logs (moves come off
    # stdout, games get their own match_games.pgn).
    engine_args = ["--hash", "256m", "--cpus", "1", "--logfile", "-"]
    with Engine(white_engine_path, f"{scratch_dir}/w", engine_args) as we, \
         Engine(black_engine_path, f"{scratch_dir}/b", engine_args) as be:
        # Disable book and pondering.  Set time control.
        # 2026-10-03: Engine.__init__ (tune_eval_dna.py:120) sends
        # "nopost" unconditionally, and root.c:248 gates the
        # per-iteration PV lines on g_Options.fShouldPost -- so those
        # lines are never emitted and local_ebf (the TRUE per-ply
        # branching ratio, as opposed to the cumulative
        # nodes**(1/depth) `ebf` that conflates aspiration/root-PV
        # re-search cost with real tree growth) could never be computed.
        # It reported 0.000 for every engine in every match since it was
        # added. "post" re-enables those lines. Opt-in via --post rather
        # than default-on: it adds roughly one stdout line per iteration
        # per move, which every engine process must then be read through,
        # and the default behavior of a harness in active use should not
        # change silently underneath a running experiment.
        startup = ["book name /none",
                   "set ThinkOnOpponentsTime false",
                   "set PostThrottle 0.0"]
        if post:
            startup.append("post")
        for command in startup:
            _dbg("W", f"send: {command}")
            we._send(command)
            _dbg("B", f"send: {command}")
            be._send(command)
        # Time control is the one setting that can legitimately differ
        # between the two engines (time odds), so it is sent separately
        # rather than as part of the shared command list above.
        if depth is not None:
            white_tc = black_tc = f"sd {depth}"
        else:
            white_tc = f"st {white_time_sec}"
            black_tc = f"st {black_time_sec}"
        _dbg("W", f"send: {white_tc}")
        we._send(white_tc)
        _dbg("B", f"send: {black_tc}")
        be._send(black_tc)

        board = chess.Board(fen)
        for ply_idx in range(max_plies):
            if board.is_game_over(claim_draw=True):
                break
            white_to_move = board.turn == chess.WHITE
            mover = we if white_to_move else be
            tag = "W" if white_to_move else "B"
            mover_stats = white_stats if white_to_move else black_stats
            _dbg(tag, f"ply {ply_idx}: send force/setboard/go")
            mover._send("force")
            mover._send(f"setboard {board.fen()}")
            mover._send("go")
            move_uci = None
            # 2026-10-03 BUGFIX: the per-iteration PV lines _ITER_RE
            # matches are printed DURING the search, i.e. BEFORE the
            # "move" line -- so they arrive in THIS loop, not in the
            # trailing stats block below. They used to be parsed only
            # down there, which meant this loop read and discarded every
            # one of them and iter_ratio_count was always 0, so
            # avg_local_ebf silently reported 0.000 for every engine in
            # every match since it was added (2026-09-18). The whole
            # point of local_ebf is that it is a TRUE per-ply branching
            # ratio, unlike the cumulative nodes**(1/depth) `ebf` which
            # conflates aspiration-window/root-PV-change re-search cost
            # with real tree growth -- so the one metric that could
            # settle "is our branching factor actually ~3.9 or ~2.0"
            # was dead on arrival. Accumulate here as well.
            last_iter_depth = last_iter_nodes = None
            iter_log_ratio_sum = 0.0
            iter_ratio_count = 0
            # Last-seen (score, pv, depth) from any _ITER_RE line this move,
            # overwritten on every line -- so once the move-scan/stats loops
            # are done, this is the search's FINAL completed iteration: the
            # score/PV that actually back the move about to be pushed. Only
            # populated when "post" is on (see play_one_game's docstring);
            # stays None otherwise, which build_pgn_game treats as "no eval
            # available" rather than guessing.
            last_eval_score = None
            last_eval_pv = None

            def _note_iter(line):
                nonlocal last_iter_depth, last_iter_nodes
                nonlocal iter_log_ratio_sum, iter_ratio_count
                nonlocal last_eval_score, last_eval_pv
                im = _ITER_RE.match(line)
                if not im:
                    return
                iter_depth = int(im.group(1))
                iter_score = int(im.group(2))
                iter_nodes = int(im.group(3))
                tail = im.group(4).strip()
                if (last_iter_depth is not None and
                        last_iter_nodes is not None and
                        last_iter_nodes > 0 and
                        iter_nodes > last_iter_nodes and
                        iter_depth > last_iter_depth):
                    iter_log_ratio_sum += (math.log(iter_nodes / last_iter_nodes)
                                            / (iter_depth - last_iter_depth))
                    iter_ratio_count += 1
                last_iter_depth = iter_depth
                last_iter_nodes = iter_nodes
                # Fail-low lines ("FL --") carry no usable PV -- the move
                # that failed low is not necessarily the move eventually
                # played, so don't let a fail-low line's score overwrite a
                # real PV from an earlier, in-window iteration at this same
                # depth. Fail-high lines ("<SAN> ++") DO have a real PV
                # (the move that beat beta), so those are kept.
                if tail == "FL --":
                    return
                last_eval_score = iter_score
                last_eval_pv = tail[:-3].rstrip() if tail.endswith(" ++") else tail

            abort_reason = None
            for _ in range(400):
                line = mover.stdout.readline()
                if not line:
                    _dbg(tag, "EOF while scanning for move line")
                    abort_reason = "eof"
                    break
                _dbg(tag, f"read(move-scan): {line!r}")
                _note_iter(line)
                if line.startswith("tellics resign"):
                    # Engine resigned instead of moving -- root.c prints
                    # this and returns without ever printing "move ...",
                    # so stop reading now rather than blocking forever.
                    # This IS a legitimate decisive result (unlike the
                    # other abort_reason cases below) -- resignation is a
                    # real chess outcome, not a harness/engine failure.
                    abort_reason = "resign"
                    break
                m = _MOVE_RE.match(line)
                if m:
                    # root.c's "move ..." line is xboard/ICS-style
                    # (MoveToIcs, ics.c), which prints the promotion
                    # piece UPPERCASE ("e7e8Q") -- correct for that
                    # protocol, but chess.Move.from_uci() below enforces
                    # strict lowercase UCI and raises ValueError on it.
                    # Unlowered, EVERY promotion move got misclassified
                    # as illegal and the game aborted right as the
                    # promoting side (almost always the winning side in
                    # a long endgame) was about to seal the win -- this
                    # was the actual root cause of the ~47%-corrupted
                    # self-play corpus (docs/NNUE.md, 2026-10-17), not an
                    # engine stall/crash. Squares/files are already
                    # lowercase, so a blanket .lower() is safe here.
                    move_uci = m.group(1).lower()
                    break
            else:
                # Read 400 lines without ever seeing "move ..." (and
                # without EOF or resignation either) -- most likely an
                # extremely verbose iteration output (deep search in a
                # sparse endgame can print far more than 400 lines for a
                # single move) exceeding this loop's hard cap while the
                # engine is still alive and about to answer.
                abort_reason = "line-cap-exceeded"
            if move_uci is None:
                if abort_reason == "resign":
                    # side to move resigned -- legitimate loss for mover
                    result = 0.0 if white_to_move else 1.0
                    return (result, moves, move_evals, white_stats,
                            black_stats, None)
                # 2026-10-17: a real bug was found producing exactly this
                # path for a non-resign reason (eof / line-cap) on a
                # position where the mover was completely winning --
                # resignation was disabled for this corpus, so there is
                # no legitimate explanation for move_uci staying None
                # other than resign. The OLD code silently recorded this
                # as an ordinary loss for the mover, which corrupted
                # ~47% of a self-play training corpus with inverted
                # results (see docs/NNUE.md). Never do that again: signal
                # the caller with an aborted (result=None) game instead
                # of guessing a winner, and capture enough context here
                # to investigate later without needing to reproduce it
                # live.
                exit_code = mover.proc.poll()
                abort_info = {
                    "reason": abort_reason,
                    "mover": "white" if white_to_move else "black",
                    "ply": ply_idx,
                    "fen_before_ply": board.fen(),
                    "mover_exit_code": exit_code,
                    "mover_alive": exit_code is None,
                }
                _dbg(tag, f"ABORT (not scored): {abort_info}")
                return (None, moves, move_evals, white_stats, black_stats,
                        abort_info)

            # PostMoveSearchReport's stats block trails the "move" line
            # (confirmed against root.c and real output), not the other
            # way around -- keep reading a bounded number of further
            # lines to pick it up before moving on to the next ply.
            ply_nodes = ply_time = ply_depth = ply_fmb = None
            # 2026-09-18: local (per-iteration) branching factor, an
            # alternative to the cumulative nodes**(1/depth) EBF below
            # that isn't sensitive to which integer depth a move's
            # search happens to finish on -- see _ITER_RE's comment and
            # EngineStats' docstring for the full rationale. Track the
            # most recently seen (depth, nodes) pair; each time depth
            # actually advances (not just a same-depth aspiration
            # re-search line), record nodes_now/nodes_prev as one
            # branching-factor sample for this ply, geometric-mean them
            # together at the end (arithmetic mean of ratios would let
            # one huge outlier iteration dominate; geometric mean is
            # the correct way to average ratios/growth-factors).
            # (state hoisted above the move-scan loop -- see the
            # 2026-10-03 bugfix comment there)
            for _ in range(_STATS_TRAILING_LINES):
                line = _readline_or_none(mover.stdout, _STATS_IDLE_TIMEOUT_SEC)
                if line is None:
                    _dbg(tag, "idle timeout reading stats block -- "
                              "assuming report done (no ReportEnd marker, "
                              "e.g. an older binary like head_reference)")
                    break
                if not line:
                    _dbg(tag, "EOF while reading stats block")
                    break
                _dbg(tag, f"read(stats): {line!r}")
                if _REPORT_END_RE.match(line):
                    break  # PostMoveSearchReport's own end-of-report marker
                sm = _SEARCHED_RE.match(line)
                if sm:
                    ply_time = float(sm.group(1))
                    ply_nodes = int(sm.group(2))
                    continue
                dm = _DEPTH_RE.match(line)
                if dm:
                    ply_depth = int(dm.group(1))
                    continue
                fm = _FIRST_MOVE_BETA_RE.match(line)
                if fm:
                    ply_fmb = float(fm.group(1))
                    continue
                _note_iter(line)
            if ply_nodes is not None and ply_time is not None:
                ply_local_ebf = (math.exp(iter_log_ratio_sum / iter_ratio_count)
                                  if iter_ratio_count else None)
                mover_stats.add_ply(ply_nodes, ply_time, ply_depth,
                                    ply_local_ebf)
            if ply_fmb is not None:
                if white_to_move:
                    white_last_fmb = ply_fmb
                else:
                    black_last_fmb = ply_fmb

            move = None
            try:
                move = chess.Move.from_uci(move_uci)
                illegal = move not in board.legal_moves
            except ValueError:
                illegal = True
            if illegal:
                # The engine printed a "move ..." line, but it wasn't a
                # legal move -- same reasoning as the move_uci-is-None
                # case above: a real engine bug (producing garbage)
                # deserves the same abort-not-score treatment, not a
                # confidently-recorded loss with no diagnostic trail.
                exit_code = mover.proc.poll()
                abort_info = {
                    "reason": "illegal-move",
                    "mover": "white" if white_to_move else "black",
                    "ply": ply_idx,
                    "fen_before_ply": board.fen(),
                    "illegal_move_uci": move_uci,
                    "mover_exit_code": exit_code,
                    "mover_alive": exit_code is None,
                }
                _dbg(tag, f"ABORT (not scored): {abort_info}")
                return (None, moves, move_evals, white_stats, black_stats,
                        abort_info)
            assert move is not None  # illegal is False only when the try succeeded
            board.push(move)
            moves.append(move_uci)
            score_white = None
            if last_eval_score is not None:
                score_white = (last_eval_score if white_to_move
                               else -last_eval_score)
            move_evals.append((score_white, last_eval_pv, last_iter_depth))

        white_stats.add_game_first_move_beta(white_last_fmb, white_stats.plies)
        black_stats.add_game_first_move_beta(black_last_fmb, black_stats.plies)

        outcome = board.outcome(claim_draw=True)
        if outcome is None or outcome.winner is None:
            result = 0.5
        else:
            result = 1.0 if outcome.winner == chess.WHITE else 0.0
        return result, moves, move_evals, white_stats, black_stats, None


def _format_eval_comment(score_white, pv, depth):
    """Render one move_evals entry as a PGN comment, e.g. "+0.34/12 Nf3
    Nc6 Bb5" (score.pawns/depth pv...), the convention most PGN viewers
    (lichess, ChessBase, scid) already render as an eval graph/annotation.
    score_white is None when no eval was captured for this move (post=False,
    or an instant book/EGTB move) -- callers should skip the comment
    entirely rather than call this with None."""
    mate_in = None
    if abs(score_white) >= NMATE_THRESHOLD:
        # Ply-distance-to-mate isn't recoverable from a single iteration
        # line (it's baked into the score magnitude, not broken out), so
        # just flag the side rather than claim a specific "#N" we can't
        # verify -- see ADJUST_MATE_IN_N/hash.c in CLAUDE.md for why mate
        # scores need care generally.
        mate_in = "+M" if score_white > 0 else "-M"
    score_str = mate_in if mate_in else f"{score_white / 100:+.2f}"
    comment = f"{score_str}/{depth}" if depth is not None else score_str
    if pv:
        comment += f" {pv}"
    return comment


# chess.h: #define NMATE (+INFINITY_SCORE - 200) -- any |score| at or past
# this is a mate-distance score, not a material/positional evaluation.
NMATE_THRESHOLD = 32000


def build_pgn_game(start_fen, moves, result_white, white_name, black_name,
                    move_evals=None):
    board = chess.Board(start_fen)
    game = chess.pgn.Game()
    game.headers["Event"] = "typhoon self-play match"
    game.headers["White"] = white_name
    game.headers["Black"] = black_name
    game.headers["FEN"] = start_fen
    game.headers["Date"] = f"{date.today()}"
    game.setup(board)
    # result_white is None for an ABORTED game (play_one_game's mover
    # failed to produce a legal move for a non-resign reason -- see its
    # abort_info return value) -- write a standard "unknown result" PGN
    # tag rather than guessing a winner. fit_k_and_label.py's
    # result_to_white_score() already treats "*" as "unknown, skip this
    # game" for training purposes, so this alone keeps an aborted game
    # out of any future training corpus without touching that script.
    result_str = ({1.0: "1-0", 0.0: "0-1", 0.5: "1/2-1/2"}[result_white]
                  if result_white is not None else "*")
    game.headers["Result"] = result_str
    node = game
    for i, move_uci in enumerate(moves):
        node = node.add_variation(chess.Move.from_uci(move_uci))
        if move_evals is not None:
            score_white, pv, depth = move_evals[i]
            if score_white is not None:
                node.comment = _format_eval_comment(score_white, pv, depth)
    return game, result_str


def elo_diff(score, n):
    """Rough Elo estimate + a 95%-ish error bar from a win rate, treating
    games as independent Bernoulli-ish trials on the score. Good enough
    as a signal, not a substitute for a real SPRT."""
    score = min(max(score, 1e-6), 1 - 1e-6)
    elo = -400 * math.log10(1 / score - 1)
    # standard error on the score, propagated through the logit
    se_score = math.sqrt(score * (1 - score) / max(n, 1))
    se_elo = (400 / math.log(10)) * se_score / (score * (1 - score))
    return elo, 1.96 * se_elo


class EngineStats:
    """Per-engine (not per-color -- candidate/baseline swap sides every
    other game) search-quality accumulator: speed and tree shape, pulled
    from the same PostMoveSearchReport block every move already prints
    (see _SEARCHED_RE/_DEPTH_RE/_FIRST_MOVE_BETA_RE), not a separate
    benchmark. EBF is computed the same way script.c's suite runs do
    (nodes**(1/depth) per move, see head_reference notes on why this can
    diverge from first-move-beta-cutoff-rate when pruning/aspiration-
    window re-searches inflate node counts independent of true tree
    shape), then averaged across all moves an engine played.

    2026-09-18: that cumulative nodes**(1/depth) EBF turned out to be
    misleading when comparing two engines with different lazy-eval-exit
    aggressiveness -- one engine measured LOWER nps and HIGHER avg_depth
    than the other while ALSO reporting HIGHER ebf, which is not
    reconcilable if ebf is really "nodes needed per unit depth" (lower
    nps + higher depth implies fewer total nodes were needed to get
    there, which should mean lower ebf, not higher). Root cause: this
    formula uses the SEARCH-WIDE cumulative node count raised to
    1/(the single integer depth reported at the end), so it's extremely
    sensitive to exactly which whole ply a move's search happens to
    finish on within its time budget -- a difference in time-management
    timing that pushes a move to complete one more (or one fewer) full
    iteration can swing this number a lot with no real change in tree
    efficiency, and this effect doesn't reliably average out because
    node-count magnitude varies hugely across game phases/positions.
    `local_ebf` (see add_ply) fixes this by computing depth-normalized
    branching ratios from CONSECUTIVE completed-iteration node counts
    within a single move's search (nodes_now/nodes_prev, depth-delta-
    normalized, geometric-meaned across a move's iterations, then
    arithmetic-meaned across moves same as avg_ebf) -- a true local
    branching-factor measurement, insensitive to which absolute depth
    the search happened to stop at. Kept the old ebf/avg_ebf fields
    alongside rather than replacing them, since the cumulative version
    is still what script.c's own suite runs report and there's value in
    being able to cross-check the two against each other. First-move
    beta cutoff rate is read as a running total inside a single engine
    process's lifetime (root.c never resets those counters mid-process),
    and since play_one_game spawns a fresh process per game, the value
    from an engine's last move of a game is already that whole game's
    rate -- summed here across games as a plies-weighted average so a
    long game doesn't count the same as a short one.
    """

    def __init__(self):
        self.total_nodes = 0
        self.total_time = 0.0
        self.plies = 0
        self.ebf_sum = 0.0
        self.ebf_count = 0
        self.local_ebf_sum = 0.0
        self.local_ebf_count = 0
        self._fmb_weighted_sum = 0.0
        self._fmb_weight = 0
        self.depth_sum = 0
        self.depth_count = 0

    def add_ply(self, nodes, elapsed, depth, local_ebf=None):
        self.total_nodes += nodes
        self.total_time += elapsed
        self.plies += 1
        if depth and depth > 0:
            self.depth_sum += depth
            self.depth_count += 1
        if depth and depth > 0 and nodes > 0:
            self.ebf_sum += nodes ** (1.0 / depth)
            self.ebf_count += 1
        if local_ebf is not None:
            self.local_ebf_sum += local_ebf
            self.local_ebf_count += 1

    def add_game_first_move_beta(self, pct, plies_in_game):
        if pct is not None and plies_in_game > 0:
            self._fmb_weighted_sum += pct * plies_in_game
            self._fmb_weight += plies_in_game

    def merge(self, other):
        self.total_nodes += other.total_nodes
        self.total_time += other.total_time
        self.plies += other.plies
        self.ebf_sum += other.ebf_sum
        self.ebf_count += other.ebf_count
        self.local_ebf_sum += other.local_ebf_sum
        self.local_ebf_count += other.local_ebf_count
        self._fmb_weighted_sum += other._fmb_weighted_sum
        self._fmb_weight += other._fmb_weight
        self.depth_sum += other.depth_sum
        self.depth_count += other.depth_count

    @property
    def nps(self):
        return self.total_nodes / self.total_time if self.total_time else 0.0

    @property
    def avg_ebf(self):
        return self.ebf_sum / self.ebf_count if self.ebf_count else 0.0

    @property
    def avg_local_ebf(self):
        return (self.local_ebf_sum / self.local_ebf_count
                if self.local_ebf_count else 0.0)

    @property
    def avg_first_move_beta(self):
        return (self._fmb_weighted_sum / self._fmb_weight
                if self._fmb_weight else 0.0)

    @property
    def avg_depth(self):
        return self.depth_sum / self.depth_count if self.depth_count else 0.0

    def summary(self):
        return (f"nps={self.nps:.0f} ebf={self.avg_ebf:.3f} "
                f"local_ebf={self.avg_local_ebf:.3f} "
                f"depth={self.avg_depth:.2f} "
                f"first_move_beta={self.avg_first_move_beta:.1f}% "
                f"nodes={self.total_nodes} plies={self.plies}")


def elo_to_score(elo):
    """Expected per-game score (0..1) for a side rated `elo` points above
    an opponent, via the standard logistic Elo model."""
    return 1.0 / (1.0 + 10.0 ** (-elo / 400.0))


def sprt_bar(llr, la, lb, width=9):
    """Render an ASCII gauge of where `llr` sits between the H0 (`la`) and
    H1 (`lb`) SPRT bounds, e.g. '|---------|V---------|' with V marking
    the rounded llr position."""
    frac = 0.5 if lb == la else (llr - la) / (lb - la)
    frac = min(max(frac, 0.0), 1.0)
    slot = round(frac * (2 * width + 3))
    bar = "|" + "-" * width + "|" + "-" * width + "|"
    idx = slot + 1
    return bar[:idx - 1] + "V" + bar[idx:]


class Sprt:
    """Sequential Probability Ratio Test for engine-vs-engine gating, same
    formulation fishtest/cutechess-cli use for exactly this problem: two
    Elo hypotheses (H0: true strength gap is elo0 or worse, H1: elo1 or
    better), tested via the log-likelihood ratio of a normal
    approximation to the per-game trinomial (W/D/L) score, with variance
    re-estimated from the running W/D/L mix as games accumulate.

    Unlike a fixed-N test (this script's original LOWER95 gate), this
    stops as soon as the evidence is conclusive in either direction --
    a handful of games if the true effect is large, potentially tens of
    thousands if it's genuinely marginal -- rather than committing to
    one game count up front regardless of the actual effect size.

    See CLAUDE.md / the "how many games" discussion this was built from:
    for a 95%/80%-power fixed-N test, resolving a 5 Elo gap needs
    ~35-40k games, 10 Elo needs ~9-10k, 20-30 Elo needs ~1-2.5k. SPRT
    doesn't remove that fundamental cost, it just avoids overpaying for
    it when the true effect is far from the boundary either way.
    """

    def __init__(self, elo0, elo1, alpha=0.05, beta=0.05):
        self.elo0 = elo0
        self.elo1 = elo1
        self.s0 = elo_to_score(elo0)
        self.s1 = elo_to_score(elo1)
        self.la = math.log(beta / (1 - alpha))
        self.lb = math.log((1 - beta) / alpha)
        self.wins = self.draws = self.losses = 0

    def update(self, result):
        """result: 1.0 (candidate win), 0.5 (draw), or 0.0 (candidate loss)."""
        if result == 1.0:
            self.wins += 1
        elif result == 0.0:
            self.losses += 1
        else:
            self.draws += 1

    @property
    def n(self):
        return self.wins + self.draws + self.losses

    def llr(self):
        n = self.n
        if n == 0:
            return 0.0
        mean = (self.wins + 0.5 * self.draws) / n
        var = (self.wins * (1 - mean) ** 2 +
               self.draws * (0.5 - mean) ** 2 +
               self.losses * (0 - mean) ** 2) / n
        if var <= 0:
            # every game so far had the identical result -- LLR is only
            # defined once there's some spread; treat as inconclusive.
            return 0.0
        return n * (self.s1 - self.s0) * (mean - (self.s0 + self.s1) / 2) / var

    def decision(self):
        """Returns 'H1' (accept candidate is >= elo1 better), 'H0' (accept
        candidate is <= elo0, i.e. not a real improvement), or None
        (inconclusive, keep playing)."""
        llr = self.llr()
        if llr >= self.lb:
            return "H1"
        if llr <= self.la:
            return "H0"
        return None


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("head_engine", help="path to the baseline/head typhoon binary")
    ap.add_argument("candidate_engine", help="path to the candidate typhoon binary")
    ap.add_argument("--pgn", required=True, nargs="+",
                     help="one or more pgn pools to sample starting "
                          "positions from (e.g. --pgn ../../pgn/"
                          "twic_filtered.pgn ../../pgn/self_play_games.pgn). "
                          "Each candidate opening's source file is picked "
                          "weighted by that file's byte size, so listing "
                          "several pools widens the set of starting "
                          "positions instead of implicitly favoring "
                          "whichever is listed first or is largest/"
                          "smallest.")
    ap.add_argument("--games", type=int, default=500,
                     help="total games; rounded down to an even number "
                          "since each opening is played as a pair "
                          "(candidate White / candidate Black). With "
                          "--sprt, this is a safety cap instead of a "
                          "fixed target -- the run stops as soon as the "
                          "SPRT concludes, or here if it never does; "
                          "pass something generous (e.g. 20000) when "
                          "using --sprt.")
    ap.add_argument("--workers", type=int, default=8)
    ap.add_argument("--sprt", action="store_true",
                     help="stop as soon as a Sequential Probability "
                          "Ratio Test concludes, instead of always "
                          "playing exactly --games games. See the "
                          "'how many games' discussion in CLAUDE.md/"
                          "session history for why this matters: a "
                          "fixed-N test either overpays for a large, "
                          "obvious effect or is simply too small to "
                          "resolve a marginal one -- SPRT adapts to "
                          "whichever it turns out to be.")
    ap.add_argument("--elo0", type=float, default=0.0,
                     help="SPRT H0: candidate is at most this many Elo "
                          "better than baseline (the 'no real "
                          "improvement' hypothesis)")
    ap.add_argument("--elo1", type=float, default=5.0,
                     help="SPRT H1: candidate is at least this many Elo "
                          "better than baseline (the 'real improvement' "
                          "hypothesis)")
    ap.add_argument("--alpha", type=float, default=0.05,
                     help="SPRT false-positive rate (probability of "
                          "accepting H1 when H0 is actually true)")
    ap.add_argument("--beta", type=float, default=0.05,
                     help="SPRT false-negative rate (probability of "
                          "accepting H0 when H1 is actually true)")
    tc = ap.add_mutually_exclusive_group()
    tc.add_argument("--sd", type=int, default=None,
                     help="fixed search depth per move for both sides "
                          "(default when neither --sd nor --st is given; "
                          "avoids machine-load noise -- see CLAUDE.md's "
                          "sd-over-st guidance). NOTE: --sd used to "
                          "default to 8 even when --st was passed, which "
                          "silently ignored --st entirely -- fixed so "
                          "the two are actually mutually exclusive now.")
    tc.add_argument("--st", type=float, default=None,
                     help="seconds per move per side instead of fixed depth")
    # Time odds. See play_one_game's docstring for why this exists: the
    # headline use is HEAD vs HEAD at 2:1 to price one extra ply of
    # depth in Elo for THIS engine, which bounds what any depth-buying
    # search change can ever be worth.
    ap.add_argument("--odds", type=float, default=None, metavar="MULT",
                     help="give the CANDIDATE engine MULT times the "
                          "head engine's seconds per move (e.g. --st 4 "
                          "--odds 2 => head 4s, candidate 8s). Requires "
                          "--st; incompatible with --sd. Use --st-head/"
                          "--st-candidate instead for full control.")
    ap.add_argument("--st-head", type=float, default=None, metavar="SEC",
                     help="explicit seconds per move for the head engine "
                          "(overrides --st for that side)")
    ap.add_argument("--st-candidate", type=float, default=None, metavar="SEC",
                     help="explicit seconds per move for the candidate "
                          "engine (overrides --st for that side)")
    ap.add_argument("--post", action="store_true",
                     help="send 'post' so engines emit per-iteration PV "
                          "lines, which is what local_ebf (the true "
                          "per-ply branching ratio) is computed from. "
                          "Without this, local_ebf reads 0.000 -- the "
                          "engines never print those lines because "
                          "Engine.__init__ sends 'nopost'.")
    ap.add_argument("--candidate-name", default="candidate",
                     help="PGN White/Black name recorded for the "
                          "candidate engine (default: 'candidate'). "
                          "2026-10-16: set this to something identifying "
                          "the eval/config actually used (e.g. "
                          "'candidate_nnue_loose') whenever the engine "
                          "isn't a plain HCE build -- self_play_games.pgn "
                          "is reused as NNUE training data later, and a "
                          "generic 'candidate' name makes it impossible "
                          "to tell which games came from a differently-"
                          "calibrated evaluator (whose scores shouldn't "
                          "be trusted as training labels) without "
                          "externally remembering run dates.")
    ap.add_argument("--baseline-name", default="baseline",
                     help="PGN White/Black name recorded for the "
                          "baseline/head engine (default: 'baseline'); "
                          "same reasoning as --candidate-name.")
    ap.add_argument("--no-color-swap", action="store_true",
                     help="play each sampled opening ONCE instead of "
                          "twice with colors swapped between head/"
                          "candidate. The swap exists to cancel out "
                          "color-dependent FIRST-MOVE ADVANTAGE in an "
                          "Elo-gating match; it is pure waste -- a byte-"
                          "for-byte duplicate game, just relabeled --  "
                          "whenever head_engine and candidate_engine are "
                          "the SAME binary (as every self-play "
                          "training-data run via "
                          "hce_training_corpus.sh is), because search "
                          "here is deterministic (--cpus 1) given a "
                          "fixed FEN+side-to-move, so swapping which "
                          "name is 'white'/'black' changes no move "
                          "actually played. Confirmed empirically "
                          "2026-10-16: consecutive game pairs in "
                          "nnue_self_play_training_data.pgn were exact "
                          "duplicates (identical FEN, identical move "
                          "list). Use this flag for self-play corpus "
                          "generation -- sample --games distinct "
                          "openings and play each once, rather than "
                          "--games/2 openings played twice -- and keep "
                          "the swap (the default) for real head-vs-"
                          "candidate Elo gating, where two different "
                          "binaries need it to cancel color bias.")
    ap.add_argument("--max-plies", type=int, default=200)
    ap.add_argument("--scratch", default="/usr/local/tmp/typhoon_match")
    ap.add_argument("--log", default="match_history.log")
    ap.add_argument("--pgn-out", default="../self_play_games.pgn",
                     help="every played game is appended here as PGN")
    ap.add_argument("--opening-seed", type=int, default=None, metavar="N",
                     help="seed for opening-position sampling. Default: "
                          "a fresh random seed every run, so successive "
                          "invocations draw a DIFFERENT set of starting "
                          "positions from --pgn rather than silently "
                          "replaying whatever sample a previous run "
                          "already cached under opening_cache/ for these "
                          "exact (pgn, n, min_ply, max_ply) params -- that "
                          "caching exists to make a single run's sampling "
                          "cheap to redo (e.g. a crashed/resumed run), not "
                          "to pin one fixed opening set across unrelated "
                          "runs. Pass an explicit value to reproduce a "
                          "specific run's exact opening set.")
    args = ap.parse_args()
    if args.opening_seed is None:
        # SystemRandom, not random.Random/time-based: back-to-back runs
        # (e.g. a shell loop launching several matches within the same
        # second) must not collide on the same seed and silently resample
        # the same "random" set.
        args.opening_seed = random.SystemRandom().randrange(2**31)
    print(f"OPENING_SEED={args.opening_seed}", file=sys.stderr)

    # --- resolve the per-side time controls once, loudly ---------------
    # Deliberately strict: a time-odds match that silently degrades into
    # an even one is the dangerous failure here, because it still scores
    # ~0.500 and looks perfectly healthy (same family of silent confound
    # as the still-live opening book and the shared logfile this script's
    # header documents). So every inconsistent combination is a hard
    # error rather than a best-effort interpretation.
    odds_requested = (args.odds is not None or
                      args.st_head is not None or
                      args.st_candidate is not None)
    if odds_requested and args.sd is not None:
        ap.error("time odds (--odds/--st-head/--st-candidate) needs a "
                 "time control, not --sd; fixed-depth odds would be a "
                 "different experiment entirely.")
    if args.sd is None and args.st is None and not odds_requested:
        args.sd = 8
    if args.odds is not None:
        if args.st is None:
            ap.error("--odds multiplies --st, so --st is required with it")
        if args.st_candidate is not None:
            ap.error("--odds and --st-candidate both set the candidate's "
                     "time; pass one or the other")
        if args.odds <= 0:
            ap.error("--odds must be positive")
    head_time = args.st_head if args.st_head is not None else args.st
    cand_time = args.st_candidate if args.st_candidate is not None else args.st
    if args.odds is not None:
        # round: 4 * 1.5 is 6.000000000000001 in binary float, which ends
        # up verbatim in the "st" command and in the result line's TC=
        # stamp. Harmless to the engine, ugly in a log that gets diffed.
        cand_time = round(args.st * args.odds, 4)
    if args.sd is None:
        if head_time is None or cand_time is None:
            ap.error("need --st (or both --st-head and --st-candidate) "
                     "when not using --sd")
        if odds_requested and head_time == cand_time:
            ap.error(f"time odds requested but both sides resolved to "
                     f"{head_time}s -- that is an even match. Check "
                     f"--odds/--st-head/--st-candidate.")
        print(f"TIME_CONTROL: head={head_time}s/move "
              f"candidate={cand_time}s/move"
              + ("  *** TIME ODDS ***" if head_time != cand_time else ""))
    else:
        print(f"TIME_CONTROL: both sides sd {args.sd}")
    # Stash on args so print_final_summary (a separate function, already
    # taking args) can stamp the time control into the result line
    # without widening its signature at both call sites.
    args.head_time = head_time
    args.cand_time = cand_time

    head_engine = str(Path(args.head_engine).resolve())
    candidate_engine = str(Path(args.candidate_engine).resolve())

    if args.no_color_swap:
        n_openings = args.games
    else:
        n_openings = args.games // 2
    openings = sample_openings(args.pgn, n_openings, seed=args.opening_seed)

    jobs = []
    for fen in openings:
        seats = (True,) if args.no_color_swap else (True, False)
        for candidate_is_white in seats:
            i = len(jobs)
            white_engine = candidate_engine if candidate_is_white else head_engine
            black_engine = head_engine if candidate_is_white else candidate_engine
            jobs.append((i, fen, candidate_is_white, white_engine, black_engine))

    candidate_points = 0.0
    wins = draws = losses = 0
    aborted = 0
    sprt = Sprt(args.elo0, args.elo1, args.alpha, args.beta) if args.sprt else None
    sprt_decision = None
    candidate_stats = EngineStats()
    head_stats = EngineStats()

    def run(job):
        i, fen, candidate_is_white, white_engine, black_engine = job
        game_start = time.monotonic()
        # The odds follow the ENGINE, not the seat. Every opening is
        # played twice with colours swapped (see the jobs[] loop above),
        # so keying the time control on White/Black instead would hand
        # the advantage to each engine in exactly half the games -- a
        # balanced match that measures nothing while scoring a healthy-
        # looking ~0.500. Hence this mapping off candidate_is_white, and
        # the per-game assertion below.
        white_time = cand_time if candidate_is_white else head_time
        black_time = head_time if candidate_is_white else cand_time
        if args.sd is None and head_time != cand_time:
            assert ((white_time == cand_time) == candidate_is_white), \
                "time-odds seat mapping inverted"
        try:
            result_white, moves, move_evals, white_stats, black_stats, abort_info = (
                play_one_game(
                    white_engine, black_engine, f"{args.scratch}/g{i}", fen,
                    depth=args.sd,
                    white_time_sec=white_time, black_time_sec=black_time,
                    max_plies=args.max_plies, post=args.post,
                )
            )
        except (BrokenPipeError, OSError) as exc:
            # One of the two engine subprocesses died mid-game (crashed,
            # was OOM-killed, etc.) -- play_one_game's own abort_info
            # machinery only covers READ-side failures (EOF/line-cap
            # while scanning for a move); a WRITE to an already-dead
            # process's stdin raises BrokenPipeError straight out of
            # Engine._send instead, uncaught anywhere below this. Left
            # unhandled, that exception propagates out of this worker
            # thread's run() through the Future and kills the entire
            # match (every other in-flight/queued game lost) over a
            # single flaky engine process -- confirmed live against a
            # polyglot/Stockfish side via stockfish_xboard. Treat it the
            # same way as any other abort: no result guessed, game
            # skipped, run continues.
            result_white = None
            moves, move_evals = [], []
            white_stats, black_stats = EngineStats(), EngineStats()
            abort_info = {
                "reason": "engine-pipe-broken",
                "exception": f"{type(exc).__name__}: {exc}",
            }
        game_elapsed = time.monotonic() - game_start
        # result_white is None for an aborted game (play_one_game's mover
        # failed to produce a legal move for a non-resign reason) -- do
        # NOT guess a winner (see abort_info's own comment at the
        # return sites in play_one_game for why this matters: it's
        # exactly the bug that silently corrupted ~47% of an earlier
        # self-play corpus). Propagate None through candidate_result too
        # so run()'s caller can skip it rather than score it.
        candidate_result = (
            None if result_white is None else
            result_white if candidate_is_white else (1.0 - result_white)
        )
        white_name = args.candidate_name if candidate_is_white else args.baseline_name
        black_name = args.baseline_name if candidate_is_white else args.candidate_name
        game, _ = build_pgn_game(fen, moves, result_white, white_name, black_name,
                                  move_evals=move_evals)
        this_candidate_stats = white_stats if candidate_is_white else black_stats
        this_head_stats = black_stats if candidate_is_white else white_stats
        return (i, candidate_result, game, game_elapsed,
                this_candidate_stats, this_head_stats, abort_info)

    game_durations = []
    job_iter = iter(jobs)
    done = 0
    # Games are processed in *completion* order (as_completed/wait), not
    # submission order -- and decisive games plausibly finish faster than
    # grindy draws/losses do (a winning side often wraps up well before
    # --max-plies; a losing/drawing side tends to run long, sometimes all
    # the way to the ply cap). That means naively printing a running
    # candidate_points/done as results arrive is a biased mid-run
    # estimator: wins arrive disproportionately early, so it reads high
    # at first and erodes toward the true value as slower non-win games
    # trickle in -- a real, reproducible artifact, not noise, and not
    # evidence the actual match methodology is broken (the *final*
    # score, summed over the complete set, is order-independent). Fixed
    # by buffering out-of-order completions and only advancing the
    # printed "score so far" through jobs in their original submission
    # order (pending[]/next_report_idx below), so the progress line
    # reflects an honest prefix rather than a completion-order-biased
    # sample.
    pending = {}
    next_report_idx = 0
    reported_points = 0.0
    reported_count = 0

    with ThreadPoolExecutor(max_workers=args.workers) as pool, \
         open(args.log, "a") as log, open(args.pgn_out, "a") as pgn_out:
        in_flight = set()
        for _ in range(args.workers):
            job = next(job_iter, None)
            if job is None:
                break
            in_flight.add(pool.submit(run, job))

        try:
            while in_flight:
                finished, in_flight = wait(in_flight, return_when=FIRST_COMPLETED)
                for fut in finished:
                    (i, candidate_result, game, game_elapsed,
                     this_candidate_stats, this_head_stats, abort_info) = fut.result()
                    print(game, file=pgn_out, end="\n\n")
                    pgn_out.flush()

                    if candidate_result is None:
                        # Aborted game (play_one_game's mover failed to
                        # produce a legal move for a non-resign reason --
                        # see abort_info). Do NOT score it (no wins/
                        # losses/draws/candidate_points/SPRT update, no
                        # contribution to the `done` denominator used for
                        # score/ETA) -- this is exactly the "silently
                        # record a decisive result with no diagnostic
                        # trail" bug that corrupted ~47% of an earlier
                        # self-play corpus (docs/NNUE.md, 2026-10-17).
                        # The game is still written to pgn_out above
                        # (Result "*", which fit_k_and_label.py's
                        # result_to_white_score() already treats as
                        # "unknown, skip" for training) so it isn't lost
                        # for post-mortem debugging, just not scored.
                        aborted += 1
                        log.write(f"game {i}: ABORTED (not scored) "
                                  f"abort_info={abort_info} "
                                  f"elapsed={game_elapsed:.1f}s\n")
                        log.flush()
                        print(f"  game {i}: ABORTED -- {abort_info}",
                              file=sys.stderr)
                        continue

                    done += 1
                    candidate_points += candidate_result
                    game_durations.append(game_elapsed)
                    candidate_stats.merge(this_candidate_stats)
                    head_stats.merge(this_head_stats)
                    if candidate_result == 1.0:
                        wins += 1
                    elif candidate_result == 0.0:
                        losses += 1
                    else:
                        draws += 1
                    if sprt is not None:
                        sprt.update(candidate_result)
                    log.write(f"game {i}: candidate_result={candidate_result} "
                              f"elapsed={game_elapsed:.1f}s\n")
                    log.flush()

                    pending[i] = candidate_result
                    while next_report_idx in pending:
                        reported_points += pending.pop(next_report_idx)
                        reported_count += 1
                        next_report_idx += 1

                    if done % 10 == 0:
                        avg_game_sec = sum(game_durations) / len(game_durations)
                        eta_sec = (avg_game_sec *
                                   max(len(jobs) - done, 0)) / args.workers
                        score = candidate_points / done
                        sprt_note = ""
                        if sprt is not None:
                            sprt_note = (f": llr={sprt.llr():+.2f}, "
                                         f"H0={sprt.la:+.2f}"
                                         f"{sprt_bar(sprt.llr(), sprt.la, sprt.lb)}"
                                         f"H1={sprt.lb:+.2f}")
                        print(f"{done} games: -{losses} ={draws} +{wins} "
                              f"({avg_game_sec:.1f}s avg, eta={eta_sec/3600:.0f}hrs, "
                              f"score={score:.3f}){sprt_note}",
                              file=sys.stderr)
                        print(f"  avg depth: candidate={candidate_stats.avg_depth:.2f} "
                              f"head={head_stats.avg_depth:.2f} "
                              f"(nps: candidate={candidate_stats.nps:.0f} "
                              f"head={head_stats.nps:.0f})",
                              file=sys.stderr)
                        print(f"  avg ebf:   candidate={candidate_stats.avg_ebf:.3f} "
                              f"head={head_stats.avg_ebf:.3f} "
                              f"(local: candidate={candidate_stats.avg_local_ebf:.3f} "
                              f"head={head_stats.avg_local_ebf:.3f})",
                              file=sys.stderr)

                if sprt is not None and sprt_decision is None:
                    sprt_decision = sprt.decision()
                    if sprt_decision is not None:
                        print(f"  SPRT concluded: {sprt_decision} after "
                              f"{sprt.n} games (llr={sprt.llr():+.3f})",
                              file=sys.stderr)

                if sprt_decision is None:
                    while len(in_flight) < args.workers:
                        job = next(job_iter, None)
                        if job is None:
                            break
                        in_flight.add(pool.submit(run, job))
                # else: stop topping up, just let whatever's already
                # in_flight finish naturally -- no wasted work beyond games
                # already launched before the decision landed.
        except KeyboardInterrupt:
            print("\n^C caught -- dumping stats for the "
                  f"{done} game(s) completed so far, then exiting "
                  "(not waiting for in-flight games to finish naturally):",
                  file=sys.stderr)
            print_final_summary(done, candidate_points, wins, draws,
                                 losses, aborted, candidate_stats, head_stats,
                                 sprt, sprt_decision, args)
            # os._exit, not sys.exit: skips the `with` block's normal
            # ThreadPoolExecutor teardown (wait=True by default -- would
            # block until every in-flight game's subprocess-driven game
            # loop finishes on its own, which is the exact wait a ^C is
            # trying to avoid) and any atexit/finally cleanup. Log/pgn
            # writes are already flushed after every completed game
            # inside the loop, so nothing buffered is lost by skipping
            # the file objects' own __exit__.
            os._exit(130)

    print_final_summary(done, candidate_points, wins, draws, losses, aborted,
                         candidate_stats, head_stats, sprt, sprt_decision,
                         args)


def print_final_summary(done, candidate_points, wins, draws, losses, aborted,
                         candidate_stats, head_stats, sprt, sprt_decision,
                         args):
    """Shared by the normal end-of-match path and the ^C handler below --
    same CANDIDATE_SCORE/CANDIDATE_STATS/HEAD_STATS block either way, just
    computed from whatever `done` games actually completed. `aborted`
    (not included in `done` or in any win/draw/loss/score arithmetic) is
    the count of games where a mover failed to produce a legal move for
    a non-resign reason -- see play_one_game's abort_info. Surfaced here
    so an elevated count is visible instead of silently discarded."""
    n_games = done
    score = candidate_points / n_games if n_games else 0.5
    elo, err = elo_diff(score, n_games)
    # 95% normal-approximation CI directly on the score (not the elo
    # transform, which distorts near 0/1) -- this is what the gate
    # should key off of: a point estimate >= 0.5 is not evidence of
    # improvement on its own when search is non-deterministic (see
    # MP=1 multithreaded search noise observed even with identical
    # DNA on both sides), only a lower bound that clears 0.5 is.
    se_score = math.sqrt(score * (1 - score) / max(n_games, 1))
    lower95 = score - 1.96 * se_score
    sprt_str = ""
    if sprt is not None:
        outcome = sprt_decision or "INCONCLUSIVE"
        sprt_str = (f" SPRT={outcome} LLR={sprt.llr():+.3f} "
                    f"BOUNDS=[{sprt.la:.3f},{sprt.lb:.3f}] "
                    f"ELO0={args.elo0:.1f} ELO1={args.elo1:.1f}")
    # Stamp the time control into the result line itself: a time-odds
    # score is NOT comparable with an even-match score, and a bare
    # CANDIDATE_SCORE in a log file gives no way to tell them apart
    # later.
    if args.sd is not None:
        tc_str = f" TC=sd{args.sd}"
    elif args.head_time != args.cand_time:
        tc_str = (f" TC=ODDS(head={args.head_time}s,"
                  f"cand={args.cand_time}s)")
    else:
        tc_str = f" TC=st{args.head_time}"
    aborted_str = f" ABORTED={aborted}" if aborted else ""
    print(f"CANDIDATE_SCORE={score:.4f} GAMES={n_games} "
          f"WINS={wins} DRAWS={draws} LOSSES={losses} "
          f"ELO={elo:+.1f}+/-{err:.1f} LOWER95={lower95:.4f}"
          f"{tc_str}{sprt_str}{aborted_str}")
    print(f"CANDIDATE_STATS: {candidate_stats.summary()}")
    print(f"HEAD_STATS:      {head_stats.summary()}")


if __name__ == "__main__":
    main()