evals/run_ai_detect_eval.py

"""Offline AI-text-detector eval — a DISAGREEMENT report, not a benchmark.

This is the detector-family sibling of ``run_eval.py`` (sentiment). It runs
*every* local detector over the *same* labeled rows and asks the question the
product actually cares about: **do the detectors agree, and when they don't,
where?** It deliberately reuses ``run_eval``'s torch-free helpers
(``compute_metrics``, ``latency_percentiles``, ``_md_cell``) so the two harnesses
share one definition of accuracy/F1/latency — but detector metrics NEVER mix
into a sentiment report and vice-versa.

Three ideas this module is built to teach:

* **Per-detector accuracy here is NOT a benchmark claim.** BOTH classes in the
  eval set were written by the AI assistant that built this project (see
  ``ai_detection_eval.csv``): the ``human`` rows are AI-authored text *styled*
  to read as human, not genuinely human writing. So a high P(ai) on a "human"
  row is not cleanly a false positive — the detector may be correctly smelling
  AI provenance. On top of that, the two hardest categories (``mixed_human_ai``,
  ``ambiguous``) have *contestable* ground truth even setting authorship aside;
  we arbitrate by ORIGIN/intent (mixed → ai, ambiguous → human), so a detector's
  score is partly a function of *our* arbitration. A real benchmark needs large,
  diverse, provenance-verified data with genuinely human text — none of which
  this is.

* **Disagreement is signal.** When three detectors trained on different data
  split on the same sentence, that split is a calibrated statement of
  uncertainty about that sentence — far more honest than any single model's
  confident label. The report surfaces the disagreement RATE and every
  disagreed-on row (each detector's label + P(ai)) as the primary deliverable.

* **Calibration matters for a probabilistic detector.** A useful P(ai) should be
  systematically *low* on genuine human text and *high* on AI text. The
  calibration hint (mean P(ai) on human rows vs ai rows, and the gap between
  them) shows whether a detector's probabilities separate the classes at all —
  a detector with a near-zero or negative gap is guessing, regardless of its
  headline accuracy.

Heavy imports (torch/transformers via DetectorModel) live INSIDE ``run()`` so
the pure helpers below stay importable and unit-testable without the ML stack,
exactly like ``run_eval``.
"""

import argparse
import csv
import io
import json
import statistics
import sys
import time
from dataclasses import dataclass
from pathlib import Path

# Shared, torch-free helpers — one definition of the numeric/formatting logic
# across both eval harnesses. Importable because run_eval keeps its heavy
# imports inside run() too.
from run_eval import (
    MAX_TEXT_CHARS,
    _md_cell,
    compute_metrics,
    latency_percentiles,
    validate_label_compatibility,
)

# --- Pure helpers (no torch, no app) -----------------------------------------

# The only two classes any detector in this project emits (registry-canonical).
DETECTOR_LABELS = ["human", "ai"]

REQUIRED_COLUMNS = ("text", "true_label")


@dataclass(frozen=True)
class DetectorRow:
    """One labeled example. ``source_type`` is the failure-analysis axis: it says
    *why* a row is interesting (``ai_edited``, ``ambiguous``, …), which is where
    detector disagreement clusters."""

    id: str
    text: str
    true_label: str
    source_type: str
    notes: str


def parse_detector_rows(csv_text: str) -> list[DetectorRow]:
    """Parse the detector CSV (``id,text,true_label,source_type,notes``).

    Same fail-loud contract as ``run_eval.parse_eval_rows``: reject a missing
    required column or an overlong row (never silently truncate the very text
    being scored), skip blank-text rows. The only difference is the
    ``source_type`` metadata column in place of ``category``.
    """
    reader = csv.DictReader(io.StringIO(csv_text))
    fields = reader.fieldnames or []
    missing = [c for c in REQUIRED_COLUMNS if c not in fields]
    if missing:
        raise ValueError(
            f"CSV missing required column(s): {missing}. Found: {list(fields)}"
        )

    rows: list[DetectorRow] = []
    for i, raw in enumerate(reader):
        text = (raw.get("text") or "").strip()
        if not text:
            continue
        if len(text) > MAX_TEXT_CHARS:
            # +2: header line + 0->1-based, so the number matches a spreadsheet.
            raise ValueError(
                f"Row {i + 2} text is {len(text)} chars, exceeds the "
                f"{MAX_TEXT_CHARS}-char limit. Fix the data; rows are never "
                "truncated."
            )
        rows.append(
            DetectorRow(
                id=(raw.get("id") or str(i + 1)).strip(),
                text=text,
                true_label=(raw.get("true_label") or "").strip(),
                source_type=(raw.get("source_type") or "").strip(),
                notes=(raw.get("notes") or "").strip(),
            )
        )
    return rows


