Spaces:

chesstb
/

explorer

Running

Deploy ChessTB Explorer on Hugging Face Spaces
e16348e 12 days ago
96.8 kB
"""
app.py: ChessTB Explorer back-end
A single-file Flask application that probes a ChessTB endgame tablebase
for a submitted FEN position and returns every legal move ranked by
Distance to Zeroing (DTZ), Distance to Conversion (DTC), Distance to
Mate (DTM), and DTM under the 50-move rule (DTM50).
DTC is 50-move-rule-aware in the same way DTM50 is (it is always probed
at the position's own halfmove clock), but unlike DTM50 it has no
"flat"/clock-independent sibling column of its own -- there is only ever
one DTC value for a position. A ``.lzdtc`` table only exists for material
with at least one pawn; for pawnless material `chess.chesstb` derives an
equivalent answer from DTZ/WDL instead (see chess.chesstb.ProbeResult),
so `has_dtc` is False only when a pawnful material's own
`.lzdtc` file genuinely hasn't been generated/shipped.
Routes
------
  POST /probe              - evaluate a FEN position (JSON)
  POST /probe/stream       - evaluate with SSE streaming progress
  GET  /                   - main UI
  GET  /health             - readiness check
  GET  /admin              - cache dashboard HTML page, or a login form if not authenticated
  POST /admin/login        - start an admin session (JSON {token})
  POST /admin/logout       - end the admin session
  POST /admin/cache/clear  - purge both LRU caches
  GET  /admin/cache/stats  - cache hit-rate statistics
  GET  /openapi.yaml       - OpenAPI 3.0 specification
/admin and /admin/cache/* require admin authentication -- either a
session cookie started via /admin/login, or an "Authorization: Bearer
<ADMIN_TOKEN>" header. See config.py's "Admin" section and README.md's
"Security notes" for how ADMIN_TOKEN is configured.
/probe and /probe/stream are rate-limited per client IP (config.py's
PROBE_RATE_LIMIT); a client over the limit gets a 429 with a JSON body.
/admin/login has its own, separate limit (config.py's
ADMIN_LOGIN_RATE_LIMIT), covering both failed and successful attempts.
Each move entry includes a child_fen so the client can pre-fetch the next
probe, and each response includes a summary of wins/draws/losses/unknown
across all legal moves. A position not covered by the loaded tablebase
raises MissingTableError, which is returned as structured JSON
({error_code, piece_count}). Root-JSON and child-probe cache sizes are set
by config.EVALUATE_CACHE_SIZE / config.PROBE_CACHE_SIZE.
Every request to /probe and /probe/stream carries an explicit "variant"
field ("standard" or "chess960", defaulting to "standard" when omitted)
alongside "fen" -- the two rulesets are never inferred from a FEN's
contents, since the orthodox starting arrangement is itself a legal
Chess960 starting position and would otherwise be ambiguous. A submitted
FEN's castling-availability field accepts "-" or any combination of "K",
"Q", "k", "q" for variant="standard", plus rook-file letters ("a"-"h" /
"A"-"H") for variant="chess960"; a right with no matching king/rook pair
is rejected as invalid input before the board is ever probed (see
_validate_and_build_board). A position with castling rights is otherwise
probed like any other: `chess.chesstb` resolves it when the loaded
tablebase set carries a table for that material (see config.py's
TABLEBASE_PATH and tablebase_router.py -- unaffected by variant,
since a Chess960 board's clean_castling_rights() reports the same way a
Standard board's does), and reports it as not covered when it doesn't.
Running this file directly (`python app.py`) serves the app via waitress
(a production-grade WSGI server) by default. Set DEBUG = True in
config.py to use Flask's own dev server instead (auto-reload +
interactive debugger) for local development. See README.md
"Installation" / "Running in production" for details.
"""

from __future__ import annotations

from collections import OrderedDict, namedtuple
from concurrent.futures import (
    ThreadPoolExecutor,
    wait as futures_wait,
    ALL_COMPLETED,
    FIRST_COMPLETED,
)
from dataclasses import dataclass
from flask import (
    Flask, render_template, request, jsonify, Response,
    stream_with_context, send_file, session,
)
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from functools import lru_cache, wraps
from typing import TypedDict
from waitress import serve as waitress_serve
from werkzeug.middleware.proxy_fix import ProxyFix
import atexit
import chess
import config
import dtz_anchor
import hmac
import json
import os
import re
import logging
import secrets
import tablebase_router
import threading
import time

# ── Logging ───────────────────────────────────────────────────────────────────

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger(__name__)


# ── ChessTB backend ───────────────────────────────────────────────────────────

# chess.chesstb comes from noobpwnftw's add-chesstb-tablebases fork of
# python-chess (see requirements.txt) - open_tablebase, ProbeResult's raw
# wdl/dtz/dtc/dtm/dtm50 fields, MissingTableError, and the WIN/CURSED_WIN/
# DRAW/BLESSED_LOSS/LOSE constants all come from that module.
import chess.chesstb as chesstb

log.info("chess.chesstb backend: noobpwnftw/python-chess (add-chesstb-tablebases)")

# Remote (HTTP) tablebase support, in two flavours -- config.REMOTE_MODE
# picks one (see tablebase_router.open_one):
#   remote/remote_direct.py   ("direct")   probe in place over byte ranges
#   remote/remote_fallback.py ("download") whole-file download, disk-cached
# See each file's module docstring for the design and the trade-off.
# looks_like_remote()/RemoteSourceError come from remote/remote_source.py.
from remote import remote_direct, remote_fallback, remote_source


# ── Type aliases ──────────────────────────────────────────────────────────────

class MoveEntry(TypedDict):
    san:         str
    plies:       int
    order:       int         # DTC only: pawn pushes still owed before a conversion (0 for every other metric)
    is_mate:     bool
    outcome:     str         # "win"|"cursed_win"|"draw"|"blessed_loss"|"loss"|"unknown"|"not_available"
    available:   bool        # False when this move's own metric table isn't present (outcome is then "not_available")
    child_fen:   str | None  # FEN after this move; None for terminal / unknown
    draw_reason: str | None  # "stalemate"|"insufficient_material"|None


class DtzMoveEntry(MoveEntry, total=False):
    rounded: bool  # DTZ rows only, when available: plies may be a ply off


@dataclass
class _DtzMove:
    row: int  # index into the DTZ rows
    my_wdl: int
    keeps_class: bool
    option: dtz_anchor.MoveOption


MoveList = list[MoveEntry]


# ── Config validation helpers ─────────────────────────────────────────────────
# Centralise the "read a config.py value, check its type, validate range"
# pattern that AppConfig.from_config() below needs once per setting.

def _validated_int(name: str, value, *, min_val: int = 1, max_val: int | None = None) -> int:
    if isinstance(value, str):
        # PORT and TRUSTED_PROXY_COUNT are the two config.py settings read
        # as a raw string rather than pre-converted at import time (see
        # config.py's own comments on each) -- specifically so a bad
        # value raises here, inside AppConfig.from_config()'s try/except,
        # rather than as an uncaught ValueError from a bare int() call
        # during module import, before this module has an exception handler
        # in place for it.
        try:
            value = int(value.strip())
        except ValueError:
            raise ValueError(f"config.{name} must be an integer, got {value!r}") from None
    if not isinstance(value, int) or isinstance(value, bool):
        raise ValueError(f"config.{name} must be an integer, got {value!r}")
    if max_val is None:
        if value < min_val:
            raise ValueError(f"config.{name} must be >= {min_val}, got {value}")
    elif not (min_val <= value <= max_val):
        raise ValueError(f"config.{name} must be {min_val}-{max_val}, got {value}")
    return value


def _validated_float(name: str, value) -> float:
    if not isinstance(value, (int, float)) or isinstance(value, bool):
        raise ValueError(f"config.{name} must be a number, got {value!r}")
    value = float(value)
    if value <= 0:
        raise ValueError(f"config.{name} must be > 0, got {value}")
    return value


def _validated_str(name: str, value) -> str:
    if not isinstance(value, str):
        raise ValueError(f"config.{name} must be a string, got {value!r}")
    return value


def _validated_path_spec(name: str, value) -> str | list[str]:
    """TABLEBASE_PATH: a
    single path (local directory or http(s):// URL) or a list of them for
    a hybrid source (see tablebase_router.py). Each list entry must
    itself be a string -- a nested list isn't a valid hybrid source."""
    if isinstance(value, str):
        return value
    if isinstance(value, (list, tuple)) and all(isinstance(v, str) for v in value):
        return list(value)
    raise ValueError(f"config.{name} must be a string or a list of strings, got {value!r}")


def _validated_choice(name: str, value, allowed: tuple) -> str:
    if value not in allowed:
        raise ValueError(
            f"config.{name} must be one of {', '.join(map(repr, allowed))}, got {value!r}")
    return value


def _validated_bool(name: str, value) -> bool:
    if not isinstance(value, bool):
        raise ValueError(f"config.{name} must be True or False, got {value!r}")
    return value


# ── Validated configuration ───────────────────────────────────────────────────

