presentation/src/evaluate.py

"""Evaluate a fine-tuned classifier on the unseen test sets (or the val split).

This is the ONLY script allowed to read data/processed/test_holdout.parquet (TREC-06) and
data/processed/test_holdout3.parquet (the three-way holdout). Everything upstream
(data_prep, tokenizer, pretrain, finetune) is forbidden from touching them, so the numbers
written here are the project's honest generalisation estimate.

Subcommand ``run --checkpoint <finetune best dir> [--split test|test3|val|test,test3]``:

1. load ``EncoderForClassification`` from ``config.json`` + ``model.safetensors`` (strict;
   ``num_labels`` comes from ``config.json["model"]``), batched inference under bf16
   autocast with length-bucketed dynamic padding and bool key-padding masks -> the full
   softmax ``P(class)`` for every row.
2. **Binary roll-up, identical to the binary-era report.** ``P(unwanted) = P(phishing) +
   P(spam)`` (for a 2-class checkpoint this is exactly ``P(class 1)``), so accuracy,
   precision, recall, F1, specificity, ROC-AUC, PR-AUC, recall at 1% / 0.1% FPR and the
   best-F1 threshold stay comparable with every earlier run. **Recall on the unwanted
   class is reported first**: in a defensive triage tool a missed phish (false negative)
   is the costly error, a false positive costs a human a glance. Plus: a credential-lure
   slice (``CREDENTIAL_LURE_PHRASES``), a per-source table, and the token-length confound
   check that RUN_LOG entry 2 asks for.
3. **Three-way metrics** (RUN_LOG entry 6) on the rows whose ``label3`` is in {0, 1, 2}:
   3x3 confusion matrix (plot + table), per-class precision / recall / F1, macro-F1, and a
   phishing-vs-spam confusion restricted to the true unwanted rows. Rows labelled
   ``-1`` / ``"unwanted"`` (TREC-05 / TREC-07 / CEAS-08 positives, whose corpora never
   separated spam from phishing) only enter the binary roll-up. The three-way decision
   rule is **roll-up first**: a row is unwanted iff ``P(unwanted) >= threshold``, and then
   the kind is ``argmax(P(phishing), P(spam))``, so the 3x3 matrix collapses exactly onto
   the binary confusion matrix above; plain argmax-of-3 is reported next to it. On the
   TREC-06 split there is no phishing/spam ground truth, so only an *informational*
   breakdown of how the flagged rows split into predicted phishing vs spam is written,
   and it says so. A 2-class checkpoint skips every three-way section with a log line.
4. single-email latency like the instructor's step 5: tokenize + forward + softmax at
   batch 1 timed with ``timeit.default_timer``, GPU bf16 and CPU fp32 (16 threads),
   printed next to the 74 ms benchmark -> ``latency.json`` (measured once per invocation).
5. plots (confusion matrices, ROC, PR), ``error_analysis.md`` (top false negatives /
   positives, phishing->spam and spam->phishing confusions with previews, lexical
   patterns, written summary), ``metrics.json`` (with a ``threeway`` block), ONE line in
   ``experiments.jsonl`` and ONE RUN_LOG.md entry per invocation covering every split
   that was run (skipped for ``--limit-rows`` dev runs, which are not evidence).
   ``reports/pretrain_curves.png`` and ``finetune_curves.png`` are written by
   ``pretrain.py plot`` / ``finetune.py plot`` and are never touched here;
   ``--finetune-run-dir`` / ``--pretrain-run-dir`` opt into extra ``eval_*_curves.png``.

Outputs for ``--split val`` get a ``_val`` suffix and for ``--split test3`` a ``_test3``
suffix so no split's report overwrites another's. ``--out-dir`` redirects every file, so a
self-test never writes into reports/; ``--data-dir`` points at fixture parquet files.
"""

from __future__ import annotations

import argparse
import json
import logging
import math
import re
import sys
import timeit
import unicodedata
from collections import Counter
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Sequence

sys.path.insert(0, str(Path(__file__).parent))

import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt  # noqa: E402
import numpy as np  # noqa: E402
import pyarrow.parquet as pq  # noqa: E402
import torch  # noqa: E402
from safetensors.torch import load_file  # noqa: E402
from sklearn.metrics import (  # noqa: E402
    ConfusionMatrixDisplay,
    average_precision_score,
    confusion_matrix,
    precision_recall_curve,
    precision_recall_fscore_support,
    roc_auc_score,
    roc_curve,
)

import common  # noqa: E402
from model import EncoderConfig, EncoderForClassification, count_parameters  # noqa: E402
from tokenizer import encode_batch, load_tokenizer  # noqa: E402

LOG = logging.getLogger("evaluate")

SPLIT_FILES: dict[str, str] = {"test": "test_holdout.parquet", "val": "finetune_val.parquet", "test3": "test_holdout3.parquet"}
SPLIT_SUFFIX: dict[str, str] = {"test": "", "val": "_val", "test3": "_test3"}
SPLIT_TITLE: dict[str, str] = {"test": "TREC-06 binary holdout", "val": "validation split", "test3": "three-way holdout"}
# Three-way label contract (RUN_LOG entry 6): 0 legitimate, 1 phishing, 2 spam; -1 = unwanted-but-unsplit
# (TREC-05 / TREC-07 / CEAS-08 positives), which only enter the binary roll-up.
LABEL_NAMES: dict[int, str] = {0: "Legitimate", 1: "Phishing", 2: "Spam"}
UNWANTED_NAME = "Unwanted (phishing or spam)"
BINARY_NAMES: dict[int, str] = {0: "Legitimate", 1: UNWANTED_NAME}
LABEL_SET_OF: dict[int, str] = {0: "legit", 1: "phish", 2: "spam", -1: "unwanted"}
DECISION_RULE = "rollup_then_argmax"
DECISION_RULE_DOC = ("unwanted iff P(phishing)+P(spam) >= threshold, then kind = argmax(P(phishing), P(spam)); "
                     "the 3x3 matrix therefore collapses exactly onto the binary confusion matrix. argmax-of-3 reported alongside.")
INSTRUCTOR_BENCHMARK_MS = 74.0
DEFAULT_THRESHOLD = 0.5
CPU_LATENCY_BUDGET_S = 60.0
CPU_LATENCY_FALLBACK_N = 10
TOP_K_ERRORS = 25
TOP_K_CONFUSIONS = 25
PREVIEW_CHARS = 160
ENCODE_BATCH_ROWS = 2048
LENGTH_BUCKETS: tuple[tuple[int, int, str], ...] = (
    (0, 64, "<=64"), (65, 128, "65-128"), (129, 256, "129-256"), (257, 512, "257-512"), (513, 10**9, ">512"),
)

# Credential-lure phrases. A row belongs to the slice when the phrase occurs as whole
# words in the lure-normalised text (NFKC -> lowercase -> every non-alphanumeric run
# becomes one space), so casing, punctuation, HTML tags and line breaks cannot hide a
# match ("Verify  your<br>account" still hits) while word boundaries hold: "urgent"
# no longer matches "insurgent"/"resurgent" (the alnum-only match did), and "urgently"
# is listed separately because urgency is the common denominator of credential lures.
# The list is documented here rather than tuned on the test set: it is the subset of
# the positive class whose cost of a miss is highest (account takeover / payment
# fraud). Phrases that were majority-legitimate on finetune_val (limited access, credit
# card number, billing/payment information, social security number) were dropped so
# the slice's "precision" says something about the model rather than about the slice
# definition; the slice's own label purity is printed next to its metrics for the same
# reason.
CREDENTIAL_LURE_PHRASES: tuple[str, ...] = (
    "verify your account", "verify your identity", "confirm your identity", "confirm your account",
    "enter your password", "update your account", "update your information", "update your billing",
    "update your payment", "click here to verify", "click the link below to verify",
    "account has been suspended", "account has been limited", "account will be suspended",
    "account has been locked", "unusual sign-in", "unusual activity", "suspicious activity",
    "paypal account", "ebay account", "password expire", "password will expire", "reset your password",
    "invoice attached", "wire transfer", "gift card", "urgent", "urgently", "security alert", "security notice",
    "login to your account", "log in to your account", "sign in to your account",
    "bank account", "failure to verify", "within 24 hours", "within 48 hours",
)
LURE_MIN_POSITIVES = 20   # below this the slice recall is reported but not compared to the overall recall
_LURE_NON_ALNUM = re.compile(r"[^a-z0-9]+")

# Palette: dataviz reference instance (slot 1 blue, 2 orange, 3 aqua; ink; surface).
C_BLUE, C_ORANGE, C_AQUA, C_VIOLET = "#2a78d6", "#eb6834", "#1baf7a", "#4a3aa7"
C_INK, C_INK2, C_GRID, C_SURFACE = "#0b0b0b", "#52514e", "#dddcd7", "#fcfcfb"
PLOT_DPI = 150


# ------------------------------------------------------------------------------- data
@dataclass
class Split:
    name: str
    path: Path
    doc_id: list[str]
    text: list[str]
    subject: list[str]
    body: list[str]
    label: np.ndarray        # binary: 1 = unwanted (phishing, spam or unsplit positive)
    label3: np.ndarray       # 0 legitimate / 1 phishing / 2 spam / -1 unwanted-but-unsplit
    label_set: list[str]     # legit / phish / spam / unwanted
    source: list[str]
    limit_rows: int | None

    @property
    def n(self) -> int:
        return len(self.text)

    @property
    def sfx(self) -> str:
        return SPLIT_SUFFIX[self.name]


def check_eval_config(cfg: dict) -> int:
    """Reconcile the documented ``eval:`` dials with this module's constants and return ``top_k_confusions``.

    ``label_names``, ``unwanted_name`` and ``threeway_decision_rule`` are the three-way label
    contract shared with finetune.py, export.py and the app, so they are fixed in code and
    config.yaml may only restate them: a drift raises here, loudly, instead of producing a card
    whose names differ from the metrics. ``top_k_confusions`` is a real dial (size of the
    phishing->spam / spam->phishing preview tables). ``eval.metrics_threeway`` is documentation
    only: it lists the sections the three-way block always contains."""
    ev = cfg.get("eval") or {}
    names = ev.get("label_names")
    if names is not None:
        got = {int(c): str(v) for c, v in dict(names).items()}
        if got != LABEL_NAMES:
            raise ValueError(f"config eval.label_names {got} != evaluate.LABEL_NAMES {LABEL_NAMES}: the label contract is fixed in code")
    if ev.get("unwanted_name") is not None and str(ev["unwanted_name"]) != UNWANTED_NAME:
        raise ValueError(f"config eval.unwanted_name {ev['unwanted_name']!r} != {UNWANTED_NAME!r}: fixed in code")
    if ev.get("threeway_decision_rule") is not None and str(ev["threeway_decision_rule"]) != DECISION_RULE:
        raise ValueError(f"config eval.threeway_decision_rule {ev['threeway_decision_rule']!r} != {DECISION_RULE!r}: only this rule is implemented")
    top_k = int(ev.get("top_k_confusions", TOP_K_CONFUSIONS))
    if top_k < 1:
        raise ValueError(f"config eval.top_k_confusions must be >= 1, got {top_k}")
    return top_k


def split_path(data_dir: Path, name: str, test3_file: str) -> Path:
    return data_dir / (test3_file if name == "test3" else SPLIT_FILES[name])


def load_split(data_dir: Path, name: str, limit_rows: int | None, seed: int, test3_file: str = SPLIT_FILES["test3"]) -> Split:
    """Read one labelled split. ``limit_rows`` takes a seeded random sample (dev only);
    the full run reads every row so the report is the whole holdout.

    ``label3`` / ``label_set`` are read when the file carries them (data_prep writes them
    for every split since RUN_LOG entry 6). A binary-era file without them is still
    evaluable: every positive is an unsplit "unwanted" row (-1) and every negative is
    legitimate (0), which is precisely the TREC-06 contract, so the binary roll-up is
    unaffected and the three-way section simply has no rows."""
    path = split_path(data_dir, name, test3_file)
    if not path.exists():
        raise FileNotFoundError(path)
    have = set(pq.read_schema(path).names)
    cols = ["doc_id", "text", "subject", "body", "source"] + [c for c in ("label", "label3", "label_set") if c in have]
    if "label" not in have and "label3" not in have:
        raise ValueError(f"{path}: neither 'label' nor 'label3' column present")
    table = pq.read_table(path, columns=cols)
    if limit_rows is not None and table.num_rows > limit_rows:
        idx = np.sort(np.random.default_rng(seed).choice(table.num_rows, size=limit_rows, replace=False))
        table = table.take(idx)
    d = table.to_pydict()
    n = table.num_rows
    if "label3" in have:
        label3 = np.asarray(d["label3"], dtype=np.int64)
        bad = sorted(set(label3.tolist()) - set(LABEL_SET_OF))
        if bad:
            raise ValueError(f"{path}: label3 values outside {{-1,0,1,2}}: {bad}")
    else:
        label3 = None
    if "label" in have:
        label = np.asarray(d["label"], dtype=np.int64)
    else:
        label = (label3 != 0).astype(np.int64)
    if label3 is None:
        label3 = np.where(label == 1, -1, 0).astype(np.int64)
        LOG.info("%s has no label3 column: positives treated as unwanted-but-unsplit (-1), negatives as legitimate (0)", path.name)
    if "label_set" in have:
        label_set = [str(s) for s in d["label_set"]]
        # label_set is the human-readable twin of label3 (finetune.read_split enforces the same
        # pairing); a disagreement means data_prep wrote inconsistent columns, and the error
        # tables below print label_set, so refuse rather than report the wrong class name.
        mismatch = [i for i in range(n) if label_set[i] != LABEL_SET_OF[int(label3[i])]]
        if mismatch:
            i = mismatch[0]
            raise ValueError(f"{path}: 'label_set' disagrees with 'label3' on {len(mismatch)} row(s) (first: row {i} "
                             f"label_set={label_set[i]!r} but label3={int(label3[i])} implies {LABEL_SET_OF[int(label3[i])]!r}); "
                             f"contract: {LABEL_SET_OF}")
    else:
        label_set = [LABEL_SET_OF[int(v)] for v in label3]
    if bool(((label3 != 0).astype(np.int64) != label).any()):
        raise ValueError(f"{path}: 'label' disagrees with 'label3 != 0' on some rows (contract: label = 1 for phish/spam/unwanted)")
    return Split(
        name=name, path=path,
        doc_id=[str(x) for x in d["doc_id"]],
        text=[t or "" for t in d["text"]],
        subject=[s or "" for s in d["subject"]],
        body=[b or "" for b in d["body"]],
        label=label, label3=label3, label_set=label_set,
        source=[str(s) for s in d["source"]],
        limit_rows=limit_rows,
    )