def compute_disagreement(labels_by_detector: dict[str, list[str]]) -> dict:
    """Per-row disagreement flag + overall rate.

    A row "disagrees" when the detectors don't all land on the same label —
    ``len({labels}) > 1``, the exact rule ``/api/ai-detect/compare`` uses, so
    the eval and the endpoint tell the same story. The rate is the share of rows
    the fleet split on; a high rate on a hand-picked hard set is expected and is
    the honest headline of this report.
    """
    detectors = list(labels_by_detector)
    n = len(labels_by_detector[detectors[0]]) if detectors else 0
    flags = [
        len({labels_by_detector[d][i] for d in detectors}) > 1 for i in range(n)
    ]
    rate = (sum(flags) / n) if n else 0.0
    return {"flags": flags, "rate": rate}


def pairwise_agreement(labels_by_detector: dict[str, list[str]]) -> list[dict]:
    """Fraction of rows each detector PAIR labels identically.

    The disagreement rate collapses the whole fleet to one number; pairwise
    agreement shows the structure underneath — which two detectors move together
    and which one is the odd one out. Reported for every unordered pair.
    """
    from itertools import combinations

    ids = list(labels_by_detector)
    out: list[dict] = []
    for a, b in combinations(ids, 2):
        la, lb = labels_by_detector[a], labels_by_detector[b]
        n = len(la)
        agree = sum(1 for x, y in zip(la, lb) if x == y)
        out.append(
            {"detector_a": a, "detector_b": b, "agreement": (agree / n) if n else 0.0}
        )
    return out


def calibration_hint(
    true_labels: list[str], p_ai_by_detector: dict[str, list[float]]
) -> dict[str, dict]:
    """Mean P(ai) on human rows vs ai rows, per detector, plus the gap.

    A probabilistic detector is only useful if its P(ai) is systematically lower
    on real human text than on AI text. ``gap = mean_on_ai - mean_on_human``
    measures that separation directly: a large positive gap means the
    probabilities carry real information; a gap near zero (or negative) means the
    detector is effectively guessing even if its thresholded accuracy looks fine.
    ``None`` when a class is absent, so a one-sided set never fabricates a gap.
    """
    human_idx = [i for i, t in enumerate(true_labels) if t == "human"]
    ai_idx = [i for i, t in enumerate(true_labels) if t == "ai"]
    out: dict[str, dict] = {}
    for d, ps in p_ai_by_detector.items():
        mh = statistics.mean(ps[i] for i in human_idx) if human_idx else None
        ma = statistics.mean(ps[i] for i in ai_idx) if ai_idx else None
        gap = (ma - mh) if (mh is not None and ma is not None) else None
        out[d] = {"mean_p_ai_on_human": mh, "mean_p_ai_on_ai": ma, "gap": gap}
    return out


def collect_disagreement_examples(
    rows: list[DetectorRow],
    labels_by_detector: dict[str, list[str]],
    p_ai_by_detector: dict[str, list[float]],
    flags: list[bool],
) -> list[dict]:
    """The disagreed-on rows — the actual product of this eval.

    For each flagged row, attach every detector's label and P(ai) side by side,
    so a reader sees not just *that* the fleet split but *how confidently* each
    side did — the case worth teaching is one detector at 0.9 against another at
    0.1 on the same sentence."""
    detectors = list(labels_by_detector)
    out: list[dict] = []
    for i, (row, flag) in enumerate(zip(rows, flags)):
        if not flag:
            continue
        per = [
            {
                "detector": d,
                "label": labels_by_detector[d][i],
                "p_ai": p_ai_by_detector[d][i],
            }
            for d in detectors
        ]
        out.append(
            {
                "id": row.id,
                "text": row.text,
                "source_type": row.source_type,
                "true": row.true_label,
                "per_detector": per,
                "notes": row.notes,
            }
        )
    return out


