eval_harness.py

#!/usr/bin/env python3
"""
eval_harness.py — ImageForge Eval Harness
==========================================
Drives the live FastAPI /generate endpoint, collects images, and computes
clean-FID + HPSv2 + GenEval2 (Soft-TIFA) for two model keys (base vs candidate).

Usage
-----
  # Base vs base (noise floor, different seeds):
  python eval_harness.py --base sd-turbo --candidate sd-turbo --noise-floor

  # Base vs LoRA candidate:
  python eval_harness.py --base sd-turbo --candidate sdxl-turbo

  # Skip GenEval2 if Qwen3-VL is not available locally:
  python eval_harness.py --base sd-turbo --candidate sdxl-turbo --skip-geneval

  # Run FID+HPS only (fastest):
  python eval_harness.py --base sd-turbo --candidate sdxl-turbo --skip-geneval

Requirements
------------
Install on the Mac (MPS box) — the serving environment:
  pip install clean-fid hpsv2 httpx Pillow tqdm

GenEval2 (runs on CUDA box — Qwen3-VL-8B needs ~19 GB VRAM at FP16, ~5 GB at INT4):
  git clone https://github.com/facebookresearch/GenEval2
  cd GenEval2
  pip install torch transformers==4.57.0 pillow tqdm scipy
  # For INT4 on 4090: pip install bitsandbytes
  # Copy geneval2_data.jsonl from the repo into this directory or pass --geneval2-data.

API
---
Drives http://127.0.0.1:8765/generate with GenerateRequest schema:
  {
    "prompt": str, "model": str|null, "steps": int|null, "guidance": float|null,
    "width": int|null, "height": int|null, "seed": int,
    "return_mode": "base64"
  }
Response: GenerationResponse.image_base64 (PNG, base64).
Seeds are pinned per-prompt for reproducibility; noise-floor run uses seed+1 offsets.
The engine uses torch.Generator(device="cpu").manual_seed(seed) — deterministic across runs.
"""

from __future__ import annotations

import argparse
import base64
import hashlib
import io
import json
import logging
import os
import shutil
import sys
import time
from pathlib import Path
from typing import Optional

# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s  %(levelname)-8s  %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("eval")

# ---------------------------------------------------------------------------
# Constants / thresholds
# ---------------------------------------------------------------------------
API_URL = "http://127.0.0.1:8765"
GENERATE_ENDPOINT = f"{API_URL}/generate"

# Optional X-API-Key sent with every request. The API's key gate is opt-in
# (app.py: auth_required defaults False), but once an operator turns it on,
# EVERY /generate call from this harness 401s and the whole eval dies at the
# first image with an opaque HTTP error -- this file previously had no way to
# authenticate at all. Populated from --api-key, else IMAGEFORGE_API_KEY, else
# the same ~/.imageforge/secrets.env the server itself reads, so an eval on a
# box with auth already enabled just works.
API_KEY: Optional[str] = None


def _load_api_key_from_secrets() -> Optional[str]:
    """Read IMAGEFORGE_API_KEY out of ~/.imageforge/secrets.env.

    Deliberately a tiny stdlib parser rather than a python-dotenv dependency:
    this harness is run from a bare eval environment (clean-fid/hpsv2/httpx
    only) and must not grow install steps just to read one KEY=value line.
    """
    path = Path.home() / ".imageforge" / "secrets.env"
    try:
        for line in path.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            key, _, value = line.partition("=")
            if key.strip() == "IMAGEFORGE_API_KEY":
                return value.strip().strip('"').strip("'") or None
    except OSError:
        return None
    return None


def _auth_headers() -> dict:
    """Headers for every outbound API call.

    X-Client-Name makes this harness show up as a named caller in the API's
    own GET /clients registry instead of an anonymous User-Agent, so a long
    eval run is attributable while it is happening.
    """
    headers = {"X-Client-Name": "eval_harness"}
    if API_KEY:
        headers["X-API-Key"] = API_KEY
    return headers

# Pass/fail gate thresholds (see emit_report for the gate model).
#
# Gate model (redesigned):
#   * HPSv2 is the PRIMARY gate — higher is better, always measurable, so the
#     gate always rests on it. Delta ≥ +0.20 AND > 2×noise_std is pass.
#   * clean-FID is a CORROBORATING signal computed ONLY against a real-image
#     reference set (--reference-dir): fid_base = FID(ref, base),
#     fid_cand = FID(ref, cand); delta = fid_cand - fid_base must be ≤ -5 (and
#     beyond the noise floor) — a LOWER candidate FID is the improvement.
#     With no reference set FID is N/A and does not gate (never fabricated).
#   * GenEval2 Soft-TIFA_AM: higher is better. Delta ≥ +2.0 pp AND > 2×noise_std.
FID_PASS_DELTA = -5.0          # candidate FID(ref,cand) lower than base FID(ref,base) by ≥5 → corroborates
HPS_PASS_DELTA = 0.20          # candidate HPS higher by ≥0.20 points (100-scale) → pass
GENEVAL_PASS_DELTA = 2.0       # candidate Soft-TIFA_AM higher by ≥2.0 pp → pass
NOISE_SIGMA_MULTIPLIER = 2.0   # delta must also exceed 2×noise_std (per metric)

# Number of images per model run. 200 images gives ±0.3 FID CI vs ±0.2 at N=1000.
# For a LoRA vs base comparison on a held-out prompt set, N=200 is adequate.
DEFAULT_N_IMAGES = 200  # use --n-images to override

# Default resolution — matches the sd-turbo/sdxl-turbo defaults in the registry.
DEFAULT_WIDTH = 512
DEFAULT_HEIGHT = 512

# Below this, an SDXL-architecture model does not merely look worse -- it
# collapses into tiled, saturated abstract mush. Scoring THAT with HPSv2
# produces a real-looking number computed over garbage, and because both the
# base and the candidate degrade the same way, the delta stays plausibly near
# zero and the gate reads as a legitimate "no improvement" verdict. A silently
# meaningless gate is worse than a loud failure, hence the guard below.
SDXL_MIN_USABLE_RESOLUTION = 768


def _looks_like_sdxl(model_key: Optional[str]) -> bool:
    """True for SDXL-architecture models that actually need >=768px.

    Turbo variants are deliberately excluded. `sdxl-turbo` is SDXL-architecture
    but is a distilled model TRAINED at 512 -- checked against the registry, it
    is the only `sdxl-*` entry with `default_size=512`, and it is exactly the
    one carrying `turbo=True`; every other sdxl-* is 1024. Flagging it would
    reject the harness's own default candidate for a problem it does not have.

    This is a string heuristic on purpose: the harness runs from a bare eval
    environment that cannot import imageforge to consult MODEL_REGISTRY.
    """
    if not model_key:
        return False
    key = model_key.lower()
    if "turbo" in key:
        return False
    return "xl" in key


def _check_resolution_sanity(
    base: Optional[str],
    candidate: Optional[str],
    width: Optional[int],
    height: Optional[int],
) -> tuple[int, int]:
    """Resolve --width/--height and refuse a silently-meaningless SDXL eval.

    Returns the (width, height) to use.

    The distinction that matters is INTENT. Explicitly passing a small size
    is a legitimate thing to want (a fast smoke run, reproducing an old
    result), so that only warns. Falling into 512 because the default was
    written for sd-turbo, while evaluating SDXL, is not a choice anyone made
    -- and it yields a gate verdict computed entirely over degenerate images.
    That is refused, with the fix spelled out.
    """
    defaulted = width is None and height is None
    w = width if width is not None else DEFAULT_WIDTH
    h = height if height is not None else DEFAULT_HEIGHT

    sdxl_models = [m for m in (base, candidate) if _looks_like_sdxl(m)]
    if sdxl_models and min(w, h) < SDXL_MIN_USABLE_RESOLUTION:
        message = (
            f"{sdxl_models} is SDXL-architecture but the eval resolution is {w}x{h}. "
            f"SDXL degenerates into tiled abstract mush below {SDXL_MIN_USABLE_RESOLUTION}px; "
            "HPSv2 would score garbage for BOTH models and the delta would look "
            "like a legitimate 'no improvement' verdict."
        )
        if defaulted:
            sys.exit(
                f"REFUSING TO RUN: {message}\n"
                f"  This resolution came from the default ({DEFAULT_WIDTH}x{DEFAULT_HEIGHT}), which is "
                "sized for sd-turbo, not from anything you asked for.\n"
                f"  Fix: pass --width 1024 --height 1024 (or any size >= {SDXL_MIN_USABLE_RESOLUTION}).\n"
                "  To override deliberately, pass the small size explicitly."
            )
        log.warning("EXPLICIT small resolution for an SDXL model: %s", message)

    return w, h

# ---------------------------------------------------------------------------
# Prompt set loader
# ---------------------------------------------------------------------------
PROMPTS_FILE = Path(__file__).parent / "prompts_heldout.jsonl"


def load_prompts(path: Path) -> list[dict]:
    """Load JSONL prompt set. Each line: {"prompt": str, "seed": int, ...}"""
    if not path.exists():
        raise FileNotFoundError(
            f"Prompt file not found: {path}\n"
            "Create prompts_heldout.jsonl next to eval_harness.py, or pass --prompts."
        )
    records = []
    with path.open() as fh:
        for line in fh:
            line = line.strip()
            if line and not line.startswith("//"):
                records.append(json.loads(line))
    log.info("Loaded %d prompts from %s", len(records), path)
    return records