@dataclass(frozen=True)
class AppConfig:
    tablebase_path:      str | list[str]
    probe_threads:       int
    parallel_threshold:  int
    probe_timeout:       float
    host:                str
    port:                int
    evaluate_cache_size: int
    probe_cache_size:    int
    block_cache_bytes:   int
    debug:               bool
    waitress_threads:    int
    # Only used for an http(s):// URL entry in tablebase_path (see "ChessTB
    # backend" above and remote/remote_source.py) -- ignored entirely for
    # a local-directory entry. Bounds an on-disk LRU of whole downloaded
    # table files cached by remote/remote_fallback.py (see that file's
    # module docstring for the full design).
    remote_mode:             str
    remote_page_cache_bytes: int
    remote_page_size_bytes:  int
    remote_timeout_secs:     float
    remote_max_retries:      int
    remote_pool_maxsize:     int
    admin_token:             str
    secret_key:              str
    source_repository_url:              str
    probe_rate_limit:        str
    prefetch_rate_limit:     str
    admin_login_rate_limit:  str
    trusted_proxy_count:     int
    session_cookie_secure:   bool
    # Hard-coded limit, not one of config.py's settings -- kept as a
    # dataclass field (rather than a bare module constant) so it reads
    # alongside every other request-validation knob above.
    max_fen_length:      int   = 100

    @classmethod
    def from_config(cls) -> "AppConfig":
        # PROBE_THREADS is the one setting with a computed (not literal)
        # default, so a None left in config.py falls back to a
        # CPU-count-based figure instead of going through _validated_int.
        threads = (
            _validated_int("PROBE_THREADS", config.PROBE_THREADS)
            if config.PROBE_THREADS is not None
            else min(16, (os.cpu_count() or 4) * 2)
        )
        # Same "None means auto" shape as PROBE_THREADS above, computed
        # from it: a connection pool smaller than the probe thread count
        # gives up connection reuse for whichever requests land once the
        # pool is exhausted (see remote/remote_source.py).
        pool_maxsize = (
            _validated_int("REMOTE_POOL_MAXSIZE", config.REMOTE_POOL_MAXSIZE)
            if config.REMOTE_POOL_MAXSIZE is not None
            else max(threads * 2, 20)
        )

        return cls(
            tablebase_path=_validated_path_spec("TABLEBASE_PATH", config.TABLEBASE_PATH),
            probe_threads=threads,
            parallel_threshold=_validated_int("PROBE_PARALLEL_THRESHOLD", config.PROBE_PARALLEL_THRESHOLD),
            probe_timeout=_validated_float("PROBE_TIMEOUT_SECS", config.PROBE_TIMEOUT_SECS),
            host=_validated_str("HOST", config.HOST),
            port=_validated_int("PORT", config.PORT, max_val=65535),
            evaluate_cache_size=_validated_int("EVALUATE_CACHE_SIZE", config.EVALUATE_CACHE_SIZE),
            probe_cache_size=_validated_int("PROBE_CACHE_SIZE", config.PROBE_CACHE_SIZE),
            block_cache_bytes=_validated_int("BLOCK_CACHE_BYTES", config.BLOCK_CACHE_BYTES),
            # DEBUG also selects the server used by
            # `if __name__ == "__main__"` below: the Flask dev server when
            # True, waitress otherwise (see README.md "Running in production").
            debug=_validated_bool("DEBUG", config.DEBUG),
            waitress_threads=_validated_int("WAITRESS_THREADS", config.WAITRESS_THREADS),
            remote_mode=_validated_choice("REMOTE_MODE", config.REMOTE_MODE, ("direct", "download")),
            remote_page_cache_bytes=_validated_int(
                "REMOTE_PAGE_CACHE_BYTES", config.REMOTE_PAGE_CACHE_BYTES),
            remote_page_size_bytes=_validated_int(
                "REMOTE_PAGE_SIZE_BYTES", config.REMOTE_PAGE_SIZE_BYTES),
            remote_timeout_secs=_validated_float("REMOTE_TIMEOUT_SECS", config.REMOTE_TIMEOUT_SECS),
            remote_max_retries=_validated_int("REMOTE_MAX_RETRIES", config.REMOTE_MAX_RETRIES),
            remote_pool_maxsize=pool_maxsize,
            admin_token=_validated_str("ADMIN_TOKEN", config.ADMIN_TOKEN),
            secret_key=_validated_str("FLASK_SECRET_KEY", config.FLASK_SECRET_KEY),
            source_repository_url=_validated_str("SOURCE_REPOSITORY_URL", config.SOURCE_REPOSITORY_URL),
            probe_rate_limit=_validated_str("PROBE_RATE_LIMIT", config.PROBE_RATE_LIMIT),
            prefetch_rate_limit=_validated_str("PREFETCH_RATE_LIMIT", config.PREFETCH_RATE_LIMIT),
            admin_login_rate_limit=_validated_str("ADMIN_LOGIN_RATE_LIMIT", config.ADMIN_LOGIN_RATE_LIMIT),
            trusted_proxy_count=_validated_int(
                "TRUSTED_PROXY_COUNT", config.TRUSTED_PROXY_COUNT, min_val=0, max_val=10),
            session_cookie_secure=_validated_bool("SESSION_COOKIE_SECURE", config.SESSION_COOKIE_SECURE),
        )


try:
    cfg = AppConfig.from_config()
except (ValueError, AttributeError) as e:
    log.error("Configuration error: %s — fix config.py and restart.", e)
    raise SystemExit(1) from e

# Exception type to treat as "couldn't reach the remote tablebase" in the
# /probe and /probe/stream handlers below.
_REMOTE_SOURCE_ERROR = remote_source.RemoteSourceError

if not cfg.tablebase_path:
    log.warning("TABLEBASE_PATH is not set — tablebase lookups will fail.")

if not cfg.admin_token:
    log.warning(
        "ADMIN_TOKEN is not set — /admin and /admin/cache/* will respond "
        "503 until it's configured (see config.py's \"Admin\" section)."
    )


# ── Flask app ─────────────────────────────────────────────────────────────────

app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 4 * 1024   # 4 KB

# secrets.token_hex(32) generated fresh at process startup when
# FLASK_SECRET_KEY isn't set (see config.py) -- every admin session is then
# invalidated by a restart, requiring a new login; it never
# breaks anything the app itself depends on, since sessions here hold
# nothing but the admin_ok flag set by /admin/login.
app.secret_key = cfg.secret_key or secrets.token_hex(32)
app.config["SESSION_COOKIE_HTTPONLY"] = True
app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
# Independent of cfg.debug -- see config.py's SESSION_COOKIE_SECURE for
# why: DEBUG is a development/runtime behavior switch, not a reliable
# indicator of whether the browser reaches this deployment over HTTPS.
app.config["SESSION_COOKIE_SECURE"] = cfg.session_cookie_secure

# Without this, request.remote_addr -- and therefore get_remote_address()
# below, which keys both rate limiters -- is whichever host made the TCP
# connection to this process. Behind a reverse proxy (Render's own
# edge/load-balancer in front of the render.yaml deployment this project
# ships with, or any other reverse proxy placed in front of it) that's the
# proxy's own address, not the actual client's, so every client sharing
# that proxy would collapse onto one rate-limit key. ProxyFix instead
# trusts exactly cfg.trusted_proxy_count X-Forwarded-For hops closest to
# this app and reads the real client IP from ahead of them -- set to 0
# this is a no-op, appropriate only for a deployment with no reverse proxy
# in front of it at all (an untrusted client's own X-Forwarded-For header
# would otherwise be trusted, letting it spoof its rate-limit key).
#
# x_proto does the equivalent correction for X-Forwarded-Proto, which a
# TLS-terminating reverse proxy sets to tell this app the scheme the
# client actually connected with -- without it, request.scheme /
# request.is_secure (and any URL this app builds with
# url_for(..., _external=True)) would read as plain http:// even when the
# client's own connection was https://, since Flask only ever sees the
# proxy's own (typically-HTTP) hop to this app. Same trusted-hop count as
# x_for, for the same reason: a value here that doesn't match the real
# proxy chain either misreads the scheme or lets a client spoof it.
#
# This only actually matters for cfg.debug's Flask-dev-server path below --
# under waitress (cfg.debug False, the production path), waitress parses
# X-Forwarded-For and corrects REMOTE_ADDR itself before this app ever
# sees the request, taking priority over -- and making redundant, but
# harmlessly so -- the correction this makes. See the waitress_serve(...)
# call below for that side of it.
if cfg.trusted_proxy_count > 0:
    app.wsgi_app = ProxyFix(
        app.wsgi_app, x_for=cfg.trusted_proxy_count, x_proto=cfg.trusted_proxy_count)


@app.errorhandler(413)
def request_too_large(e: Exception) -> tuple:
    return jsonify({"error": "Request body too large (max 4 KB)."}), 413


# ── Rate limiting ────────────────────────────────────────────────────────────
# Per-IP limits on /probe, /probe/stream (config.py's PROBE_RATE_LIMIT, or
# PREFETCH_RATE_LIMIT for a request tagged "X-Prefetch: 1" -- see
# _request_probe_limit below) and, separately, /admin/login (config.py's
# ADMIN_LOGIN_RATE_LIMIT -- see README.md's "Security notes").
# default_limits=[] means nothing is limited except the routes explicitly
# decorated with @_probe_rate_limit / @_admin_login_rate_limit below; the
# in-memory storage is per-process, which is enough for a single-container
# deployment but doesn't share state across multiple replicas. key_func
# reads request.remote_addr, which ProxyFix (see the Flask app section
# above) has already corrected for cfg.trusted_proxy_count reverse-proxy
# hops.
limiter = Limiter(
    key_func=get_remote_address, app=app, default_limits=[], storage_uri="memory://",
    headers_enabled=True)


def _request_probe_limit() -> str:
    # X-Prefetch is client-asserted and therefore untrusted as a basis for
    # relaxing enforcement -- it only ever selects a separate, tighter
    # budget than the default, never bypasses PROBE_RATE_LIMIT. A request
    # without the header, or when PREFETCH_RATE_LIMIT is unset, always
    # falls back to the normal probe budget.
    if request.headers.get("X-Prefetch") == "1" and cfg.prefetch_rate_limit:
        return cfg.prefetch_rate_limit
    return cfg.probe_rate_limit


def _probe_rate_limit(view):
    # A plain pass-through when neither limit is configured, rather than
    # a zero/near-zero limiter.limit() call, so "disabled" actually means
    # unlimited rather than "limited to approximately nothing." A
    # callable limit_value (rather than the plain string
    # _admin_login_rate_limit below uses) is what lets a single decorator
    # route a hover-prefetch request to its own, tighter budget -- see
    # _request_probe_limit above -- flask-limiter re-evaluates it on
    # every request rather than once at decoration time.
    if not cfg.probe_rate_limit and not cfg.prefetch_rate_limit:
        return view
    return limiter.limit(_request_probe_limit)(view)


def _admin_login_rate_limit(view):
    # Independent of _probe_rate_limit above -- /admin/login is a
    # brute-forceable shared-secret check, not a CPU-bound probe, so it
    # gets its own (much tighter) limit rather than reusing or being
    # covered by PROBE_RATE_LIMIT.
    if not cfg.admin_login_rate_limit:
        return view
    return limiter.limit(cfg.admin_login_rate_limit)(view)


@app.errorhandler(429)
def rate_limit_exceeded(e: Exception) -> tuple:
    return jsonify({
        "error": "Too many requests. Please slow down and try again shortly.",
    }), 429


# ── Tablebase + thread pool ───────────────────────────────────────────────────
# All material, including castling and recursive dependencies, uses the
# same ordered source list. The adapter shares one decoded-block cache.
try:
    _tb_source = tablebase_router.open_many(
        cfg.tablebase_path, label="TABLEBASE_PATH", cfg=cfg,
        remote_source=remote_source, remote_direct=remote_direct,
        remote_download=remote_fallback,
    )
except Exception as e:
    log.warning("Failed to open TABLEBASE_PATH: %s", e)
    _tb_source = None

TB = (
    tablebase_router.SourcedTablebase(
        _tb_source, remote_source=remote_source,
        block_cache_bytes=cfg.block_cache_bytes,
    )
    if _tb_source is not None else None
)

_executor = ThreadPoolExecutor(max_workers=cfg.probe_threads, thread_name_prefix="tb_probe")
log.info(
    "Thread pool: %d workers, parallel_threshold=%d, timeout=%.0fs, "
    "eval_cache=%d, probe_cache=%d, block_cache_bytes=%d",
    cfg.probe_threads, cfg.parallel_threshold, cfg.probe_timeout,
    cfg.evaluate_cache_size, cfg.probe_cache_size, cfg.block_cache_bytes,
)


@atexit.register
def _close_tb() -> None:
    try:
        if TB is not None:
            TB.close()
    except Exception as e:
        log.warning("Error closing tablebase: %s", e)


@atexit.register
def _shutdown_executor() -> None:
    try:
        _executor.shutdown(wait=True, cancel_futures=True)
    except Exception as e:
        log.warning("Error shutting down thread pool: %s", e)


# ── Security headers ──────────────────────────────────────────────────────────