@dataclass
class Encoded:
    """Per-row token ids as the model sees them, plus the untruncated length for the confound check."""

    ids: list[list[int]]
    n_tokens: np.ndarray        # fed to the model (incl. [CLS]/[SEP]), <= max_len
    n_tokens_full: np.ndarray   # before truncation
    truncated: np.ndarray


def encode_split(tok, texts: Sequence[str], max_len: int) -> Encoded:
    """Encode once without truncation (so the full length is known), then truncate the
    content to ``max_len - 2`` and keep ``[CLS] ... [SEP]``, which is exactly what
    ``tokenizers`` truncation + the BERT template produce."""
    tok.no_truncation()
    tok.no_padding()
    sep_id = tok.token_to_id("[SEP]")
    ids: list[list[int]] = []
    full = np.empty(len(texts), dtype=np.int64)
    for i in range(0, len(texts), ENCODE_BATCH_ROWS):
        for j, enc in enumerate(tok.encode_batch_fast(list(texts[i:i + ENCODE_BATCH_ROWS]))):
            seq = enc.ids
            full[i + j] = len(seq)
            if len(seq) > max_len:
                seq = seq[: max_len - 1] + [sep_id]
            ids.append(seq)
    n_tok = np.minimum(full, max_len)
    return Encoded(ids, n_tok, full, full > max_len)


# ------------------------------------------------------------------------------ model
def load_classifier(ckpt_dir: Path, device: torch.device) -> tuple[EncoderForClassification, EncoderConfig, dict]:
    """Rebuild the classifier from ``config.json["model"]`` (including ``num_labels``) and
    load ``model.safetensors`` strictly.

    Strict loading is non-negotiable: a half-loaded encoder still produces confident
    probabilities, and nothing downstream could tell the difference. A head whose shape
    disagrees with ``num_labels`` fails here, loudly. Weights may be fp32 (training
    checkpoint) or bf16 (export); ``load_state_dict`` casts into the fp32 module, and the
    GPU path runs under bf16 autocast either way.
    """
    cfg_json = common.read_json(ckpt_dir / "config.json")
    mcfg = EncoderConfig.from_dict(cfg_json["model"])
    mcfg.grad_checkpointing = False
    if mcfg.num_labels not in (2, 3):
        raise ValueError(f"num_labels={mcfg.num_labels}: this evaluator understands 2 (binary) or 3 (legitimate/phishing/spam)")
    with torch.device(device):
        model = EncoderForClassification(mcfg)
    sd = load_file(str(ckpt_dir / "model.safetensors"), device="cpu")
    model.load_state_dict(sd, strict=True)
    model.eval()
    dtypes = sorted({str(v.dtype).replace("torch.", "") for v in sd.values()})
    LOG.info("loaded %s: %d tensors (%s), %s params, num_labels=%d", ckpt_dir / "model.safetensors", len(sd), dtypes,
             f"{count_parameters(model)['total']:,}", mcfg.num_labels)
    return model, mcfg, cfg_json


def _pad_batch(enc: Encoded, idx: np.ndarray, multiple: int) -> tuple[torch.Tensor, torch.Tensor]:
    L = int(enc.n_tokens[idx].max())
    L = int(math.ceil(L / multiple) * multiple)
    arr = np.zeros((len(idx), L), dtype=np.int64)
    mask = np.zeros((len(idx), L), dtype=np.bool_)
    for r, i in enumerate(idx):
        seq = enc.ids[i]
        arr[r, : len(seq)] = seq
        mask[r, : len(seq)] = True
    return torch.from_numpy(arr), torch.from_numpy(mask)


@torch.inference_mode()
def predict_probs(model: EncoderForClassification, enc: Encoded, device: torch.device,
                  batch_size: int, pad_multiple: int) -> tuple[np.ndarray, dict[str, Any]]:
    """Full softmax ``[n, num_labels]`` per row. Rows are sorted longest-first so each batch
    pads to its own longest member (dynamic padding; ~2x fewer pad tokens than padding to
    512) and an OOM would surface on the first batch rather than after ten minutes."""
    n = len(enc.ids)
    k = int(model.config.num_labels)
    order = np.argsort(-enc.n_tokens, kind="stable")
    probs = np.empty((n, k), dtype=np.float32)
    use_amp = device.type == "cuda"
    padded_tokens = 0
    n_batches = math.ceil(n / batch_size)
    t0 = common.timer()
    for b, start in enumerate(range(0, n, batch_size)):
        idx = order[start: start + batch_size]
        ids, mask = _pad_batch(enc, idx, pad_multiple)
        padded_tokens += ids.numel()
        ids, mask = ids.to(device, non_blocking=True), mask.to(device, non_blocking=True)
        with torch.autocast(device_type="cuda", dtype=torch.bfloat16, enabled=use_amp):
            logits = model(ids, mask)
        probs[idx] = logits.float().softmax(-1).cpu().numpy()
        if b % 20 == 0 or b == n_batches - 1:
            LOG.info("inference batch %d/%d (L=%d) %.1fs", b + 1, n_batches, ids.shape[1], common.timer() - t0)
    if use_amp:
        torch.cuda.synchronize()
    secs = common.timer() - t0
    info = {
        "device": str(device), "autocast_bf16": use_amp, "batch_size": batch_size, "pad_to_multiple_of": pad_multiple,
        "num_labels": k, "rows": n, "seconds": round(secs, 2), "rows_per_s": round(n / secs, 1),
        "padded_tokens": int(padded_tokens), "real_tokens": int(enc.n_tokens.sum()),
        "padded_tokens_per_s": round(padded_tokens / secs),
        "peak_vram_gb": round(common.vram_stats()["max_allocated_gb"], 2),
    }
    LOG.info("inference: %s", info)
    return probs, info


def p_unwanted(probs: np.ndarray) -> np.ndarray:
    """Binary roll-up: P(unwanted) = sum of every non-legitimate class. For a 2-class head
    this is exactly P(class 1), so the binary report is byte-for-byte the old definition."""
    return np.clip(probs[:, 1:].sum(axis=1), 0.0, 1.0).astype(np.float32)


# ---------------------------------------------------------------------------- metrics
def _safe_div(a: float, b: float) -> float:
    return float(a) / float(b) if b else 0.0


def threshold_metrics(y: np.ndarray, p: np.ndarray, thr: float) -> dict[str, Any]:
    """Confusion-matrix metrics at ``p >= thr``. An empty slice (e.g. no credential-lure
    rows in a small ``--limit-rows`` sample) returns all-zero counts instead of hitting
    sklearn's minimum-one-sample check."""
    if len(y) == 0:
        tn = fp = fn = tp = 0
    else:
        pred = (p >= thr).astype(np.int64)
        cm = confusion_matrix(y, pred, labels=[0, 1])
        tn, fp, fn, tp = (int(v) for v in cm.ravel())
    precision = _safe_div(tp, tp + fp)
    recall = _safe_div(tp, tp + fn)
    return {
        "threshold": thr,
        "accuracy": _safe_div(tp + tn, len(y)),
        "precision": precision,
        "recall": recall,
        "f1": _safe_div(2 * precision * recall, precision + recall),
        "specificity": _safe_div(tn, tn + fp),
        "false_positive_rate": _safe_div(fp, tn + fp),
        "false_negative_rate": _safe_div(fn, tp + fn),
        "confusion_matrix": {"tn": tn, "fp": fp, "fn": fn, "tp": tp},
        "n": int(len(y)), "n_positive": int(y.sum()), "n_negative": int(len(y) - y.sum()),
    }


def _both_classes(y: np.ndarray) -> bool:
    return bool(y.min() == 0 and y.max() == 1) if len(y) else False


def threshold_free_metrics(y: np.ndarray, p: np.ndarray) -> dict[str, Any]:
    """ROC-AUC, PR-AUC (average precision), recall at a fixed FPR budget read off the ROC
    curve, and the threshold that maximises F1. The FPR-budget numbers are the ones an
    inbox operator actually asks ("if I may bother users with 1 false alarm in 100
    legitimate emails, how many phish do I catch?"). Thresholds are to be applied as
    ``p >= threshold``; when the only point inside the budget is the ROC origin, sklearn's
    sentinel threshold is +inf and ``threshold`` is reported as None (no finite score
    cut reaches that budget) rather than clamped to 1.0, which bf16 probabilities that
    round to exactly 1.0 would still pass."""
    if not _both_classes(y):
        LOG.warning("only one class present: ROC/PR metrics undefined")
        return {"roc_auc": None, "pr_auc": None, "recall_at_fpr_1pct": None, "recall_at_fpr_0_1pct": None,
                "best_f1": None, "curves": None}
    fpr, tpr, roc_thr = roc_curve(y, p)
    prec, rec, pr_thr = precision_recall_curve(y, p)

    def recall_at_fpr(budget: float) -> dict[str, float]:
        ok = np.nonzero(fpr <= budget)[0]
        i = ok[np.argmax(tpr[ok])]
        t = float(roc_thr[i])
        return {"fpr_budget": budget, "recall": float(tpr[i]), "fpr": float(fpr[i]),
                "threshold": t if math.isfinite(t) else None, "threshold_rule": "flag when p >= threshold"}

    f1 = 2 * prec[:-1] * rec[:-1] / np.clip(prec[:-1] + rec[:-1], 1e-12, None)
    j = int(np.argmax(f1))
    return {
        "roc_auc": float(roc_auc_score(y, p)),
        "pr_auc": float(average_precision_score(y, p)),
        "recall_at_fpr_1pct": recall_at_fpr(0.01),
        "recall_at_fpr_0_1pct": recall_at_fpr(0.001),
        "best_f1": {"threshold": float(pr_thr[j]), "f1": float(f1[j]), "precision": float(prec[j]), "recall": float(rec[j])},
        "curves": {"fpr": fpr, "tpr": tpr, "precision": prec, "recall": rec},
    }


def lure_normalize(s: str) -> str:
    """Word-boundary-preserving sibling of ``common.normalize_text``: HTML tags removed,
    NFKC -> lowercase -> every non-alphanumeric run collapses to one space, padded with a
    space on both ends so a phrase match ``" ph "`` is a whole-word match
    ("Verify your<br>account" -> " verify your account ")."""
    s = _HTML_TAG.sub(" ", s or "")
    return " " + _LURE_NON_ALNUM.sub(" ", unicodedata.normalize("NFKC", s).lower()).strip() + " "


def lure_mask(texts: Sequence[str], labels: np.ndarray | None = None) -> tuple[np.ndarray, Counter, dict[str, dict[str, int]]]:
    """Rows containing any lure phrase as whole words, per-phrase hit counts, and (when
    labels are given) per-phrase positive/negative hits so the slice's purity is visible."""
    norm_phrases = [(ph, lure_normalize(ph)) for ph in CREDENTIAL_LURE_PHRASES]
    hits = np.zeros(len(texts), dtype=np.bool_)
    per_phrase: Counter = Counter()
    by_label: dict[str, dict[str, int]] = {ph: {"positive": 0, "negative": 0} for ph in CREDENTIAL_LURE_PHRASES}
    for i, t in enumerate(texts):
        n = lure_normalize(t)
        for ph, nph in norm_phrases:
            if nph in n:
                hits[i] = True
                per_phrase[ph] += 1
                if labels is not None:
                    by_label[ph]["positive" if int(labels[i]) == 1 else "negative"] += 1
    return hits, per_phrase, {ph: c for ph, c in by_label.items() if per_phrase[ph]}


def slice_metrics(y: np.ndarray, p: np.ndarray, thr: float) -> dict[str, Any]:
    m = threshold_metrics(y, p, thr)
    m["roc_auc"] = float(roc_auc_score(y, p)) if _both_classes(y) else None
    return m


def per_source_table(split: Split, p: np.ndarray, thr: float) -> dict[str, dict[str, Any]]:
    out: dict[str, dict[str, Any]] = {}
    src = np.asarray(split.source)
    for s in sorted(set(split.source)):
        sel = src == s
        m = slice_metrics(split.label[sel], p[sel], thr)
        m["label3_counts"] = _label3_counts(split.label3[sel])
        out[s] = m
    return out