# ---------------------------------------------------------------------------
# Image generation driver
# ---------------------------------------------------------------------------
def generate_images(
    prompts: list[dict],
    model: Optional[str],
    out_dir: Path,
    *,
    seed_offset: int = 0,
    steps: Optional[int] = None,
    guidance: Optional[float] = None,
    width: int = DEFAULT_WIDTH,
    height: int = DEFAULT_HEIGHT,
    n_images: int = DEFAULT_N_IMAGES,
) -> list[tuple[Path, dict]]:
    """
    POST each prompt to /generate, save returned PNG to out_dir.

    Respects the engine's single-inference-lock: sends one request at a time
    (no concurrency) with a 300s per-image timeout.

    Seeds are pinned per-prompt (prompt["seed"] + seed_offset) for reproducibility.
    The engine uses torch.Generator(device="cpu").manual_seed(seed) which is
    deterministic across MPS restarts (pipeline.py:573).

    Returns a list of (image_path, prompt_record) pairs -- NOT just paths.
    Bug fixed here: this used to return a bare list[Path], and any single
    failed request (network blip, one bad base64) was `continue`d past,
    shrinking the list by one with no marker of WHICH prompt was skipped.
    Downstream, score_hps()/build_geneval_image_map() re-derived prompt
    alignment by re-tiling the original prompt list to len(images) and
    zip()-ping positionally -- so every image after a single dropped one
    silently shifted onto the WRONG prompt for the rest of the batch, with
    no exception and no warning, just a plausible-looking but wrong score.
    Pairing the prompt with its image at the moment it's actually saved
    makes misalignment structurally impossible instead of re-derived.
    """
    try:
        import httpx
    except ImportError:
        sys.exit("httpx not installed. Run: pip install httpx")

    out_dir.mkdir(parents=True, exist_ok=True)
    saved: list[tuple[Path, dict]] = []

    # Tile prompts up to n_images if fewer prompts than requested
    tiled: list[dict] = []
    while len(tiled) < n_images:
        tiled.extend(prompts)
    tiled = tiled[:n_images]

    if len(prompts) < n_images:
        # Transparency, not a behavior change: each unique prompt repeats
        # exactly (n_images // len(prompts)) times, and every repeat reuses
        # that record's OWN "seed" field verbatim (idx only affects the
        # fallback default when a record has no explicit seed) -- so with
        # the default prompts_heldout.jsonl (30 records, every one already
        # carries an explicit seed) and DEFAULT_N_IMAGES=200, most of "N=200
        # samples" are pixel-identical duplicates of ~30 real generations,
        # not independent samples. This understates true variance for both
        # FID (near-singular covariance risk) and HPS. Not silently changing
        # seed derivation here (that would break reproducibility of any
        # past eval run compared against this code) -- just surfacing it.
        log.warning(
            "n_images=%d > %d available prompts -- each prompt repeats ~%.1fx "
            "with its own seed reused verbatim on every repeat (pixel-identical "
            "duplicates, not independent samples). Pass more prompts or reduce "
            "--n-images for a truly independent sample.",
            n_images, len(prompts), n_images / len(prompts),
        )

    label = model or "default"
    log.info("Generating %d images with model=%s → %s", len(tiled), label, out_dir)

    with httpx.Client(base_url=API_URL, timeout=300.0, headers=_auth_headers()) as client:
        # Health check first — fail fast rather than timeout N times
        try:
            health = client.get("/health").json()
            log.info("Engine health: status=%s device=%s", health.get("status"), health.get("device"))
        except Exception as e:
            sys.exit(f"Cannot reach ImageForge at {API_URL}. Start with: ./run_api.sh\nError: {e}")

        for idx, rec in enumerate(tiled):
            seed = (rec.get("seed", idx * 1000) + seed_offset) & 0xFFFFFFFF
            payload: dict = {
                "prompt": rec["prompt"],
                "seed": seed,
                "width": width,
                "height": height,
                "return_mode": "base64",
                "use_cache": False,  # always generate fresh for eval
            }
            if model:
                payload["model"] = model
            if steps is not None:
                payload["steps"] = steps
            if guidance is not None:
                payload["guidance"] = guidance

            out_file = out_dir / f"{idx:04d}_seed{seed}.png"
            if out_file.exists():
                log.debug("[%d/%d] cached %s", idx + 1, len(tiled), out_file.name)
                saved.append((out_file, rec))
                continue

            t0 = time.perf_counter()
            try:
                resp = client.post("/generate", json=payload)
                resp.raise_for_status()
            except httpx.HTTPStatusError as e:
                log.error("[%d/%d] HTTP %s: %s", idx + 1, len(tiled), e.response.status_code, e.response.text[:200])
                continue
            except Exception as e:
                log.error("[%d/%d] request failed: %s", idx + 1, len(tiled), e)
                continue

            data = resp.json()
            b64 = data.get("image_base64")
            if not b64:
                log.warning("[%d/%d] no image_base64 in response: %s", idx + 1, len(tiled), list(data.keys()))
                continue

            img_bytes = base64.b64decode(b64)
            out_file.write_bytes(img_bytes)
            elapsed = time.perf_counter() - t0
            log.info("[%d/%d] seed=%d %.1fs → %s", idx + 1, len(tiled), seed, elapsed, out_file.name)
            saved.append((out_file, rec))

    if len(saved) < len(tiled):
        log.warning(
            "Only %d/%d images generated successfully -- the real sample size "
            "for scoring is %d, not the requested %d.",
            len(saved), len(tiled), len(saved), len(tiled),
        )
    log.info("Generated %d/%d images", len(saved), len(tiled))
    return saved


# ---------------------------------------------------------------------------
# clean-FID scoring (on-Mac, CPU or MPS via torch device override)
# ---------------------------------------------------------------------------
def score_fid(dir_base: Path, dir_candidate: Path, *, device_str: str = "cpu") -> float:
    """
    Compute clean-FID between two image directories.

    Uses mode='clean' (PIL-bicubic resizing) to avoid the Parmar 2022 resize bug
    where OpenCV/PyTorch bicubic produced FID ≥6 offset vs correct PIL-bicubic.

    device_str: 'cpu' for Mac MPS box (Inception runs fine on CPU, ~2min for N=200).
                'cuda' for the CUDA training box.
    """
    try:
        from cleanfid import fid
    except ImportError:
        sys.exit("clean-fid not installed. Run: pip install clean-fid")

    import torch

    device = torch.device(device_str)
    log.info("Computing FID: %s vs %s (device=%s) …", dir_base.name, dir_candidate.name, device_str)
    score = fid.compute_fid(
        str(dir_base),
        str(dir_candidate),
        mode="clean",          # correct PIL-bicubic resize — never legacy_pytorch/legacy_tensorflow
        model_name="inception_v3",
        num_workers=0,         # 0 = no multiprocessing (avoids fork issues on macOS)
        batch_size=32,
        device=device,
        verbose=True,
    )
    return float(score)


# ---------------------------------------------------------------------------
# HPSv2 scoring (on-Mac, CPU or MPS)
# ---------------------------------------------------------------------------
def score_hps_per_image(
    image_prompt_pairs: list[tuple[Path, dict]],
    *,
    hps_version: str = "v2.1",
) -> list[tuple[Path, dict, float]]:
    """Return per-image HPSv2 scores as ``(image_path, prompt_record, score)``.

    This is the raw, per-image basis both :func:`score_hps` (which averages it)
    and the paired bootstrap CI (which needs the individual samples, not just
    the mean) consume. Keeping alignment with the source ``(path, prompt)``
    pair here means the bootstrap pairs candidate-vs-base by IMAGE INDEX with
    no positional re-derivation. Images that fail to score are omitted from
    the returned list (logged), so a caller can detect a shrunk sample.
    """
    try:
        import hpsv2
    except ImportError:
        sys.exit("hpsv2 not installed. Run: pip install hpsv2")

    out: list[tuple[Path, dict, float]] = []
    log.info("Scoring %d images with HPSv2.%s …", len(image_prompt_pairs), hps_version)
    for img_path, rec in image_prompt_pairs:
        try:
            result = hpsv2.score(str(img_path), rec["prompt"], hps_version=hps_version)
            # hpsv2.score returns a list when given a list of images; a scalar for a single path
            val = result[0] if isinstance(result, (list, tuple)) else float(result)
            out.append((img_path, rec, float(val)))
        except Exception as e:
            log.warning("HPSv2 error on %s: %s", img_path.name, e)
    return out


def score_hps(
    image_prompt_pairs: list[tuple[Path, dict]],
    *,
    hps_version: str = "v2.1",
    dump_scores_path: "Optional[Path]" = None,
) -> float:
    """
    Compute mean HPSv2 score for a list of (image_path, prompt_record) pairs.

    hpsv2.score(img, prompt) accepts: PIL.Image, file path str, or list.
    Returns a logit score; typical range for SDXL-class models: 26–30 on 100-scale.
    We use v2.1 (released Sep 2024, trained on higher-quality HPD v2.1 dataset).

    Runs on CPU perfectly. MPS is NOT supported -- hpsv2's own img_score.py
    hardcodes device selection to "cuda" or "cpu" and never checks
    torch.backends.mps (verified directly against the installed hpsv2
    package's source; this docstring previously claimed otherwise).
    The reward model is a CLIP-ViT-H variant (~900MB download on first run).

    Takes (path, prompt) pairs directly from generate_images() -- previously
    took separate images/prompts lists and RE-DERIVED alignment by re-tiling
    prompts to len(images) and zip()-ping positionally. Any single dropped
    image during generation shifted every subsequent image onto the wrong
    prompt for the rest of the batch, silently. Pairing at the source makes
    that class of bug structurally impossible here.

    Batch 1: ``dump_scores_path`` -- when given, the per-image scores are
    persisted as JSON (a list of ``{"image": name, "prompt": text, "hps":
    score}`` records) so a paired bootstrap CI can be recomputed offline from
    the raw samples, not just the mean. Best-effort: a write failure is logged
    and never fails scoring.
    """
    scored = score_hps_per_image(image_prompt_pairs, hps_version=hps_version)

    if dump_scores_path is not None:
        try:
            dump_scores_path.parent.mkdir(parents=True, exist_ok=True)
            dump_scores_path.write_text(
                json.dumps(
                    [
                        {"image": p.name, "prompt": rec.get("prompt"), "hps": s}
                        for p, rec, s in scored
                    ],
                    indent=2,
                ),
                encoding="utf-8",
            )
            log.info("Persisted %d per-image HPS scores → %s", len(scored), dump_scores_path)
        except OSError as e:
            log.warning("Could not persist per-image HPS scores to %s: %s", dump_scores_path, e)

    if not scored:
        return float("nan")
    scores = [s for _p, _rec, s in scored]
    mean_score = sum(scores) / len(scores)
    log.info("HPSv2 mean=%.4f (N=%d)", mean_score, len(scores))
    return mean_score


# ---------------------------------------------------------------------------
# Paired bootstrap CI on the base-vs-candidate HPS delta (Batch 1)
# ---------------------------------------------------------------------------
def paired_bootstrap_ci(
    base_scores: "list[float]",
    cand_scores: "list[float]",
    *,
    pass_delta: float = HPS_PASS_DELTA,
    n_boot: int = 10_000,
    ci_level: float = 0.95,
    seed: int = 1234,
) -> dict:
    """Paired bootstrap confidence interval on the mean ``cand - base`` delta.

    The two inputs are aligned per-image: ``delta_i = cand_scores[i] -
    base_scores[i]``. We resample the *paired* deltas (not the two batches
    independently) ``n_boot`` times with replacement and take the empirical
    percentile interval of the resampled mean delta. Pairing preserves the
    per-prompt correlation between base and candidate (same prompt+seed), which
    an unpaired two-sample bootstrap would throw away, inflating the CI.

    Gate (matches the roadmap): the candidate PASSES iff the CI lies entirely
    ABOVE zero (``ci_low > 0`` -- a real, direction-correct improvement, not
    merely "excludes zero") AND the observed mean delta is at least
    ``pass_delta`` in magnitude. ``ci_excludes_zero`` is reported separately
    for transparency (a CI entirely below zero is a real regression, not a
    pass). Deterministic for a fixed ``seed`` -- uses ``random.Random`` so the
    result never depends on numpy's RNG version.

    Returns a dict with: n, delta_mean, ci_low, ci_high, ci_level, n_boot,
    seed, pass_delta, ci_excludes_zero, passes, error (None on success).
    A length mismatch or empty input returns an all-NaN result with a set
    ``error`` string rather than raising, so the report degrades gracefully.
    """
    import math
    import random as _random

    n = len(base_scores)
    base_ok = all(x == x for x in base_scores)  # not NaN
    cand_ok = all(x == x for x in cand_scores)
    if n == 0 or len(cand_scores) != n or not base_ok or not cand_ok:
        return {
            "n": n,
            "delta_mean": float("nan"),
            "ci_low": float("nan"),
            "ci_high": float("nan"),
            "ci_level": ci_level,
            "n_boot": n_boot,
            "seed": seed,
            "pass_delta": pass_delta,
            "ci_excludes_zero": False,
            "passes": False,
            "error": (
                "length mismatch"
                if (n and len(cand_scores) != n)
                else "empty input" if n == 0
                else "NaN score(s) present"
            ),
        }

    deltas = [c - b for b, c in zip(base_scores, cand_scores)]
    delta_mean = sum(deltas) / n

    rng = _random.Random(seed)
    boot_means: list[float] = []
    for _ in range(n_boot):
        acc = 0.0
        for _ in range(n):
            acc += deltas[rng.randrange(n)]
        boot_means.append(acc / n)
    boot_means.sort()

    alpha = (1.0 - ci_level) / 2.0
    lo_idx = int(math.floor(alpha * n_boot))
    hi_idx = int(math.ceil((1.0 - alpha) * n_boot)) - 1
    lo_idx = max(0, min(lo_idx, n_boot - 1))
    hi_idx = max(0, min(hi_idx, n_boot - 1))
    ci_low = boot_means[lo_idx]
    ci_high = boot_means[hi_idx]

    ci_excludes_zero = ci_low > 0.0 or ci_high < 0.0
    passes = (ci_low > 0.0) and (delta_mean >= pass_delta)

    return {
        "n": n,
        "delta_mean": delta_mean,
        "ci_low": ci_low,
        "ci_high": ci_high,
        "ci_level": ci_level,
        "n_boot": n_boot,
        "seed": seed,
        "pass_delta": pass_delta,
        "ci_excludes_zero": ci_excludes_zero,
        "passes": passes,
        "error": None,
    }