@app.after_request
def set_security_headers(response: Response) -> Response:
    # base-uri and form-action are set explicitly since default-src doesn't
    # cover them; object-src, though covered, is set to 'none' since the app
    # has no use for <object>/<embed>. Allow Hugging Face's Space page
    # to embed the application; CSP controls the permitted ancestors.
    response.headers["Content-Security-Policy"] = (
        "default-src 'self'; script-src 'self'; "
        "style-src 'self' 'unsafe-inline'; "
        "font-src 'self'; img-src 'self' data:; connect-src 'self'; "
        "object-src 'none'; base-uri 'self'; form-action 'self'; "
        "frame-ancestors 'self' https://huggingface.co;"
    )
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
    return response


# ── Admin authentication ────────────────────────────────────────────────────
# Gates /admin and /admin/cache/* behind cfg.admin_token (see config.py's
# "Admin" section). Two ways to authenticate:
#   - Browser: POST /admin/login with the token starts a signed session
#     cookie (see app.secret_key above); subsequent requests from the same
#     browser are authenticated via that cookie automatically.
#   - API client: send "Authorization: Bearer <ADMIN_TOKEN>" directly,
#     no session/cookie needed.
# If cfg.admin_token is empty, every admin route responds 503 rather than
# running with no password -- there is no "admin panel open to everyone"
# fallback.

def _admin_token_matches(candidate: str) -> bool:
    # hmac.compare_digest for constant-time comparison -- a plain `==`
    # would leak the token's length/prefix through response-timing
    # differences to an attacker making repeated guesses.
    return bool(cfg.admin_token) and hmac.compare_digest(candidate, cfg.admin_token)


def _is_admin_authenticated() -> bool:
    if session.get("admin_ok"):
        return True
    auth_header = request.headers.get("Authorization", "")
    if auth_header.startswith("Bearer "):
        return _admin_token_matches(auth_header[len("Bearer "):])
    return False


def _require_admin(view):
    @wraps(view)
    def wrapped(*args, **kwargs):
        if not cfg.admin_token:
            return jsonify({
                "error": "Admin panel disabled: ADMIN_TOKEN is not configured.",
            }), 503
        if not _is_admin_authenticated():
            return jsonify({"error": "Admin authentication required."}), 401
        return view(*args, **kwargs)
    return wrapped


# ── FEN normalisation ─────────────────────────────────────────────────────────

_FEN_FIELD_DEFAULTS = ("8/8/8/8/8/8/8/8", "w", "-", "-", "0", "1")


def _normalize_fen(fen: str) -> str:
    parts = fen.split()
    while len(parts) < 6:
        parts.append(_FEN_FIELD_DEFAULTS[len(parts)])
    return " ".join(parts[:6])


# ── Sort helpers ──────────────────────────────────────────────────────────────

_OUTCOME_RANK: dict[int, int] = {2: 0, 1: 1, 0: 2, -1: 3, -2: 4}
_PROMOTION_RANK = {"Q": 0, "R": 1, "B": 2, "N": 3}


def _move_tiebreak(san: str) -> tuple[int, str, int, str]:
    """Prefer immediate mate, then SAN order with Q/R/B/N promotions.
    The SAN prefix groups promotions along the same path. This key is
    used only after the metric's outcome, DTC order, and distance, so a
    better underpromotion still beats a worse queen promotion. Immediate
    mate must beat a tied zeroing/conversion move (DTZ/DTC can score both
    as one ply), including when the mating move is an underpromotion.
    """
    mate_rank = 0 if san.endswith("#") else 1
    prefix, separator, promotion = san.partition("=")
    if separator:
        return mate_rank, prefix, _PROMOTION_RANK[promotion[0]], promotion[1:]
    return mate_rank, san, 0, ""


def _outcome_rank(wdl_val: int) -> int:
    return _OUTCOME_RANK[wdl_val]


def _ply_rank(wdl_val: int, eff_val: int) -> int:
    if wdl_val > 0:  return eff_val
    if wdl_val < 0:  return -abs(eff_val)
    return 0


def _order_rank(wdl_val: int, order: int) -> int:
    """Sort key for DTC's "pushes still owed" component. DTC's own notion
    of a better line is pushes first, then plies (see ProbeResult's
    docstring in chess.chesstb): fewer pushes owed is better while
    winning, more is better while losing (it's more delay bought before
    the opponent's own clock forces a conversion). order is already an
    unsigned count, so - unlike _ply_rank - there's no magnitude to
    unwrap, only a direction to apply."""
    if wdl_val > 0:  return order
    if wdl_val < 0:  return -order
    return 0


_WDL_TO_OUTCOME_LABEL: dict[int, str] = {
    2: "win", 1: "cursed_win", 0: "draw", -1: "blessed_loss", -2: "loss",
}


def _outcome_label(wdl_val: int) -> str:
    return _WDL_TO_OUTCOME_LABEL[wdl_val]


def _effective_move_wdl(move_wdl: int, root_wdl: int, eff_distance: int) -> int:
    """The win/cursed-win (or loss/blessed-loss) bucket a move belongs in,
    used for both its outcome label and its sort-key rank. A raw win/loss
    (move_wdl 2/-2) downgrades to cursed_win/blessed_loss (1/-1) when its
    own eff_distance exceeds chesstb.MAX_NON_CURSED_DTZ. The result is then capped
    at root_wdl, since a move can never be better than the root position's
    own optimal value - this cap is one-directional, as a move can be
    worse than root_wdl (e.g. a blunder turning a blessed_loss root into a
    pure -2 loss). Draws pass through unchanged."""
    if abs(eff_distance) > chesstb.MAX_NON_CURSED_DTZ:
        if move_wdl == 2:
            move_wdl = 1
        elif move_wdl == -2:
            move_wdl = -1
    return _capped_wdl(move_wdl, root_wdl)


def _effective_distance(wdl: int, raw_distance: int) -> int:
    if wdl != 0 and raw_distance == 0:
        return 1 if wdl > 0 else -1
    if raw_distance > 0:
        return raw_distance + 1
    if raw_distance < 0:
        return raw_distance - 1
    return 0


def _is_cursed_exit(parent: chess.Board, child: chess.Board, child_wdl: int) -> bool:
    """True when a move into a cursed or blessed child (signed child_wdl)
    gives up castling rights."""
    return (abs(child_wdl) == 1
            and chess.popcount(child.clean_castling_rights())
            < chess.popcount(parent.clean_castling_rights()))


def _capped_wdl(move_wdl: int, root_wdl: int) -> int:
    """move_wdl capped at root_wdl: a move can never be better than the root
    position's own optimal value (see _effective_move_wdl). Used by the
    clock-independent DTZ/DTM columns."""
    return root_wdl if move_wdl > root_wdl else move_wdl


# ── Variant ────────────────────────────────────────────────────────────────
#
# "standard" | "chess960". Chess960 changes only the interpretation of
# castling rights/moves -- board geometry, piece movement, and tablebase
# material selection are unaffected (see tablebase_router.py, unchanged by
# variant). The two rulesets are never inferred from a FEN's contents: the
# orthodox starting array is itself one of the 960 legal Chess960 starting
# arrangements, so the same FEN text is genuinely ambiguous without this
# explicit field. An omitted variant defaults to "standard".
_ALLOWED_VARIANTS = frozenset({"standard", "chess960"})


def _normalize_variant(value: object) -> str:
    if value is None:
        return "standard"
    if not isinstance(value, str):
        raise ValueError("variant must be a string.")
    normalized = value.strip().lower()
    if normalized not in _ALLOWED_VARIANTS:
        raise ValueError("Unsupported variant: expected 'standard' or 'chess960'.")
    return normalized


def _build_board(variant: str, fen: str) -> chess.Board:
    """The single place a chess.Board is constructed from a request- or
    move-generation-supplied FEN string. chess960=True changes only how
    the fork parses/serializes the castling field and how it represents
    castling moves -- everything else about the resulting Board (piece
    movement, check/mate detection, piece_map, etc.) behaves identically
    to the variant="standard" path."""
    return chess.Board(fen, chess960=(variant == "chess960"))


# ── FEN pre-validation ────────────────────────────────────────────────────────

_EP_PATTERN = re.compile(r"^(-|[a-h][36])$")
_CASTLING_PATTERN_STANDARD = re.compile(r"^K?Q?k?q?$")
# Chess960 castling rights use conventional K/Q/k/q where applicable and
# rook-file letters for explicit rook identification. Semantic validity is
# checked after board construction.
_CASTLING_PATTERN_CHESS960 = re.compile(r"^[A-Ha-hKQkq]{1,4}$")


def _validate_fen_parts(variant: str, parts: list[str]) -> str | None:
    if len(parts) < 2:
        return "FEN must have at least a piece-placement field and a side-to-move field."
    if parts[1] not in ("w", "b"):
        return f"Invalid side to move: '{parts[1]}' (expected 'w' or 'b')."
    if len(parts) >= 3 and parts[2] != "-":
        if variant == "chess960":
            if not _CASTLING_PATTERN_CHESS960.match(parts[2]):
                return (
                    f"Invalid castling availability: '{parts[2]}' (expected some combination "
                    "of 'K', 'Q', 'k', 'q', rook-file letters 'a'-'h'/'A'-'H', or '-')."
                )
        elif not _CASTLING_PATTERN_STANDARD.match(parts[2]):
            return (
                f"Invalid castling availability: '{parts[2]}' "
                "(expected some combination of 'K', 'Q', 'k', 'q', or '-')."
            )
    if len(parts) >= 4 and not _EP_PATTERN.match(parts[3]):
        return (
            f"Invalid en-passant square: '{parts[3]}' "
            "(expected '-' or a valid target square such as 'e3' or 'd6')."
        )
    if len(parts) >= 5:
        try:
            hmc = int(parts[4])
            if hmc < 0:
                return "Halfmove clock must be a non-negative integer."
        except ValueError:
            return f"Invalid halfmove clock: '{parts[4]}' (expected a non-negative integer)."
    if len(parts) >= 6:
        try:
            fmc = int(parts[5])
            if fmc < 1:
                return "Fullmove number must be a positive integer."
        except ValueError:
            return f"Invalid fullmove number: '{parts[5]}' (expected a positive integer)."
    return None


# ── Signed-value helpers ──────────────────────────────────────────────────────
#
# WIN/CURSED_WIN/DRAW/BLESSED_LOSS/LOSE -> +2/+1/0/-1/-2, matching
# chess.syzygy's convention -- built from chess.chesstb's public class
# constants rather than its private _WDL_SIGNED, since this app has no
# reason to depend on a leading-underscore upstream symbol for a five-entry
# map it can just as easily own. This is what _probe_board below builds its
# signed wdl/dtc_wdl/dtm50_wdl fields on.

_SIGNED_WDL: dict[int, int] = {
    chesstb.WIN: 2, chesstb.CURSED_WIN: 1, chesstb.DRAW: 0,
    chesstb.BLESSED_LOSS: -1, chesstb.LOSE: -2,
}


def _signed_wdl(wdl: int) -> int | None:
    """Signed WDL for a raw wdl/dtc_wdl/dtm50_wdl class value."""
    return _SIGNED_WDL.get(wdl)


# ── Raw probe ─────────────────────────────────────────────────────────────────