def _label3_counts(label3: np.ndarray) -> dict[str, int]:
    return {LABEL_SET_OF[k]: int((label3 == k).sum()) for k in (0, 1, 2, -1)}


def length_confound(split: Split, enc: Encoded, p: np.ndarray, thr: float) -> dict[str, Any]:
    """Mean token length of FP/FN vs correct rows, by label, and recall/FPR per length
    bucket. If errors concentrate in the length range where the classes' length
    distributions overlap (long positives, short legitimate), the model is at least
    partly reading length rather than content. ``by_class3`` adds the phishing / spam /
    unsplit split of the positive class so a phishing-vs-spam length gap is visible too."""
    y, pred = split.label, (p >= thr).astype(np.int64)
    groups = {"tp": (y == 1) & (pred == 1), "tn": (y == 0) & (pred == 0),
              "fp": (y == 0) & (pred == 1), "fn": (y == 1) & (pred == 0)}

    def stats(sel: np.ndarray) -> dict[str, Any]:
        if not sel.any():
            return {"n": 0, "mean_tokens_full": None, "mean_tokens_fed": None, "median_tokens_full": None, "pct_truncated": None}
        return {"n": int(sel.sum()), "mean_tokens_full": round(float(enc.n_tokens_full[sel].mean()), 1),
                "mean_tokens_fed": round(float(enc.n_tokens[sel].mean()), 1),
                "median_tokens_full": float(np.median(enc.n_tokens_full[sel])),
                "pct_truncated": round(100 * float(enc.truncated[sel].mean()), 1)}

    by_outcome = {k: stats(v) for k, v in groups.items()}
    by_outcome["correct"] = stats(groups["tp"] | groups["tn"])
    by_outcome["wrong"] = stats(groups["fp"] | groups["fn"])
    by_label = {"legitimate": stats(y == 0), "unwanted": stats(y == 1)}
    by_class3 = {LABEL_SET_OF[k]: stats(split.label3 == k) for k in (0, 1, 2, -1) if (split.label3 == k).any()}
    buckets = []
    for lo, hi, name in LENGTH_BUCKETS:
        sel = (enc.n_tokens_full >= lo) & (enc.n_tokens_full <= hi)
        if not sel.any():
            continue
        m = threshold_metrics(y[sel], p[sel], thr)
        buckets.append({"bucket": name, "n": m["n"], "n_positive": m["n_positive"], "positive_rate": _safe_div(m["n_positive"], m["n"]),
                        "recall": m["recall"] if m["n_positive"] else None,
                        "fpr": m["false_positive_rate"] if m["n_negative"] else None, "accuracy": m["accuracy"]})
    return {"by_outcome": by_outcome, "by_label": by_label, "by_class3": by_class3, "by_length_bucket": buckets}


# ------------------------------------------------------------------------- three-way
@dataclass
class ThreeWay:
    """Three-way result for one split. ``metrics`` is the JSON block; ``pred3`` / ``kind``
    are per-row arrays (full length) that the error analysis uses, None for a 2-class head."""

    evaluated: bool
    metrics: dict[str, Any]
    pred3: np.ndarray | None
    kind: np.ndarray | None


def predict3(probs: np.ndarray, p_unw: np.ndarray, thr: float) -> tuple[np.ndarray, np.ndarray]:
    """Roll-up-first decision: ``kind`` = argmax over the unwanted classes (1 phishing /
    2 spam) for every row; ``pred3`` = kind when P(unwanted) >= thr else 0 (legitimate).
    Choosing the kind *conditionally* on being unwanted (rather than argmax-of-3) keeps the
    three-way prediction consistent with the binary verdict the app shows: a row the app
    flags can never be "legitimate" in the 3x3 table, and vice versa."""
    kind = 1 + probs[:, 1:3].argmax(axis=1)
    pred3 = np.where(p_unw >= thr, kind, 0).astype(np.int64)
    return pred3, kind.astype(np.int64)


def _cm_dict(mat: np.ndarray, labels: Sequence[int]) -> dict[str, dict[str, int]]:
    return {f"true_{LABEL_NAMES[a].lower()}": {f"pred_{LABEL_NAMES[b].lower()}": int(mat[i, j]) for j, b in enumerate(labels)}
            for i, a in enumerate(labels)}


def _macro_block(y: np.ndarray, pred: np.ndarray) -> dict[str, Any]:
    p, r, f, s = precision_recall_fscore_support(y, pred, labels=[0, 1, 2], zero_division=0)
    per_class = {LABEL_NAMES[k]: {"precision": float(p[k]), "recall": float(r[k]), "f1": float(f[k]), "support": int(s[k])} for k in range(3)}
    present = [k for k in range(3) if s[k] > 0]
    return {"accuracy": float((y == pred).mean()) if len(y) else 0.0,
            "macro_f1": float(np.mean([f[k] for k in range(3)])),
            "macro_f1_present_classes": float(np.mean([f[k] for k in present])) if present else 0.0,
            "classes_present": [LABEL_NAMES[k] for k in present],
            "per_class": per_class}


def threeway_metrics(split: Split, probs: np.ndarray, p_unw: np.ndarray, thr: float) -> ThreeWay:
    """Three-way block for one split.

    * 2-class head: nothing to evaluate (the checkpoint has no phishing-vs-spam output).
    * 3-class head: ``predicted_kind_breakdown`` is ALWAYS written (how the flagged rows
      split into predicted phishing vs spam); it is informational on TREC-06, where every
      positive is an unsplit "unwanted" row and no ground truth exists, and it says so.
    * rows with ``label3`` in {0, 1, 2}: 3x3 confusion matrix, per-class P/R/F1, macro-F1
      (over all three classes, zero F1 for an absent class, plus the mean over the classes
      actually present), argmax-of-3 as a secondary rule, a phishing-vs-spam confusion on
      the true unwanted rows (the kind decision alone, i.e. "given it is unwanted, which
      kind", plus the roll-up view that also shows how many were missed as legitimate),
      and per-source class counts / accuracy / macro-F1.
    """
    k = probs.shape[1]
    base: dict[str, Any] = {"num_labels": k, "label_names": {str(c): LABEL_NAMES[c] for c in range(k) if c in LABEL_NAMES},
                            "unwanted_name": UNWANTED_NAME, "decision_rule": DECISION_RULE, "decision_rule_doc": DECISION_RULE_DOC,
                            "threshold": thr, "label3_counts": _label3_counts(split.label3)}
    if k != 3:
        base["label_names"] = {"0": BINARY_NAMES[0], "1": BINARY_NAMES[1]}
        return ThreeWay(False, {**base, "evaluated": False, "reason": f"{k}-class checkpoint: no phishing-vs-spam head, three-way sections skipped"}, None, None)
    pred3, kind = predict3(probs, p_unw, thr)
    argmax3 = probs.argmax(axis=1)
    y, y3 = split.label, split.label3
    flagged = p_unw >= thr

    def kinds(sel: np.ndarray) -> dict[str, int]:
        return {"n": int(sel.sum()), "predicted_phishing": int((kind[sel] == 1).sum()), "predicted_spam": int((kind[sel] == 2).sum())}

    base["predicted_kind_breakdown"] = {
        "note": "How the model's flagged rows split into predicted phishing vs spam (kind = argmax(P(phishing), P(spam))). "
                "Informational: for rows labelled -1/unwanted (all TREC-06 positives) there is NO phishing-vs-spam ground truth.",
        "true_positives_flagged": kinds((y == 1) & flagged), "false_positives_flagged": kinds((y == 0) & flagged),
        "missed_positives": kinds((y == 1) & ~flagged), "all_flagged": kinds(flagged),
        "unsplit_unwanted_rows_flagged": kinds((y3 == -1) & flagged),
    }
    full = (y3 >= 0) & (y3 <= 2)
    base["rows_with_full_label"] = int(full.sum())
    base["rows_unwanted_only"] = int((y3 == -1).sum())
    # Legitimate rows alone (label3 == 0) say nothing about phishing-vs-spam: a split whose every
    # positive is unsplit (TREC-06) would otherwise yield a degenerate 3x3 with zero phishing/spam
    # support and a meaningless macro-F1. At least one row labelled phishing or spam is required.
    if not (full & (y3 >= 1)).any():
        return ThreeWay(False, {**base, "evaluated": False,
                                "reason": "no rows labelled phishing or spam: every positive is unwanted-but-unsplit (no phishing-vs-spam ground truth)"},
                        pred3, kind)
    yf, pf, af = y3[full], pred3[full], argmax3[full]
    mat = confusion_matrix(yf, pf, labels=[0, 1, 2])
    main = _macro_block(yf, pf)
    alt = _macro_block(yf, af)
    tu = full & (y3 >= 1)
    if tu.any():
        kmat = confusion_matrix(y3[tu], kind[tu], labels=[1, 2])
        rmat = confusion_matrix(y3[tu], pred3[tu], labels=[0, 1, 2])[1:, :]
        pvs = {"n_true_unwanted": int(tu.sum()),
               "kind_only": {"doc": "given the row is unwanted, is argmax(P(phishing), P(spam)) right? (independent of the roll-up threshold)",
                             "labels": ["Phishing", "Spam"], "matrix": kmat.tolist(), "cells": _cm_dict(kmat, [1, 2]),
                             "accuracy": float(np.trace(kmat) / max(kmat.sum(), 1)),
                             "phishing_as_spam": int(kmat[0, 1]), "spam_as_phishing": int(kmat[1, 0])},
               "rollup_rule": {"doc": "rows = true Phishing / Spam; columns = predicted Legitimate (missed) / Phishing / Spam under the roll-up rule",
                               "rows": ["Phishing", "Spam"], "columns": ["Legitimate", "Phishing", "Spam"], "matrix": rmat.tolist(),
                               "missed_as_legitimate": {"phishing": int(rmat[0, 0]), "spam": int(rmat[1, 0])}}}
    else:
        pvs = {"n_true_unwanted": 0, "kind_only": None, "rollup_rule": None}
    src = np.asarray(split.source)
    per_source: dict[str, Any] = {}
    for s in sorted(set(split.source)):
        sel = full & (src == s)
        if not sel.any():
            continue
        b = _macro_block(y3[sel], pred3[sel])
        per_source[s] = {"n": int(sel.sum()), "counts": {LABEL_NAMES[c]: int((y3[sel] == c).sum()) for c in range(3)},
                         "accuracy": b["accuracy"], "macro_f1": b["macro_f1"], "macro_f1_present_classes": b["macro_f1_present_classes"],
                         "per_class_recall": {c: v["recall"] for c, v in b["per_class"].items() if v["support"]}}
    metrics = {**base, "evaluated": True, "reason": None,
               "confusion_matrix": {"labels": [LABEL_NAMES[c] for c in range(3)], "matrix": mat.tolist(), "cells": _cm_dict(mat, [0, 1, 2])},
               "accuracy": main["accuracy"], "macro_f1": main["macro_f1"], "macro_f1_present_classes": main["macro_f1_present_classes"],
               "classes_present": main["classes_present"], "per_class": main["per_class"],
               "argmax_rule": {"doc": "plain argmax over the three probabilities (secondary; can disagree with the binary verdict near the threshold)",
                               "accuracy": alt["accuracy"], "macro_f1": alt["macro_f1"], "per_class": alt["per_class"],
                               "rows_where_rules_differ": int((pf != af).sum())},
               "phish_vs_spam_on_true_unwanted": pvs, "per_source": per_source}
    return ThreeWay(True, metrics, pred3, kind)


# ---------------------------------------------------------------------------- latency
def _time_single(model, tok, texts: Sequence[str], device: torch.device, max_len: int,
                 use_amp: bool, warmup: int = 3) -> list[float]:
    """Wall-clock ms per email for tokenize + forward + softmax at batch 1, measured with
    ``timeit.default_timer`` exactly like the instructor's step-5 inference timer.
    ``.item()`` forces the GPU sync so the stop time includes the kernel work (the
    roll-up sum is part of what the app computes per request, so it is timed too)."""
    times: list[float] = []
    with torch.inference_mode():
        for k, t in enumerate(texts):
            t0 = timeit.default_timer()
            ids, mask = encode_batch(tok, [t], max_len)
            ids, mask = ids.to(device), mask.to(device)
            with torch.autocast(device_type="cuda", dtype=torch.bfloat16, enabled=use_amp):
                logits = model(ids, mask)
            _ = logits.float().softmax(-1)[0, 1:].sum().item()
            if k >= warmup:
                times.append((timeit.default_timer() - t0) * 1000.0)
    return times


def _summ(times: list[float], n_tokens: Sequence[int]) -> dict[str, Any]:
    arr = np.asarray(times)
    return {"n": int(len(arr)), "avg_ms": round(float(arr.mean()), 2), "p50_ms": round(float(np.median(arr)), 2),
            "p95_ms": round(float(np.percentile(arr, 95)), 2), "max_ms": round(float(arr.max()), 2),
            "avg_s": round(float(arr.mean()) / 1000.0, 4), "mean_tokens": round(float(np.mean(n_tokens)), 1),
            "ratio_to_benchmark": round(float(arr.mean()) / INSTRUCTOR_BENCHMARK_MS, 3)}