# ---------------------------------------------------------------------------
# GenEval2 Soft-TIFA scoring (run on CUDA box via SSH or local if 4090 present)
# ---------------------------------------------------------------------------
def build_geneval_image_map(image_prompt_pairs: list[tuple[Path, dict]], run_dir: Path) -> Path:
    """
    Build the image_filepath_data.json that GenEval2's evaluation.py expects.
    Format: {"<prompt_text>": "<absolute_path_to_image>"}

    Takes (path, prompt) pairs directly from generate_images() -- see
    score_hps()'s docstring for why re-deriving alignment via re-tiling was
    a real bug, fixed the same way here.

    GenEval2's own expected format keys by prompt TEXT, an external
    constraint of the tool this project doesn't control -- so when
    n_images exceeds the number of unique prompts (the default config
    does: DEFAULT_N_IMAGES=200 vs 30 heldout prompts), a later repeat of an
    already-seen prompt still overwrites the map entry for an earlier one.
    That's not fixable without changing GenEval2's own input contract, but
    it was previously silent; now it's logged so a caller can tell GenEval2
    scored fewer effective images than HPS/FID did in the same report.
    """
    mapping: dict[str, str] = {}
    for img_path, rec in image_prompt_pairs:
        prompt_text = rec["prompt"]
        # GenEval2 uses prompt as key; use last image if duplicated prompts
        mapping[prompt_text] = str(img_path.resolve())

    dropped = len(image_prompt_pairs) - len(mapping)
    if dropped > 0:
        log.warning(
            "%d of %d images share a prompt with another image already in the "
            "map -- GenEval2 will only score %d unique prompts, not %d images "
            "(an inherent limit of GenEval2's own prompt-keyed input format, "
            "not a bug in this mapping step).",
            dropped, len(image_prompt_pairs), len(mapping), len(image_prompt_pairs),
        )

    out_path = run_dir / "geneval_image_map.json"
    out_path.write_text(json.dumps(mapping, indent=2), encoding="utf-8")
    log.info("Wrote GenEval2 image map → %s (%d entries)", out_path, len(mapping))
    return out_path


def score_geneval2(
    image_map_path: Path,
    geneval2_repo: str,
    *,
    method: str = "soft_tifa_am",
    geneval2_data: Optional[str] = None,
    output_file: Optional[Path] = None,
    timeout: float = 1800.0,
) -> float:
    """
    Run GenEval2 evaluation.py as a subprocess.

    geneval2_repo: path to the cloned facebookresearch/GenEval2 directory.
    method: 'soft_tifa_am' (atom-level arithmetic mean, recommended for LoRA delta).
            'soft_tifa_gm' is more conservative (geometric mean, sensitive to zero atoms).
    timeout: seconds before the subprocess is killed (default 30 min) -- this
        previously had NO timeout at all, so a VLM-scoring hang (model load
        stall, GPU contention) blocked the whole harness indefinitely with no
        way to recover short of manually killing the process. The caller
        already wraps this call in try/except, so a TimeoutExpired here
        degrades gracefully to a skipped GenEval2 row, same as any other
        scoring failure -- it just no longer hangs forever getting there.
    Returns: aggregate score (0–100).

    NOTE: GenEval2 requires Qwen3-VL-8B-Instruct (~19GB VRAM at FP16, ~5GB at INT4).
    Run on the RTX 4090 (CUDA box). Pass --skip-geneval on the Mac if not available.
    """
    import subprocess

    repo = Path(geneval2_repo).resolve()
    eval_script = repo / "evaluation.py"
    if not eval_script.exists():
        raise FileNotFoundError(f"GenEval2 evaluation.py not found at {eval_script}")

    data_path = geneval2_data or str(repo / "geneval2_data.jsonl")
    out_path = output_file or image_map_path.parent / "geneval2_scores.json"

    cmd = [
        sys.executable, str(eval_script),
        "--benchmark_data", data_path,
        "--image_filepath_data", str(image_map_path),
        "--method", method,
        "--output_file", str(out_path),
    ]
    log.info("Running GenEval2: %s", " ".join(cmd))
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, cwd=str(repo), timeout=timeout)
    except subprocess.TimeoutExpired as exc:
        raise RuntimeError(f"GenEval2 evaluation.py timed out after {timeout:.0f}s") from exc
    if result.returncode != 0:
        log.error("GenEval2 stderr:\n%s", result.stderr[-2000:])
        raise RuntimeError(f"GenEval2 evaluation.py exited {result.returncode}")

    log.info("GenEval2 stdout:\n%s", result.stdout[-2000:])

    if not out_path.exists():
        raise FileNotFoundError(f"GenEval2 output not written: {out_path}")

    scores_data = json.loads(out_path.read_text(encoding="utf-8"))
    # Expected structure: list of per-atom scores or a dict with "score" key.
    # Aggregate: arithmetic mean of all atom scores (0–1), then ×100.
    if isinstance(scores_data, list):
        flat = [v for entry in scores_data for v in (entry if isinstance(entry, list) else [entry]) if isinstance(v, (int, float))]
    elif isinstance(scores_data, dict) and "score" in scores_data:
        return float(scores_data["score"])
    else:
        flat = list(scores_data.values()) if isinstance(scores_data, dict) else []

    if not flat:
        return float("nan")
    return 100.0 * sum(flat) / len(flat)


# ---------------------------------------------------------------------------
# Noise floor estimation (base vs base, different seeds)
# ---------------------------------------------------------------------------
def estimate_noise_floor(
    prompts: list[dict],
    model: Optional[str],
    run_dir: Path,
    *,
    width: int,
    height: int,
    n_images: int,
    hps_version: str,
    seed_offset_a: int = 0,
    seed_offset_b: int = 100_000,
) -> dict[str, float]:
    """
    Generate two batches from the same model with offset seeds.
    Returns FID and HPS std-dev between the two runs as noise floor estimates.
    """
    log.info("Estimating noise floor: %s run A vs run B …", model or "default")
    dir_a = run_dir / "noise_a"
    dir_b = run_dir / "noise_b"

    imgs_a = generate_images(prompts, model, dir_a, seed_offset=seed_offset_a, width=width, height=height, n_images=n_images)
    imgs_b = generate_images(prompts, model, dir_b, seed_offset=seed_offset_b, width=width, height=height, n_images=n_images)

    fid_noise = score_fid(dir_a, dir_b)
    hps_a = score_hps(imgs_a, hps_version=hps_version)
    hps_b = score_hps(imgs_b, hps_version=hps_version)

    # std of two-point estimate = |a - b| / sqrt(2) (conservative)
    import math
    hps_noise_std = abs(hps_a - hps_b) / math.sqrt(2)

    return {
        "fid_a": fid_noise,     # FID between same-model different-seed runs
        "hps_a": hps_a,
        "hps_b": hps_b,
        "hps_noise_std": hps_noise_std,
        "fid_noise_floor": fid_noise,  # treat same-model FID as noise floor for FID
    }


# ---------------------------------------------------------------------------
# Batch 23: Anatomy plausibility (hands/limbs) via mediapipe hand landmarks
# ---------------------------------------------------------------------------
#
# IMPORTANT SCOPE / LIMITATION (stated plainly, not overclaimed):
#   mediapipe's hand landmarker fits a FIXED 21-point / 5-finger template onto
#   whatever it detects. It does NOT count fingers -- it structurally CANNOT
#   flag "this hand has six fingers" or "two fingers are fused", because it
#   always returns exactly 21 points regardless of the true digit count. No
#   permissively-licensed fully-automated polydactyly/finger-count detector
#   exists today (OpenPose weights are CMU non-commercial; MMPose is Apache-2.0
#   but a far heavier torch/mmcv chain for no clear win here). So this is a
#   HEURISTIC SANITY CHECK, not a correctness guarantee, built from the two
#   real, cheap, deterministic signals mediapipe DOES give:
#     (a) detection confidence / detect-vs-no-detect -- broken or fused AI
#         hands frequently fail to trigger detection at all, or trigger only
#         at low confidence. A real plausibility proxy.
#     (b) geometric plausibility of the 21 points it DID return -- finger
#         segment-length ratios within human proportion bounds, and no
#         impossible interior joint angles.
#   Thresholds below are a FIRST CALIBRATION, not authoritative numbers, and
#   are labelled as such in the report (same spirit as "UNVERIFIED NOISE
#   FLOOR").
#
# mediapipe is CPU-only and an OPTIONAL dependency: absent it, --skip-anatomy-
# check (or the automatic availability guard) lets the rest of the harness run,
# mirroring the clean-fid / hpsv2 optional-dependency pattern.

# MediaPipe 21-landmark hand topology (fixed indices):
#   0 wrist; thumb 1-4 (CMC,MCP,IP,TIP); index 5-8; middle 9-12; ring 13-16;
#   pinky 17-20 (each: MCP, PIP, DIP, TIP).
_FINGERS: "dict[str, tuple[int, int, int, int]]" = {
    "thumb": (1, 2, 3, 4),
    "index": (5, 6, 7, 8),
    "middle": (9, 10, 11, 12),
    "ring": (13, 14, 15, 16),
    "pinky": (17, 18, 19, 20),
}
# Non-thumb fingers have a clean proximal>middle>distal phalanx ordering; the
# thumb's metacarpal/phalanx geometry differs, so proportion checks skip it.
_NON_THUMB = ("index", "middle", "ring", "pinky")

# First-calibration heuristic bounds (loose on purpose -- tuned to NOT flag
# real, correctly-articulated human hands, at the cost of missing subtle
# defects; see the module note above on why this is a sanity check only).
_RATIO_MID_OVER_PROX = (0.30, 1.05)   # middle-phalanx / proximal-phalanx length
_RATIO_DIST_OVER_MID = (0.30, 1.15)   # distal-phalanx / middle-phalanx length
_MIN_INTERIOR_ANGLE_DEG = 18.0        # sharper than this at a joint = implausible kink