def _probe_board(board: chess.Board) -> chesstb.ProbeResult | None:
    # A single combined probe() call computes WDL + DTZ + DTC + DTM + DTM50
    # together. The raw ProbeResult is returned as-is: its constructor
    # zero-initialises dtz/dtc_order/dtc/dtm/dtm50 and sets
    # wdl/dtc_wdl/dtm50_wdl to ILLEGAL whenever the matching has_* flag is
    # False, and _signed_wdl() maps ILLEGAL to None. WDL is the only metric
    # required - a position whose WDL can't be resolved is not probeable and
    # returns None. DTZ, DTC, DTM, and DTM50 are each independently
    # optional; has_dtz/has_dtc/has_dtm/has_dtm50 report which of the four
    # the loaded tables answered for this position.
    r = TB.probe(board)
    if r.status != "ok" or _signed_wdl(r.wdl) is None:
        return None
    return r


# ── Child-position probe cache ────────────────────────────────────────────────

class ProbeInFlightTimeout(Exception):
    """A probe of this FEN was already running and didn't finish in time.
    Retryable, exactly like the probe having timed out directly."""


class _InFlightProbe:
    """Slot for one probe in progress. The thread that creates it does the
    probing; every other thread asking for the same FEN waits on it."""

    __slots__ = ("done", "result", "error")

    def __init__(self) -> None:
        self.done   = threading.Event()
        self.result: chesstb.ProbeResult | None = None
        self.error:  BaseException | None = None


_inflight_lock = threading.Lock()
_inflight: dict[tuple[str, str], _InFlightProbe] = {}   # (variant, fen) -> slot

# Set (per calling thread) by _probe_fen_with_deadline below, so
# _probe_fen_deduped's in-flight wait can use a request's own remaining
# deadline instead of a fresh cfg.probe_timeout -- see that function's
# docstring for why a plain module-level value wouldn't work here. .board
# is set the same way, for the same reason, when the caller already has a
# chess.Board built from this exact fen in scope -- see
# _probe_fen_with_deadline's board parameter.
_probe_tls = threading.local()


def _probe_fen_deduped(variant: str, fen: str) -> chesstb.ProbeResult | None:
    """Runs at most one probe per (variant, fen) at a time, however many
    callers ask.
    _probe_fen's lru_cache only dedupes probes that have already finished:
    concurrent callers all miss it and all probe the same position. That
    happens routinely: /probe/stream pre-warms a child FEN that
    evaluate_all_moves then asks for again, and a retry after a timeout
    re-asks for a FEN whose first probe is still running. Each duplicate costs
    a full remote probe and occupies a pool worker.
    A non-owner's wait is bounded by the calling thread's own remaining
    deadline (see _probe_fen_with_deadline) so one hung probe can't pin a
    waiter beyond its own request's budget -- falling back to a fresh
    cfg.probe_timeout when no deadline was set, for a caller reached
    outside a deadline-aware path. The wait expiring raises, which reads
    downstream as a retryable failure just like the probe itself timing out.
    The owner builds the chess.Board to probe from (variant, fen) itself
    unless the calling thread already left one in _probe_tls.board (see
    _probe_fen_with_deadline's board parameter) -- for a root-fen probe,
    the caller has usually already built one anyway (to validate the FEN,
    render the board, etc.), and re-parsing the same string a second time
    here is pure waste. variant is required even when board is supplied,
    since it is also this function's in-flight/dedup key -- the same FEN
    text probes to a different (and possibly incompatible) result under
    Standard vs Chess960 rules (see the module-level variant comment), so
    a key of fen alone would let one variant's probe be served to the
    other's caller.
    """
    deadline = getattr(_probe_tls, "deadline", None)
    if deadline is not None and time.monotonic() >= deadline:
        raise ProbeInFlightTimeout("Probe deadline expired before work started")
    key = (variant, fen)
    with _inflight_lock:
        slot = _inflight.get(key)
        if slot is None:
            slot = _inflight[key] = _InFlightProbe()
            owner = True
        else:
            owner = False

    if not owner:
        deadline = getattr(_probe_tls, "deadline", None)
        timeout = (deadline - time.monotonic()) if deadline is not None else cfg.probe_timeout
        if timeout <= 0 or not slot.done.wait(max(0, timeout)):
            raise ProbeInFlightTimeout(f"Timed out waiting on an in-flight probe of variant={variant} {fen}")
        if slot.error is not None:
            raise slot.error
        return slot.result

    try:
        board = getattr(_probe_tls, "board", None)
        slot.result = _probe_board(board if board is not None else _build_board(variant, fen))
        return slot.result
    except BaseException as exc:
        slot.error = exc
        raise
    finally:
        # Drop the slot before waking the waiters, so a caller arriving in
        # between starts a fresh probe rather than joining a finished one.
        with _inflight_lock:
            _inflight.pop(key, None)
        slot.done.set()


@lru_cache(maxsize=cfg.probe_cache_size)
def _probe_fen(variant: str, fen: str) -> chesstb.ProbeResult | None:
    return _probe_fen_deduped(variant, fen)


def _probe_fen_with_deadline(
    variant: str, fen: str, deadline: float | None, board: chess.Board | None = None,
) -> chesstb.ProbeResult | None:
    """Calls _probe_fen(variant, fen), but first makes `deadline` (and,
    optionally, `board`) visible to _probe_fen_deduped's in-flight wait /
    owner branch on this thread.
    _probe_fen must stay a (variant, fen)-only lru_cache -- adding deadline
    as a real parameter would make it part of the cache key, and since a
    deadline is different on every call, that would turn the cache into a
    permanent miss. A plain thread-local set by the caller doesn't work
    either when this runs via _executor.submit(_probe_fen_with_deadline,
    variant, fen, ...): submit() runs the call on a pool worker thread, not
    the submitting thread, so a thread-local the submitter set wouldn't be
    visible there. Passing deadline as a real argument to this wrapper --
    called either directly or as the submitted callable itself -- sidesteps
    both problems: it's a normal argument up to here, then becomes this
    worker thread's own thread-local for the duration of the call.
    board is optional and exists purely to avoid re-parsing a FEN this
    thread already has a chess.Board for -- pass it only when `board` was
    itself built from this exact (variant, fen) pair, right here, in this
    same function call's caller, moments before this call. _probe_fen_deduped
    has no way to verify the two actually match (board.fen() isn't
    reliably equal to a shortened/unpadded input fen string, so comparing
    them isn't a safety net either), so passing a board that doesn't
    correspond to (variant, fen) silently returns a wrong probe result.
    Child boards copied by _collect_move_info() satisfy the same contract.
    """
    _probe_tls.deadline = deadline
    _probe_tls.board = board
    try:
        return _probe_fen(variant, fen)
    finally:
        _probe_tls.deadline = None
        _probe_tls.board = None



# ── Move evaluation ───────────────────────────────────────────────────────────

def _apply_sign(magnitude: int, sign_wdl: int) -> int:
    """Gives an unsigned ply magnitude the sign implied by a signed WDL
    value: positive for a winning sign_wdl, negative for a losing one,
    zero for a draw. Shared by every DTZ/DTC/DTM/DTM50 assembly step in
    evaluate_all_moves below, so the four metrics can't drift apart on
    how a magnitude gets its sign."""
    if sign_wdl > 0:
        return magnitude
    if sign_wdl < 0:
        return -magnitude
    return 0


def _collect_move_info(board: chess.Board) -> list[tuple[str, bool, bool, bool, str | None, str | None, chess.Board | None]]:
    move_info: list[tuple[str, bool, bool, bool, str | None, str | None, chess.Board | None]] = []
    for move in board.legal_moves:
        san = board.san(move)
        # Bypasses expensive Board.is_checkmate() checks by parsing the SAN string directly
        move_is_mate = san.endswith("#")
        move_is_zeroing = board.is_zeroing(move)
        # DTC's own notion of "conversion" (a capture or a promotion) is
        # narrower than move_is_zeroing (capture OR any pawn move): a quiet
        # pawn push zeroes the clock too, but chess.chesstb.Tablebase.
        # _dtc_move_kind does not treat it as a conversion, only as one of
        # the pushes the winning side still owes (see the DTC row below,
        # this field's only consumer). is_capture() rather than a bare
        # to-square occupancy check so an
        # en passant capture - whose destination square is empty - still
        # counts as a conversion here.
        move_is_conversion = board.is_capture(move) or move.promotion is not None
        draw_reason: str | None = None
        child_board: chess.Board | None = None
        board.push(move)
        try:
            if not move_is_mate:
                # Bypasses Board.is_stalemate() by stopping at the first legal move found
                if not any(board.generate_legal_moves()):
                    draw_reason = "stalemate"
                elif board.is_insufficient_material():
                    draw_reason = "insufficient_material"
            child_fen: str | None = None if (move_is_mate or draw_reason) else board.fen()
            if child_fen is not None:
                # Upstream probes positions, not repetition history. Preserve
                # the clock, en passant and Chess960 state without the stack.
                child_board = board.copy(stack=False)
        finally:
            board.pop()
        move_info.append((san, move_is_mate, move_is_zeroing, move_is_conversion, draw_reason, child_fen, child_board))
    return move_info