def measure_latency(model, tok, split: Split, enc: Encoded, device: torch.device, max_len: int,
                    n_samples: int, cpu_threads: int, seed: int) -> dict[str, Any]:
    """GPU bf16 then CPU fp32 on the same seeded sample of real emails. The CPU pass is
    cut to 10 samples when 3 warmups predict more than 60 s for the full count (the
    smoke test measured ~0.7 s per 512-token email on 16 threads). Moves the model to
    the CPU, so it is the last thing an invocation does with the model."""
    rng = np.random.default_rng(seed)
    n = min(n_samples, split.n)
    idx = rng.choice(split.n, size=n, replace=False)
    texts = [split.text[i] for i in idx]
    toks = [int(enc.n_tokens[i]) for i in idx]
    out: dict[str, Any] = {"instructor_benchmark_ms": INSTRUCTOR_BENCHMARK_MS, "samples_requested": n_samples,
                           "sample_split": split.name, "max_len": max_len}
    warm = texts[:3] + texts  # 3 warmups, then every sample timed
    if device.type == "cuda":
        torch.cuda.synchronize()
        gpu_times = _time_single(model, tok, warm, device, max_len, use_amp=True)
        out["gpu_bf16"] = _summ(gpu_times, toks)
        out["gpu_bf16"]["device"] = torch.cuda.get_device_name(device)
        LOG.info("GPU bf16 latency: %s", out["gpu_bf16"])
    else:
        out["gpu_bf16"] = None
    model.cpu().float()
    torch.set_num_threads(cpu_threads)
    cpu = torch.device("cpu")
    probe = _time_single(model, tok, texts[:3] + texts[:3], cpu, max_len, use_amp=False, warmup=3)
    est_s = float(np.mean(probe)) / 1000.0 * n
    n_cpu = n if est_s <= CPU_LATENCY_BUDGET_S or n <= CPU_LATENCY_FALLBACK_N else CPU_LATENCY_FALLBACK_N
    if n_cpu != n:
        LOG.info("CPU latency: %d samples would take ~%.0f s (> %.0f s budget); using %d", n, est_s, CPU_LATENCY_BUDGET_S, n_cpu)
    cpu_times = _time_single(model, tok, texts[:3] + texts[:n_cpu], cpu, max_len, use_amp=False)
    out["cpu_fp32"] = _summ(cpu_times, toks[:n_cpu])
    out["cpu_fp32"]["threads"] = torch.get_num_threads()
    out["cpu_fp32"]["estimated_full_run_s"] = round(est_s, 1)
    LOG.info("CPU fp32 latency: %s", out["cpu_fp32"])
    return out


# ------------------------------------------------------------------------------ plots
def _style(ax, title: str, xlabel: str, ylabel: str) -> None:
    ax.set_title(title, color=C_INK, fontsize=11, loc="left")
    ax.set_xlabel(xlabel, color=C_INK2)
    ax.set_ylabel(ylabel, color=C_INK2)
    ax.grid(True, color=C_GRID, linewidth=0.6)
    for s in ("top", "right"):
        ax.spines[s].set_visible(False)
    for s in ("left", "bottom"):
        ax.spines[s].set_color(C_GRID)
    ax.tick_params(colors=C_INK2)
    ax.set_facecolor(C_SURFACE)


def _save(fig, path: Path) -> None:
    fig.patch.set_facecolor(C_SURFACE)
    fig.tight_layout()
    fig.savefig(path, dpi=PLOT_DPI)
    plt.close(fig)
    LOG.info("wrote %s", path)


def plot_confusion(cm: dict[str, int], path: Path, title: str) -> None:
    mat = np.array([[cm["tn"], cm["fp"]], [cm["fn"], cm["tp"]]])
    fig, ax = plt.subplots(figsize=(5.6, 4.8))
    ConfusionMatrixDisplay(mat, display_labels=[BINARY_NAMES[0], BINARY_NAMES[1]]).plot(
        ax=ax, cmap="Blues", values_format="d", colorbar=False)
    ax.set_title(title, color=C_INK, fontsize=11, loc="left")
    ax.set_xlabel("Predicted label", color=C_INK2)
    ax.set_ylabel("True label", color=C_INK2)
    ax.tick_params(colors=C_INK2)
    plt.setp(ax.get_xticklabels(), rotation=15, ha="right", fontsize=8)
    plt.setp(ax.get_yticklabels(), fontsize=8)
    _save(fig, path)


def plot_confusion3(mat: Sequence[Sequence[int]], path: Path, title: str) -> None:
    fig, ax = plt.subplots(figsize=(5.6, 4.8))
    ConfusionMatrixDisplay(np.asarray(mat), display_labels=[LABEL_NAMES[c] for c in range(3)]).plot(
        ax=ax, cmap="Blues", values_format="d", colorbar=False)
    ax.set_title(title, color=C_INK, fontsize=11, loc="left")
    ax.set_xlabel("Predicted label (roll-up first, then argmax phishing/spam)", color=C_INK2, fontsize=9)
    ax.set_ylabel("True label (corpus provenance)", color=C_INK2)
    ax.tick_params(colors=C_INK2)
    _save(fig, path)


def plot_roc(tf: dict[str, Any], path: Path, title: str) -> None:
    c = tf["curves"]
    fig, axes = plt.subplots(1, 2, figsize=(10, 4.4))
    for ax, logx in zip(axes, (False, True)):
        ax.plot(c["fpr"], c["tpr"], color=C_BLUE, linewidth=2, label=f"ROC (AUC {tf['roc_auc']:.4f})")
        if not logx:
            ax.plot([0, 1], [0, 1], color=C_GRID, linewidth=1, linestyle="--")
        for key, col, lab, dy in (("recall_at_fpr_1pct", C_ORANGE, "1% FPR", 16), ("recall_at_fpr_0_1pct", C_AQUA, "0.1% FPR", 4)):
            r = tf[key]
            x = max(r["fpr"], 1e-4 if logx else 0)
            ax.scatter([x], [r["recall"]], color=col, s=48, zorder=3)
            ax.annotate(f"recall {r['recall']:.3f} @ {lab}", (x, r["recall"]), textcoords="offset points",
                        xytext=(8, dy), color=C_INK2, fontsize=8, annotation_clip=False)
        if logx:
            ax.set_xscale("log")
            ax.set_xlim(1e-4, 1)
            _style(ax, "ROC, log false-positive rate", "False positive rate (log)", "Recall (true positive rate)")
        else:
            ax.set_xlim(0, 1)
            _style(ax, title, "False positive rate", "Recall (true positive rate)")
        ax.set_ylim(0, 1.02)
        ax.legend(loc="lower right", frameon=False, labelcolor=C_INK2)
    _save(fig, path)


def plot_pr(tf: dict[str, Any], base_rate: float, path: Path, title: str) -> None:
    c = tf["curves"]
    fig, ax = plt.subplots(figsize=(5.6, 4.4))
    ax.plot(c["recall"], c["precision"], color=C_BLUE, linewidth=2, label=f"PR (AP {tf['pr_auc']:.4f})")
    ax.axhline(base_rate, color=C_GRID, linewidth=1, linestyle="--")
    ax.text(0.01, base_rate + 0.01, f"positive rate {base_rate:.3f}", color=C_INK2, fontsize=8)
    b = tf["best_f1"]
    ax.scatter([b["recall"]], [b["precision"]], color=C_ORANGE, s=48, zorder=3)
    ax.annotate(f"best F1 {b['f1']:.3f} @ thr {b['threshold']:.3f}", (b["recall"], b["precision"]),
                textcoords="offset points", xytext=(-8, -14), ha="right", color=C_INK2, fontsize=8)
    ax.set_xlim(0, 1)
    ax.set_ylim(0, 1.02)
    _style(ax, title, "Recall (unwanted)", "Precision (unwanted)")
    ax.legend(loc="lower left", frameon=False, labelcolor=C_INK2)
    _save(fig, path)


def _read_csv(path: Path):
    import pandas as pd

    try:
        df = pd.read_csv(path)
    except Exception as e:  # noqa: BLE001 - a corrupt CSV must not fail the evaluation
        LOG.warning("could not read %s: %s", path, e)
        return None
    return df if len(df) else None


def plot_pretrain_curves(run_dir: Path, path: Path) -> bool:
    """Loss (train + val) and learning rate from the pretraining run's metrics.csv; two
    panels, one axis each (never a twin axis)."""
    df = _read_csv(run_dir / "metrics.csv")
    if df is None or "step" not in df or "loss" not in df:
        return False
    fig, axes = plt.subplots(1, 2, figsize=(10, 4.2))
    ax = axes[0]
    ax.plot(df["step"], df["loss"], color=C_BLUE, linewidth=1.6, label="train MLM loss")
    if "val_mlm_loss" in df and df["val_mlm_loss"].notna().any():
        v = df[df["val_mlm_loss"].notna()]
        ax.plot(v["step"], v["val_mlm_loss"], color=C_ORANGE, linewidth=2, marker="o", markersize=4, label="val MLM loss")
    ax.axhline(math.log(32768), color=C_GRID, linewidth=1, linestyle="--")
    ax.text(float(df["step"].min()), math.log(32768) + 0.05, "ln(V) = uniform guess", color=C_INK2, fontsize=8)
    _style(ax, f"MLM pretraining loss ({run_dir.name})", "optimizer step", "cross-entropy")
    ax.legend(frameon=False, labelcolor=C_INK2)
    ax = axes[1]
    if "lr" in df:
        ax.plot(df["step"], df["lr"], color=C_AQUA, linewidth=2)
        _style(ax, "Learning rate schedule", "optimizer step", "lr")
        ax.ticklabel_format(axis="y", style="sci", scilimits=(0, 0))
    else:
        ax.axis("off")
    _save(fig, path)
    return True


def plot_finetune_curves(run_dir: Path, path: Path) -> bool:
    """Training loss per logging step (metrics.csv) and val metrics per evaluation
    (epochs.csv). finetune.py appends an epochs.csv row every ``eval_every`` steps, not
    only at epoch ends, so both panels use the optimizer ``step`` as the x-axis (plotting
    against ``epoch`` stacks every mid-epoch evaluation on one vertical line); rows with
    ``is_epoch_end == 1`` get a hollow marker."""
    m = _read_csv(run_dir / "metrics.csv")
    e = _read_csv(run_dir / "epochs.csv")
    if m is None and e is None:
        return False
    fig, axes = plt.subplots(1, 2, figsize=(10, 4.2))
    ax = axes[0]
    if m is not None and "loss" in m:
        x = m["step"] if "step" in m else np.arange(len(m))
        ax.plot(x, m["loss"], color=C_BLUE, linewidth=1.4, label="train loss")
        for col, colr in (("val_loss", C_ORANGE),):
            if col in m and m[col].notna().any():
                v = m[m[col].notna()]
                ax.plot(v["step"] if "step" in v else v.index, v[col], color=colr, linewidth=2, marker="o", markersize=4, label=col)
        _style(ax, f"Fine-tuning loss ({run_dir.name})", "optimizer step", "cross-entropy")
        ax.legend(frameon=False, labelcolor=C_INK2)
    else:
        ax.axis("off")
    ax = axes[1]
    if e is not None:
        x = e["step"] if "step" in e else (e["epoch"] if "epoch" in e else np.arange(1, len(e) + 1))
        series = [c for c in ("f1", "recall", "precision", "accuracy", "roc_auc", "pr_auc", "macro_f1_3way") if c in e]
        series = series or [c for c in e.columns if c.startswith("val_") and c != "val_loss"][:6]
        colors = [C_BLUE, C_ORANGE, C_AQUA, "#eda100", "#e87ba4", "#008300", C_VIOLET]
        for col, colr in zip(series, colors):
            ax.plot(x, e[col], color=colr, linewidth=2, marker="o", markersize=4, label=col)
            ax.annotate(f"{float(e[col].iloc[-1]):.3f}", (float(x.iloc[-1]), float(e[col].iloc[-1])),
                        textcoords="offset points", xytext=(6, 0), color=C_INK2, fontsize=8)
        if "is_epoch_end" in e and (e["is_epoch_end"] == 1).any():
            ends = e[e["is_epoch_end"] == 1]
            for xe in (ends["step"] if "step" in ends else ends.index):
                ax.axvline(float(xe), color=C_GRID, linewidth=1, linestyle="--")
            ax.text(float((ends["step"] if "step" in ends else ends.index).iloc[0]), 0.03, "epoch end", color=C_INK2,
                    fontsize=8, rotation=90, va="bottom", ha="right", transform=ax.get_xaxis_transform())
        _style(ax, "Validation metrics per evaluation", "optimizer step", "score")
        if series:
            ax.legend(frameon=False, labelcolor=C_INK2, fontsize=8)
    else:
        ax.axis("off")
    _save(fig, path)
    return True


# --------------------------------------------------------------------- error analysis
_WORD = re.compile(r"[a-z][a-z0-9'\-]{2,}")
_HTML_TAG = re.compile(r"<\s*/?\s*[a-z!][^>]{0,80}>", re.I)
_LIST_BODY = re.compile(r"unsubscribe|mailing list|listserv|majordomo|listinfo|to be removed from this list|digest", re.I)
_LIST_SUBJ = re.compile(r"^\s*(re:\s*)?\[[^\]]{1,40}\]")


def preview(subject: str, body: str, n: int = PREVIEW_CHARS) -> str:
    """Subject and body on one line, whitespace collapsed, pipes escaped for the markdown table."""
    s = re.sub(r"\s+", " ", f"{subject.strip()} │ {body.strip()}").replace("|", "\\|")
    return (s[:n] + "...") if len(s) > n else s


def is_mailing_list(subject: str, body: str) -> bool:
    return bool(_LIST_SUBJ.search(subject) or _LIST_BODY.search(body))