# Report gate thresholds (FIRST CALIBRATION -- see note above).
ANATOMY_DETECT_RATE_MIN = 0.60        # >=60% of hand prompts should trigger a detection
ANATOMY_PLAUSIBILITY_MIN = 0.70       # mean geometric-plausibility of detected hands
FIDELITY_STRUCT_DELTA_MAX = 0.35      # per-image edit structural delta above this = degradation


def _euclid(a: "tuple[float, float, float]", b: "tuple[float, float, float]") -> float:
    return ((a[0] - b[0]) ** 2 + (a[1] - b[1]) ** 2 + (a[2] - b[2]) ** 2) ** 0.5


def _interior_angle_deg(
    a: "tuple[float, float, float]",
    b: "tuple[float, float, float]",
    c: "tuple[float, float, float]",
) -> float:
    """Angle at vertex ``b`` formed by segments b->a and b->c, in degrees.
    Returns 180.0 (straight, maximally plausible) for a degenerate zero-length
    segment rather than raising."""
    import math

    v1 = (a[0] - b[0], a[1] - b[1], a[2] - b[2])
    v2 = (c[0] - b[0], c[1] - b[1], c[2] - b[2])
    n1 = (v1[0] ** 2 + v1[1] ** 2 + v1[2] ** 2) ** 0.5
    n2 = (v2[0] ** 2 + v2[1] ** 2 + v2[2] ** 2) ** 0.5
    if n1 == 0.0 or n2 == 0.0:
        return 180.0
    dot = v1[0] * v2[0] + v1[1] * v2[1] + v1[2] * v2[2]
    cos = max(-1.0, min(1.0, dot / (n1 * n2)))
    return math.degrees(math.acos(cos))


def _hand_scale(landmarks: "list[tuple[float, float, float]]") -> float:
    """A rotation/translation-invariant size reference for the hand: the
    wrist(0)->middle-MCP(9) distance. Used to normalise segment lengths so the
    signature is comparable across image scales (input vs edited output)."""
    s = _euclid(landmarks[0], landmarks[9])
    return s if s > 1e-9 else 1e-9


def geometric_plausibility(landmarks: "list[tuple[float, float, float]]") -> float:
    """Heuristic geometric-plausibility score in [0, 1] for one detected hand.

    1.0 = every proportion + joint-angle check passed (looks like a well-
    articulated human hand); 0.0 = every check failed (impossible geometry, or
    a degenerate all-identical / NaN landmark set). This is the (b) signal from
    the module note -- it operates purely on the 21 returned points and knows
    NOTHING about how many real fingers were in the image.

    Returns 0.0 for anything other than a well-formed 21-point list (defensive:
    a degenerate/blank detection must read as implausible, never crash)."""
    import math

    if not landmarks or len(landmarks) != 21:
        return 0.0
    for p in landmarks:
        if len(p) != 3 or any(math.isnan(c) or math.isinf(c) for c in p):
            return 0.0

    scale = _hand_scale(landmarks)
    checks: list[bool] = []

    for name in _NON_THUMB:
        mcp, pip, dip, tip = _FINGERS[name]
        prox = _euclid(landmarks[mcp], landmarks[pip]) / scale
        mid = _euclid(landmarks[pip], landmarks[dip]) / scale
        dist = _euclid(landmarks[dip], landmarks[tip]) / scale
        # A collapsed finger (any near-zero phalanx -- e.g. an all-identical
        # degenerate landmark set) is anatomically impossible: fail ALL four of
        # this finger's checks rather than letting the zero-length angle default
        # to "straight" and score it half-plausible.
        if prox <= 1e-6 or mid <= 1e-6 or dist <= 1e-6:
            checks.extend([False, False, False, False])
            continue
        # Proportion checks: successive phalanges shrink, ratios within bounds.
        r_mp = mid / prox
        checks.append(_RATIO_MID_OVER_PROX[0] <= r_mp <= _RATIO_MID_OVER_PROX[1])
        r_dm = dist / mid
        checks.append(_RATIO_DIST_OVER_MID[0] <= r_dm <= _RATIO_DIST_OVER_MID[1])
        # Interior joint angles: no impossibly-sharp kink at PIP or DIP.
        ang_pip = _interior_angle_deg(landmarks[mcp], landmarks[pip], landmarks[dip])
        ang_dip = _interior_angle_deg(landmarks[pip], landmarks[dip], landmarks[tip])
        checks.append(ang_pip >= _MIN_INTERIOR_ANGLE_DEG)
        checks.append(ang_dip >= _MIN_INTERIOR_ANGLE_DEG)

    if not checks:
        return 0.0
    return sum(1 for c in checks if c) / len(checks)


def structural_signature(landmarks: "list[tuple[float, float, float]]") -> "list[float]":
    """Scale-normalised structural feature vector for before/after comparison.

    Concatenates, for all 5 fingers: the 3 phalanx segment lengths (each
    divided by hand scale) and the 2 interior joint angles (normalised to
    [0, 1] by /180). Length = 5*(3+2) = 25. This is what the img2img/inpaint
    FIDELITY test diffs between the ORIGINAL photo's hand and the EDITED
    output's hand: a large delta means the edit moved real anatomy.

    Returns an empty list for a malformed landmark set (caller treats an empty
    signature as "no comparable structure")."""
    import math

    if not landmarks or len(landmarks) != 21:
        return []
    for p in landmarks:
        if len(p) != 3 or any(math.isnan(c) or math.isinf(c) for c in p):
            return []

    scale = _hand_scale(landmarks)
    sig: list[float] = []
    for name in ("thumb",) + _NON_THUMB:
        a, b, c, d = _FINGERS[name]
        sig.append(_euclid(landmarks[a], landmarks[b]) / scale)
        sig.append(_euclid(landmarks[b], landmarks[c]) / scale)
        sig.append(_euclid(landmarks[c], landmarks[d]) / scale)
        sig.append(_interior_angle_deg(landmarks[a], landmarks[b], landmarks[c]) / 180.0)
        sig.append(_interior_angle_deg(landmarks[b], landmarks[c], landmarks[d]) / 180.0)
    return sig


def structural_delta(
    sig_before: "list[float]",
    sig_after: "list[float]",
) -> float:
    """Mean absolute difference between two structural signatures.

    0.0 = identical structure (edit preserved the hand perfectly); larger =
    more structural drift. NaN when the signatures are empty or mismatched
    (no comparable hand on one side) so the caller can skip, not fabricate a
    zero. Monotonic in perturbation magnitude: a bigger geometric change
    yields a strictly larger delta -- the property the fidelity test asserts."""
    if not sig_before or not sig_after or len(sig_before) != len(sig_after):
        return float("nan")
    return sum(abs(a - b) for a, b in zip(sig_before, sig_after)) / len(sig_before)


# Canonical Google-hosted hand-landmarker asset (Apache-2.0). The mediapipe
# Tasks API needs this .task file (the legacy solutions.hands API that bundled
# weights in the wheel was removed in mediapipe 0.10.x). Resolution order:
#   1. $IMAGEFORGE_HAND_LANDMARKER_TASK (an explicit local path)
#   2. a cached copy under ~/.cache/imageforge/
#   3. one-time best-effort download from the URL below into that cache
# -- the same "download real weights on first use" pattern hpsv2/clean-fid
# already use. When it can't be resolved, _mediapipe_hands raises and the
# anatomy section auto-skips (never fabricated, never fatal to the main eval).
_HAND_LANDMARKER_URL = (
    "https://storage.googleapis.com/mediapipe-models/hand_landmarker/"
    "hand_landmarker/float16/1/hand_landmarker.task"
)


def _resolve_hand_landmarker_task() -> "Path":
    env = os.environ.get("IMAGEFORGE_HAND_LANDMARKER_TASK")
    if env:
        p = Path(env)
        if not p.is_file():
            raise ImportError(
                f"IMAGEFORGE_HAND_LANDMARKER_TASK={env} is not a file. Point it at a "
                "downloaded hand_landmarker.task, or unset it to use the cache."
            )
        return p
    cache = Path.home() / ".cache" / "imageforge" / "hand_landmarker.task"
    if cache.is_file():
        return cache
    cache.parent.mkdir(parents=True, exist_ok=True)
    log.info("Downloading hand_landmarker.task (one-time, ~7.5MB) → %s", cache)
    try:
        import urllib.request

        tmp = cache.with_suffix(".task.part")
        urllib.request.urlretrieve(_HAND_LANDMARKER_URL, tmp)
        tmp.replace(cache)
    except Exception as exc:  # noqa: BLE001
        raise ImportError(
            "Could not obtain hand_landmarker.task automatically "
            f"({exc}). Download it from {_HAND_LANDMARKER_URL} and set "
            "IMAGEFORGE_HAND_LANDMARKER_TASK to its path, or pass "
            "--skip-anatomy-check."
        ) from exc
    return cache


def _mediapipe_hands():
    """Construct a mediapipe Tasks-API ``HandLandmarker`` detector.

    Returns 21 fixed 3D landmarks + handedness (with a confidence score) per
    detected hand. Raises ImportError with an install hint when mediapipe is
    absent OR the model asset cannot be resolved, so the caller degrades to a
    skipped anatomy section exactly like a missing clean-fid/hpsv2. A low
    detection floor (0.3) is deliberate -- we WANT to observe low-confidence
    hits, since low confidence is itself a plausibility signal."""
    try:
        import mediapipe as mp  # type: ignore  # noqa: F401
        from mediapipe.tasks import python as mp_python  # type: ignore
        from mediapipe.tasks.python import vision  # type: ignore
    except ImportError as exc:  # pragma: no cover - exercised via --skip-anatomy-check
        raise ImportError(
            "mediapipe not installed. Run: pip install mediapipe  "
            "(CPU-only; or pass --skip-anatomy-check to run the rest of the harness)."
        ) from exc
    task_path = _resolve_hand_landmarker_task()
    options = vision.HandLandmarkerOptions(
        base_options=mp_python.BaseOptions(model_asset_path=str(task_path)),
        num_hands=2,
        min_hand_detection_confidence=0.3,
    )
    return vision.HandLandmarker.create_from_options(options)


def detect_hand_landmarks(image_path: "Path", detector=None) -> "list[dict]":
    """Run mediapipe hand detection on one image.

    Returns a list (one entry per detected hand) of
    ``{"landmarks": [(x,y,z)*21], "confidence": float, "handedness": str}``.
    An empty list means NO hand was detected -- itself the primary plausibility
    signal (broken AI hands often fail to detect at all). ``confidence`` is the
    Tasks-API handedness classification score, mediapipe's per-hand confidence.

    ``detector`` may be injected (tests pass a fake exposing ``.detect``);
    otherwise a fresh HandLandmarker is built and closed around the call."""
    import numpy as np  # local: numpy is always present in the serving venv
    import mediapipe as mp  # type: ignore
    from PIL import Image

    owns = detector is None
    if owns:
        detector = _mediapipe_hands()
    try:
        img = Image.open(image_path).convert("RGB")
        arr = np.asarray(img)
        mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=arr)
        results = detector.detect(mp_image)
        out: list[dict] = []
        hand_landmarks = getattr(results, "hand_landmarks", None) or []
        handedness = getattr(results, "handedness", None) or []
        for i, hand in enumerate(hand_landmarks):
            lms = [(lm.x, lm.y, lm.z) for lm in hand]
            conf = 0.0
            label = ""
            if i < len(handedness) and handedness[i]:
                conf = float(handedness[i][0].score)
                label = handedness[i][0].category_name
            out.append({"landmarks": lms, "confidence": conf, "handedness": label})
        return out
    finally:
        if owns:
            try:
                detector.close()
            except Exception:  # noqa: BLE001
                pass