def evaluate_all_moves(
    board: chess.Board,
    root_wdl: int,
    variant: str = "standard",
    bypass_parallel: bool = False,
    precomputed_move_info: list[tuple[str, bool, bool, bool, str | None, str | None, chess.Board | None]] | None = None,
    deadline: float | None = None,
    root_dtz: int | None = None,
    root_dtz_rounded: bool = False,
) -> tuple[MoveList, MoveList, MoveList, MoveList, bool, dict[str, int], int | None]:
    """Returns the four ranked move lists (DTZ, DTC, DTM, DTM50, in that
    order), a "complete" flag: False when at least one move shows "unknown"
    for a retryable reason (probe timeout or error) rather than because its
    table is genuinely absent (callers use it to decide whether the result
    is worth caching - see evaluate_fen()), the wins/draws/losses/unknown
    summary across all legal moves, and the root's exact DTZ when
    cross-checking the moves resolved it (None otherwise).
    root_dtz/root_dtz_rounded are the root's own DTZ, None when its material
    has no DTZ table. The DTZ rows' rounded values are resolved against them
    by dtz_anchor.resolve().
    variant applies to every child position probed below, matching the
    ruleset `board` itself was built under (see _build_board) -- board.push()/
    pop() inside _collect_move_info already preserves board.chess960 for
    move generation and child.fen() serialization for free, since it's the
    same Board object throughout; variant only needs to be threaded through
    explicitly here because child positions are (re-)probed from their FEN
    string, not from the Board object itself (see _probe_fen_with_deadline).
    deadline is an absolute time.monotonic() value bounding child probing."""
    effective_deadline = deadline if deadline is not None else time.monotonic() + cfg.probe_timeout

    # Phase 1: collect move metadata (no TB calls) - reuses the caller's own
    # pass over board.legal_moves() when one is supplied, instead of
    # re-walking every legal move a second time for the same board.
    move_info = precomputed_move_info if precomputed_move_info is not None else _collect_move_info(board)

    # Phase 2: probe child positions
    unique_fens: set[str] = {child_fen for _, _, _, _, _, child_fen, _ in move_info if child_fen is not None}
    probe_cache: dict[str, chesstb.ProbeResult | None] = {}
    child_boards: dict[str, chess.Board] = {
        child_fen: child_board
        for _, _, _, _, _, child_fen, child_board in move_info
        if child_fen is not None and child_board is not None
    }

    # Child FENs that failed for a retryable reason. A probe that timed out is
    # still running in the pool and will populate _probe_fen's own cache, so a
    # later attempt at this position resolves it; a probe that raised may
    # succeed on a retry too. Neither is the same as _probe_fen returning None,
    # which means tb_not_found and won't change however often it's re-probed.
    transient_failures: set[str] = set()

    # Probe sufficiently large uncached batches through the thread pool.
    if len(unique_fens) >= cfg.parallel_threshold and not bypass_parallel:
        futures = {
            fen: _executor.submit(
                _probe_fen_with_deadline, variant, fen, effective_deadline, child_boards.get(fen)
            )
            for fen in unique_fens
        }
        remaining = max(0.0, effective_deadline - time.monotonic())
        done, _ = futures_wait(
            futures.values(), timeout=remaining, return_when=ALL_COMPLETED
        )
        for fen, fut in futures.items():
            if fut in done:
                try:
                    probe_cache[fen] = fut.result()
                except Exception as exc:
                    log.warning("Probe error for variant=%s %s: %s", variant, fen, exc)
                    probe_cache[fen] = None
                    transient_failures.add(fen)
            else:
                fut.cancel()  # Drop queued work; running probes can still warm the cache.
                log.warning("Probe timed out for variant=%s %s", variant, fen)
                probe_cache[fen] = None
                transient_failures.add(fen)
    else:
        # Probe smaller batches, and batches whose children are already cached,
        # sequentially to avoid unnecessary thread-pool overhead.
        for fen in unique_fens:
            try:
                probe_cache[fen] = _probe_fen_with_deadline(
                    variant, fen, effective_deadline, child_boards.get(fen)
                )
            except Exception as exc:
                log.warning("Probe error for variant=%s %s: %s", variant, fen, exc)
                probe_cache[fen] = None
                transient_failures.add(fen)

    # Phase 3: assemble and sort
    dtz_rows:   list[tuple] = []
    dtc_rows:   list[tuple] = []
    dtm_rows:   list[tuple] = []
    dtm50_rows: list[tuple] = []

    # Summary outcomes respect each child's halfmove clock. DTM50 and DTC
    # already resolve the rule; clock-independent WDL cannot establish a
    # rule50 win or loss when both of those metrics are unavailable.
    move_outcomes: list[str] = []

    # Sort key/entry pair for a move whose own metric table doesn't cover
    # it: flatly "Not Available", the same shape for DTZ, DTC, DTM, and
    # DTM50 alike rather than a value borrowed from another metric.
    # has_order adds DTC's extra pushes-owed slot to the sort key.
    def _not_available(
        san: str, child_fen: str | None, move_is_mate: bool,
        draw_reason: str | None, has_order: bool = False,
    ) -> tuple[tuple, MoveEntry]:
        key = (
            (_outcome_rank(0) + 10, 0, 0, _move_tiebreak(san)) if has_order
            else (_outcome_rank(0) + 10, 0, _move_tiebreak(san))
        )
        entry: MoveEntry = {
            "san": san, "plies": 0, "order": 0, "is_mate": move_is_mate, "available": False,
            "outcome": "not_available", "child_fen": child_fen, "draw_reason": draw_reason,
        }
        return key, entry

    def _dtz_row(
        san: str, my_wdl: int, eff_dtz: int, rounded: bool, move_is_mate: bool,
        child_fen: str | None, draw_reason: str | None,
    ) -> tuple[tuple, DtzMoveEntry, int]:
        eff_wdl = _effective_move_wdl(my_wdl, root_wdl, eff_dtz)
        key = (_outcome_rank(eff_wdl), _ply_rank(eff_wdl, eff_dtz), _move_tiebreak(san))
        entry: DtzMoveEntry = {
            "san": san, "plies": abs(eff_dtz), "order": 0, "is_mate": move_is_mate, "available": True,
            "outcome": _outcome_label(eff_wdl), "child_fen": child_fen, "draw_reason": draw_reason,
            "rounded": rounded,
        }
        return key, entry, eff_wdl

    dtz_moves: list[_DtzMove] = []
    dtz_complete = True

    for san, move_is_mate, move_is_zeroing, move_is_conversion, draw_reason, child_fen, child_board in move_info:

        if move_is_mate:
            opp_wdl = opp_dtc_wdl = opp_dtm50_wdl = -2
            opp_dtz = opp_dtc = opp_dtc_order = opp_dtm = opp_dtm50 = 0
            opp_has_dtz = opp_has_dtc = opp_has_dtm = opp_has_dtm50 = True
            opp_dtz_rounded = False

        elif draw_reason:
            opp_wdl = opp_dtc_wdl = opp_dtm50_wdl = 0
            opp_dtz = opp_dtc = opp_dtc_order = opp_dtm = opp_dtm50 = 0
            opp_has_dtz = opp_has_dtc = opp_has_dtm = opp_has_dtm50 = True
            opp_dtz_rounded = False

        else:
            probe_result = probe_cache.get(child_fen)
            if probe_result is None:
                log.warning("No probe result for %s (%s); showing unknown.", child_fen, san)
                stub: MoveEntry = {
                    "san": san, "plies": 0, "order": 0, "is_mate": False, "available": False,
                    "outcome": "unknown", "child_fen": child_fen, "draw_reason": None,
                }
                stub_key = (_outcome_rank(0) + 10, 0, _move_tiebreak(san))
                dtc_stub_key = (_outcome_rank(0) + 10, 0, 0, _move_tiebreak(san))
                dtz_rows.append((stub_key, stub))
                dtc_rows.append((dtc_stub_key, stub))
                dtm_rows.append((stub_key, stub))
                dtm50_rows.append((stub_key, stub))
                move_outcomes.append("unknown")
                dtz_complete = False
                continue
            opp_wdl = _signed_wdl(probe_result.wdl)
            opp_has_dtz, opp_dtz = probe_result.has_dtz, probe_result.dtz
            opp_dtz_rounded = probe_result.dtz_rounded
            opp_has_dtc = probe_result.has_dtc
            opp_dtc_wdl = _signed_wdl(probe_result.dtc_wdl)
            opp_dtc_order, opp_dtc = probe_result.dtc_order, probe_result.dtc
            opp_has_dtm, opp_dtm = probe_result.has_dtm, probe_result.dtm
            opp_has_dtm50 = probe_result.has_dtm50
            opp_dtm50_wdl = _signed_wdl(probe_result.dtm50_wdl)
            opp_dtm50 = probe_result.dtm50

        my_wdl = -opp_wdl
        if opp_has_dtm50:
            summary_wdl = -opp_dtm50_wdl
        elif opp_has_dtc:
            summary_wdl = -opp_dtc_wdl
        else:
            # Draws, cursed wins and blessed losses are draws under the
            # rule even without a clock-aware table. True wins/losses
            # need one of those tables to resolve the remaining budget.
            summary_wdl = 0 if abs(my_wdl) <= 1 else None
        move_outcomes.append(
            "win" if summary_wdl == 2 else
            "loss" if summary_wdl == -2 else
            "unknown" if summary_wdl is None else "draw"
        )

        # DTZ row.
        if opp_has_dtz:
            my_dtz = _apply_sign(opp_dtz, my_wdl)
            # If this move is itself a zeroing move (capture/pawn-push/
            # promotion), it is the zeroing move, so its DTZ is 1 ply - not
            # 1 + the child position's own distance to its next
            # zeroing move, which is what feeding my_dtz straight into
            # _effective_distance would compute (double-counting a second
            # zeroing event nobody asked for). _effective_distance
            # already has a branch for exactly this ("no further distance
            # needed, the move itself resolves it"), so call it with
            # raw_distance=0 instead of reimplementing that branch here.
            cursed_exit = (not move_is_zeroing and child_board is not None
                           and _is_cursed_exit(board, child_board, opp_wdl))
            option = dtz_anchor.move_option(opp_dtz, opp_dtz_rounded, move_is_zeroing, cursed_exit)
            if move_is_zeroing:
                eff_dtz = _effective_distance(my_wdl, 0)
            elif cursed_exit:
                # CORRECTNESS.md Theorem 12.15: a rounded derived child
                # can be even. Its true exit values then span y+1..y+3;
                # report y+2 so the displayed error stays within one ply.
                # option.value still carries the possible true distances
                # for the anchor, rather than this representative value.
                exit_dtz = opp_dtz + 2 if opp_dtz_rounded else option.value[0]
                eff_dtz = _apply_sign(exit_dtz, my_wdl)
            else:
                eff_dtz = _effective_distance(my_wdl, my_dtz)
            dtz_key, dtz_entry, eff_wdl = _dtz_row(
                san, my_wdl, eff_dtz, len(set(option.value)) > 1, move_is_mate, child_fen, draw_reason,
            )
            outcome = dtz_entry["outcome"]
            dtz_rows.append((dtz_key, dtz_entry))
            dtz_moves.append(_DtzMove(len(dtz_rows) - 1, my_wdl, eff_wdl == root_wdl, option))
        else:
            dtz_rows.append(_not_available(san, child_fen, move_is_mate, draw_reason))
            dtz_complete = False

        # DTC row. DTC's own notion of a better move is pushes-owed first, then
        # plies (see _order_rank's docstring), so the sort key carries both,
        # unlike DTZ/DTM/DTM50 above and below which only ever compare
        # plies.
        #
        # A capture or a promotion (move_is_conversion) is the conversion,
        # at this move: 0 pushes still owed, 1 ply to get there - the
        # child's own pushes/plies describe its next, later conversion,
        # which is irrelevant once this one has already happened.
        #
        # A quiet pawn push zeroes the clock too (move_is_zeroing) but is
        # not a conversion - chess.chesstb.Tablebase._dtc_move_kind treats
        # it as spending one of the pushes still owed rather than settling
        # the count, so the wait still resets to 1 ply the same way a
        # conversion's does, but pushes owed does not reset to 0.
        #
        # The child is probed at its own halfmove clock (see _probe_board)
        # and its dtc_order counts pushes owed from that position onward,
        # so it does not include the push spent getting there. For a root
        # reporting dtc_order=N whose best move is a push, the child
        # reports dtc_order=N-1; the +1 below restores the push this move
        # spent.
        #
        # Whether the +1 applies depends on which side is pushing: dtc_order
        # counts the pushes still owed by whichever side is winning, and
        # only the WINNING side's own push spends one of them. A defensive
        # push by the losing side leaves that budget untouched, so it
        # carries the child's order over unchanged, same as a quiet move;
        # hence the my_dtc_wdl > 0 guard below.
        if opp_has_dtc:
            my_dtc_wdl = -opp_dtc_wdl
            if move_is_mate or draw_reason or move_is_conversion:
                # A capture, a promotion, checkmate, or an immediate draw
                # are all fully resolved right here: 0 pushes still owed,
                # 1 ply to get there. move_is_mate/draw_reason already
                # hardwire opp_dtc_order/opp_dtc to 0 above, so this is a
                # no-op arithmetically for those in the ordinary case - it
                # matters for the case where the move is also a quiet
                # pawn push, e.g. checkmate delivered by discovered check
                # from a pawn advance that neither captures nor promotes.
                # Without this guard that would fall into the push branch
                # below and pick up a spurious +1: mate ends the game, so
                # nothing is owed no matter how the pawn got there.
                raw_order = 0
                eff_dtc = _effective_distance(my_dtc_wdl, 0)
            elif move_is_zeroing:  # a quiet pawn push, not a conversion
                raw_order = opp_dtc_order + (1 if my_dtc_wdl > 0 else 0)
                eff_dtc = _effective_distance(my_dtc_wdl, 0)
            else:
                raw_order = opp_dtc_order
                my_dtc = _apply_sign(opp_dtc, my_dtc_wdl)
                eff_dtc = _effective_distance(my_dtc_wdl, my_dtc)
            # A draw has nothing owed regardless of the raw push count
            # above - e.g. a defensive push into a dead draw (stalemate,
            # insufficient material, or a budget that no longer fits the
            # clock) must not report a nonzero push count just because it
            # happened to be a push; _order_rank sorts it correctly either
            # way (it treats wdl_val == 0 as rank 0 regardless), but the
            # value shown to the client should read 0 like plies does.
            eff_order = raw_order if my_dtc_wdl != 0 else 0
            dtc_key = (
                _outcome_rank(my_dtc_wdl),
                _order_rank(my_dtc_wdl, eff_order),
                _ply_rank(my_dtc_wdl, eff_dtc),
                _move_tiebreak(san),
            )
            dtc_rows.append((dtc_key, {
                "san": san, "plies": abs(eff_dtc), "order": eff_order, "is_mate": move_is_mate,
                "available": True, "outcome": _outcome_label(my_dtc_wdl),
                "child_fen": child_fen, "draw_reason": draw_reason,
            }))
        else:
            dtc_rows.append(_not_available(san, child_fen, move_is_mate, draw_reason, has_order=True))

        # DTM row.
        if opp_has_dtm:
            my_dtm = _apply_sign(opp_dtm, my_wdl)
            eff_dtm = _effective_distance(my_wdl, my_dtm)
            if opp_has_dtz:
                # DTZ and DTM share the same win/cursed-win (loss/blessed-
                # loss) bucket and label as the DTZ row above, since that
                # status is a single fact about eff_dtz and root_wdl rather
                # than a separate one per metric.
                dtm_key = (_outcome_rank(eff_wdl), _ply_rank(eff_wdl, eff_dtm), _move_tiebreak(san))
                dtm_outcome = outcome
            else:
                # Without DTZ the bucket comes from WDL alone.
                dtm_eff_wdl = _capped_wdl(my_wdl, root_wdl)
                dtm_key = (_outcome_rank(dtm_eff_wdl), _ply_rank(dtm_eff_wdl, eff_dtm), _move_tiebreak(san))
                dtm_outcome = _outcome_label(dtm_eff_wdl)
            dtm_rows.append((dtm_key, {
                "san": san, "plies": abs(eff_dtm), "order": 0, "is_mate": move_is_mate, "available": True,
                "outcome": dtm_outcome, "child_fen": child_fen, "draw_reason": draw_reason,
            }))
        else:
            dtm_rows.append(_not_available(san, child_fen, move_is_mate, draw_reason))

        # DTM50 row.
        if opp_has_dtm50:
            my_dtm50_wdl = -opp_dtm50_wdl
            my_dtm50 = _apply_sign(opp_dtm50, my_dtm50_wdl)
            eff_dtm50 = _effective_distance(my_dtm50_wdl, my_dtm50)
            dtm50_key = (_outcome_rank(my_dtm50_wdl), _ply_rank(my_dtm50_wdl, eff_dtm50), _move_tiebreak(san))
            dtm50_rows.append((dtm50_key, {
                "san": san, "plies": abs(eff_dtm50), "order": 0, "is_mate": move_is_mate, "available": True,
                "outcome": _outcome_label(my_dtm50_wdl), "child_fen": child_fen, "draw_reason": draw_reason,
            }))
        else:
            dtm50_rows.append(_not_available(san, child_fen, move_is_mate, draw_reason))

    # Cross-check rounded DTZ values against each other; skipped when any move
    # lacks a DTZ, since that move could be the one supplying the root's value.
    resolved_dtz: int | None = None
    if dtz_complete and root_dtz is not None and root_wdl != 0:
        keepers = [m for m in dtz_moves if m.keeps_class]
        if root_dtz_rounded or any(len(set(m.option.value)) > 1 for m in keepers):
            resolution = dtz_anchor.resolve(
                root_dtz, root_dtz_rounded, [m.option for m in keepers], root_wdl > 0,
            )
            resolved_dtz = resolution.root
            for k, value in resolution.pinned.items():
                move = keepers[k]
                row = dtz_rows[move.row][1]
                dtz_rows[move.row] = _dtz_row(
                    row["san"], move.my_wdl, _apply_sign(value, move.my_wdl), False,
                    row["is_mate"], row["child_fen"], row["draw_reason"],
                )[:2]

    dtz_rows.sort(key=lambda r: r[0])
    dtc_rows.sort(key=lambda r: r[0])
    dtm_rows.sort(key=lambda r: r[0])
    dtm50_rows.sort(key=lambda r: r[0])

    summary = {
        "wins":    sum(1 for o in move_outcomes if o in ("win", "cursed_win")),
        "draws":   sum(1 for o in move_outcomes if o == "draw"),
        "losses":  sum(1 for o in move_outcomes if o in ("loss", "blessed_loss")),
        "unknown": sum(1 for o in move_outcomes if o == "unknown"),
    }

    return (
        [r[1] for r in dtz_rows],
        [r[1] for r in dtc_rows],
        [r[1] for r in dtm_rows],
        [r[1] for r in dtm50_rows],
        not transient_failures,
        summary,
        resolved_dtz,
    )