def is_html_heavy(body: str) -> bool:
    return len(_HTML_TAG.findall(body)) >= 10 or bool(re.search(r"<\s*(html|table|td|tr)\b", body, re.I))


def over_represented(a_texts: Sequence[str], b_texts: Sequence[str], top: int = 20, alpha: float = 0.5) -> list[dict[str, Any]]:
    """Tokens with the highest document-frequency ratio in group A vs group B (smoothed).
    Simple lexical statistics, not a model explanation: it says what the missed emails
    tend to contain that caught ones do not."""
    if not a_texts or not b_texts:
        return []
    df_a: Counter = Counter()
    df_b: Counter = Counter()
    for t in a_texts:
        df_a.update(set(_WORD.findall(t.lower())))
    for t in b_texts:
        df_b.update(set(_WORD.findall(t.lower())))
    min_df = max(3, int(0.05 * len(a_texts)))
    rows = []
    for w, c in df_a.items():
        if c < min_df:
            continue
        ra, rb = (c + alpha) / (len(a_texts) + alpha), (df_b.get(w, 0) + alpha) / (len(b_texts) + alpha)
        rows.append({"token": w, "share_a": round(c / len(a_texts), 3), "share_b": round(df_b.get(w, 0) / len(b_texts), 3),
                     "ratio": round(ra / rb, 2)})
    rows.sort(key=lambda r: (-r["ratio"], -r["share_a"]))
    return rows[:top]


def error_analysis(split: Split, enc: Encoded, probs: np.ndarray, p: np.ndarray, thr: float, at: dict[str, Any], tf: dict[str, Any],
                   lure: dict[str, Any], conf: dict[str, Any], tw: ThreeWay, run_name: str, ts: str) -> tuple[str, dict[str, Any]]:
    y = split.label
    pred = (p >= thr).astype(np.int64)
    fn_idx = np.nonzero((y == 1) & (pred == 0))[0]
    fp_idx = np.nonzero((y == 0) & (pred == 1))[0]
    tp_idx = np.nonzero((y == 1) & (pred == 1))[0]
    tn_idx = np.nonzero((y == 0) & (pred == 0))[0]
    fn_top = fn_idx[np.argsort(p[fn_idx])][:TOP_K_ERRORS]          # most confidently "legitimate"
    fp_top = fp_idx[np.argsort(-p[fp_idx])][:TOP_K_ERRORS]         # most confidently "unwanted"

    def share(idx: np.ndarray, fn) -> float | None:
        return round(float(np.mean([fn(i) for i in idx])), 3) if len(idx) else None

    comp = {
        grp: {
            "n": int(len(idx)),
            "mailing_list_or_digest": share(idx, lambda i: is_mailing_list(split.subject[i], split.body[i])),
            "html_heavy": share(idx, lambda i: is_html_heavy(split.body[i])),
            "very_short_lt20_tokens": share(idx, lambda i: enc.n_tokens_full[i] - 2 < 20),
            "truncated_gt512": share(idx, lambda i: bool(enc.truncated[i])),
            "lure_phrase": share(idx, lambda i: bool(lure["mask"][i])),
        }
        for grp, idx in (("fp", fp_idx), ("tn", tn_idx), ("fn", fn_idx), ("tp", tp_idx))
    }
    fn_vs_tp = over_represented([split.text[i] for i in fn_idx], [split.text[i] for i in tp_idx])
    fp_vs_tn = over_represented([split.text[i] for i in fp_idx], [split.text[i] for i in tn_idx])

    cm = at["confusion_matrix"]
    k = probs.shape[1]
    L = [f"# Error analysis — {split.name} split ({SPLIT_TITLE[split.name]}), run `{run_name}` ({ts})", "",
         "All quoted emails are rows of public corpora "
         f"(source {', '.join(sorted(set(split.source)))}); nothing here is private mail. "
         f"Threshold {thr}. Binary positive class = {UNWANTED_NAME}, P(unwanted) = "
         + ("P(phishing) + P(spam)." if k == 3 else "P(class 1) of the 2-class head."), "",
         "## Counts (binary roll-up)", "",
         f"- rows {at['n']:,} (positives {at['n_positive']:,}, {100 * at['n_positive'] / max(at['n'], 1):.1f}%); "
         f"label3 counts {lure_free_counts(split)}",
         f"- **recall (unwanted) {at['recall']:.4f}** — {cm['fn']:,} missed unwanted emails (false negatives) out of {at['n_positive']:,}",
         f"- precision {at['precision']:.4f} — {cm['fp']:,} false alarms out of {cm['fp'] + cm['tp']:,} flagged",
         f"- F1 {at['f1']:.4f}, accuracy {at['accuracy']:.4f}, specificity {at['specificity']:.4f}",
         f"- ROC-AUC {_fmt(tf['roc_auc'])}, PR-AUC {_fmt(tf['pr_auc'])}; "
         f"recall at 1% FPR {_fmt(tf['recall_at_fpr_1pct'] and tf['recall_at_fpr_1pct']['recall'])}, "
         f"at 0.1% FPR {_fmt(tf['recall_at_fpr_0_1pct'] and tf['recall_at_fpr_0_1pct']['recall'])}",
         f"- credential-lure slice: {lure['metrics']['n']:,} rows, {lure['metrics']['n_positive']:,} positive "
         f"({100 * lure['label_purity']:.1f}% label purity: slice precision cannot exceed what the phrase list selects), "
         f"recall {_fmt(lure['metrics']['recall'])}, precision {_fmt(lure['metrics']['precision'])}, "
         f"{lure['metrics']['confusion_matrix']['fn']} missed", ""]

    def table(title: str, idx: np.ndarray) -> list[str]:
        rows = [f"## {title}", "", "| # | doc_id | P(unwanted) | tokens (full) | source | label_set | preview (subject │ body, 160 chars) |",
                "|---:|---|---:|---:|---|---|---|"]
        for kk, i in enumerate(idx, 1):
            rows.append(f"| {kk} | `{split.doc_id[i]}` | {p[i]:.4f} | {enc.n_tokens_full[i]} | {split.source[i]} | {split.label_set[i]} | "
                        f"{preview(split.subject[i], split.body[i])} |")
        if not len(idx):
            rows.append("| - | none | | | | | |")
        return rows + [""]

    L += table(f"Top-{TOP_K_ERRORS} false negatives (missed unwanted mail, most confident first)", fn_top)
    L += table(f"Top-{TOP_K_ERRORS} false positives (legitimate flagged, most confident first)", fp_top)

    # three-way section
    stats3: dict[str, Any] = {}
    L += ["## Three-way (legitimate / phishing / spam)", ""]
    if k != 3:
        L += [f"Skipped: {tw.metrics.get('reason')}.", ""]
    else:
        m3 = tw.metrics
        bd = m3["predicted_kind_breakdown"]
        if tw.evaluated:
            cmx = m3["confusion_matrix"]
            L += [f"Decision rule: {DECISION_RULE_DOC}", "",
                  f"Rows with a full label: {m3['rows_with_full_label']:,}; unwanted-but-unsplit rows (binary roll-up only): {m3['rows_unwanted_only']:,}.", "",
                  "| true \\ predicted | " + " | ".join(cmx["labels"]) + " |", "|---|" + "---:|" * 3]
            for i, lab in enumerate(cmx["labels"]):
                L.append(f"| **{lab}** | " + " | ".join(f"{v:,}" for v in cmx["matrix"][i]) + " |")
            L += ["", f"- accuracy {m3['accuracy']:.4f}, **macro-F1 {m3['macro_f1']:.4f}** (over 3 classes; over the "
                  f"{len(m3['classes_present'])} present: {m3['macro_f1_present_classes']:.4f}); argmax-of-3 rule: macro-F1 "
                  f"{m3['argmax_rule']['macro_f1']:.4f}, {m3['argmax_rule']['rows_where_rules_differ']:,} rows differ between the rules", "",
                  "| class | precision | recall | F1 | support |", "|---|---:|---:|---:|---:|"]
            for lab, v in m3["per_class"].items():
                L.append(f"| {lab} | {v['precision']:.4f} | {v['recall']:.4f} | {v['f1']:.4f} | {v['support']:,} |")
            pvs = m3["phish_vs_spam_on_true_unwanted"]
            if pvs["kind_only"]:
                ko, rr = pvs["kind_only"], pvs["rollup_rule"]
                L += ["", f"### Phishing vs spam on the {pvs['n_true_unwanted']:,} true unwanted rows", "",
                      f"Kind decision alone (given the row is unwanted, argmax(P(phishing), P(spam))): accuracy {ko['accuracy']:.4f}; "
                      f"phishing read as spam {ko['phishing_as_spam']:,}, spam read as phishing {ko['spam_as_phishing']:,}.", "",
                      "| true \\ predicted (roll-up rule) | Legitimate (missed) | Phishing | Spam |", "|---|---:|---:|---:|"]
                for i, lab in enumerate(rr["rows"]):
                    L.append(f"| **{lab}** | " + " | ".join(f"{v:,}" for v in rr["matrix"][i]) + " |")
            L += ["", "### Per source (three-way rows)", "", "| source | n | Legitimate | Phishing | Spam | accuracy | macro-F1 (present classes) |",
                  "|---|---:|---:|---:|---:|---:|---:|"]
            for s, v in m3["per_source"].items():
                c3 = v["counts"]
                L.append(f"| {s} | {v['n']:,} | {c3['Legitimate']:,} | {c3['Phishing']:,} | {c3['Spam']:,} | {v['accuracy']:.4f} | {v['macro_f1_present_classes']:.4f} |")
            # top confusions with previews
            y3, pred3 = split.label3, tw.pred3
            ps_idx = np.nonzero((y3 == 1) & (pred3 == 2))[0]
            sp_idx = np.nonzero((y3 == 2) & (pred3 == 1))[0]
            ps_top = ps_idx[np.argsort(-probs[ps_idx, 2])][:TOP_K_CONFUSIONS]
            sp_top = sp_idx[np.argsort(-probs[sp_idx, 1])][:TOP_K_CONFUSIONS]

            def table3(title: str, idx: np.ndarray) -> list[str]:
                rows = [f"### {title}", "", "| # | doc_id | P(legit) | P(phish) | P(spam) | tokens (full) | source | preview (subject │ body, 160 chars) |",
                        "|---:|---|---:|---:|---:|---:|---|---|"]
                for kk, i in enumerate(idx, 1):
                    rows.append(f"| {kk} | `{split.doc_id[i]}` | {probs[i, 0]:.4f} | {probs[i, 1]:.4f} | {probs[i, 2]:.4f} | {enc.n_tokens_full[i]} | "
                                f"{split.source[i]} | {preview(split.subject[i], split.body[i])} |")
                if not len(idx):
                    rows.append("| - | none | | | | | | |")
                return rows + [""]

            L += [""]
            L += table3(f"Top-{TOP_K_CONFUSIONS} phishing -> spam confusions ({len(ps_idx):,} total, most confident first)", ps_top)
            L += table3(f"Top-{TOP_K_CONFUSIONS} spam -> phishing confusions ({len(sp_idx):,} total, most confident first)", sp_top)
            ps_vs_pp = over_represented([split.text[i] for i in ps_idx], [split.text[i] for i in np.nonzero((y3 == 1) & (pred3 == 1))[0]])
            sp_vs_ss = over_represented([split.text[i] for i in sp_idx], [split.text[i] for i in np.nonzero((y3 == 2) & (pred3 == 2))[0]])
            L += ["### Tokens over-represented in phishing read as spam vs phishing read correctly", ""] + _ratio_table(ps_vs_pp, "phish->spam", "phish->phish")
            L += ["", "### Tokens over-represented in spam read as phishing vs spam read correctly", ""] + _ratio_table(sp_vs_ss, "spam->phish", "spam->spam") + [""]
            stats3 = {"n_phish_as_spam": int(len(ps_idx)), "n_spam_as_phish": int(len(sp_idx)),
                      "top_phish_as_spam_doc_ids": [split.doc_id[i] for i in ps_top], "top_spam_as_phish_doc_ids": [split.doc_id[i] for i in sp_top],
                      "phish_as_spam_tokens": ps_vs_pp, "spam_as_phish_tokens": sp_vs_ss}
        else:
            L += [f"No three-way ground truth on this split: {m3['reason']}.", ""]
        tpf, fpf, mis = bd["true_positives_flagged"], bd["false_positives_flagged"], bd["missed_positives"]
        L += ["### Predicted kind of the flagged rows (informational — "
              + ("no phishing-vs-spam ground truth on this split)" if not tw.evaluated else "includes the unsplit rows, which have no kind label)"), "",
              "| group | n | predicted Phishing | predicted Spam |", "|---|---:|---:|---:|",
              f"| true positives flagged | {tpf['n']:,} | {tpf['predicted_phishing']:,} | {tpf['predicted_spam']:,} |",
              f"| false positives flagged | {fpf['n']:,} | {fpf['predicted_phishing']:,} | {fpf['predicted_spam']:,} |",
              f"| missed positives (kind the model would have given) | {mis['n']:,} | {mis['predicted_phishing']:,} | {mis['predicted_spam']:,} |", ""]

    L += ["## Patterns (simple lexical statistics, not a model explanation)", "",
          "### Tokens over-represented in false negatives vs true positives (document frequency, smoothed ratio)", ""]
    L += _ratio_table(fn_vs_tp, "FN", "TP")
    L += ["", "### Tokens over-represented in false positives vs true negatives", ""]
    L += _ratio_table(fp_vs_tn, "FP", "TN")
    L += ["", "### Composition of the error groups (share of rows)", "",
          "| group | n | mailing-list / digest | HTML-heavy | very short (<20 content tokens) | truncated (>512) | lure phrase |",
          "|---|---:|---:|---:|---:|---:|---:|"]
    for grp in ("fp", "tn", "fn", "tp"):
        c = comp[grp]
        L.append(f"| {grp.upper()} | {c['n']:,} | {_pct(c['mailing_list_or_digest'])} | {_pct(c['html_heavy'])} | "
                 f"{_pct(c['very_short_lt20_tokens'])} | {_pct(c['truncated_gt512'])} | {_pct(c['lure_phrase'])} |")
    L += ["", "### Length confound (RUN_LOG entry 2)", "",
          "| group | n | mean tokens (full) | mean tokens fed | median (full) | % truncated |", "|---|---:|---:|---:|---:|---:|"]
    for grp in ("tp", "tn", "fp", "fn", "correct", "wrong"):
        s = conf["by_outcome"][grp]
        L.append(f"| {grp.upper()} | {s['n']:,} | {_fmt(s['mean_tokens_full'], 1)} | {_fmt(s['mean_tokens_fed'], 1)} | "
                 f"{_fmt(s['median_tokens_full'], 0)} | {_fmt(s['pct_truncated'], 1)} |")
    for lab in ("legitimate", "unwanted"):
        s = conf["by_label"][lab]
        L.append(f"| all {lab} | {s['n']:,} | {_fmt(s['mean_tokens_full'], 1)} | {_fmt(s['mean_tokens_fed'], 1)} | "
                 f"{_fmt(s['median_tokens_full'], 0)} | {_fmt(s['pct_truncated'], 1)} |")
    for lab, s in conf["by_class3"].items():
        if lab in ("phish", "spam", "unwanted"):
            L.append(f"| all label3={lab} | {s['n']:,} | {_fmt(s['mean_tokens_full'], 1)} | {_fmt(s['mean_tokens_fed'], 1)} | "
                     f"{_fmt(s['median_tokens_full'], 0)} | {_fmt(s['pct_truncated'], 1)} |")
    L += ["", "| length bucket (full tokens) | n | positive rate | recall | FPR | accuracy |", "|---|---:|---:|---:|---:|---:|"]
    for b in conf["by_length_bucket"]:
        L.append(f"| {b['bucket']} | {b['n']:,} | {b['positive_rate']:.3f} | {_fmt(b['recall'])} | {_fmt(b['fpr'])} | {b['accuracy']:.4f} |")

    L += ["", "## Summary", ""] + summary_sentences(at, tf, lure, conf, comp, fn_vs_tp, fp_vs_tn, tw)
    stats = {"n_fn": int(len(fn_idx)), "n_fp": int(len(fp_idx)), "composition": comp,
             "fn_vs_tp_tokens": fn_vs_tp, "fp_vs_tn_tokens": fp_vs_tn,
             "top_fn_doc_ids": [split.doc_id[i] for i in fn_top], "top_fp_doc_ids": [split.doc_id[i] for i in fp_top],
             "threeway": stats3 or None}
    return "\n".join(L) + "\n", stats