def anatomy_plausibility_for_image(image_path: "Path", detector=None) -> dict:
    """Per-image anatomy plausibility: combines detection (a) + geometry (b).

    Returns ``{"detected": bool, "confidence": float, "plausibility": float,
    "score": float}`` where ``score`` folds both signals into one number in
    [0, 1]: 0.0 when no hand is detected at all (the degenerate/blank/corrupt
    case -- correctly flagged implausible), otherwise
    ``best_confidence * best_geometric_plausibility`` over the detected hands.
    Multiplicative so a hand that detects but has impossible geometry, OR one
    with fine geometry but near-zero detection confidence, both score low."""
    hands = detect_hand_landmarks(image_path, detector=detector)
    if not hands:
        return {"detected": False, "confidence": 0.0, "plausibility": 0.0, "score": 0.0}
    best_score = 0.0
    best_conf = 0.0
    best_plaus = 0.0
    for h in hands:
        plaus = geometric_plausibility(h["landmarks"])
        s = h["confidence"] * plaus
        if s >= best_score:
            best_score = s
            best_conf = h["confidence"]
            best_plaus = plaus
    return {
        "detected": True,
        "confidence": best_conf,
        "plausibility": best_plaus,
        "score": best_score,
    }


def edit_fidelity_delta(
    input_path: "Path",
    output_path: "Path",
    detector=None,
) -> float:
    """Structural delta between the hand in the ORIGINAL photo and the hand in
    the EDITED output. NaN when either side has no detectable hand (nothing to
    compare -- skipped, never fabricated). Larger = the edit degraded real
    anatomy more. Uses the best-detection-confidence hand on each side."""
    def _best_sig(path: "Path") -> "list[float]":
        hands = detect_hand_landmarks(path, detector=detector)
        if not hands:
            return []
        best = max(hands, key=lambda h: h["confidence"])
        return structural_signature(best["landmarks"])

    return structural_delta(_best_sig(input_path), _best_sig(output_path))


def summarize_anatomy_text2img(per_image: "list[dict]") -> dict:
    """Aggregate per-image anatomy_plausibility_for_image() results into the
    row the report shows. ``detect_rate`` = fraction of images where a hand was
    detected at all; ``mean_plausibility`` / ``mean_score`` are averaged over
    the DETECTED images only (an undetected image contributes to detect_rate,
    not to a geometry average computed on points that don't exist)."""
    n = len(per_image)
    if n == 0:
        return {"n": 0, "detect_rate": float("nan"), "mean_plausibility": float("nan"),
                "mean_score": float("nan"), "n_detected": 0}
    detected = [r for r in per_image if r.get("detected")]
    nd = len(detected)
    return {
        "n": n,
        "n_detected": nd,
        "detect_rate": nd / n,
        "mean_plausibility": (sum(r["plausibility"] for r in detected) / nd) if nd else float("nan"),
        "mean_score": (sum(r["score"] for r in detected) / nd) if nd else float("nan"),
    }


def summarize_anatomy_fidelity(deltas: "list[float]") -> dict:
    """Aggregate per-image edit_fidelity_delta() values. ``comparable`` counts
    images where BOTH input and output had a detectable hand (a NaN delta means
    one side didn't, so it's excluded from the mean). ``n_degraded`` counts
    comparable images whose structural delta exceeded the calibration
    threshold -- the edit measurably moved anatomy there."""
    import math

    comparable = [d for d in deltas if not math.isnan(d)]
    nc = len(comparable)
    return {
        "n": len(deltas),
        "comparable": nc,
        "mean_delta": (sum(comparable) / nc) if nc else float("nan"),
        "max_delta": (max(comparable) if nc else float("nan")),
        "n_degraded": sum(1 for d in comparable if d > FIDELITY_STRUCT_DELTA_MAX),
    }


def emit_anatomy_report(
    run_dir: "Path",
    *,
    text2img: "Optional[dict]" = None,
    fidelity: "Optional[dict]" = None,
) -> str:
    """Render the Batch-23 anatomy section and append it to eval_report.md
    (creating the file if the main report has not written it). Returns the
    section markdown. Explicitly labelled a heuristic FIRST CALIBRATION -- the
    same honesty convention as the "UNVERIFIED NOISE FLOOR" tag -- so a reader
    never mistakes these numbers for an authoritative anatomy-correctness gate.
    """
    import math

    def _fmt(x: float, nd: int = 3) -> str:
        return "N/A" if (x is None or (isinstance(x, float) and math.isnan(x))) else f"{x:.{nd}f}"

    lines = [
        "",
        "## Anatomy Plausibility (Batch 23 — hands/limbs)",
        "",
        "> **HEURISTIC SANITY CHECK, FIRST CALIBRATION — not authoritative.** "
        "mediapipe fits a fixed 21-point/5-finger template and CANNOT count "
        "fingers or detect polydactyly/fusion; these rows use detection "
        "confidence + geometric proportion/angle plausibility only. Thresholds "
        "are an initial calibration, not a verified gate (cf. UNVERIFIED NOISE "
        "FLOOR).",
        "",
        "| Check | Metric | Value | Threshold | Flag |",
        "|-------|--------|-------|-----------|------|",
    ]
    verdicts: list[bool] = []
    if text2img is not None:
        dr = text2img.get("detect_rate", float("nan"))
        mp_ = text2img.get("mean_plausibility", float("nan"))
        dr_ok = not math.isnan(dr) and dr >= ANATOMY_DETECT_RATE_MIN
        mp_ok = not math.isnan(mp_) and mp_ >= ANATOMY_PLAUSIBILITY_MIN
        verdicts += [dr_ok, mp_ok]
        lines.append(
            f"| TEXT2IMG stress | hand detect-rate ({text2img.get('n_detected','?')}/"
            f"{text2img.get('n','?')}) | {_fmt(dr)} | >= {ANATOMY_DETECT_RATE_MIN} | "
            f"{'ok' if dr_ok else 'LOW'} |"
        )
        lines.append(
            f"| TEXT2IMG stress | mean geom. plausibility | {_fmt(mp_)} | "
            f">= {ANATOMY_PLAUSIBILITY_MIN} | {'ok' if mp_ok else 'LOW'} |"
        )
    if fidelity is not None:
        md = fidelity.get("mean_delta", float("nan"))
        nd = fidelity.get("n_degraded", 0)
        nc = fidelity.get("comparable", 0)
        md_ok = not math.isnan(md) and md <= FIDELITY_STRUCT_DELTA_MAX
        verdicts.append(md_ok)
        lines.append(
            f"| IMG2IMG fidelity | mean struct. delta ({nc} comparable) | {_fmt(md)} | "
            f"<= {FIDELITY_STRUCT_DELTA_MAX} | {'ok' if md_ok else 'HIGH'} |"
        )
        lines.append(
            f"| IMG2IMG fidelity | images degraded | {nd}/{nc} | 0 | "
            f"{'ok' if nd == 0 else 'DEGRADED'} |"
        )
    if not verdicts:
        lines.append("| — | (no anatomy checks ran) | — | — | — |")
    verdict = "PASS" if verdicts and all(verdicts) else ("FAIL" if verdicts else "SKIPPED")
    lines.append("")
    lines.append(f"### ANATOMY VERDICT: {verdict} (FIRST CALIBRATION — heuristic, not authoritative)")
    lines.append("")
    section = "\n".join(lines)

    report_path = run_dir / "eval_report.md"
    existing = report_path.read_text(encoding="utf-8") if report_path.exists() else ""
    report_path.write_text(existing + section, encoding="utf-8")
    log.info("Anatomy section written → %s", report_path)
    return section