# ── Root Probe Calculation ────────────────────────────────────────────────────

_evaluate_tls = threading.local()


def _validate_and_build_board(variant: str, fen: str) -> chess.Board:
    """
    Single authoritative implementation of root-FEN validation, shared by
    /probe (via evaluate_fen) and /probe/stream (which needs a validated
    board up front, before it can start emitting SSE progress events).
    Raises RuntimeError if the tablebase isn't loaded, or ValueError for any
    FEN/board problem - callers are expected to catch and format those.
    Legality (castling rights matching an actual king/rook pair, no pawns
    on the back rank, exactly one king per side, etc.) is entirely
    delegated to board.is_valid() for both variants -- this function adds
    no Standard-only assumptions of its own (e.g. requiring castling rooks
    on a1/h1/a8/h8) that would misclassify a legal Chess960 position as
    invalid. See _build_board for how the variant affects construction.
    """
    if TB is None:
        raise RuntimeError(
            "Tablebase not initialised. Check TABLEBASE_PATH.")

    parts = fen.split()
    fmt_error = _validate_fen_parts(variant, parts)
    if fmt_error:
        raise ValueError(fmt_error)

    try:
        board = _build_board(variant, fen)
    except ValueError as e:
        raise ValueError(f"Malformed FEN: {e}") from e

    castling_field = parts[2] if len(parts) >= 3 else "-"
    if variant == "chess960" and castling_field != "-":
        if board.castling_rights.bit_count() != len(castling_field):
            raise ValueError("Invalid castling availability: duplicate castling rights.")

    if not board.is_valid():
        raise ValueError(
            "Invalid board position (missing kings, adjacent kings, pawn on "
            "back rank, a castling right with no matching king/rook pair, etc.)."
        )

    return board


def _missing_table_error_payload(variant: str, fen: str) -> dict:
    """
    Single authoritative shape for the "position not covered by the loaded
    tablebase" error, shared by /probe and /probe/stream so the two
    transports (plain JSON vs SSE) can't drift out of sync on wording,
    error_code, or how piece_count is computed.
    """
    try:
        piece_count = len(_build_board(variant, fen).piece_map())
    except Exception:
        piece_count = None
    return {
        "error":       "Position not covered by the loaded tablebase.",
        "error_code":  "missing_table",
        "piece_count": piece_count,
    }


class _IncompleteResult(Exception):
    """Carries a probe result that must not be cached - see evaluate_fen()."""

    def __init__(self, json_str: str) -> None:
        super().__init__("probe incomplete")
        self.json_str = json_str


def _evaluate_fen_impl(variant: str, fen: str) -> str:
    """
    Probes the tablebase for a root FEN and returns a pre-serialized JSON string.
    Routes the root probe through the same shared _probe_fen cache used for
    child positions, so a position that was just probed as a child (e.g. the
    position the user just moved into) is served from cache instead of
    re-hitting the tablebase.
    (variant, fen) are the cache key, so a caller that already built a
    board/move_info for this exact pair -- e.g. probe_stream()'s SSE handler,
    which needs both up front to size its progress bar -- hands them over via _evaluate_tls
    instead, letting this call skip redoing that work. Ignored on a cache
    hit, since the function body below never runs in that case.
    Raises _IncompleteResult (carrying the same JSON) when a move came back
    unknown for a retryable reason, so evaluate_fen() can serve it without
    caching it.
    """
    precomputed_board     = getattr(_evaluate_tls, "board", None)
    precomputed_move_info = getattr(_evaluate_tls, "move_info", None)
    board = precomputed_board if precomputed_board is not None else _validate_and_build_board(variant, fen)

    # A fresh cfg.probe_timeout-sized budget for the root probe alone,
    # independent of how long any earlier phase of the same request (e.g.
    # probe_stream()'s own pre-warm step) already took, so a slow earlier
    # phase cannot erode the time this one gets. The child batch below
    # carries its own equally-sized fresh budget (see child_deadline),
    # matching the independent-budget-per-phase pattern probe_stream()
    # applies between its pre-warm step and the root probe above it. The
    # two independently-budgeted phases put a single /probe request's
    # worst case at up to 2 * cfg.probe_timeout.
    root_deadline = time.monotonic() + cfg.probe_timeout

    if board.is_checkmate():
        return json.dumps({
            "wdl": -2, "dtz": 0, "dtz_rounded": False, "dtc": [-2, 0, 0], "dtm": 0, "dtm50": [-2, 0],
            "dtz_available": True, "dtc_available": True, "dtm_available": True, "dtm50_available": True,
            "moves_dtz": [], "moves_dtc": [], "moves_dtm": [], "moves_dtm50": [],
            "summary": {"wins": 0, "draws": 0, "losses": 0, "unknown": 0},
            "draw_reason": None,
        })

    if board.is_stalemate() or board.is_insufficient_material():
        return json.dumps({
            "wdl": 0, "dtz": 0, "dtz_rounded": False, "dtc": [0, 0, 0], "dtm": 0, "dtm50": [0, 0],
            "dtz_available": True, "dtc_available": True, "dtm_available": True, "dtm50_available": True,
            "moves_dtz": [], "moves_dtc": [], "moves_dtm": [], "moves_dtm50": [],
            "summary": {"wins": 0, "draws": 0, "losses": 0, "unknown": 0},
            "draw_reason": "stalemate" if board.is_stalemate() else "insufficient_material",
        })

    # Route the root probe through the shared child-probe cache. board was
    # either handed over via _evaluate_tls or just built above from this
    # exact fen, so it's passed through here too rather than making
    # _probe_fen_deduped re-parse the same string a second time.
    root = _probe_fen_with_deadline(variant, fen, root_deadline, board)
    if root is None:
        raise chesstb.MissingTableError(f"Position not in tablebase: variant={variant} {fen}")

    bypass_parallel = getattr(_evaluate_tls, "bypass_parallel", False)
    # Own fresh budget, not the remainder of root_deadline -- see the
    # comment above root_deadline.
    child_deadline = time.monotonic() + cfg.probe_timeout
    moves_dtz, moves_dtc, moves_dtm, moves_dtm50, complete, summary, resolved_dtz = evaluate_all_moves(
        board, _signed_wdl(root.wdl), variant, bypass_parallel=bypass_parallel,
        precomputed_move_info=precomputed_move_info, deadline=child_deadline,
        root_dtz=root.dtz if root.has_dtz else None, root_dtz_rounded=root.dtz_rounded,
    )
    root_dtz, root_dtz_rounded = (
        (root.dtz, root.dtz_rounded) if resolved_dtz is None else (resolved_dtz, False)
    )

    # dtz/dtc/dtm/dtm50 are the root's own metrics, each independently
    # unavailable when this material has no DTZ/DTC/DTM/DTM50 table - see
    # _probe_board. The paired *_available flag is the authoritative signal
    # for that; the value fields fall back to null/[null, null(, null)]
    # rather than being omitted, so every response has the same fixed shape
    # regardless of coverage. dtc is a 3-tuple [wdl, pushes_owed, plies]
    # rather than dtm50's pair, since - unlike dtm50 - its distance is
    # priced in two components, not one; see chess.chesstb.ProbeResult and
    # the DTC row comment in evaluate_all_moves. root is the raw
    # ProbeResult _probe_board returns, so its wdl/dtc_wdl/dtm50_wdl are
    # still the unsigned chess.chesstb class values here - each goes
    # through _signed_wdl() below, same as every other consumer of a raw
    # ProbeResult in this file.
    json_str = json.dumps({
        "wdl":              _signed_wdl(root.wdl),
        "dtz":              root_dtz if root.has_dtz else None,
        "dtz_available":    root.has_dtz,
        "dtz_rounded":      root_dtz_rounded if root.has_dtz else None,
        "dtc":              [_signed_wdl(root.dtc_wdl), root.dtc_order, root.dtc] if root.has_dtc else [None, None, None],
        "dtc_available":    root.has_dtc,
        "dtm":              root.dtm if root.has_dtm else None,
        "dtm_available":    root.has_dtm,
        "dtm50":            [_signed_wdl(root.dtm50_wdl), root.dtm50] if root.has_dtm50 else [None, None],
        "dtm50_available":  root.has_dtm50,
        "moves_dtz":        moves_dtz,
        "moves_dtc":        moves_dtc,
        "moves_dtm":        moves_dtm,
        "moves_dtm50":      moves_dtm50,
        "summary":          summary,
        "draw_reason":      None,   # root always has legal moves here, so never terminal
    })
    if not complete:
        raise _IncompleteResult(json_str)
    return json_str