def lure_free_counts(split: Split) -> str:
    c = _label3_counts(split.label3)
    return ", ".join(f"{k} {v:,}" for k, v in c.items() if v)


def summary_sentences(at, tf, lure, conf, comp, fn_vs_tp, fp_vs_tn, tw: ThreeWay | None = None) -> list[str]:
    cm = at["confusion_matrix"]
    out = [f"- The classifier catches **{100 * at['recall']:.2f}% of unwanted mail (phishing or spam)** at threshold {at['threshold']} "
           f"({cm['fn']:,} of {at['n_positive']:,} missed) while flagging {100 * at['false_positive_rate']:.2f}% of legitimate mail "
           f"({cm['fp']:,} false alarms). Recall is the headline because a missed phish is the costly error for a small-business inbox."]
    if tf["recall_at_fpr_1pct"]:
        r1 = tf["recall_at_fpr_1pct"]
        thr_txt = f"threshold {r1['threshold']:.3f}, flag when p >= threshold" if r1["threshold"] is not None else "no finite threshold reaches this budget"
        out.append(f"- Holding false alarms to 1 in 100 legitimate emails keeps recall at {100 * r1['recall']:.2f}% "
                   f"({thr_txt}); at 1 in 1,000 it is "
                   f"{100 * tf['recall_at_fpr_0_1pct']['recall']:.2f}%. The best-F1 threshold is {tf['best_f1']['threshold']:.3f} "
                   f"(F1 {tf['best_f1']['f1']:.4f}).")
    if tw is not None and tw.evaluated:
        m3 = tw.metrics
        pc = m3["per_class"]
        pvs = m3["phish_vs_spam_on_true_unwanted"]
        s = (f"- Three-way: macro-F1 **{m3['macro_f1']:.4f}** over {m3['rows_with_full_label']:,} rows with a full label "
             f"(recall legitimate {100 * pc['Legitimate']['recall']:.1f}%, phishing {100 * pc['Phishing']['recall']:.1f}%, spam {100 * pc['Spam']['recall']:.1f}%).")
        if pvs["kind_only"]:
            ko = pvs["kind_only"]
            s += (f" Given that an email is unwanted, the model tells phishing from spam {100 * ko['accuracy']:.1f}% of the time "
                  f"({ko['phishing_as_spam']:,} phishing read as spam, {ko['spam_as_phishing']:,} spam read as phishing). "
                  "The phishing/spam labels follow corpus provenance (Nazario + Nigerian fraud + synthetic lures = phishing; "
                  "SpamAssassin + Enron-Spam spam folders = spam), so part of this confusion is corpus style, not lure content.")
        out.append(s)
    elif tw is not None and tw.metrics.get("num_labels") == 3:
        bd = tw.metrics["predicted_kind_breakdown"]["true_positives_flagged"]
        out.append(f"- Three-way: this split has no phishing-vs-spam ground truth; of the {bd['n']:,} true positives the model flagged, "
                   f"it called {bd['predicted_phishing']:,} phishing and {bd['predicted_spam']:,} spam (informational only).")
    lm = lure["metrics"]
    if lm["n_positive"] >= LURE_MIN_POSITIVES:
        out.append(f"- On the {lm['n']:,}-row credential-lure slice ({len(CREDENTIAL_LURE_PHRASES)} documented phrases, "
                   f"{100 * lure['label_purity']:.1f}% of its rows are labelled positive) recall is "
                   f"{100 * lm['recall']:.2f}% with precision {100 * lm['precision']:.2f}%: "
                   f"{'the highest-cost phish are caught at least as well as the average' if lm['recall'] >= at['recall'] else 'the highest-cost phish are caught LESS often than the average positive, which is the first thing to fix'}.")
    elif lm["n"]:
        out.append(f"- The credential-lure slice has only {lm['n']:,} rows ({lm['n_positive']:,} positive, {lm['confusion_matrix']['fn']} missed): "
                   f"too few rows to compare its recall with the overall recall (needs >= {LURE_MIN_POSITIVES} positives).")
    else:
        out.append("- No row of this sample contains a credential-lure phrase; the slice is empty.")
    bo, bl = conf["by_outcome"], conf["by_label"]
    mean = {g: bo[g]["mean_tokens_full"] for g in ("tp", "tn", "fp", "fn")}
    out.append(f"- Length confound: missed unwanted mail averages {_fmt(mean['fn'], 0)} tokens vs {_fmt(mean['tp'], 0)} for caught; "
               f"false positives average {_fmt(mean['fp'], 0)} tokens vs {_fmt(mean['tn'], 0)} for correct legitimate mail "
               f"(all legitimate {_fmt(bl['legitimate']['mean_tokens_full'], 0)}, all unwanted {_fmt(bl['unwanted']['mean_tokens_full'], 0)}).")
    if all(mean[g] is not None for g in mean):
        follows_length = mean["fn"] > mean["tp"] and mean["fp"] < mean["tn"]
        out.append(f"  Legitimate mail is ~{bl['legitimate']['mean_tokens_full'] / max(bl['unwanted']['mean_tokens_full'] or 1, 1):.1f}x longer than unwanted mail on average; "
                   + ("the errors sit on long unwanted mail and short legitimate mail, so the model partly reads length rather than content."
                      if follows_length else "the error groups do not follow the length gap, so length is not the main shortcut."))
    c3 = conf.get("by_class3") or {}
    if "phish" in c3 and "spam" in c3:
        out.append(f"  Phishing rows average {_fmt(c3['phish']['mean_tokens_full'], 0)} tokens vs spam {_fmt(c3['spam']['mean_tokens_full'], 0)}: "
                   "a length gap between the two unwanted kinds is a second shortcut to watch in the phishing-vs-spam confusion.")
    buckets = [b for b in conf["by_length_bucket"] if b["recall"] is not None and b["n_positive"] >= 20]
    if buckets:
        worst = min(buckets, key=lambda b: b["recall"])
        out.append(f"- Recall by length is lowest in the {worst['bucket']}-token bucket ({100 * worst['recall']:.1f}%, {worst['n_positive']:,} positives).")
    fp_c = comp["fp"]
    if fp_c["n"]:
        out.append(f"- Of the {fp_c['n']:,} false positives, {_pct(fp_c['mailing_list_or_digest'])} look like mailing-list or digest traffic, "
                   f"{_pct(fp_c['html_heavy'])} are HTML-heavy and {_pct(fp_c['very_short_lt20_tokens'])} are very short "
                   f"(TN reference: {_pct(comp['tn']['mailing_list_or_digest'])} / {_pct(comp['tn']['html_heavy'])} / {_pct(comp['tn']['very_short_lt20_tokens'])}).")
    if fn_vs_tp:
        out.append("- Tokens most over-represented in missed unwanted mail vs caught: " + ", ".join(f"`{r['token']}`" for r in fn_vs_tp[:10]) + ".")
    if fp_vs_tn:
        out.append("- Tokens most over-represented in false positives vs correct legitimate mail: " + ", ".join(f"`{r['token']}`" for r in fp_vs_tn[:10]) + ".")
    return out


def _ratio_table(rows: list[dict[str, Any]], a: str, b: str) -> list[str]:
    if not rows:
        return ["(too few rows in one of the groups)"]
    out = [f"| token | share of {a} docs | share of {b} docs | ratio |", "|---|---:|---:|---:|"]
    out += [f"| `{r['token']}` | {r['share_a']:.3f} | {r['share_b']:.3f} | {r['ratio']:.2f} |" for r in rows]
    return out


def _fmt(v: Any, nd: int = 4) -> str:
    if v is None or v is False:
        return "n/a"
    return f"{v:.{nd}f}"


def _pct(v: float | None) -> str:
    return "n/a" if v is None else f"{100 * v:.1f}%"


# ------------------------------------------------------------------------ run log
def next_run_log_number(path: Path) -> int:
    if not path.exists():
        return 0
    nums = [int(m.group(1)) for m in re.finditer(r"^## (\d+)\.", path.read_text(encoding="utf-8"), re.M)]
    return (max(nums) + 1) if nums else 0


@dataclass
class SplitResult:
    """Everything one split produced, kept until the invocation-level artifacts are written."""

    split: Split
    enc: Encoded
    probs: np.ndarray
    p_unw: np.ndarray
    inf: dict[str, Any]
    at: dict[str, Any]
    tf: dict[str, Any]
    lure: dict[str, Any]
    per_source: dict[str, Any]
    conf: dict[str, Any]
    tw: ThreeWay
    plots: dict[str, str]
    err_path: Path
    err_stats: dict[str, Any]