# ---------------------------------------------------------------------------
# Report emitter
# ---------------------------------------------------------------------------
def emit_report(
    run_dir: Path,
    *,
    base_model: str,
    cand_model: str,
    fid_base: Optional[float],
    fid_cand: Optional[float],
    fid_noise: float,
    hps_base: float,
    hps_cand: float,
    hps_noise_std: float,
    geneval_base: Optional[float],
    geneval_cand: Optional[float],
    geneval_noise_std: float,
    n_images: "int | str",
    noise_floor_measured: bool = False,
    hps_bootstrap: "Optional[dict]" = None,
) -> None:
    """Print and write the markdown report table.

    n_images: pass the caller's REAL count of successfully-generated images
    actually scored (e.g. "196/200" when a caller wants to show requested vs
    actual, or a plain int when they matched) -- this used to always be the
    CLI's requested --n-images, even when generation partially failed, so a
    report could claim "N=200 images" while HPS/FID were computed over a
    materially smaller surviving sample with no indication of that anywhere
    in the report.

    Gate semantics (redesigned — see module docstring "Gate model"):

    * **HPSv2 is the PRIMARY gate.** It is a higher-is-better learned human
      preference score; the candidate must beat the base by ≥ HPS_PASS_DELTA
      AND by ≥ 2×noise_std. HPSv2 is always measurable (no reference set
      needed), so the gate can ALWAYS rest on it.

    * **clean-FID is a SECONDARY / corroborating signal, and only when a real
      reference set was provided.** FID is a distribution distance to a real
      reference: ``fid_base = FID(reference, base)``, ``fid_cand =
      FID(reference, candidate)`` against the SAME reference. A LOWER candidate
      FID is the improvement, so ``fid_delta = fid_cand - fid_base`` must be
      ≤ FID_PASS_DELTA (and beyond the noise floor) to corroborate.
      When no reference set is available, ``fid_base``/``fid_cand`` are ``None``
      → FID is reported as **N/A** and the gate rests on HPSv2 alone (with a
      logged note). FID is never fabricated to 0.0.

    The overall GATE passes iff every *available* metric passes. FID rows that
    are N/A do not count against the gate (they are corroborating, not required).
    """
    import math

    def fid_moved(delta: float, noise: float) -> str:
        return "YES" if delta <= FID_PASS_DELTA and delta <= -NOISE_SIGMA_MULTIPLIER * noise else "no"

    def hps_moved(delta: float, noise_std: float) -> str:
        return "YES" if delta >= HPS_PASS_DELTA and delta >= NOISE_SIGMA_MULTIPLIER * noise_std else "no"

    def ge_moved(delta: float, noise_std: float) -> str:
        return "YES" if delta >= GENEVAL_PASS_DELTA and delta >= NOISE_SIGMA_MULTIPLIER * noise_std else "no"

    # score_hps() returns nan when EVERY image in a batch failed to score
    # (network error, corrupt image, reward-model crash, etc.) -- NaN
    # comparisons are always False in Python, so hps_moved(nan, ...) silently
    # returned "no" and the gate printed a plain FAIL, indistinguishable from
    # a genuine regression. Scoring failure is a DIFFERENT outcome from "the
    # candidate is worse" and must be surfaced as its own state.
    hps_scoring_failed = math.isnan(hps_base) or math.isnan(hps_cand)
    hps_delta = float("nan") if hps_scoring_failed else hps_cand - hps_base

    # Batch 1: a paired bootstrap CI on the per-image (cand - base) delta is
    # the stronger gate when the raw per-image scores are available -- it uses
    # the actual sampling variability of THIS run rather than a fixed noise_std
    # constant. When a valid bootstrap is supplied, the PRIMARY HPS gate rests
    # on bootstrap["passes"] (CI entirely above zero AND |delta| ≥ pass_delta);
    # otherwise it falls back to the delta-vs-noise_std rule below.
    boot_ok = (
        hps_bootstrap is not None
        and not hps_scoring_failed
        and hps_bootstrap.get("error") is None
        and hps_bootstrap.get("ci_low") == hps_bootstrap.get("ci_low")  # not NaN
    )

    # HPSv2 — PRIMARY gate, always present.
    if hps_scoring_failed:
        rows = [
            ("HPSv2.1 (PRIMARY)", "ERROR" if math.isnan(hps_base) else f"{hps_base:.4f}",
             "ERROR" if math.isnan(hps_cand) else f"{hps_cand:.4f}", "—", "—", "ERROR"),
        ]
    elif boot_ok:
        ci_low = hps_bootstrap["ci_low"]
        ci_high = hps_bootstrap["ci_high"]
        ci_pct = int(round(hps_bootstrap.get("ci_level", 0.95) * 100))
        rows = [
            ("HPSv2.1 (PRIMARY, bootstrap)", f"{hps_base:.4f}", f"{hps_cand:.4f}",
             f"{hps_delta:+.4f}", f"{ci_pct}% CI [{ci_low:+.4f}, {ci_high:+.4f}]",
             "YES" if hps_bootstrap["passes"] else "no"),
        ]
    else:
        rows = [
            ("HPSv2.1 (PRIMARY)", f"{hps_base:.4f}", f"{hps_cand:.4f}", f"{hps_delta:+.4f}", f"±{hps_noise_std:.4f}", hps_moved(hps_delta, hps_noise_std)),
        ]

    # clean-FID — corroborating signal, only when a real reference set produced
    # both scores. N/A (None) means no reference set → does not gate.
    fid_available = fid_base is not None and fid_cand is not None and not (
        math.isnan(fid_base) or math.isnan(fid_cand)
    )
    if fid_available:
        fid_delta = fid_cand - fid_base
        rows.append(
            ("clean-FID (corroborating)", f"{fid_base:.2f}", f"{fid_cand:.2f}", f"{fid_delta:+.2f}", f"{fid_noise:.2f}", fid_moved(fid_delta, fid_noise))
        )
    else:
        rows.append(("clean-FID (corroborating)", "N/A", "N/A", "—", "—", "—"))

    # Bug: this previously only checked math.isnan(geneval_base), never
    # geneval_cand -- the exact same "NaN comparisons are always False in
    # Python" trap already fixed for HPSv2 above, just left unfixed here.
    # If ONLY the candidate failed to score (score_geneval2's own empty-flat
    # path can return nan independent of the base score), ge_delta became
    # nan, ge_moved(nan, ...) silently returned "no", and the row rendered
    # as a normal FAIL instead of the same non-gating SKIPPED state a
    # cleanly-unavailable GenEval2 already gets below.
    geneval_available = (
        geneval_base is not None and geneval_cand is not None
        and not math.isnan(geneval_base) and not math.isnan(geneval_cand)
    )
    if geneval_available:
        ge_delta = geneval_cand - geneval_base
        rows.append(
            ("GenEval2 Soft-TIFA_AM", f"{geneval_base:.1f}", f"{geneval_cand:.1f}", f"{ge_delta:+.1f}", f"±{geneval_noise_std:.1f}", ge_moved(ge_delta, geneval_noise_std))
        )
    else:
        rows.append(("GenEval2 Soft-TIFA_AM", "SKIPPED", "SKIPPED", "—", "—", "—"))

    # Gate logic: every AVAILABLE metric (Moved? != "—") must pass. The PRIMARY
    # HPSv2 metric is always available, so the gate is never vacuous.
    # ERROR (total scoring failure, not a real regression) short-circuits to
    # its own distinct gate state -- must never render as a plain FAIL.
    if hps_scoring_failed:
        gate = "ERROR"
    else:
        gate_results = [r[5] for r in rows if r[5] != "—"]
        gate = "PASS" if gate_results and all(g == "YES" for g in gate_results) else "FAIL"

    fid_note = (
        ""
        if fid_available
        else "\n> **Note:** No real-image reference set provided (`--reference-dir`); "
             "clean-FID is N/A and the gate rests on HPSv2 alone. FID is a "
             "distribution distance to a real reference, not a base-vs-candidate "
             "quantity — supply `--reference-dir` for the corroborating FID signal.\n"
    )
    error_note = (
        "\n> **ERROR, not FAIL:** HPSv2 scoring failed for every image in this "
        "batch (network error, corrupt image, or reward-model crash) rather "
        "than producing a real (possibly worse) score. This is NOT a "
        "regression verdict -- re-run the eval before drawing any conclusion.\n"
        if hps_scoring_failed else ""
    )
    # Audit-found bug (M-1): the "beyond noise floor" half of every gate
    # condition (NOISE_SIGMA_MULTIPLIER x noise_std) is only measured
    # empirically when the caller passes --noise-floor. Without it, this
    # falls back to hardcoded "conservative default" constants
    # (hps_noise_std=0.05 etc) -- but the HPSv2 paper's own measured
    # between-group std can be as high as 0.226 for some styles (4x higher),
    # making the default fallback trivially satisfied by a real HPS_PASS_DELTA
    # of 0.20 regardless of whether the change is actually distinguishable
    # from noise. A PASS/FAIL gated on an unmeasured floor is not the same
    # confidence level as one gated on a real, measured floor -- label it
    # rather than let it look identical to a verified result.
    # Batch 1: when the PRIMARY HPS gate rests on a paired bootstrap CI, its
    # significance test uses THIS run's own resampled per-image variability --
    # a measured floor, not the hardcoded conservative default. So the
    # "unverified noise floor" caveat does not apply to a bootstrap-gated run.
    hps_floor_measured = noise_floor_measured or boot_ok

    bootstrap_note = (
        "\n> **PRIMARY gate = paired bootstrap CI:** the HPSv2 verdict below "
        f"rests on a {int(round(hps_bootstrap.get('ci_level', 0.95) * 100))}% "
        f"paired bootstrap CI ({hps_bootstrap.get('n_boot')} resamples, "
        f"N={hps_bootstrap.get('n')} paired images) on the per-image "
        "(candidate − base) HPS delta. PASS requires the CI to lie entirely "
        f"above zero AND the mean delta ≥ {hps_bootstrap.get('pass_delta')}.\n"
        if boot_ok else ""
    )
    noise_floor_note = (
        ""
        if hps_floor_measured or hps_scoring_failed  # ERROR state has its own, unrelated note
        else "\n> **UNVERIFIED NOISE FLOOR:** this run did not pass `--noise-floor`, "
             "so the gate below rests on hardcoded conservative-default noise "
             "estimates, not a measured same-model/different-seed baseline. "
             "These defaults can understate the real noise floor by several "
             "times for some prompt styles (see research/08_weights_research.md) "
             "-- re-run with `--noise-floor` before treating this gate as authoritative.\n"
    )

    header = f"""
# ImageForge Eval Report
Base: `{base_model}` | Candidate: `{cand_model}` | N={n_images} images
{fid_note}{error_note}{bootstrap_note}{noise_floor_note}
| Metric | Base | Candidate | Delta | Noise Floor | Moved? |
|--------|------|-----------|-------|-------------|--------|
"""
    table_rows = "\n".join(f"| {r[0]} | {r[1]} | {r[2]} | {r[3]} | {r[4]} | {r[5]} |" for r in rows)
    thresholds = f"""
## Pass/Fail Thresholds
- HPSv2.1 (PRIMARY): delta ≥ **+{HPS_PASS_DELTA}** AND delta ≥ **{NOISE_SIGMA_MULTIPLIER}×noise_std**
- clean-FID (corroborating, only with `--reference-dir`): delta ≤ **{FID_PASS_DELTA}** AND delta ≤ **-{NOISE_SIGMA_MULTIPLIER}×noise_floor**, where delta = FID(ref,cand) − FID(ref,base)
- GenEval2: delta ≥ **+{GENEVAL_PASS_DELTA} pp** AND delta ≥ **{NOISE_SIGMA_MULTIPLIER}×noise_std**
"""
    gate_label = gate if (hps_floor_measured or gate == "ERROR") else f"{gate} (UNVERIFIED NOISE FLOOR)"
    report = header + table_rows + thresholds + f"\n## GATE: {gate_label}\n"

    # PERSIST BEFORE PRINTING. The report contains non-ASCII (>=, minus,
    # multiplication signs) and stdout on Windows defaults to cp1252, so
    # print() can raise UnicodeEncodeError -- and it used to do so BEFORE the
    # file was written, destroying the entire eval's result after every image
    # had been generated and scored. D3.1 fixed this function's file encoding
    # but left the print ahead of the write, so the same character class could
    # still take the whole run down.
    report_path = run_dir / "eval_report.md"
    report_path.write_text(report, encoding="utf-8")
    log.info("Report written -> %s", report_path)

    try:
        print(report)
    except UnicodeEncodeError:
        # Console can't represent it; the file already has the full text.
        enc = (getattr(sys.stdout, "encoding", None) or "ascii")
        print(report.encode(enc, errors="replace").decode(enc, errors="replace"))


# ---------------------------------------------------------------------------
# Batch 23: live-API anatomy check drivers (text2img stress + img2img fidelity)
# ---------------------------------------------------------------------------
def _edit_image_via_api(
    client,
    image_path: "Path",
    prompt: str,
    out_path: "Path",
    *,
    model: Optional[str],
    strength: float,
    seed: int,
) -> "Optional[Path]":
    """POST one image through /edit at a fixed strength; save the returned PNG.

    Sends the source as base64 (``init_image``) so the API needs no shared
    filesystem. Returns the saved output path, or None on any failure (logged,
    never raised) so one bad edit doesn't abort the fidelity sweep."""
    b64_in = base64.b64encode(image_path.read_bytes()).decode()
    payload: dict = {
        "prompt": prompt,
        "init_image": b64_in,
        "strength": strength,
        "seed": seed & 0xFFFFFFFF,
        "return_mode": "base64",
        "use_cache": False,
    }
    if model:
        payload["model"] = model
    try:
        resp = client.post("/edit", json=payload)
        resp.raise_for_status()
    except Exception as e:  # noqa: BLE001
        log.error("edit failed for %s: %s", image_path.name, e)
        return None
    b64 = resp.json().get("image_base64")
    if not b64:
        log.warning("no image_base64 in /edit response for %s", image_path.name)
        return None
    out_path.write_bytes(base64.b64decode(b64))
    return out_path


def run_anatomy_text2img_stress(
    prompts: "list[dict]",
    model: Optional[str],
    out_dir: "Path",
    *,
    width: int,
    height: int,
    n_images: int,
    detector=None,
) -> dict:
    """TEXT2IMG stress test: generate the known-hard hand prompts, score each
    output's mediapipe detection confidence + geometric plausibility, and
    aggregate into the report row. Reuses generate_images() (same lock-
    respecting, one-at-a-time driver the main eval uses)."""
    pairs = generate_images(
        prompts, model, out_dir,
        width=width, height=height, n_images=n_images,
    )
    per_image: list[dict] = []
    for img_path, rec in pairs:
        res = anatomy_plausibility_for_image(img_path, detector=detector)
        res["image"] = img_path.name
        res["prompt"] = rec.get("prompt")
        per_image.append(res)
        log.info(
            "anatomy[t2i] %s detected=%s conf=%.3f plaus=%.3f",
            img_path.name, res["detected"], res["confidence"], res["plausibility"],
        )
    summary = summarize_anatomy_text2img(per_image)
    summary["per_image"] = per_image
    return summary