# An explicit LRU lets SSE check for a complete response before building
# boards, walking legal moves or submitting child probes. Failed/incomplete
# evaluations never enter this cache. Computation stays outside the lock.
_evaluation_cache: OrderedDict[tuple[str, str], str] = OrderedDict()
_evaluation_cache_lock = threading.Lock()
_evaluation_hits = _evaluation_misses = 0
_EvaluationCacheInfo = namedtuple("CacheInfo", "hits misses maxsize currsize")


def _cached_evaluation(variant: str, fen: str, *, count_miss: bool = True) -> str | None:
    global _evaluation_hits, _evaluation_misses
    with _evaluation_cache_lock:
        key = (variant, fen)
        result = _evaluation_cache.get(key)
        if result is not None:
            _evaluation_cache.move_to_end(key)
            _evaluation_hits += 1
        elif count_miss:
            _evaluation_misses += 1
        return result


def evaluate_fen(variant: str, fen: str) -> str:
    cached = _cached_evaluation(variant, fen)
    if cached is not None:
        return cached
    try:
        result = _evaluate_fen_impl(variant, fen)
    except _IncompleteResult as incomplete:
        return incomplete.json_str
    with _evaluation_cache_lock:
        key = (variant, fen)
        _evaluation_cache[key] = result
        _evaluation_cache.move_to_end(key)
        while len(_evaluation_cache) > cfg.evaluate_cache_size:
            _evaluation_cache.popitem(last=False)
    return result


def _clear_evaluation_cache() -> None:
    global _evaluation_hits, _evaluation_misses
    with _evaluation_cache_lock:
        _evaluation_cache.clear()
        _evaluation_hits = _evaluation_misses = 0


def _evaluation_cache_info():
    with _evaluation_cache_lock:
        return _EvaluationCacheInfo(
            _evaluation_hits, _evaluation_misses, cfg.evaluate_cache_size, len(_evaluation_cache),
        )


evaluate_fen.cache_clear = _clear_evaluation_cache  # type: ignore[attr-defined]
evaluate_fen.cache_info = _evaluation_cache_info  # type: ignore[attr-defined]


# ── Common (variant, FEN) extraction + validation helper ──────────────────────

def _extract_probe_request(data: dict | None) -> tuple[str, str, Response | tuple[Response, int] | None]:
    """Single authoritative request-body parser shared by /probe and
    /probe/stream, so the two transports can't drift on what counts as a
    valid variant/FEN or how each is rejected. Returns (variant, fen,
    error_response); error_response is None on success and variant/fen are
    both "" when it isn't."""
    if not data or "fen" not in data:
        return "", "", (jsonify({"error": "Missing FEN in request body."}), 400)
    fen_raw = data["fen"]
    if not isinstance(fen_raw, str):
        return "", "", (jsonify({"error": "FEN must be a string."}), 400)
    fen = fen_raw.strip()
    if not fen:
        return "", "", (jsonify({"error": "FEN string is empty."}), 400)
    if len(fen) > cfg.max_fen_length:
        return "", "", (jsonify({"error": f"FEN string too long (max {cfg.max_fen_length})."}), 400)

    try:
        variant = _normalize_variant(data.get("variant"))
    except ValueError as e:
        return "", "", (jsonify({"error": str(e)}), 400)

    return variant, _normalize_fen(fen), None


# ── Routes ────────────────────────────────────────────────────────────────────

@app.route("/")
def index() -> str:
    return render_template("index.html", source_repository_url=cfg.source_repository_url)


@app.route("/probe", methods=["POST"])
@_probe_rate_limit
def probe() -> Response | tuple[Response, int]:
    body = request.get_json(silent=True)
    variant, fen, err = _extract_probe_request(body)
    if err:
        return err

    try:
        json_str = evaluate_fen(variant, fen)
        return app.response_class(json_str, status=200, mimetype="application/json")

    except chesstb.MissingTableError:
        return jsonify(_missing_table_error_payload(variant, fen)), 400

    except RuntimeError as e:
        # Tablebase not initialised (TABLEBASE_PATH failed to load) --
        # same condition and status code as /health's own "degraded"
        # response for this, and the same message _validate_and_build_board
        # raises for /probe/stream to report over SSE.
        return jsonify({"error": str(e)}), 503

    except ValueError as e:
        return jsonify({"error": str(e)}), 400

    except _REMOTE_SOURCE_ERROR as e:
        log.warning("Remote tablebase fetch failed probing variant=%s FEN %s: %s", variant, fen, e)
        return jsonify({
            "error": "Could not reach the remote tablebase. Try again shortly.",
            "error_code": "remote_unavailable",
        }), 502

    except ProbeInFlightTimeout as e:
        log.warning("%s", e)
        return jsonify({
            "error": "Probe timed out. Try again shortly.",
            "error_code": "probe_timeout",
        }), 503

    except Exception:
        log.exception("Unhandled error probing variant=%s FEN: %s", variant, fen)
        return jsonify({"error": "Internal server error."}), 500


def _stream_error_payload(variant: str, fen: str, exc: BaseException) -> dict:
    """SSE error payload for an exception raised while probing (variant, fen),
    keeping wording, error_code, and logging identical no matter which probe
    step inside probe_stream() raised it."""
    if isinstance(exc, chesstb.MissingTableError):
        return {"status": "error", **_missing_table_error_payload(variant, fen)}
    if isinstance(exc, _REMOTE_SOURCE_ERROR):
        log.warning("Remote tablebase fetch failed in probe/stream for variant=%s %s: %s", variant, fen, exc)
        return {
            "status": "error",
            "error": "Could not reach the remote tablebase. Try again shortly.",
            "error_code": "remote_unavailable",
        }
    if isinstance(exc, ProbeInFlightTimeout):
        log.warning("%s", exc)
        return {
            "status": "error",
            "error": "Probe timed out. Try again shortly.",
            "error_code": "probe_timeout",
        }
    log.exception("Error in probe/stream for variant=%s %s", variant, fen)
    return {"status": "error", "error": str(exc)}