def _split_table(r: SplitResult) -> list[str]:
    at, tf, lure, tw = r.at, r.tf, r.lure, r.tw
    cm = at["confusion_matrix"]
    lines = [f"### {r.split.name} split ({SPLIT_TITLE[r.split.name]}): `{r.split.path.name}`, {r.split.n:,} rows "
             f"({at['n_positive']:,} unwanted, {100 * at['n_positive'] / max(r.split.n, 1):.1f}%; label3 {lure_free_counts(r.split)}). "
             f"Inference {r.inf['rows_per_s']} rows/s on {r.inf['device']} (bf16 autocast, batch {r.inf['batch_size']}, length-bucketed), "
             f"peak VRAM {r.inf['peak_vram_gb']} GB.", "",
             "| binary roll-up metric (P(unwanted) = P(phishing) + P(spam)) | value |", "|---|---:|",
             f"| **recall (unwanted)** | **{at['recall']:.4f}** ({cm['fn']:,} missed of {at['n_positive']:,}) |",
             f"| precision | {at['precision']:.4f} ({cm['fp']:,} false alarms) |",
             f"| F1 | {at['f1']:.4f} |", f"| accuracy | {at['accuracy']:.4f} |", f"| specificity | {at['specificity']:.4f} |",
             f"| ROC-AUC | {_fmt(tf['roc_auc'])} |", f"| PR-AUC | {_fmt(tf['pr_auc'])} |",
             f"| recall @ 1% FPR | {_fmt(tf['recall_at_fpr_1pct'] and tf['recall_at_fpr_1pct']['recall'])} |",
             f"| recall @ 0.1% FPR | {_fmt(tf['recall_at_fpr_0_1pct'] and tf['recall_at_fpr_0_1pct']['recall'])} |",
             f"| best-F1 threshold | {_fmt(tf['best_f1'] and tf['best_f1']['threshold'], 3)} (F1 {_fmt(tf['best_f1'] and tf['best_f1']['f1'])}) |",
             f"| credential-lure slice | {lure['metrics']['n']:,} rows ({100 * lure['label_purity']:.1f}% labelled positive), "
             f"recall {_fmt(lure['metrics']['recall'])}, precision {_fmt(lure['metrics']['precision'])} |", ""]
    if tw.evaluated:
        m3 = tw.metrics
        cmx = m3["confusion_matrix"]
        pc = m3["per_class"]
        lines += [f"**Three-way** on the {m3['rows_with_full_label']:,} rows with label3 in {{0,1,2}} ({m3['rows_unwanted_only']:,} unsplit rows "
                  f"enter the roll-up only; rule: {DECISION_RULE}): accuracy {m3['accuracy']:.4f}, **macro-F1 {m3['macro_f1']:.4f}**; "
                  + "; ".join(f"{lab} P {v['precision']:.3f} / R {v['recall']:.3f} / F1 {v['f1']:.3f} (n={v['support']:,})" for lab, v in pc.items())
                  + ". 3x3 (rows true, cols predicted Legitimate/Phishing/Spam): "
                  + "; ".join(f"{lab} {'/'.join(f'{v:,}' for v in cmx['matrix'][i])}" for i, lab in enumerate(cmx["labels"])) + "."]
        pvs = m3["phish_vs_spam_on_true_unwanted"]
        if pvs["kind_only"]:
            ko = pvs["kind_only"]
            lines[-1] += (f" Phishing-vs-spam on the {pvs['n_true_unwanted']:,} true unwanted rows: kind accuracy {ko['accuracy']:.4f} "
                          f"({ko['phishing_as_spam']:,} phishing->spam, {ko['spam_as_phishing']:,} spam->phishing).")
        lines += ["", "| source | n | Legitimate | Phishing | Spam | accuracy | macro-F1 (present classes) |", "|---|---:|---:|---:|---:|---:|---:|"]
        for s, v in m3["per_source"].items():
            c3 = v["counts"]
            lines.append(f"| {s} | {v['n']:,} | {c3['Legitimate']:,} | {c3['Phishing']:,} | {c3['Spam']:,} | {v['accuracy']:.4f} | {v['macro_f1_present_classes']:.4f} |")
        lines.append("")
    elif tw.metrics.get("num_labels") == 3:
        bd = tw.metrics["predicted_kind_breakdown"]["true_positives_flagged"]
        lines += [f"**Three-way:** no phishing-vs-spam ground truth on this split ({tw.metrics['reason']}); informational: of the "
                  f"{bd['n']:,} flagged true positives the model called {bd['predicted_phishing']:,} phishing and {bd['predicted_spam']:,} spam.", ""]
    else:
        lines += [f"**Three-way:** skipped ({tw.metrics.get('reason')}).", ""]
    return lines


def run_log_entry(num: int, ts: str, run_name: str, results: list[SplitResult], ckpt: Path, lat: dict, out_dir: Path, dev: bool, k: int) -> str:
    g, c = lat.get("gpu_bf16"), lat.get("cpu_fp32")
    names = " + ".join(r.split.name for r in results)
    lines = [f"## {num}. {ts} — Evaluation on the {names} split{'s' if len(results) > 1 else ''} (evaluate.py run, `{run_name}`)"
             f"{' [DEV RUN, --limit-rows]' if dev else ''}", "",
             f"**Checkpoint.** `{ckpt}` ({k}-class head{'; three-way label contract of RUN_LOG entry 6' if k == 3 else ''}). "
             "Binary roll-up metrics use the same definitions as the binary-era runs so they stay comparable.", ""]
    for r in results:
        lines += _split_table(r)
    lines += ["**Latency (batch 1, tokenize + forward + softmax, timeit.default_timer) vs the instructor's 74 ms.** "
              + (f"GPU bf16: {g['avg_ms']} ms avg / {g['p50_ms']} ms p50 over {g['n']} emails ({g['ratio_to_benchmark']:.2f}x). " if g else "GPU: not available. ")
              + (f"CPU fp32 ({c['threads']} threads): {c['avg_ms']} ms avg / {c['p50_ms']} ms p50 over {c['n']} emails ({c['ratio_to_benchmark']:.2f}x)." if c else ""),
              ""]
    r0 = results[0]
    conf = r0.conf
    lines += [f"**Length confound ({r0.split.name}).** FN mean {_fmt(conf['by_outcome']['fn']['mean_tokens_full'], 0)} tokens vs TP {_fmt(conf['by_outcome']['tp']['mean_tokens_full'], 0)}; "
              f"FP {_fmt(conf['by_outcome']['fp']['mean_tokens_full'], 0)} vs TN {_fmt(conf['by_outcome']['tn']['mean_tokens_full'], 0)} "
              f"(see error_analysis{r0.split.sfx}.md for the per-bucket table).", "",
              f"**Artifacts.** `{out_dir}`: " + "; ".join(
                  f"metrics{r.split.sfx}.json, latency{r.split.sfx}.json, confusion_matrix{r.split.sfx}.png"
                  + (f", confusion_matrix3{r.split.sfx}.png" if r.tw.evaluated else "")
                  + f", roc_curve{r.split.sfx}.png, pr_curve{r.split.sfx}.png, error_analysis{r.split.sfx}.md" for r in results)
              + " (training curves: reports/pretrain_curves.png and finetune_curves.png from pretrain.py / finetune.py `plot`)."]
    return "\n".join(lines)


# -------------------------------------------------------------------------------- main
def _summary(r: SplitResult) -> dict[str, Any]:
    """The experiments.jsonl summary for one split: the binary keys of the binary-era line
    plus a ``threeway`` sub-dict (None when not evaluated)."""
    at, tf, lure, tw = r.at, r.tf, r.lure, r.tw
    s = {"accuracy": at["accuracy"], "precision": at["precision"], "recall": at["recall"], "f1": at["f1"],
         "specificity": at["specificity"], "roc_auc": tf["roc_auc"], "pr_auc": tf["pr_auc"],
         "recall_at_fpr_1pct": tf["recall_at_fpr_1pct"] and tf["recall_at_fpr_1pct"]["recall"],
         "recall_at_fpr_0_1pct": tf["recall_at_fpr_0_1pct"] and tf["recall_at_fpr_0_1pct"]["recall"],
         "confusion_matrix": at["confusion_matrix"], "n": r.split.n, "threshold": at["threshold"],
         "lure_recall": lure["metrics"]["recall"] if lure["metrics"]["n_positive"] else None,
         "f1_unwanted": at["f1"], "threeway": None}
    if tw.evaluated:
        m3 = tw.metrics
        pvs = m3["phish_vs_spam_on_true_unwanted"]
        s["threeway"] = {"n": m3["rows_with_full_label"], "accuracy": m3["accuracy"], "macro_f1": m3["macro_f1"],
                         "macro_f1_present_classes": m3["macro_f1_present_classes"],
                         "per_class": {lab: {"precision": v["precision"], "recall": v["recall"], "f1": v["f1"], "support": v["support"]}
                                       for lab, v in m3["per_class"].items()},
                         "confusion_matrix_3x3": m3["confusion_matrix"]["matrix"],
                         "kind_accuracy_on_true_unwanted": pvs["kind_only"]["accuracy"] if pvs["kind_only"] else None}
        s["macro_f1_3way"] = m3["macro_f1"]
    return s


def evaluate_split(name: str, model, mcfg: EncoderConfig, tok, cfg: dict, args: argparse.Namespace, device: torch.device,
                   max_len: int, out_dir: Path, run_name: str, ts: str, data_dir: Path) -> SplitResult:
    dev = args.limit_rows is not None
    test3_file = str((cfg.get("eval") or {}).get("test3_file") or SPLIT_FILES["test3"])
    split = load_split(data_dir, name, args.limit_rows, int(cfg["seed"]), test3_file)
    LOG.info("%s split: %d rows, %d unwanted (%.1f%%), label3 %s, sources %s%s", split.name, split.n, int(split.label.sum()),
             100 * split.label.mean(), _label3_counts(split.label3), sorted(set(split.source)), f" [limit_rows={args.limit_rows}]" if dev else "")
    t0 = common.timer()
    enc = encode_split(tok, split.text, max_len)
    LOG.info("encoded %d rows in %.1fs: mean %.0f tokens (full), %.1f%% truncated at %d",
             split.n, common.timer() - t0, enc.n_tokens_full.mean(), 100 * enc.truncated.mean(), max_len)

    probs, inf = predict_probs(model, enc, device, args.batch_size, int(cfg["finetune"].get("dynamic_padding_multiple", 8)))
    p_unw = p_unwanted(probs)
    thr = args.threshold
    at = threshold_metrics(split.label, p_unw, thr)
    tf = threshold_free_metrics(split.label, p_unw)
    lmask, per_phrase, phrase_by_label = lure_mask(split.text, split.label)
    lure = {"mask": lmask, "phrases": list(CREDENTIAL_LURE_PHRASES), "phrase_hits": dict(per_phrase.most_common()),
            "phrase_hits_by_label": phrase_by_label,
            "metrics": slice_metrics(split.label[lmask], p_unw[lmask], thr),
            "label3_counts": _label3_counts(split.label3[lmask])}
    # Share of slice rows labelled positive: the slice is defined by phrases, not labels, so its
    # "precision" is bounded by this purity and must be read next to it.
    lure["label_purity"] = _safe_div(lure["metrics"]["n_positive"], lure["metrics"]["n"])
    per_source = per_source_table(split, p_unw, thr)
    conf = length_confound(split, enc, p_unw, thr)
    tw = threeway_metrics(split, probs, p_unw, thr)
    if tw.evaluated and lmask.any():
        full = lmask & (split.label3 >= 0)
        lure["threeway_accuracy"] = float((tw.pred3[full] == split.label3[full]).mean()) if full.any() else None
    if not tw.evaluated:
        LOG.info("three-way sections skipped on %s: %s", split.name, tw.metrics.get("reason"))

    cm = at["confusion_matrix"]
    print(f"\n=== {split.name} split / {SPLIT_TITLE[split.name]} ({split.n:,} rows, {at['n_positive']:,} unwanted) — run {run_name} ===")
    print(f"RECALL (unwanted = phishing or spam, the metric that matters): {at['recall']:.4f}   missed: {cm['fn']:,} / {at['n_positive']:,}")
    print(f"precision {at['precision']:.4f}  F1 {at['f1']:.4f}  accuracy {at['accuracy']:.4f}  specificity {at['specificity']:.4f}")
    print(f"confusion matrix [[TN {cm['tn']:,}, FP {cm['fp']:,}], [FN {cm['fn']:,}, TP {cm['tp']:,}]]")
    print(f"ROC-AUC {_fmt(tf['roc_auc'])}  PR-AUC {_fmt(tf['pr_auc'])}  "
          f"recall@1%FPR {_fmt(tf['recall_at_fpr_1pct'] and tf['recall_at_fpr_1pct']['recall'])}  "
          f"recall@0.1%FPR {_fmt(tf['recall_at_fpr_0_1pct'] and tf['recall_at_fpr_0_1pct']['recall'])}  "
          f"best-F1 thr {_fmt(tf['best_f1'] and tf['best_f1']['threshold'], 3)}")
    lm = lure["metrics"]
    print(f"credential-lure slice: {lm['n']:,} rows ({lm['n_positive']:,} positive = {100 * lure['label_purity']:.1f}% label purity)  "
          f"recall {_fmt(lm['recall'])}  precision {_fmt(lm['precision'])}"
          + (f"  [< {LURE_MIN_POSITIVES} positives: not compared to overall recall]" if lm['n_positive'] < LURE_MIN_POSITIVES else ""))
    print("per source: " + "; ".join(f"{s}: n={m['n']:,} recall={_fmt(m['recall'])} prec={_fmt(m['precision'])}" for s, m in per_source.items()))
    bo = conf["by_outcome"]
    print(f"mean tokens (full): TP {_fmt(bo['tp']['mean_tokens_full'], 0)}  FN {_fmt(bo['fn']['mean_tokens_full'], 0)}  "
          f"TN {_fmt(bo['tn']['mean_tokens_full'], 0)}  FP {_fmt(bo['fp']['mean_tokens_full'], 0)}")
    if tw.evaluated:
        m3 = tw.metrics
        print(f"THREE-WAY ({m3['rows_with_full_label']:,} rows with label3 in {{0,1,2}}): accuracy {m3['accuracy']:.4f}  MACRO-F1 {m3['macro_f1']:.4f}  "
              + "  ".join(f"{lab[:5]} P/R/F1 {v['precision']:.3f}/{v['recall']:.3f}/{v['f1']:.3f} (n={v['support']:,})" for lab, v in m3["per_class"].items()))
        print("3x3 [rows true L/P/S, cols pred L/P/S]: " + str(m3["confusion_matrix"]["matrix"]))
        pvs = m3["phish_vs_spam_on_true_unwanted"]
        if pvs["kind_only"]:
            ko = pvs["kind_only"]
            print(f"phishing vs spam on {pvs['n_true_unwanted']:,} true unwanted rows: kind accuracy {ko['accuracy']:.4f}  "
                  f"phish->spam {ko['phishing_as_spam']:,}  spam->phish {ko['spam_as_phishing']:,}")
        print("per source (3-way): " + "; ".join(f"{s}: n={v['n']:,} L/P/S={v['counts']['Legitimate']}/{v['counts']['Phishing']}/{v['counts']['Spam']} "
                                                  f"acc={v['accuracy']:.3f}" for s, v in m3["per_source"].items()))
    elif mcfg.num_labels == 3:
        bd = tw.metrics["predicted_kind_breakdown"]["true_positives_flagged"]
        print(f"three-way: no ground truth on this split ({tw.metrics['reason']}); flagged true positives split "
              f"{bd['predicted_phishing']:,} phishing / {bd['predicted_spam']:,} spam (informational)")
    else:
        print(f"three-way: skipped ({tw.metrics.get('reason')})")

    sfx = split.sfx
    plots: dict[str, str] = {}
    p_cm = out_dir / f"confusion_matrix{sfx}.png"
    plot_confusion(cm, p_cm, f"Confusion matrix (binary roll-up) — {split.name} split, threshold {thr}")
    plots["confusion_matrix"] = str(p_cm)
    if tw.evaluated:
        p_cm3 = out_dir / f"confusion_matrix3{sfx}.png"
        plot_confusion3(tw.metrics["confusion_matrix"]["matrix"], p_cm3, f"Three-way confusion matrix — {split.name} split, threshold {thr}")
        plots["confusion_matrix3"] = str(p_cm3)
    if tf["curves"] is not None:
        p_roc, p_pr = out_dir / f"roc_curve{sfx}.png", out_dir / f"pr_curve{sfx}.png"
        plot_roc(tf, p_roc, f"ROC (binary roll-up) — {split.name} split")
        plot_pr(tf, split.label.mean(), p_pr, f"Precision-recall (binary roll-up) — {split.name} split")
        plots["roc_curve"], plots["pr_curve"] = str(p_roc), str(p_pr)
    # Training curves are pretrain.py / finetune.py `plot` outputs (reports/pretrain_curves.png,
    # reports/finetune_curves.png) and are never regenerated here: evaluate.py cannot know which
    # pretraining run a scratch-initialised fine-tune "belongs" to, and a wrong guess would silently
    # overwrite the authoritative plot. Explicit run dirs opt into extra eval_* renderings.
    for flag, plotter, key in ((args.finetune_run_dir, plot_finetune_curves, "finetune"), (args.pretrain_run_dir, plot_pretrain_curves, "pretrain")):
        if not flag:
            continue
        p_curve = out_dir / f"eval_{key}_curves{sfx}.png"
        if plotter(Path(flag), p_curve):
            plots[f"eval_{key}_curves"] = str(p_curve)
        else:
            LOG.warning("no usable metrics.csv under %s: %s skipped", flag, p_curve.name)

    md, err_stats = error_analysis(split, enc, probs, p_unw, thr, at, tf, lure, conf, tw, run_name, ts)
    p_err = out_dir / f"error_analysis{sfx}.md"
    p_err.write_text(md, encoding="utf-8")
    LOG.info("wrote %s", p_err)
    return SplitResult(split, enc, probs, p_unw, inf, at, tf, lure, per_source, conf, tw, plots, p_err, err_stats)