def collect_wrong_examples(
    rows: list[DetectorRow], y_pred: list[str], p_ai: list[float]
) -> list[dict]:
    """One detector's misclassified rows, with P(ai) attached.

    Confidently wrong is the dangerous failure: a detector calling genuine human
    text "ai" at P(ai)=0.9 is worse than a hesitant 0.55. The ``source_type``
    tells you which kind of text tripped it (non-native? edited AI?)."""
    wrong: list[dict] = []
    for row, pred, p in zip(rows, y_pred, p_ai):
        if pred != row.true_label:
            wrong.append(
                {
                    "id": row.id,
                    "text": row.text,
                    "source_type": row.source_type,
                    "true": row.true_label,
                    "predicted": pred,
                    "p_ai": p,
                    "notes": row.notes,
                }
            )
    return wrong


def build_detector_summary(
    data_file: str,
    rows: list[DetectorRow],
    detector_results: dict[str, dict],
    warning: str,
) -> dict:
    """Assemble the machine-readable summary from raw per-detector predictions.

    ``detector_results[id]`` carries ``name``/``y_pred``/``p_ai``/
    ``latencies_ms`` (produced by ``run()``); every metric — per-detector scores
    AND the cross-detector disagreement/pairwise/calibration blocks — is derived
    here from those torch-free arrays, so this whole function is unit-testable
    without the model stack.
    """
    y_true = [r.true_label for r in rows]
    labels = list(DETECTOR_LABELS)
    labels_by_detector = {d: res["y_pred"] for d, res in detector_results.items()}
    p_ai_by_detector = {d: res["p_ai"] for d, res in detector_results.items()}

    detectors: dict[str, dict] = {}
    for d, res in detector_results.items():
        metrics = compute_metrics(y_true, res["y_pred"], labels)
        lat = latency_percentiles(res["latencies_ms"])
        detectors[d] = {
            "name": res["name"],
            "accuracy": round(metrics["accuracy"], 4),
            "macro_f1": round(metrics["macro_f1"], 4),
            "latency_p50_ms": round(lat["p50"], 2),
            "latency_p95_ms": round(lat["p95"], 2),
            "per_class": metrics["per_class"],
            "confusion_matrix": metrics["confusion_matrix"],
            "wrong_examples": collect_wrong_examples(rows, res["y_pred"], res["p_ai"]),
        }

    dis = compute_disagreement(labels_by_detector)
    return {
        "data_file": data_file,
        "n_examples": len(rows),
        "labels": labels,
        "detectors": detectors,
        "disagreement_rate": round(dis["rate"], 4),
        "disagreement_examples": collect_disagreement_examples(
            rows, labels_by_detector, p_ai_by_detector, dis["flags"]
        ),
        "pairwise_agreement": pairwise_agreement(labels_by_detector),
        "calibration": calibration_hint(y_true, p_ai_by_detector),
        "warning": warning,
    }


# The brief's honest-scope paragraph — a DIFFERENT string from the API's
# DETECTOR_WARNING (both are required in the report). This one explains what the
# eval does and does not claim; the API warning is the product-facing promise.
_HONEST_LIMITATIONS = (
    "AI text detection is not proof of authorship. These models estimate whether "
    "text resembles patterns seen in AI-generated or human-written training data. "
    "Short text, edited AI text, non-native writing, and formal human writing can "
    "all confuse detectors."
)


def _fmt(x: float | None, nd: int = 3) -> str:
    """Format a possibly-None metric for a table cell."""
    return "—" if x is None else f"{x:.{nd}f}"