@app.route("/probe/stream", methods=["POST"])
@_probe_rate_limit
def probe_stream() -> Response | tuple[Response, int]:
    body = request.get_json(silent=True)
    variant, fen, err = _extract_probe_request(body)
    if err:
        return err

    def _sse(obj: dict) -> str:
        return f"data: {json.dumps(obj)}\n\n"

    def generate():
        cached = _cached_evaluation(variant, fen, count_miss=False)
        if cached is not None:
            yield 'data: {"status":"done",' + cached[1:] + '\n\n'
            return
        # Validated up front (rather than only inside evaluate_fen below) so
        # the validation error can be reported immediately over SSE instead of
        # only finding out after pre-warming child positions. Uses the same
        # _validate_and_build_board() that evaluate_fen() itself calls, so
        # there's one authoritative implementation of "what makes a FEN
        # probeable" instead of two hand-written copies that could disagree.
        try:
            root_board = _validate_and_build_board(variant, fen)
        except (RuntimeError, ValueError) as e:
            yield _sse({"status": "error", "error": str(e)})
            return

        # A terminal root (checkmate, stalemate, insufficient material) never
        # reaches the tablebase - see evaluate_fen()'s own early return for
        # these - so only a non-terminal root needs its coverage confirmed
        # here. Checking the root before pre-warming any child means a root
        # the tablebase doesn't cover is caught immediately: no child of an
        # uncovered root is coverable either, so pre-warming first would only
        # probe the entire move list to discover that after the fact. This
        # reads through the same cache evaluate_fen() itself reads below, so
        # a covered root costs nothing extra downstream.
        terminal = (
            root_board.is_checkmate()
            or root_board.is_stalemate()
            or root_board.is_insufficient_material()
        )
        if not terminal:
            try:
                # root_board was just built above from this exact (variant,
                # fen), so it's passed through here too rather than making
                # _probe_fen_deduped re-parse the same string a second
                # time -- see _probe_fen_with_deadline's board parameter.
                root_probe = _probe_fen_with_deadline(
                    variant, fen, time.monotonic() + cfg.probe_timeout, root_board,
                )
            except Exception as e:
                yield _sse(_stream_error_payload(variant, fen, e))
                return
            if root_probe is None:
                yield _sse({"status": "error", **_missing_table_error_payload(variant, fen)})
                return

        # Insufficient-material draws can still have legal moves, but
        # evaluate_fen returns a terminal response without using children.
        move_info = [] if terminal else _collect_move_info(root_board)
        child_fens: set[str] = {
            child_fen for _, _, _, _, _, child_fen, _ in move_info if child_fen is not None
        }
        child_boards: dict[str, chess.Board] = {
            child_fen: child_board
            for _, _, _, _, _, child_fen, child_board in move_info
            if child_fen is not None and child_board is not None
        }

        total = len(child_fens)
        yield _sse({"status": "probing", "completed": 0, "total": total})

        # Pre-warm child FENs in parallel using the native non-blocking cache.
        # Always via the thread pool (regardless of cfg.parallel_threshold, which
        # only governs evaluate_all_moves' own sequential/parallel choice below):
        # a bare sequential loop here would have no way to bound a single hung
        # _probe_fen_with_deadline() call, since there'd be no worker thread to
        # poll against; only a background thread lets this deadline actually
        # cut the wait short. This budget is independent of the root probe's
        # own, above - a slow root probe doesn't eat into pre-warm's time.
        deadline = time.monotonic() + cfg.probe_timeout
        futures = {
            cf: _executor.submit(
                _probe_fen_with_deadline, variant, cf, deadline, child_boards.get(cf)
            )
            for cf in child_fens
        }
        remaining = set(futures.values())
        done_count = 0
        prewarm_complete = True
        last_event = time.monotonic()
        try:
            while done_count < total:
                time_left = deadline - time.monotonic()
                if time_left <= 0:
                    log.warning(
                        "Probe pre-warm timed out for variant=%s %s (%d/%d child positions still pending)",
                        variant, fen, total - done_count, total,
                    )
                    prewarm_complete = False
                    break
                newly_done, remaining = futures_wait(
                    remaining, timeout=min(0.3, time_left), return_when=FIRST_COMPLETED
                )
                if any(f.cancelled() or f.exception() is not None for f in newly_done):
                    prewarm_complete = False
                if newly_done:
                    done_count += len(newly_done)
                    yield _sse({"status": "probing", "completed": done_count, "total": total})
                    last_event = time.monotonic()
                elif time.monotonic() - last_event >= 1.0:
                    # Keep the stream live and detect disconnects without
                    # making the browser render identical progress updates.
                    yield ": keep-alive\n\n"
                    last_event = time.monotonic()
        finally:
            # Also runs when the streaming client disconnects. Running probes
            # may finish, but queued work should not delay newer requests.
            for future in remaining:
                future.cancel()

        # If pre-warming above finished, every child FEN is already cached, so
        # evaluate_fen() below can fetch them all as cache hits sequentially,
        # without the overhead of resubmitting each one to the thread pool. If
        # pre-warming timed out instead, skip that shortcut: evaluate_all_moves()
        # inside evaluate_fen() then takes its own parallel path, with its own
        # fresh cfg.probe_timeout budget (see _evaluate_fen_impl), and reports
        # "unknown" for whatever's still unresolved once that budget passes.
        _evaluate_tls.bypass_parallel = prewarm_complete
        # Already built above (root_board, move_info) to size the progress
        # bar and know which child FENs to pre-warm - handing them to
        # evaluate_fen() here means a cache-miss FEN doesn't pay for a
        # second board build + legal-move walk right after this one.
        _evaluate_tls.board     = root_board
        _evaluate_tls.move_info = move_info
        try:
            json_str = evaluate_fen(variant, fen)
            yield 'data: {"status":"done",' + json_str[1:] + '\n\n'

        except Exception as e:
            yield _sse(_stream_error_payload(variant, fen, e))
        finally:
            _evaluate_tls.bypass_parallel = False
            _evaluate_tls.board     = None
            _evaluate_tls.move_info = None

    return Response(
        stream_with_context(generate()),
        content_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@app.route("/health")
def health() -> Response:
    if TB is None:
        return jsonify({"status": "degraded", "reason": "Tablebase not initialised"}), 503
    return jsonify({"status": "ok"}), 200


def _cache_info_dict(info) -> dict:
    hits, misses       = info.hits, info.misses
    maxsize, currsize  = info.maxsize, info.currsize
    total = hits + misses
    return {
        "hits":     hits,
        "misses":   misses,
        "hit_rate": round(hits / total, 4) if total else 0.0,
        "maxsize":  maxsize,
        "currsize": currsize,
    }


@app.route("/admin/cache/clear", methods=["POST"])
@_require_admin
def clear_cache() -> Response:
    evaluate_fen.cache_clear()
    _probe_fen.cache_clear()
    # The two LRUs above are app.py-level caches keyed on (variant, FEN)
    # pairs. They are not the only cache in the probing pipeline: each tier's
    # underlying tablebase can keep its own internal decoded-block cache
    # and, in remote mode, a downloaded-file cache (remote/remote_fallback.py).
    # SourcedTablebase.clear_caches()/cache_stats() are exposed on TB
    # itself, but only actually clear/report something for a tier backed
    # by a remote/remote_fallback.py Tablebase -- a plain local-directory
    # Tablebase has no such cache. cache_stats() coming back empty is what
    # distinguishes "nothing to clear" from "cleared something", since
    # every tier is asked unconditionally either way.
    tb_cache_info: dict = {}
    if TB is not None:
        TB.clear_caches()
        tb_cache_info = TB.cache_stats()
    tb_cache_cleared = bool(tb_cache_info)
    log.info(
        "All probe caches cleared (tablebase-internal cache %s).",
        "cleared" if tb_cache_cleared else "not applicable for this tablebase",
    )
    response = {
        "status":             "ok",
        "message":            "All probe caches cleared (root JSON cache + child probe cache"
                               + (" + tablebase-internal caches)." if tb_cache_cleared else ")."),
        "evaluate_fen_cache": _cache_info_dict(evaluate_fen.cache_info()),
        "probe_fen_cache":    _cache_info_dict(_probe_fen.cache_info()),
        "tablebase_internal_cache_cleared": tb_cache_cleared,
    }
    if tb_cache_info:
        response["tablebase_cache"] = tb_cache_info
    return jsonify(response)


@app.route("/admin/cache/stats", methods=["GET"])
@_require_admin
def cache_stats() -> Response:
    stats = {
        "evaluate_fen_cache": _cache_info_dict(evaluate_fen.cache_info()),
        "probe_fen_cache":    _cache_info_dict(_probe_fen.cache_info()),
        "thread_pool": {
            "max_workers":        cfg.probe_threads,
            "parallel_threshold": cfg.parallel_threshold,
            "probe_timeout_secs": cfg.probe_timeout,
        },
        "config": {
            "evaluate_cache_size": cfg.evaluate_cache_size,
            "probe_cache_size":    cfg.probe_cache_size,
            "block_cache_bytes":   cfg.block_cache_bytes,
        },
        "tablebase": {
            "configured": TB is not None,
        },
    }
    # SourcedTablebase.cache_stats() is exposed on TB and reports the configured source
    # -- a source contributes no key when it has no remote-backed cache to
    # report (see clear_cache() above for why a plain local-directory
    # Tablebase has none).
    if TB is not None:
        tb_cache = TB.cache_stats()
        if tb_cache:
            stats["tablebase_cache"] = tb_cache
    return jsonify(stats)


@app.route("/admin/login", methods=["POST"])
@_admin_login_rate_limit
def admin_login() -> Response:
    if not cfg.admin_token:
        return jsonify({"error": "Admin panel disabled: ADMIN_TOKEN is not configured."}), 503
    body = request.get_json(silent=True) or {}
    token = body.get("token", "")
    if not isinstance(token, str) or not _admin_token_matches(token):
        return jsonify({"error": "Incorrect admin token."}), 401
    session["admin_ok"] = True
    return jsonify({"status": "ok"})


@app.route("/admin/logout", methods=["POST"])
def admin_logout() -> Response:
    session.pop("admin_ok", None)
    return jsonify({"status": "ok"})


@app.route("/admin", methods=["GET"])
def admin_dashboard() -> str | tuple[str, int]:
    # Renders the dashboard once authenticated (session cookie from
    # /admin/login); a login form otherwise. Unlike /admin/cache/* below,
    # this route can't just use _require_admin -- a browser visiting /admin
    # needs the login form itself, not a bare 401/503 JSON body. Matches
    # /admin/login and /admin/cache/*'s own 503 when ADMIN_TOKEN is unset,
    # rather than serving the dashboard shell with a 200.
    if not cfg.admin_token:
        return render_template(
            "admin.html", authenticated=False, admin_configured=False,
            source_repository_url=cfg.source_repository_url,
        ), 503
    return render_template(
        "admin.html",
        authenticated=_is_admin_authenticated(),
        admin_configured=True,
        source_repository_url=cfg.source_repository_url,
    )


# ── OpenAPI specification ─────────────────────────────────────────────────────
# The specification is maintained in openapi.yaml (project root).

_OPENAPI_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "openapi.yaml")


@app.route("/openapi.yaml")
def openapi_spec() -> Response:
    return send_file(_OPENAPI_PATH, mimetype="application/yaml")


if __name__ == "__main__":
    if cfg.debug:
        # Flask's own dev server: auto-reload + interactive debugger, for
        # local development only. `threaded=True` is what lets a long-lived
        # /probe/stream SSE connection coexist with other concurrent
        # requests on this single-process server (see README.md "Running in
        # production").
        log.info("DEBUG = True (config.py) — starting Flask development server on %s:%s", cfg.host, cfg.port)
        app.run(debug=True, host=cfg.host, port=cfg.port, threaded=True)
    else:
        # Production: serve via waitress, a pure-Python production-grade
        # WSGI server. `threads` gives waitress its own pool of
        # request-handling threads, playing the same role `threaded=True`
        # plays for the Flask dev server above - without it, a single
        # open /probe/stream connection could starve every other request
        # (see README.md "Running in production").
        #
        # trusted_proxy=* / trusted_proxy_count / trusted_proxy_headers:
        # waitress parses X-Forwarded-For and X-Forwarded-Proto and
        # rewrites REMOTE_ADDR / wsgi.url_scheme itself, before the
        # request ever reaches the Flask app object -- so it sits in front
        # of, and takes priority over, the ProxyFix layer wired onto
        # app.wsgi_app above. Without this, waitress's default
        # (trusted_proxy=None) strips both headers entirely as untrusted,
        # silently discarding them before ProxyFix could ever see them --
        # ProxyFix alone is not sufficient here. `"*"` trusts whichever
        # peer connects, rather than one specific IP, because Render's
        # edge/load-balancer's own address isn't a fixed, documented
        # value; this is safe under the same assumption
        # TRUSTED_PROXY_COUNT's docstring already states -- that nothing
        # but that one proxy hop can reach this process directly. Skipped
        # entirely when cfg.trusted_proxy_count is 0 (no reverse proxy in
        # front at all), leaving waitress's own default in place.
        log.info(
            "Starting waitress production server on %s:%s (threads=%s)",
            cfg.host, cfg.port, cfg.waitress_threads,
        )
        waitress_kwargs = {}
        if cfg.trusted_proxy_count > 0:
            waitress_kwargs.update(
                trusted_proxy="*",
                trusted_proxy_count=cfg.trusted_proxy_count,
                trusted_proxy_headers={"x-forwarded-for", "x-forwarded-proto"},
            )
        waitress_serve(
            app, host=cfg.host, port=cfg.port, threads=cfg.waitress_threads, **waitress_kwargs)