def run(args: argparse.Namespace) -> int:
    cfg = common.load_config(args.config)
    global LOG, TOP_K_CONFUSIONS
    LOG = common.setup_logging("evaluate", cfg["paths"]["logs"])
    TOP_K_CONFUSIONS = check_eval_config(cfg)
    common.set_seed(int(cfg["seed"]))
    common.print_device_info()
    device = torch.device("cuda" if torch.cuda.is_available() and not args.cpu else "cpu")
    ts = datetime.now().strftime("%Y-%m-%d %H:%M")
    dev = args.limit_rows is not None
    splits = [s.strip() for s in args.split.split(",") if s.strip()]
    bad = [s for s in splits if s not in SPLIT_FILES]
    if bad or not splits:
        raise ValueError(f"--split must be a comma list of {sorted(SPLIT_FILES)}, got {args.split!r}")
    if len(set(splits)) != len(splits):
        raise ValueError(f"--split lists a split twice: {args.split!r}")

    ckpt = Path(args.checkpoint).resolve()
    run_name = args.run_name or (ckpt.parent.name if ckpt.name in ("best", "last") else ckpt.name)
    out_dir = Path(args.out_dir) if args.out_dir else Path(cfg["paths"]["reports"])
    out_dir.mkdir(parents=True, exist_ok=True)
    data_dir = Path(args.data_dir) if args.data_dir else Path(cfg["paths"]["data_processed"])
    LOG.info("run_name=%s splits=%s checkpoint=%s out_dir=%s data_dir=%s device=%s", run_name, splits, ckpt, out_dir, data_dir, device)

    model, mcfg, ckpt_cfg = load_classifier(ckpt, device)
    k = int(mcfg.num_labels)
    max_len = int(args.max_len or mcfg.max_seq_len)
    tok = load_tokenizer(cfg["paths"]["tokenizer"])
    if tok.get_vocab_size() > mcfg.vocab_size:
        raise RuntimeError(f"tokenizer vocab {tok.get_vocab_size()} > model vocab {mcfg.vocab_size}")
    if k == 2:
        LOG.info("2-class checkpoint: binary roll-up only; every three-way section is skipped")
    else:
        LOG.info("3-class checkpoint: labels %s; binary roll-up P(unwanted) = P(phishing) + P(spam); rule %s", LABEL_NAMES, DECISION_RULE)

    # Inference + per-split artifacts first (the model stays on `device`); latency last (it
    # moves the model to the CPU), measured once per invocation on the first split's sample.
    results = [evaluate_split(name, model, mcfg, tok, cfg, args, device, max_len, out_dir, run_name, ts, data_dir) for name in splits]
    r0 = results[0]
    lat = measure_latency(model, tok, r0.split, r0.enc, device, max_len, args.latency_samples, args.cpu_threads, int(cfg["seed"]))
    lat.update({"checkpoint": str(ckpt), "run_name": run_name, "params": count_parameters(model)["total"], "timestamp": ts,
                "num_labels": k, "measured_once_for_splits": splits})
    g, c = lat.get("gpu_bf16"), lat["cpu_fp32"]
    print(f"\nlatency per email (batch 1, tokenize+forward+softmax) vs instructor benchmark {INSTRUCTOR_BENCHMARK_MS:.0f} ms:")
    if g:
        print(f"  GPU bf16 : avg {g['avg_ms']:.2f} ms  p50 {g['p50_ms']:.2f} ms  (n={g['n']}, {g['ratio_to_benchmark']:.2f}x benchmark)")
    print(f"  CPU fp32 : avg {c['avg_ms']:.1f} ms  p50 {c['p50_ms']:.1f} ms  (n={c['n']}, {c['threads']} threads, {c['ratio_to_benchmark']:.2f}x benchmark)")

    for r in results:
        split, sfx = r.split, r.split.sfx
        common.write_json(lat, out_dir / f"latency{sfx}.json")
        tf_out = {kk: v for kk, v in r.tf.items() if kk != "curves"}
        metrics = {
            "timestamp": ts, "run_name": run_name, "split": split.name, "split_file": str(split.path), "split_title": SPLIT_TITLE[split.name],
            "limit_rows": args.limit_rows, "dev_run": dev, "checkpoint": str(ckpt), "splits_in_invocation": splits,
            "model_config": mcfg.to_dict(), "num_labels": k, "params": count_parameters(model)["total"],
            "checkpoint_config": {kk: v for kk, v in ckpt_cfg.items() if kk != "model"},
            "tokenizer": str(cfg["paths"]["tokenizer"]), "max_len": max_len,
            "label_names": {str(c3): LABEL_NAMES[c3] for c3 in range(k)} if k == 3 else {str(c2): BINARY_NAMES[c2] for c2 in range(2)},
            "binary_rollup": {"unwanted_name": UNWANTED_NAME, "classes": list(range(1, k)), "rule": "P(unwanted) = sum of P(class) for class >= 1; flag when >= threshold"},
            "split_sizes": {"rows": split.n, "positive": r.at["n_positive"], "negative": r.at["n_negative"],
                            "positive_rate": round(_safe_div(r.at["n_positive"], split.n), 4),
                            "label3": _label3_counts(split.label3),
                            "by_source": {s: {"rows": m["n"], "positive": m["n_positive"], "label3": m["label3_counts"]} for s, m in r.per_source.items()},
                            "mean_tokens_full": round(float(r.enc.n_tokens_full.mean()), 1),
                            "pct_truncated": round(100 * float(r.enc.truncated.mean()), 2)},
            "threshold": args.threshold, "at_threshold": r.at, "threshold_free": tf_out,
            "threeway": r.tw.metrics,
            "credential_lure_slice": {"n_phrases": len(CREDENTIAL_LURE_PHRASES), "phrases": r.lure["phrases"],
                                      "match_rule": "whole-word match on NFKC-lowercased text with non-alphanumerics collapsed to spaces",
                                      "phrase_hits": r.lure["phrase_hits"], "phrase_hits_by_label": r.lure["phrase_hits_by_label"],
                                      "label_purity": round(r.lure["label_purity"], 4), "min_positives_for_comparison": LURE_MIN_POSITIVES,
                                      "label3_counts": r.lure["label3_counts"], "threeway_accuracy": r.lure.get("threeway_accuracy"),
                                      "metrics": r.lure["metrics"]},
            "per_source": r.per_source, "length_confound": r.conf, "inference": r.inf, "latency": lat,
            "error_analysis": {"path": str(r.err_path), **r.err_stats}, "plots": r.plots,
        }
        p_metrics = out_dir / f"metrics{sfx}.json"
        common.write_json(metrics, p_metrics)
        LOG.info("wrote %s", p_metrics)

    # experiments.jsonl (contract: one line per invocation with run_name + the metrics of every split run)
    by_name = {r.split.name: _summary(r) for r in results}
    line = {"run_name": run_name, "timestamp": ts, "stage": "evaluate", "checkpoint": str(ckpt), "splits": splits, "num_labels": k,
            "dev_run": dev, "limit_rows": args.limit_rows,
            "val": by_name.get("val"), "test": by_name.get("test"), "test3": by_name.get("test3"),
            "latency_ms": {"gpu_bf16_avg": g and g["avg_ms"], "cpu_fp32_avg": c["avg_ms"]}}
    p_exp = out_dir / "experiments.jsonl"
    with p_exp.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(line, default=common._json_default) + "\n")
    LOG.info("appended %s", p_exp)

    # RUN_LOG.md: one entry per invocation covering every split that was run
    num = next_run_log_number(common.project_root() / "RUN_LOG.md")
    entry = run_log_entry(num, ts, run_name, results, ckpt, lat, out_dir, dev, k)
    p_entry = out_dir / f"run_log_entry_{'_'.join(splits)}.md"   # test -> _test, test,test3 -> _test_test3 (no collision with test3)
    p_entry.write_text(entry + "\n", encoding="utf-8")
    if dev or args.no_run_log:
        LOG.info("RUN_LOG.md not appended (%s); entry saved to %s", "dev run" if dev else "--no-run-log", p_entry)
    else:
        common.append_run_log(entry)
        LOG.info("appended entry %d to RUN_LOG.md", num)
    return 0


def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("--config", default="config.yaml")
    sub = ap.add_subparsers(dest="cmd", required=True)
    r = sub.add_parser("run", help="evaluate a fine-tuned checkpoint on the test / test3 / val split(s)")
    r.add_argument("--checkpoint", required=True, help="fine-tune best/ dir with model.safetensors + config.json")
    r.add_argument("--split", default="test",
                   help=f"one of {sorted(SPLIT_FILES)} or a comma list (e.g. test,test3): one invocation, one experiments line, one RUN_LOG entry")
    r.add_argument("--limit-rows", type=int, default=None, help="dev only: seeded random subset per split; skips the RUN_LOG append")
    r.add_argument("--batch-size", type=int, default=64)
    r.add_argument("--latency-samples", type=int, default=None, help="default config.eval.latency_samples (100)")
    r.add_argument("--out-dir", default=None, help="default config.paths.reports")
    r.add_argument("--data-dir", default=None, help="default config.paths.data_processed (fixture dir for self-tests)")
    r.add_argument("--run-name", default=None, help="default: the run dir name (parent of best/)")
    r.add_argument("--threshold", type=float, default=DEFAULT_THRESHOLD)
    r.add_argument("--max-len", type=int, default=None, help="default model max_seq_len")
    r.add_argument("--cpu-threads", type=int, default=16)
    r.add_argument("--cpu", action="store_true", help="force CPU inference (dev)")
    r.add_argument("--finetune-run-dir", default=None,
                   help="opt-in: also render <out-dir>/eval_finetune_curves.png from this run's metrics.csv/epochs.csv "
                        "(reports/finetune_curves.png itself belongs to finetune.py plot)")
    r.add_argument("--pretrain-run-dir", default=None,
                   help="opt-in: also render <out-dir>/eval_pretrain_curves.png from this run's metrics.csv "
                        "(reports/pretrain_curves.png itself belongs to pretrain.py plot)")
    r.add_argument("--no-run-log", action="store_true", help="do not append to RUN_LOG.md")
    args = ap.parse_args(argv)
    if args.latency_samples is None:
        args.latency_samples = int(common.load_config(args.config)["eval"]["latency_samples"])
    return args


def main(argv: Sequence[str] | None = None) -> int:
    args = parse_args(argv)
    return run(args)


if __name__ == "__main__":
    raise SystemExit(main())