def render_detector_report(summary: dict) -> str:
    """Render the human-facing Markdown disagreement report from a summary."""
    labels = summary["labels"]
    detectors = summary["detectors"]
    lines: list[str] = []

    lines.append("# AI Text Detector Evaluation — Disagreement Report")
    lines.append("")
    lines.append(f"- **Dataset:** `{summary['data_file']}` "
                 f"({summary['n_examples']} labeled examples)")
    lines.append(f"- **Detectors:** {', '.join(detectors)}")
    lines.append(f"- **Labels:** {', '.join(labels)}")
    lines.append("")

    # Two distinct honest texts, both required.
    lines.append(f"> **Limitations.** {_HONEST_LIMITATIONS}")
    lines.append("")
    lines.append("> **Product warning (verbatim, shown on every detector API "
                 f"response):** {summary['warning']}")
    lines.append("")

    lines.append("## Why this is not a benchmark")
    lines.append("")
    lines.append("Every row in this set — **both the `human`-labeled and the "
                 "`ai`-labeled examples — was written by the AI assistant that "
                 "built this project.** There is no genuinely human-authored text "
                 "here: the `human` rows are AI-authored text *styled* to read as "
                 "human (casual, formal, non-native, marketing, \"sounds-AI\"). "
                 "That makes this a teaching fixture for *failure analysis and "
                 "disagreement*, **not a benchmark** of detector quality, and it "
                 "colors every number below:")
    lines.append("")
    lines.append("1. **The `human` class is an AI-authored stand-in, not ground "
                 "truth.** A high P(ai) on those rows is therefore not cleanly a "
                 "false positive — the detector may be correctly smelling AI "
                 "provenance. Read \"flags a human row as AI\" as \"flags our "
                 "AI-authored human *impersonation* as AI,\" which is as much a "
                 "statement about the fixture as about the detector. The small "
                 "calibration gaps below follow from the same fact: AI-styled-human "
                 "and AI text are *both* AI-authored, so their P(ai) distributions "
                 "naturally sit close together.")
    lines.append("2. **Contestable ground truth.** Even setting authorship aside, "
                 "the `mixed_human_ai` and `ambiguous` rows have no objective gold "
                 "label. We arbitrate by **origin/intent** (`mixed_human_ai` → "
                 "`ai`, `ambiguous` → `human`), so each detector's accuracy is "
                 "partly a function of *our* labeling calls, not just its skill.")
    lines.append("3. **Tiny, single-author, English-only.** Real evaluation needs "
                 "large, diverse, provenance-verified data with genuinely human "
                 "writing. Read the numbers below as *illustrations of failure "
                 "modes and detector disagreement*, never as scores.")
    lines.append("")

    # --- The primary deliverable: disagreement --------------------------------
    lines.append("## Detector disagreement")
    lines.append("")
    rate = summary["disagreement_rate"]
    lines.append(f"The fleet split on **{rate:.1%}** of rows (a row \"disagrees\" "
                 "when the detectors do not all pick the same label — the same rule "
                 "`/api/ai-detect/compare` uses). Disagreement is not noise to "
                 "average away: when models trained on different data split on a "
                 "sentence, that split *is* a statement of uncertainty about it.")
    lines.append("")
    ex = summary["disagreement_examples"]
    if not ex:
        lines.append("_No disagreements — every detector agreed on every row._")
    else:
        det_ids = list(detectors)
        header = ["Text", "Source", "True"] + [f"{d} label / P(ai)" for d in det_ids]
        lines.append("| " + " | ".join(header) + " |")
        lines.append("| " + " | ".join("---" for _ in header) + " |")
        for e in ex:
            per = {d["detector"]: d for d in e["per_detector"]}
            cells = [_md_cell(e["text"]), _md_cell(e["source_type"]), e["true"]]
            for d in det_ids:
                cells.append(f"{per[d]['label']} / {per[d]['p_ai']:.2f}")
            lines.append("| " + " | ".join(cells) + " |")
    lines.append("")

    lines.append("## Pairwise agreement")
    lines.append("")
    lines.append("How often each PAIR of detectors lands on the same label. This "
                 "shows the structure the single disagreement rate hides — which "
                 "detectors move together, and which is the outlier.")
    lines.append("")
    lines.append("| Detector A | Detector B | Agreement |")
    lines.append("| --- | --- | --- |")
    for p in summary["pairwise_agreement"]:
        lines.append(f"| {p['detector_a']} | {p['detector_b']} | "
                     f"{p['agreement']:.1%} |")
    lines.append("")

    lines.append("## Calibration hint")
    lines.append("")
    lines.append("Mean P(ai) on the genuinely-human rows vs the AI rows. A useful "
                 "probabilistic detector keeps P(ai) **low on human** text and "
                 "**high on AI** text; the **gap** is that separation. A gap near "
                 "zero means the probabilities carry little information even when "
                 "the thresholded label happens to be right.")
    lines.append("")
    lines.append("| Detector | Mean P(ai) on human | Mean P(ai) on AI | Gap |")
    lines.append("| --- | --- | --- | --- |")
    for d, c in summary["calibration"].items():
        lines.append(f"| {d} | {_fmt(c['mean_p_ai_on_human'])} | "
                     f"{_fmt(c['mean_p_ai_on_ai'])} | {_fmt(c['gap'])} |")
    lines.append("")

    # --- Per-detector metrics -------------------------------------------------
    lines.append("## Per-detector metrics")
    lines.append("")
    lines.append("Accuracy and macro F1 against the seed labels, plus single-text "
                 "latency (p50 = typical, p95 = the slow tail users feel).")
    lines.append("")
    lines.append("| Detector | Accuracy | Macro F1 | Latency p50 (ms) | "
                 "Latency p95 (ms) |")
    lines.append("| --- | --- | --- | --- | --- |")
    for d, m in detectors.items():
        lines.append(f"| {d} | {m['accuracy']:.3f} | {m['macro_f1']:.3f} | "
                     f"{m['latency_p50_ms']:.1f} | {m['latency_p95_ms']:.1f} |")
    lines.append("")

    for d, m in detectors.items():
        lines.append(f"### `{d}` — {m['name']}")
        lines.append("")
        lines.append("Per-class precision / recall / F1:")
        lines.append("")
        lines.append("| Class | Precision | Recall | F1 | Support |")
        lines.append("| --- | --- | --- | --- | --- |")
        for label in labels:
            pc = m["per_class"][label]
            lines.append(f"| {label} | {pc['precision']:.2f} | {pc['recall']:.2f} "
                         f"| {pc['f1']:.2f} | {pc['support']} |")
        lines.append("")
        lines.append("Confusion matrix (rows = true, cols = predicted):")
        lines.append("")
        lines.append("| true \\ pred | " + " | ".join(labels) + " |")
        lines.append("| --- | " + " | ".join("---" for _ in labels) + " |")
        for i, label in enumerate(labels):
            cells = " | ".join(str(c) for c in m["confusion_matrix"][i])
            lines.append(f"| **{label}** | {cells} |")
        lines.append("")
        wrong = m["wrong_examples"]
        if not wrong:
            lines.append("_No misclassifications for this detector._")
        else:
            lines.append(f"Misclassified ({len(wrong)} of {summary['n_examples']}) — "
                         "P(ai) shows how confidently it was wrong:")
            lines.append("")
            lines.append("| Text | Source | True | Predicted | P(ai) |")
            lines.append("| --- | --- | --- | --- | --- |")
            for w in wrong:
                lines.append(f"| {_md_cell(w['text'])} | {_md_cell(w['source_type'])} "
                             f"| {w['true']} | {w['predicted']} | {w['p_ai']:.2f} |")
        lines.append("")

    lines.append("## Machine-readable summary")
    lines.append("")
    lines.append("```json")
    lines.append(json.dumps({
        "n_examples": summary["n_examples"],
        "disagreement_rate": summary["disagreement_rate"],
        "pairwise_agreement": summary["pairwise_agreement"],
        "calibration": summary["calibration"],
        "detectors": {
            d: {k: m[k] for k in ("accuracy", "macro_f1",
                                  "latency_p50_ms", "latency_p95_ms")}
            for d, m in detectors.items()
        },
    }, indent=2))
    lines.append("```")
    lines.append("")
    return "\n".join(lines)