def run_anatomy_img2img_fidelity(
    image_dir: "Path",
    out_dir: "Path",
    *,
    model: Optional[str],
    strength: float,
    prompt: str,
    seed: int,
    detector=None,
) -> dict:
    """IMG2IMG FIDELITY test: run each vendored reference photo through /edit at
    a fixed transform, then compare mediapipe landmark-derived structure before
    vs after. A large structural delta means the edit degraded real anatomy.

    Only compares images whose ORIGINAL already has a detectable hand (no point
    measuring fidelity on a photo mediapipe can't read to begin with)."""
    try:
        import httpx
    except ImportError:
        sys.exit("httpx not installed. Run: pip install httpx")

    out_dir.mkdir(parents=True, exist_ok=True)
    images = sorted(p for p in image_dir.glob("*.jpg"))
    deltas: list[float] = []
    per_image: list[dict] = []
    with httpx.Client(base_url=API_URL, timeout=300.0, headers=_auth_headers()) as client:
        for img_path in images:
            edited = _edit_image_via_api(
                client, img_path, prompt, out_dir / f"edited_{img_path.stem}.png",
                model=model, strength=strength, seed=seed,
            )
            if edited is None:
                continue
            delta = edit_fidelity_delta(img_path, edited, detector=detector)
            deltas.append(delta)
            per_image.append({"image": img_path.name, "delta": delta})
            log.info("anatomy[i2i] %s strength=%.2f struct_delta=%s",
                     img_path.name, strength, "N/A" if delta != delta else f"{delta:.4f}")
    summary = summarize_anatomy_fidelity(deltas)
    summary["per_image"] = per_image
    summary["strength"] = strength
    return summary


# ---------------------------------------------------------------------------
# Dry-run validator (checks config + API reachability; no generation)
# ---------------------------------------------------------------------------
def _dry_run(args: "argparse.Namespace") -> None:
    """
    Validate configuration and probe API reachability.

    Prints a structured summary and exits:
      0 — config valid AND API reachable
      1 — config valid BUT API unreachable (clear message printed)
    The caller has already validated numeric/string constraints via argparse
    before reaching here, so those checks are not repeated.
    """
    print("=== eval_harness --dry-run ===")
    print(f"  base model       : {args.base}")
    print(f"  candidate model  : {args.candidate}")
    print(f"  n-images         : {args.n_images}")
    print(f"  resolution       : {args.width}×{args.height}")
    print(f"  prompts file     : {args.prompts}")
    print(f"  api-url          : {args.api_url}")
    print(f"  noise-floor      : {args.noise_floor}")
    print(f"  skip-geneval     : {args.skip_geneval}")
    print(f"  skip-anatomy-chk : {args.skip_anatomy_check}")
    print(f"  hps-version      : {args.hps_version}")
    print()

    # Optional-dep availability report (informational, not fatal in dry-run)
    dep_status: list[tuple[str, bool, str]] = []
    for pkg, import_name, install_hint in [
        ("httpx",     "httpx",    "pip install httpx"),
        ("clean-fid", "cleanfid", "pip install clean-fid"),
        ("hpsv2",     "hpsv2",    "pip install hpsv2"),
        ("mediapipe", "mediapipe", "pip install mediapipe  (CPU-only; for --skip-anatomy-check anatomy checks)"),
    ]:
        try:
            __import__(import_name)
            dep_status.append((pkg, True, ""))
        except ImportError:
            dep_status.append((pkg, False, install_hint))

    print("  Optional dependencies:")
    all_deps_ok = True
    for pkg, ok, hint in dep_status:
        # ASCII arrow: the dep listing prints to a Windows cp1252 console where
        # a U+2192 '→' raises UnicodeEncodeError and aborts the whole dry-run.
        status_str = "OK" if ok else f"MISSING  ->  {hint}"
        print(f"    {pkg:<12} {status_str}")
        if not ok:
            all_deps_ok = False
    print()

    # Prompts file check (not fatal; generation would fail later if missing)
    prompts_path = Path(args.prompts)
    if prompts_path.exists():
        print(f"  Prompts file     : FOUND ({prompts_path})")
    else:
        print(f"  Prompts file     : NOT FOUND — create {prompts_path} before running eval")
    print()

    # API reachability check
    try:
        import httpx as _httpx
    except ImportError:
        print("  API reachability : SKIPPED (httpx not installed — run: pip install httpx)")
        print()
        print("Config valid. Install missing dependencies before running eval.")
        sys.exit(0)

    api_url = args.api_url
    print(f"  Checking API reachability at {api_url}/health …")
    try:
        with _httpx.Client(timeout=5.0, headers=_auth_headers()) as client:
            resp = client.get(f"{api_url}/health")
            resp.raise_for_status()
            health = resp.json()
        status = health.get("status", "?")
        device = health.get("device", "?")
        model  = health.get("model", "?")
        print(f"  API status       : REACHABLE  (status={status} device={device} model={model})")
        print()
        print("Config valid and API reachable. Ready to run eval.")
        sys.exit(0)
    except Exception as exc:
        print(f"  API status       : UNREACHABLE")
        print(f"  Error            : {exc}")
        print()
        print(
            f"Cannot reach ImageForge at {api_url}.\n"
            "Start the server with:  ./run_api.sh\n"
            "Then re-run eval_harness.py (without --dry-run) to begin evaluation."
        )
        sys.exit(1)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main() -> None:
    global API_URL, GENERATE_ENDPOINT, API_KEY
    parser = argparse.ArgumentParser(description="ImageForge eval harness — clean-FID + HPSv2 + GenEval2")
    parser.add_argument("--base", default="sd-turbo", help="Base model key (e.g. sd-turbo)")
    parser.add_argument("--candidate", default="sdxl-turbo", help="Candidate model key (e.g. sdxl-turbo)")
    parser.add_argument("--prompts", default=str(PROMPTS_FILE), help="Path to prompts_heldout.jsonl")
    parser.add_argument("--run-dir", default=None, help="Directory for this eval run (auto-named if omitted)")
    parser.add_argument("--n-images", type=int, default=DEFAULT_N_IMAGES, help=f"Images per model (default {DEFAULT_N_IMAGES})")
    # default=None so we can tell "user chose 512" from "512 because nobody
    # said otherwise" -- only the latter is refused for SDXL. See
    # _check_resolution_sanity.
    parser.add_argument("--width", type=int, default=None, help=f"(default {DEFAULT_WIDTH})")
    parser.add_argument("--height", type=int, default=None, help=f"(default {DEFAULT_HEIGHT})")
    parser.add_argument("--steps", type=int, default=None, help="Inference steps override")
    parser.add_argument("--guidance", type=float, default=None, help="CFG guidance override")
    parser.add_argument("--hps-version", default="v2.1", choices=["v2", "v2.1"], help="HPSv2 model version")
    parser.add_argument("--fid-device", default="cpu", help="Device for FID Inception forward pass (cpu or cuda)")
    parser.add_argument(
        "--reference-dir",
        default=None,
        help=(
            "Directory of REAL held-out images used as the clean-FID reference "
            "distribution. FID is computed as FID(reference, base) and "
            "FID(reference, candidate) against this SAME set; a lower candidate "
            "FID is the improvement. If omitted, FID is reported as N/A and the "
            "gate rests on HPSv2 alone (FID is never fabricated)."
        ),
    )
    parser.add_argument("--skip-geneval", action="store_true", help="Skip GenEval2 scoring (e.g. Qwen3-VL not available)")
    parser.add_argument("--geneval2-repo", default=None, help="Path to cloned facebookresearch/GenEval2 repo")
    parser.add_argument("--geneval2-data", default=None, help="Path to geneval2_data.jsonl (defaults to <repo>/geneval2_data.jsonl)")
    parser.add_argument("--geneval-method", default="soft_tifa_am", choices=["soft_tifa_am", "soft_tifa_gm", "tifa", "vqascore"])
    parser.add_argument("--noise-floor", action="store_true", help="Estimate noise floor (base vs base, different seeds) before comparison")
    # Batch 23: anatomy (hands/limbs) plausibility check.
    parser.add_argument(
        "--skip-anatomy-check", action="store_true",
        help=("Skip the Batch-23 anatomy plausibility checks (mediapipe hand "
              "landmarks). The rest of the harness still runs. Auto-skipped with "
              "a hint when mediapipe is not installed."),
    )
    parser.add_argument(
        "--anatomy-prompts", default=str(Path(__file__).parent / "prompts_hands_heldout.jsonl"),
        help="JSONL of known-hard hand/limb prompts for the TEXT2IMG anatomy stress test.",
    )
    parser.add_argument(
        "--benchmark-images-dir",
        default=str(Path(__file__).parent / "tests" / "fixtures" / "benchmark_images"),
        help="Directory of vendored real reference photos for the IMG2IMG fidelity test.",
    )
    parser.add_argument(
        "--anatomy-edit-strength", type=float, default=0.35,
        help="Fixed /edit denoising strength for the IMG2IMG fidelity test (default 0.35).",
    )
    parser.add_argument("--api-url", default=API_URL, help=f"ImageForge API base URL (default {API_URL})")
    parser.add_argument(
        "--api-key",
        default=None,
        help=(
            "X-API-Key for an API with auth enabled. Defaults to $IMAGEFORGE_API_KEY, "
            "then ~/.imageforge/secrets.env. Without it, an auth-enabled server 401s "
            "every request and the eval dies on the first image."
        ),
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help=(
            "Validate config and check API reachability without generating images. "
            "Exits 0 if the API is reachable and config is valid; exits 1 with a "
            "clear message if the API is unreachable or config is invalid."
        ),
    )
    args = parser.parse_args()

    # ---------------------------------------------------------------------------
    # Argparse validation — reject bad values early so errors surface before any
    # expensive work (image gen, scoring) begins.
    # ---------------------------------------------------------------------------

    # --n-images must be ≥ 1
    if args.n_images < 1:
        parser.error(f"--n-images must be ≥ 1 (got {args.n_images})")

    # Resolve --width/--height from their defaults FIRST (they parse as None so
    # an explicit small size can be told apart from an inherited one), refusing
    # an SDXL eval that silently inherited sd-turbo's 512 default -- that would
    # score garbage for both models and report it as a real verdict. Must come
    # before the multiple-of-8 check below, which needs concrete ints.
    args.width, args.height = _check_resolution_sanity(
        args.base, args.candidate, args.width, args.height
    )

    # --width / --height must be multiples of 8 (diffusion latent-space requirement)
    if args.width % 8 != 0:
        parser.error(f"--width must be a multiple of 8 (got {args.width})")
    if args.height % 8 != 0:
        parser.error(f"--height must be a multiple of 8 (got {args.height})")

    # --base / --candidate must be non-empty strings
    if not args.base or not args.base.strip():
        parser.error("--base must be a non-empty model key string")
    if not args.candidate or not args.candidate.strip():
        parser.error("--candidate must be a non-empty model key string")

    # Override global API URL if provided
    API_URL = args.api_url
    GENERATE_ENDPOINT = f"{API_URL}/generate"

    # Resolve the API key once, most-explicit source first. Only the presence
    # of a key is ever logged -- never the value, since this report and this
    # log are meant to be pasteable into a handoff doc.
    API_KEY = args.api_key or os.environ.get("IMAGEFORGE_API_KEY") or _load_api_key_from_secrets()
    log.info("API auth: %s", "X-API-Key will be sent" if API_KEY else "no key (fine if the server has auth off)")

    # ---------------------------------------------------------------------------
    # --dry-run: validate config + check API reachability, then exit without gen.
    # ---------------------------------------------------------------------------
    if args.dry_run:
        _dry_run(args)

    # Load prompts
    prompts = load_prompts(Path(args.prompts))

    # Run directory
    run_name = f"eval_{args.base}_vs_{args.candidate}_{int(time.time())}"
    run_dir = Path(args.run_dir) if args.run_dir else Path(__file__).parent / "eval_runs" / run_name
    run_dir.mkdir(parents=True, exist_ok=True)
    log.info("Run dir: %s", run_dir)

    # Noise floor estimation
    fid_noise_floor = 10.0    # conservative default if not measured
    hps_noise_std = 0.05      # conservative default
    geneval_noise_std = 1.5   # conservative default (1.5 pp)

    if args.noise_floor:
        nf = estimate_noise_floor(
            prompts, args.base, run_dir / "noise_floor",
            width=args.width, height=args.height, n_images=min(50, args.n_images),
            hps_version=args.hps_version,
        )
        fid_noise_floor = nf["fid_noise_floor"]
        hps_noise_std = nf["hps_noise_std"]
        log.info("Noise floor — FID: %.2f  HPS std: %.4f", fid_noise_floor, hps_noise_std)

    # Generate base images
    dir_base = run_dir / f"imgs_{args.base}"
    imgs_base = generate_images(
        prompts, args.base, dir_base,
        seed_offset=0, width=args.width, height=args.height, n_images=args.n_images,
        steps=args.steps, guidance=args.guidance,
    )

    # Generate candidate images
    dir_cand = run_dir / f"imgs_{args.candidate}"
    imgs_cand = generate_images(
        prompts, args.candidate, dir_cand,
        seed_offset=0, width=args.width, height=args.height, n_images=args.n_images,
        steps=args.steps, guidance=args.guidance,
    )

    if not imgs_base or not imgs_cand:
        sys.exit("No images generated. Check that ImageForge is running: ./run_api.sh")

    # ---------------------------------------------------------------------------
    # clean-FID — coherent before/after ONLY when a real reference set exists.
    #
    # FID is a distance between a REAL-image reference distribution and a
    # GENERATED distribution. The only meaningful before/after is:
    #     fid_base = FID(reference, base)      (lower = base closer to real)
    #     fid_cand = FID(reference, candidate) (lower = candidate closer to real)
    #     delta    = fid_cand - fid_base       (negative = candidate improved)
    # against the SAME reference set. Without a real reference there is NO
    # coherent FID-as-quality signal, so FID is reported as N/A (None) and the
    # gate rests on HPSv2 alone — we never fabricate fid_base = 0.0.
    # ---------------------------------------------------------------------------
    fid_base_score: Optional[float] = None
    fid_cand_score: Optional[float] = None

    ref_dir_arg = getattr(args, "reference_dir", None)
    if ref_dir_arg:
        ref_dir = Path(ref_dir_arg)
        if not ref_dir.is_dir():
            sys.exit(
                f"--reference-dir {ref_dir} is not a directory. Provide a folder of "
                "real held-out images for the FID reference, or omit it (FID will be N/A)."
            )
        log.info("Computing clean-FID against real reference set %s", ref_dir)
        fid_base_score = score_fid(ref_dir, dir_base, device_str=args.fid_device)
        fid_cand_score = score_fid(ref_dir, dir_cand, device_str=args.fid_device)
    else:
        log.warning(
            "No --reference-dir provided: clean-FID is N/A (a coherent FID needs a "
            "real-image reference distribution). The gate will rest on HPSv2 alone."
        )

    # HPSv2 -- per-image scores so a paired bootstrap CI can be computed on the
    # base-vs-candidate delta (Batch 1). Scores are keyed by image filename
    # (``{idx:04d}_seed{seed}.png``, identical for base and candidate), so the
    # pairing survives any single dropped image on either side without the
    # positional-shift misalignment score_hps()'s docstring warns about.
    base_scored = score_hps_per_image(imgs_base, hps_version=args.hps_version)
    cand_scored = score_hps_per_image(imgs_cand, hps_version=args.hps_version)

    # Persist raw per-image scores so the bootstrap is reproducible offline.
    try:
        (run_dir / "hps_base_scores.json").write_text(
            json.dumps([{"image": p.name, "prompt": r.get("prompt"), "hps": s}
                        for p, r, s in base_scored], indent=2), encoding="utf-8")
        (run_dir / "hps_cand_scores.json").write_text(
            json.dumps([{"image": p.name, "prompt": r.get("prompt"), "hps": s}
                        for p, r, s in cand_scored], indent=2), encoding="utf-8")
    except OSError as e:
        log.warning("Could not persist per-image HPS scores: %s", e)

    hps_base_score = (
        sum(s for _p, _r, s in base_scored) / len(base_scored) if base_scored else float("nan")
    )
    hps_cand_score = (
        sum(s for _p, _r, s in cand_scored) / len(cand_scored) if cand_scored else float("nan")
    )

    # Pair by image filename (same idx+seed on both sides). Only images that
    # scored on BOTH sides enter the paired vector.
    base_by_name = {p.name: s for p, _r, s in base_scored}
    cand_by_name = {p.name: s for p, _r, s in cand_scored}
    common_names = [n for n in base_by_name if n in cand_by_name]
    paired_base = [base_by_name[n] for n in common_names]
    paired_cand = [cand_by_name[n] for n in common_names]

    hps_bootstrap = paired_bootstrap_ci(paired_base, paired_cand)
    try:
        (run_dir / "hps_bootstrap.json").write_text(
            json.dumps(hps_bootstrap, indent=2), encoding="utf-8")
    except OSError as e:
        log.warning("Could not persist HPS bootstrap result: %s", e)
    if hps_bootstrap.get("error") is None:
        log.info(
            "HPS paired bootstrap: delta=%.4f  %d%% CI=[%.4f, %.4f]  passes=%s (N=%d)",
            hps_bootstrap["delta_mean"], int(round(hps_bootstrap["ci_level"] * 100)),
            hps_bootstrap["ci_low"], hps_bootstrap["ci_high"],
            hps_bootstrap["passes"], hps_bootstrap["n"],
        )
    else:
        log.warning("HPS paired bootstrap not computed: %s", hps_bootstrap["error"])

    # GenEval2
    geneval_base_score: Optional[float] = None
    geneval_cand_score: Optional[float] = None

    if not args.skip_geneval:
        if not args.geneval2_repo:
            log.warning("--geneval2-repo not set; skipping GenEval2. Pass path to cloned repo.")
            args.skip_geneval = True
        else:
            try:
                map_base = build_geneval_image_map(imgs_base, run_dir / "geneval_base")
                geneval_base_score = score_geneval2(
                    map_base, args.geneval2_repo,
                    method=args.geneval_method, geneval2_data=args.geneval2_data,
                    output_file=run_dir / "geneval_base" / "scores.json",
                )
                map_cand = build_geneval_image_map(imgs_cand, run_dir / "geneval_cand")
                geneval_cand_score = score_geneval2(
                    map_cand, args.geneval2_repo,
                    method=args.geneval_method, geneval2_data=args.geneval2_data,
                    output_file=run_dir / "geneval_cand" / "scores.json",
                )
            except Exception as e:
                log.error("GenEval2 scoring failed: %s", e)
                log.info("Re-run with --skip-geneval to skip this metric.")

    # Report
    emit_report(
        run_dir,
        base_model=args.base,
        cand_model=args.candidate,
        fid_base=fid_base_score,
        fid_cand=fid_cand_score,
        fid_noise=fid_noise_floor,
        hps_base=hps_base_score,
        hps_cand=hps_cand_score,
        hps_noise_std=hps_noise_std,
        geneval_base=geneval_base_score,
        geneval_cand=geneval_cand_score,
        geneval_noise_std=geneval_noise_std,
        # The real surviving sample size actually scored, not just the
        # requested --n-images -- see emit_report's own docstring for why
        # that distinction matters. Shown as "requested" only when a real
        # partial-generation-failure gap exists, to avoid noise on the
        # (usual) case where everything succeeded.
        n_images=(
            args.n_images
            if len(imgs_base) == len(imgs_cand) == args.n_images
            else f"{len(imgs_base)} base / {len(imgs_cand)} candidate (requested {args.n_images})"
        ),
        noise_floor_measured=args.noise_floor,
        hps_bootstrap=hps_bootstrap,
    )

    # ---------------------------------------------------------------------------
    # Batch 23: anatomy plausibility (hands/limbs) -- additive report section.
    # Scores the CANDIDATE model on the known-hard hand prompts (TEXT2IMG stress)
    # and the vendored real reference photos (IMG2IMG fidelity). Optional and
    # non-fatal: skipped cleanly (with an install hint) when --skip-anatomy-check
    # is set or mediapipe is absent, so the primary eval above always stands.
    # ---------------------------------------------------------------------------
    if args.skip_anatomy_check:
        log.info("Anatomy check skipped (--skip-anatomy-check).")
    else:
        try:
            detector = _mediapipe_hands()
        except ImportError as e:
            log.warning("Anatomy check auto-skipped: %s", e)
            detector = None
        if detector is not None:
            try:
                text2img_summary = None
                fidelity_summary = None
                anatomy_prompts_path = Path(args.anatomy_prompts)
                if anatomy_prompts_path.exists():
                    anatomy_prompts = load_prompts(anatomy_prompts_path)
                    text2img_summary = run_anatomy_text2img_stress(
                        anatomy_prompts, args.candidate, run_dir / "anatomy_t2i",
                        width=args.width, height=args.height,
                        n_images=len(anatomy_prompts), detector=detector,
                    )
                else:
                    log.warning("Anatomy prompts file %s not found; skipping TEXT2IMG stress.",
                                anatomy_prompts_path)
                bench_dir = Path(args.benchmark_images_dir)
                if bench_dir.is_dir() and any(bench_dir.glob("*.jpg")):
                    fidelity_summary = run_anatomy_img2img_fidelity(
                        bench_dir, run_dir / "anatomy_i2i",
                        model=args.candidate, strength=args.anatomy_edit_strength,
                        prompt="a photograph of a hand, same pose, high detail",
                        seed=4242, detector=detector,
                    )
                else:
                    log.warning("Benchmark image dir %s empty/missing; skipping IMG2IMG fidelity.",
                                bench_dir)
                (run_dir / "anatomy_summary.json").write_text(
                    json.dumps({"text2img": text2img_summary, "fidelity": fidelity_summary},
                               indent=2, default=str),
                    encoding="utf-8",
                )
                emit_anatomy_report(run_dir, text2img=text2img_summary, fidelity=fidelity_summary)
            finally:
                try:
                    detector.close()
                except Exception:  # noqa: BLE001
                    pass


if __name__ == "__main__":
    main()