# --- CLI + model integration (heavy imports live here) -----------------------

_BACKEND_DIR = Path(__file__).resolve().parents[1] / "backend"


def _backend_imports():
    """Import the backend registry/model factory + the API's warning string.

    torch/transformers are pulled in only when a detector actually loads (the
    backend's own lazy design), so importing this module for the pure helpers
    stays cheap. The warning is sourced from ``app.routes`` — its single source
    of truth — so the report can never drift from what the API promises."""
    if str(_BACKEND_DIR) not in sys.path:
        sys.path.insert(0, str(_BACKEND_DIR))
    from app.model import build_model
    from app.model_registry import ModelTask, get_model_config, models_for_task
    from app.routes import DETECTOR_WARNING

    return build_model, get_model_config, models_for_task, ModelTask, DETECTOR_WARNING


def run(data_path: str, out_path: str, detector_ids: list[str] | None = None) -> dict:
    """End-to-end eval: load data, run every detector over the SAME rows, score
    each, compute cross-detector disagreement, and write the report."""
    (build_model, get_model_config, models_for_task,
     ModelTask, warning) = _backend_imports()

    rows = parse_detector_rows(Path(data_path).read_text(encoding="utf-8"))
    if not rows:
        raise SystemExit(f"No labeled rows found in {data_path}")

    # Guard once: the seed labels must be predictable by the detectors. All
    # detectors share the ("human", "ai") label set, so one check covers them.
    dataset_labels = {r.true_label for r in rows}
    validate_label_compatibility(dataset_labels, set(DETECTOR_LABELS), allow_mismatch=False)

    ids = detector_ids or list(models_for_task(ModelTask.AI_TEXT_DETECTION))

    detector_results: dict[str, dict] = {}
    for model_id in ids:
        cfg = get_model_config(model_id)
        if cfg.task != ModelTask.AI_TEXT_DETECTION:
            raise SystemExit(
                f"Model '{model_id}' is not an AI text detector (task={cfg.task})."
            )
        model = build_model(cfg)
        print(f"Loading {model_id} ({cfg.name})...")
        model.load()
        print(f"  loaded on device: {model.device}")

        # One untimed warmup: the first MPS inference pays one-off graph/compile
        # cost that would otherwise masquerade as a p95 spike.
        model.predict([rows[0].text])

        y_pred: list[str] = []
        p_ai: list[float] = []
        latencies_ms: list[float] = []
        for row in rows:
            start = time.perf_counter()
            pred = model.predict([row.text])[0]
            latencies_ms.append((time.perf_counter() - start) * 1000)
            y_pred.append(pred["label"])
            p_ai.append(pred["scores"]["ai"])

        detector_results[model_id] = {
            "name": cfg.name,
            "y_pred": y_pred,
            "p_ai": p_ai,
            "latencies_ms": latencies_ms,
        }
        # Drop the model before loading the next one — desklib alone is ~1.6GB;
        # we only need its predictions from here on, not its weights.
        del model

    summary = build_detector_summary(data_path, rows, detector_results, warning)
    Path(out_path).write_text(render_detector_report(summary), encoding="utf-8")
    print(f"\nWrote report to {out_path}")
    print(json.dumps({
        "n_examples": summary["n_examples"],
        "disagreement_rate": summary["disagreement_rate"],
        "detectors": {
            d: {"accuracy": m["accuracy"], "macro_f1": m["macro_f1"]}
            for d, m in summary["detectors"].items()
        },
    }, indent=2))
    return summary


def main(argv: list[str] | None = None) -> None:
    parser = argparse.ArgumentParser(
        description="Run every local AI text detector over a labeled CSV and "
        "write a Markdown disagreement report.",
    )
    parser.add_argument(
        "--data", default="evals/data/ai_detection_eval.csv",
        help="Path to the labeled detector eval CSV.",
    )
    parser.add_argument(
        "--out", default="evals/ai_detection_report.md",
        help="Path to write the report.",
    )
    parser.add_argument(
        "--detectors", nargs="*", default=None,
        help="Detector model ids to run (default: all registry detectors).",
    )
    args = parser.parse_args(argv)
    run(args.data, args.out, args.detectors)


if __name__ == "__main__":
    main()