scripts/analyze_confirmatory.py

#!/usr/bin/env python3
"""Audit E01-E05 and produce deterministic paper tables, tests, and figures."""

from __future__ import annotations

import argparse
from collections import Counter
from hashlib import sha256
import itertools
import json
import math
import os
from pathlib import Path
import subprocess
import warnings
from typing import Any, Callable, Iterable, Sequence

import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from scipy.stats import binomtest

from agent_harness.specs import load_harnesses, load_tasks
from agent_harness.tokenization import QwenTokenCounter


E01_REVISION = "e59145ac"
E02_ONE_SHOT_REVISION = "b08e027a"
E02_ITERATIVE_REVISION = "1a7066f6"
E03_REVISIONS = {"e78637b7", "6debf9e0"}
E04_REVISION = "6a0f103b"
E05_REVISION = "d1b9828e"
BOOTSTRAPS = 20_000
ANALYSIS_SEED = 20260718


class AnalysisFailure(RuntimeError):
    """Raised when raw artifacts do not match the declared analysis population."""


def git(root: Path, *arguments: str) -> str:
    result = subprocess.run(
        ["git", *arguments], cwd=root, check=False, capture_output=True, text=True, timeout=60
    )
    if result.returncode:
        raise AnalysisFailure(result.stderr.strip() or "git command failed")
    return result.stdout.strip()


def seed_for(label: str) -> int:
    return ANALYSIS_SEED + int(sha256(label.encode()).hexdigest()[:8], 16)


def bootstrap_mean(values: Sequence[float], label: str) -> tuple[float, float]:
    data = np.asarray(values, dtype=float)
    if not len(data):
        return (math.nan, math.nan)
    rng = np.random.default_rng(seed_for(label))
    draws = data[rng.integers(0, len(data), size=(BOOTSTRAPS, len(data)))].mean(axis=1)
    return tuple(float(value) for value in np.quantile(draws, [0.025, 0.975]))


def paired_result(
    left: pd.Series,
    right: pd.Series,
    label: str,
    test: bool = True,
) -> dict[str, Any]:
    paired = pd.concat([left.rename("left"), right.rename("right")], axis=1).dropna()
    differences = (paired["left"] - paired["right"]).to_numpy(dtype=float)
    low, high = bootstrap_mean(differences, label)
    result = {
        "n_tasks": int(len(differences)),
        "left_mean": float(paired["left"].mean()),
        "right_mean": float(paired["right"].mean()),
        "mean_difference": float(differences.mean()),
        "ci95_low": low,
        "ci95_high": high,
    }
    if test:
        result["p_exact_two_sided"] = exact_sign_flip(differences)
    return result


def exact_sign_flip(values: Sequence[float]) -> float:
    data = np.asarray(values, dtype=float)
    data = data[np.abs(data) > 1e-15]
    if not len(data):
        return 1.0
    if len(data) > 20:
        raise AnalysisFailure("exact sign-flip enumeration is bounded to 20 task units")
    observed = abs(float(data.mean()))
    extreme = 0
    total = 2 ** len(data)
    magnitudes = np.abs(data)
    for bits in itertools.product((-1.0, 1.0), repeat=len(data)):
        statistic = abs(float(np.mean(magnitudes * np.asarray(bits))))
        if statistic >= observed - 1e-15:
            extreme += 1
    return extreme / total


def holm(records: list[dict[str, Any]], p_key: str = "p_exact_two_sided") -> None:
    ordered = sorted(enumerate(records), key=lambda item: item[1][p_key])
    running = 0.0
    count = len(records)
    for order, (index, record) in enumerate(ordered):
        adjusted = min(1.0, (count - order) * float(record[p_key]))
        running = max(running, adjusted)
        records[index]["p_holm"] = running


def native(value: Any) -> Any:
    if isinstance(value, dict):
        return {str(key): native(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [native(item) for item in value]
    if isinstance(value, (np.integer,)):
        return int(value)
    if isinstance(value, (np.floating,)):
        return None if not np.isfinite(value) else float(value)
    if isinstance(value, float) and not math.isfinite(value):
        return None
    return value


def write_json(path: Path, value: Any) -> None:
    path.write_text(json.dumps(native(value), indent=2, sort_keys=True) + "\n", encoding="utf-8")


def manifest_for(final_path: Path) -> dict[str, Any]:
    return json.loads((final_path.parent / "run_manifest.json").read_text(encoding="utf-8"))


def load_raw(root: Path, experiment_id: str) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for path in sorted((root / "results" / "raw" / experiment_id).glob("*/*/*/final_metrics.json")):
        final = json.loads(path.read_text(encoding="utf-8"))
        manifest = manifest_for(path)
        identity = manifest["identity"]
        rows.append(
            {
                "final": final,
                "manifest": manifest,
                "path": path,
                "task_id": identity["task_id"],
                "harness_id": identity["harness_id"],
                "seed": identity["seed"],
                "revision": identity["code_revision"],
                "run_id": manifest["run_id"],
            }
        )
    return rows


def audit_trajectory(row: dict[str, Any]) -> None:
    path = row["path"].parent / "trajectory.jsonl"
    events = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line]
    if not events:
        raise AnalysisFailure(f"empty trajectory: {path}")
    if [event["sequence"] for event in events] != list(range(len(events))):
        raise AnalysisFailure(f"non-contiguous trajectory: {path}")
    if events[0]["event_type"] != "run_started" or events[-1]["event_type"] != "run_finished":
        raise AnalysisFailure(f"trajectory boundary failure: {path}")
    if any(event["run_id"] != row["run_id"] for event in events):
        raise AnalysisFailure(f"trajectory run ID mismatch: {path}")


def audit_artifacts(root: Path, raw: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
    selected = {
        "E01": [row for row in raw["E01"] if row["revision"].startswith(E01_REVISION)],
        "E02": [
            row
            for row in raw["E02"]
            if row["revision"].startswith((E02_ONE_SHOT_REVISION, E02_ITERATIVE_REVISION))
        ],
        "E03": [row for row in raw["E03"] if row["revision"][:8] in E03_REVISIONS],
        "E04": [row for row in raw["E04"] if row["revision"].startswith(E04_REVISION)],
        "E05": [row for row in raw["E05"] if row["revision"].startswith(E05_REVISION)],
    }
    expected = {"E01": 150, "E02": 105, "E03": 100, "E04": 135, "E05": 405}
    observed = {experiment: len(rows) for experiment, rows in selected.items()}
    if observed != expected:
        raise AnalysisFailure(f"confirmatory count mismatch: expected {expected}, observed {observed}")
    ids = [row["run_id"] for rows in selected.values() for row in rows]
    if len(ids) != len(set(ids)):
        raise AnalysisFailure("confirmatory run IDs are not globally unique")
    for rows in selected.values():
        for row in rows:
            audit_trajectory(row)

    e02_counts = Counter(row["harness_id"] for row in selected["E02"])
    if e02_counts != Counter({item: 15 for item in ("H008", "H010", "H011", "H012", "H013", "H014", "H015")}):
        raise AnalysisFailure(f"E02 harness imbalance: {e02_counts}")
    e02_superseded = [row for row in raw["E02"] if row not in selected["E02"]]
    if len(e02_superseded) != 3:
        raise AnalysisFailure(f"expected three superseded E02 development rows, found {len(e02_superseded)}")
    staging = root / "results" / "staging" / "E02" / "1a7066f6c7682793b6f04445c776a93cc4fac895"
    if len(list(staging.glob("*/*/query_stage.json"))) != 30 or len(list(staging.glob("*/*/refined_ranking.json"))) != 30:
        raise AnalysisFailure("E02 iterative staging artifacts are incomplete")

    stale = Counter(row["final"]["stale_index"]["shared_key"] for row in selected["E04"])
    if len(stale) != 45 or set(stale.values()) != {3}:
        raise AnalysisFailure("E04 stale-index units were not copied exactly three times")
    doses = Counter(
        (row["final"]["seed"], row["final"]["plausible_distractors"]["severity"])
        for row in selected["E04"]
    )
    if doses != Counter({(0, 1): 45, (1, 5): 45, (2, 10): 45}):
        raise AnalysisFailure(f"E04 dose mapping mismatch: {doses}")
    for row in selected["E04"]:
        source_path = row["path"].parent / "distractor_sources.json"
        for source in json.loads(source_path.read_text(encoding="utf-8")):
            if sha256(source["text"].encode()).hexdigest() != source["sha256"]:
                raise AnalysisFailure(f"E04 distractor hash mismatch: {source_path}")

    return {
        "confirmatory_cells": sum(observed.values()),
        "counts": observed,
        "unique_run_ids": len(set(ids)),
        "e02_superseded_development_cells_excluded": len(e02_superseded),
        "e02_query_stage_count": 30,
        "e02_refined_ranking_count": 30,
        "e04_unique_stale_units": len(stale),
        "e04_stale_copy_multiplicity": 3,
        "e04_dose_counts": {f"seed_{seed}_dose_{dose}": count for (seed, dose), count in sorted(doses.items())},
        "development_only": {
            "E00_cells": len(load_raw(root, "E00")),
            "E06_smoke_cells": len(load_raw(root, "E06")),
        },
        "selected": selected,
    }


def ci_summary(group: pd.DataFrame, metric: str, label: str) -> dict[str, Any]:
    values = group[metric].astype(float).to_numpy()
    low, high = bootstrap_mean(values, label)
    return {
        "n": int(len(values)),
        "mean": float(values.mean()),
        "ci95_low": low,
        "ci95_high": high,
        "median": float(np.median(values)),
    }


def analyze_e01(rows: list[dict[str, Any]], harness_names: dict[str, str]) -> dict[str, Any]:
    records = []
    for row in rows:
        records.append({"task_id": row["task_id"], "harness_id": row["harness_id"], **row["final"]})
    frame = pd.DataFrame(records).sort_values(["task_id", "harness_id"])
    summary_rows = []
    for harness_id, group in frame.groupby("harness_id", sort=True):
        item = {"harness_id": harness_id, "harness_name": harness_names[harness_id], "n_tasks": len(group)}
        for metric in ("file_recall_at_10", "function_recall_at_10", "mrr", "ndcg_at_10", "all_gold_in_top_10", "all_gold_within_token_budget"):
            values = group[metric].astype(float).to_numpy()
            low, high = bootstrap_mean(values, f"E01:{harness_id}:{metric}")
            item[f"{metric}_mean"] = float(values.mean())
            item[f"{metric}_ci_low"] = low
            item[f"{metric}_ci_high"] = high
        item["query_seconds_median"] = float(group["query_seconds"].median())
        item["packed_tokens_median"] = float(group["packed_tokens"].median())
        summary_rows.append(item)
    summary = pd.DataFrame(summary_rows)

    pivot = frame.pivot(index="task_id", columns="harness_id", values="file_recall_at_10")
    definitions = [
        ("lexical_vs_exact", "H001", "H000"),
        ("syntax_vs_exact", "H002", "H000"),
        ("dense_vs_exact", "H003", "H000"),
        ("full_stack_vs_exact", "H007", "H000"),
        ("one_hop_vs_zero_hop", "H008", "H007"),
        ("two_hop_vs_one_hop", "H009", "H008"),
    ]
    contrasts = []
    for label, left, right in definitions:
        result = paired_result(pivot[left], pivot[right], f"E01:contrast:{label}")
        contrasts.append({"contrast": label, "left": left, "right": right, **result})
    holm(contrasts)

    factor_ids = [f"H{index:03d}" for index in range(8)]
    factors = {
        "H000": (0, 0, 0), "H001": (1, 0, 0), "H002": (0, 1, 0), "H003": (0, 0, 1),
        "H004": (1, 1, 0), "H005": (1, 0, 1), "H006": (0, 1, 1), "H007": (1, 1, 1),
    }
    task_effects: dict[str, list[float]] = {key: [] for key in ("L", "S", "D", "L:S", "L:D", "S:D", "L:S:D")}
    for _, task in frame[frame["harness_id"].isin(factor_ids)].pivot(index="task_id", columns="harness_id", values="file_recall_at_10").iterrows():
        cube = {factors[harness]: float(task[harness]) for harness in factor_ids}
        task_effects["L"].append(np.mean([cube[(1, s, d)] - cube[(0, s, d)] for s in (0, 1) for d in (0, 1)]))
        task_effects["S"].append(np.mean([cube[(l, 1, d)] - cube[(l, 0, d)] for l in (0, 1) for d in (0, 1)]))
        task_effects["D"].append(np.mean([cube[(l, s, 1)] - cube[(l, s, 0)] for l in (0, 1) for s in (0, 1)]))
        task_effects["L:S"].append(np.mean([cube[(1, 1, d)] - cube[(1, 0, d)] - cube[(0, 1, d)] + cube[(0, 0, d)] for d in (0, 1)]))
        task_effects["L:D"].append(np.mean([cube[(1, s, 1)] - cube[(1, s, 0)] - cube[(0, s, 1)] + cube[(0, s, 0)] for s in (0, 1)]))
        task_effects["S:D"].append(np.mean([cube[(l, 1, 1)] - cube[(l, 1, 0)] - cube[(l, 0, 1)] + cube[(l, 0, 0)] for l in (0, 1)]))
        task_effects["L:S:D"].append(
            cube[(1, 1, 1)] - cube[(1, 1, 0)] - cube[(1, 0, 1)] - cube[(0, 1, 1)]
            + cube[(1, 0, 0)] + cube[(0, 1, 0)] + cube[(0, 0, 1)] - cube[(0, 0, 0)]
        )
    factorial = []
    for effect, values in task_effects.items():
        low, high = bootstrap_mean(values, f"E01:factorial:{effect}")
        factorial.append(
            {
                "effect": effect,
                "n_tasks": len(values),
                "estimate": float(np.mean(values)),
                "ci95_low": low,
                "ci95_high": high,
                "p_exact_two_sided": exact_sign_flip(values),
            }
        )
    holm(factorial)

    mixed: dict[str, Any]
    try:
        import statsmodels.formula.api as smf

        mixed_frame = frame[frame["harness_id"].isin(factor_ids)][["task_id", "harness_id", "file_recall_at_10"]].copy()
        mixed_frame[["L", "S", "D"]] = [
            tuple(value - 0.5 for value in factors[harness]) for harness in mixed_frame["harness_id"]
        ]
        with warnings.catch_warnings(record=True) as caught:
            warnings.simplefilter("always")
            fit = smf.mixedlm("file_recall_at_10 ~ L * S * D", mixed_frame, groups=mixed_frame["task_id"]).fit(
                reml=True, method="powell", maxiter=1000, disp=False
            )
        fixed_names = list(fit.fe_params.index)
        intervals = fit.conf_int().loc[fixed_names]
        mixed = {
            "status": "fit",
            "converged": bool(fit.converged),
            "fixed_effects": {
                name: {
                    "estimate": float(fit.fe_params[name]),
                    "standard_error": float(fit.bse_fe[name]),
                    "wald_ci95_low": float(intervals.loc[name, 0]),
                    "wald_ci95_high": float(intervals.loc[name, 1]),
                }
                for name in fixed_names
            },
            "task_random_intercept_variance": float(fit.cov_re.iloc[0, 0]),
            "warnings": [str(item.message) for item in caught],
        }
    except Exception as exc:  # The exact paired factorial analysis remains primary.
        mixed = {"status": "failed", "error": str(exc)}

    return {"runs": frame, "summary": summary, "contrasts": contrasts, "factorial": factorial, "mixed_model": mixed}


def usage_totals(usage: dict[str, Any]) -> dict[str, int]:
    parts = [usage]
    if "first" in usage or "second" in usage:
        parts = [value for key in ("first", "second") if isinstance((value := usage.get(key)), dict)]
    result = {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0, "reasoning_tokens": 0}
    for part in parts:
        result["prompt_tokens"] += int(part.get("prompt_tokens", 0))
        result["completion_tokens"] += int(part.get("completion_tokens", 0))
        result["total_tokens"] += int(part.get("total_tokens", 0))
        result["reasoning_tokens"] += int(part.get("completion_tokens_details", {}).get("reasoning_tokens", 0))
    return result


def analyze_e02(rows: list[dict[str, Any]], harness_names: dict[str, str]) -> dict[str, Any]:
    records = []
    for row in rows:
        final = row["final"]
        records.append(
            {
                "task_id": row["task_id"], "harness_id": row["harness_id"],
                **final["metrics"], "context_tokens": final["context_tokens"],
                "elapsed_seconds": final["elapsed_seconds"], "model_calls": final["model_calls"],
                "protocol_violation": final.get("protocol_violation") is not None,
                **usage_totals(final.get("usage", {})),
            }
        )
    frame = pd.DataFrame(records).sort_values(["task_id", "harness_id"])
    summary_rows = []
    for harness_id, group in frame.groupby("harness_id", sort=True):
        recall = ci_summary(group, "file_recall_at_10", f"E02:{harness_id}:recall")
        summary_rows.append(
            {
                "harness_id": harness_id, "harness_name": harness_names[harness_id], "n_tasks": len(group),
                "file_recall_at_10_mean": recall["mean"], "file_recall_at_10_ci_low": recall["ci95_low"],
                "file_recall_at_10_ci_high": recall["ci95_high"], "mrr_mean": float(group["mrr"].mean()),
                "all_gold_in_top_10_rate": float(group["all_gold_in_top_10"].mean()),
                "protocol_violations": int(group["protocol_violation"].sum()),
                "protocol_violation_rate": float(group["protocol_violation"].mean()),
                "context_tokens_mean": float(group["context_tokens"].mean()),
                "total_tokens_mean": float(group["total_tokens"].mean()),
                "reasoning_tokens_mean": float(group["reasoning_tokens"].mean()),
                "elapsed_seconds_median": float(group["elapsed_seconds"].median()),
                "model_calls_mean": float(group["model_calls"].mean()),
            }
        )
    summary = pd.DataFrame(summary_rows)
    recall = frame.pivot(index="task_id", columns="harness_id", values="file_recall_at_10")
    tokens = frame.pivot(index="task_id", columns="harness_id", values="total_tokens")
    definitions = [
        ("iterative_vs_one_shot_unified", "H010", "H008"),
        ("iterative_vs_one_shot_specialized", "H012", "H011"),
        ("specialized_vs_unified_one_shot", "H011", "H008"),
        ("specialized_vs_unified_iterative", "H012", "H010"),
        ("skeletons_vs_snippets", "H013", "H008"),
        ("whole_files_vs_snippets", "H014", "H008"),
        ("role_summaries_vs_snippets", "H015", "H008"),
        ("skeletons_vs_whole_files", "H013", "H014"),
    ]
    contrasts = []
    for label, left, right in definitions:
        result = paired_result(recall[left], recall[right], f"E02:contrast:{label}")
        token_result = paired_result(tokens[left], tokens[right], f"E02:tokens:{label}", test=False)
        contrasts.append(
            {
                "contrast": label, "left": left, "right": right, **result,
                "mean_total_token_difference": token_result["mean_difference"],
                "token_difference_ci95_low": token_result["ci95_low"],
                "token_difference_ci95_high": token_result["ci95_high"],
            }
        )
    holm(contrasts)
    return {"runs": frame, "summary": summary, "contrasts": contrasts}


def analyze_e03(rows: list[dict[str, Any]], harness_names: dict[str, str]) -> dict[str, Any]:
    records = []
    for row in rows:
        final = row["final"]
        records.append(
            {
                "task_id": row["task_id"], "harness_id": row["harness_id"],
                "resolved_at_1": bool(final["resolved_at_1"]), "failure_stage": final["failure_stage"],
                "patch_applied": bool(final["patch_applied"]), "fail_to_pass": bool(final["fail_to_pass"]),
                "pass_to_pass": bool(final["pass_to_pass"]),
                "file_recall_at_10": final["localization_metrics"]["file_recall_at_10"],
                "mrr": final["localization_metrics"]["mrr"], "context_tokens": final["context_tokens"],
                "elapsed_seconds": final["elapsed_seconds"],
                "protocol_violation": final.get("protocol_violation") is not None,
                **usage_totals(final.get("usage", {})),
            }
        )
    frame = pd.DataFrame(records).sort_values(["task_id", "harness_id"])
    summary_rows = []
    for harness_id, group in frame.groupby("harness_id", sort=True):
        success = group["resolved_at_1"].astype(float).to_numpy()
        low, high = bootstrap_mean(success, f"E03:{harness_id}:resolved")
        summary_rows.append(
            {
                "harness_id": harness_id, "harness_name": harness_names[harness_id], "n_tasks": len(group),
                "resolved_count": int(group["resolved_at_1"].sum()), "resolved_rate": float(success.mean()),
                "resolved_ci_low": low, "resolved_ci_high": high,
                "patch_apply_rate": float(group["patch_applied"].mean()),
                "fail_to_pass_rate": float(group["fail_to_pass"].mean()),
                "pass_to_pass_rate": float(group["pass_to_pass"].mean()),
                "localization_recall_at_10_mean": float(group["file_recall_at_10"].mean()),
                "protocol_violations": int(group["protocol_violation"].sum()),
                "context_tokens_median": float(group["context_tokens"].median()),
                "total_tokens_median": float(group["total_tokens"].median()),
                "elapsed_seconds_median": float(group["elapsed_seconds"].median()),
            }
        )
    summary = pd.DataFrame(summary_rows)
    success = frame.pivot(index="task_id", columns="harness_id", values="resolved_at_1").astype(int)
    contrasts = []
    for harness_id in sorted(set(frame["harness_id"]) - {"H000"}):
        differences = (success[harness_id] - success["H000"]).to_numpy(dtype=float)
        discordant = int(np.count_nonzero(differences))
        treatment_only = int(np.sum(differences == 1))
        control_only = int(np.sum(differences == -1))
        p_value = 1.0 if not discordant else float(binomtest(treatment_only, discordant, 0.5).pvalue)
        low, high = bootstrap_mean(differences, f"E03:{harness_id}:risk_difference")
        treatment_rate = float(success[harness_id].mean())
        control_rate = float(success["H000"].mean())
        contrasts.append(
            {
                "contrast": f"{harness_id}_vs_H000", "left": harness_id, "right": "H000",
                "n_tasks": len(success), "risk_difference": treatment_rate - control_rate,
                "risk_difference_ci95_low": low, "risk_difference_ci95_high": high,
                "treatment_rate": treatment_rate, "control_rate": control_rate,
                "risk_ratio_haldane_anscombe": (treatment_rate + 0.05) / (control_rate + 0.05),
                "discordant_pairs": discordant, "treatment_only_successes": treatment_only,
                "control_only_successes": control_only, "p_exact_mcnemar": p_value,
            }
        )
    holm(contrasts, p_key="p_exact_mcnemar")
    failure = (
        frame.groupby(["harness_id", "failure_stage"], dropna=False).size().rename("count").reset_index()
    )
    localized = frame["file_recall_at_10"] > 0
    association = {
        "localized_cells": int(localized.sum()),
        "localized_successes": int(frame.loc[localized, "resolved_at_1"].sum()),
        "localized_success_rate": float(frame.loc[localized, "resolved_at_1"].mean()),
        "nonlocalized_cells": int((~localized).sum()),
        "nonlocalized_successes": int(frame.loc[~localized, "resolved_at_1"].sum()),
        "nonlocalized_success_rate": float(frame.loc[~localized, "resolved_at_1"].mean()),
        "interpretation": "descriptive pooled association; not a mediation estimate",
    }
    return {"runs": frame, "summary": summary, "contrasts": contrasts, "failure": failure, "association": association}


def analyze_e04(rows: list[dict[str, Any]], harness_names: dict[str, str]) -> dict[str, Any]:
    records = []
    for row in rows:
        final = row["final"]
        records.append(
            {
                "task_id": final["task_id"], "harness_id": final["harness_id"], "seed": final["seed"],
                "query_source": final["query_source"],
                "baseline_recall_at_10": final["baseline"]["metrics"]["file_recall_at_10"],
                "baseline_mrr": final["baseline"]["metrics"]["mrr"],
                "stale_shared_key": final["stale_index"]["shared_key"],
                "stale_recall_at_10": final["stale_index"]["metrics"]["file_recall_at_10"],
                "stale_mrr": final["stale_index"]["metrics"]["mrr"],
                "stale_recall_delta": final["stale_index"]["shift"]["file_recall_at_10_delta"],
                "stale_rank_delta_censored": final["stale_index"]["shift"]["first_gold_rank_displacement_censored"],
                "severity": final["plausible_distractors"]["severity"],
                "distractor_recall_at_10": final["plausible_distractors"]["metrics"]["file_recall_at_10"],
                "distractor_mrr": final["plausible_distractors"]["metrics"]["mrr"],
                "distractor_recall_delta": final["plausible_distractors"]["shift"]["file_recall_at_10_delta"],
                "distractor_mrr_delta": final["plausible_distractors"]["shift"]["mrr_delta"],
                "distractor_rank_delta_censored": final["plausible_distractors"]["shift"]["first_gold_rank_displacement_censored"],
                "synthetic_in_top_10_count": final["plausible_distractors"]["synthetic_in_top_10_count"],
            }
        )
    frame = pd.DataFrame(records).sort_values(["task_id", "harness_id", "seed"])
    stale_frame = frame.drop_duplicates("stale_shared_key")
    stale_summary = []
    tests = []
    for harness_id, group in stale_frame.groupby("harness_id", sort=True):
        paired = paired_result(
            group.set_index("task_id")["stale_recall_at_10"],
            group.set_index("task_id")["baseline_recall_at_10"],
            f"E04:stale:{harness_id}",
        )
        stale_summary.append(
            {
                "harness_id": harness_id, "harness_name": harness_names[harness_id],
                "baseline_recall_at_10_mean": float(group["baseline_recall_at_10"].mean()),
                "stale_recall_at_10_mean": float(group["stale_recall_at_10"].mean()),
                "mean_rank_displacement_censored": float(group["stale_rank_delta_censored"].mean()),
                **paired,
            }
        )
        tests.append({"scenario": "stale_parent", "harness_id": harness_id, **paired})
    distractor_summary = []
    for (harness_id, severity), group in frame.groupby(["harness_id", "severity"], sort=True):
        paired = paired_result(
            group.set_index("task_id")["distractor_recall_at_10"],
            group.set_index("task_id")["baseline_recall_at_10"],
            f"E04:distractor:{harness_id}:{severity}",
        )
        distractor_summary.append(
            {
                "harness_id": harness_id, "harness_name": harness_names[harness_id], "severity": int(severity),
                "baseline_recall_at_10_mean": float(group["baseline_recall_at_10"].mean()),
                "distractor_recall_at_10_mean": float(group["distractor_recall_at_10"].mean()),
                "distractor_mrr_delta_mean": float(group["distractor_mrr_delta"].mean()),
                "mean_rank_displacement_censored": float(group["distractor_rank_delta_censored"].mean()),
                "synthetic_top_10_count_mean": float(group["synthetic_in_top_10_count"].mean()),
                "any_synthetic_top_10_rate": float((group["synthetic_in_top_10_count"] > 0).mean()),
                **paired,
            }
        )
        tests.append({"scenario": f"distractors_{severity}", "harness_id": harness_id, **paired})
    holm(tests)
    adjusted = {(row["scenario"], row["harness_id"]): row["p_holm"] for row in tests}
    for row in stale_summary:
        row["p_holm_family_12"] = adjusted[("stale_parent", row["harness_id"])]
    for row in distractor_summary:
        row["p_holm_family_12"] = adjusted[(f"distractors_{row['severity']}", row["harness_id"])]
    return {
        "runs": frame, "stale_units": stale_frame, "stale_summary": pd.DataFrame(stale_summary),
        "distractor_summary": pd.DataFrame(distractor_summary), "tests": tests,
    }


def analyze_e05(rows: list[dict[str, Any]]) -> dict[str, Any]:
    records = []
    for row in rows:
        final = row["final"]
        records.append({"task_id": row["task_id"], "harness_id": row["harness_id"], **final})
    frame = pd.DataFrame(records).sort_values(["task_id", "harness_id", "backend_id", "seed"])
    collapsed = (
        frame.groupby(["task_id", "harness_id", "backend_id"], as_index=False)
        .agg(
            backend_recall_at_10_vs_flat=("backend_recall_at_10_vs_flat", "mean"),
            file_recall_at_10=("file_recall_at_10", "mean"),
            index_build_seconds=("index_build_seconds", "median"),
            index_ram_bytes_delta=("index_ram_bytes_delta", "median"),
            index_disk_bytes=("index_disk_bytes", "median"),
            query_p50_ms=("query_p50_ms", "median"),
            query_p95_ms=("query_p95_ms", "median"),
        )
    )
    names = {"B001": "FAISS FlatIP", "B002": "FAISS HNSW", "B003": "sqlite-vec"}
    summary_rows = []
    for backend, group in collapsed.groupby("backend_id", sort=True):
        summary_rows.append(
            {
                "backend_id": backend, "backend_name": names[backend], "n_task_harness_units": len(group),
                "ranking_recall_at_10_vs_flat_mean": float(group["backend_recall_at_10_vs_flat"].mean()),
                "ranking_recall_at_10_vs_flat_min": float(group["backend_recall_at_10_vs_flat"].min()),
                "file_recall_at_10_mean": float(group["file_recall_at_10"].mean()),
                "index_build_seconds_median": float(group["index_build_seconds"].median()),
                "index_build_seconds_p95": float(group["index_build_seconds"].quantile(0.95)),
                "query_p50_ms_median": float(group["query_p50_ms"].median()),
                "query_p95_ms_median": float(group["query_p95_ms"].median()),
                "index_ram_mib_median": float(group["index_ram_bytes_delta"].median() / 2**20),
                "index_disk_mib_median": float(group["index_disk_bytes"].median() / 2**20),
            }
        )
    return {"runs": frame, "collapsed": collapsed, "summary": pd.DataFrame(summary_rows)}


def repository_statistics(root: Path, representative_commit: str) -> dict[str, Any]:
    repository = root / "data" / "repos" / "gitlab-runner"
    paths_raw = subprocess.run(
        ["git", "ls-tree", "-r", "--name-only", "-z", representative_commit],
        cwd=repository, check=True, capture_output=True, timeout=60,
    ).stdout
    paths = [item for item in paths_raw.decode("utf-8", "surrogateescape").split("\0") if item]
    tokenizer = QwenTokenCounter()
    stats = {
        "representative_commit": representative_commit, "tracked_files": len(paths), "text_files": 0,
        "binary_files": 0, "text_bytes": 0, "text_lines": 0, "text_tokens": 0,
        "go_files": 0, "go_bytes": 0, "go_lines": 0, "go_tokens": 0,
        "tokenizer_sha256": tokenizer.sha256,
    }
    for path in paths:
        payload = subprocess.run(
            ["git", "show", f"{representative_commit}:{path}"], cwd=repository,
            check=True, capture_output=True, timeout=60,
        ).stdout
        if b"\0" in payload[:8192]:
            stats["binary_files"] += 1
            continue
        text = payload.decode("utf-8", "replace")
        tokens = tokenizer.count(text)
        stats["text_files"] += 1
        stats["text_bytes"] += len(payload)
        stats["text_lines"] += len(text.splitlines())
        stats["text_tokens"] += tokens
        if path.endswith(".go"):
            stats["go_files"] += 1
            stats["go_bytes"] += len(payload)
            stats["go_lines"] += len(text.splitlines())
            stats["go_tokens"] += tokens
    stats["text_tokens_over_model_context"] = stats["text_tokens"] / 262_144
    stats["go_tokens_over_model_context"] = stats["go_tokens"] / 262_144
    return stats


def task_statistics(root: Path) -> pd.DataFrame:
    tasks = load_tasks(root)
    e03 = set(
        line.strip()
        for line in (root / "tasks" / "splits" / "end_to_end_confirmatory.txt").read_text().splitlines()
        if line.strip() and not line.startswith("#")
    )
    records = []
    for task_id in sorted(item for item in tasks if item.startswith("TASK_CR_")):
        task = tasks[task_id]
        patch = (root / "tasks" / task.gold_patch).read_text(encoding="utf-8")
        added = sum(line.startswith("+") and not line.startswith("+++") for line in patch.splitlines())
        deleted = sum(line.startswith("-") and not line.startswith("---") for line in patch.splitlines())
        records.append(
            {
                "task_id": task_id, "base_commit": task.base_commit, "gold_commit": task.gold_commit,
                "gold_file_count": len(task.gold_files), "gold_symbol_count": len(task.gold_symbols),
                "patch_added_lines": added, "patch_deleted_lines": deleted,
                "difficulty": task.difficulty, "e03_eligible": task_id in e03,
            }
        )
    return pd.DataFrame(records)


def save_figure(fig: plt.Figure, output: Path, name: str) -> None:
    fig.savefig(output / f"{name}.pdf", bbox_inches="tight")
    fig.savefig(output / f"{name}.png", dpi=220, bbox_inches="tight")
    plt.close(fig)


def make_figures(output: Path, e01: dict[str, Any], e02: dict[str, Any], e03: dict[str, Any], e04: dict[str, Any], e05: dict[str, Any]) -> None:
    plt.rcParams.update({"font.size": 9, "axes.spines.top": False, "axes.spines.right": False})

    table = e01["summary"].sort_values("harness_id")
    fig, ax = plt.subplots(figsize=(7.2, 3.4))
    values = table["file_recall_at_10_mean"].to_numpy()
    low = values - table["file_recall_at_10_ci_low"].to_numpy()
    high = table["file_recall_at_10_ci_high"].to_numpy() - values
    ax.bar(table["harness_id"], values, color="#3b82b8", yerr=np.vstack([low, high]), capsize=3)
    ax.set(ylabel="Mean file recall@10", xlabel="Harness", ylim=(0, 1.06), title="Static retrieval quality (E01; 95% task bootstrap CI)")
    save_figure(fig, output, "figure_e01_retrieval")

    table = e02["summary"].sort_values("harness_id")
    fig, ax = plt.subplots(figsize=(6.2, 4.0))
    ax.scatter(table["total_tokens_mean"], table["file_recall_at_10_mean"], s=48, color="#8e5eb7")
    for row in table.itertuples():
        ax.annotate(row.harness_id, (row.total_tokens_mean, row.file_recall_at_10_mean), xytext=(4, 4), textcoords="offset points")
    ax.set(xlabel="Mean LM Studio total tokens", ylabel="Mean selected-file recall@10", title="Localization quality–cost trade-off (E02)")
    save_figure(fig, output, "figure_e02_quality_cost")

    table = e03["summary"].sort_values("harness_id")
    fig, ax = plt.subplots(figsize=(7.2, 3.4))
    values = table["resolved_rate"].to_numpy()
    low = values - table["resolved_ci_low"].to_numpy()
    high = table["resolved_ci_high"].to_numpy() - values
    ax.bar(table["harness_id"], values, color="#c46a4a", yerr=np.vstack([low, high]), capsize=3)
    ax.set(ylabel="Resolved@1", xlabel="Harness", ylim=(0, max(0.45, float((values + high).max()) + 0.05)), title="End-to-end repair (E03; n=10 paired tasks)")
    save_figure(fig, output, "figure_e03_resolution")

    table = e04["distractor_summary"]
    fig, ax = plt.subplots(figsize=(6.2, 3.8))
    colors = {"H000": "#333333", "H008": "#3b82b8", "H010": "#c46a4a"}
    for harness, group in table.groupby("harness_id"):
        group = group.sort_values("severity")
        values = group["mean_difference"].to_numpy()
        low = values - group["ci95_low"].to_numpy()
        high = group["ci95_high"].to_numpy() - values
        ax.errorbar(
            group["severity"],
            values,
            yerr=np.vstack([low, high]),
            marker="o",
            capsize=3,
            label=harness,
            color=colors[harness],
        )
    ax.axhline(0, color="#999999", linewidth=0.8)
    ax.set(xlabel="Synthetic distractor files", ylabel="Change in file recall@10", title="Robustness to nested plausible distractors (E04)", xticks=[1, 5, 10])
    ax.legend(frameon=False)
    save_figure(fig, output, "figure_e04_distractors")

    table = e05["summary"]
    fig, ax = plt.subplots(figsize=(5.8, 3.8))
    sizes = 45 + 7 * table["index_disk_mib_median"].to_numpy()
    ax.scatter(table["query_p50_ms_median"], table["ranking_recall_at_10_vs_flat_mean"], s=sizes, color=["#3b82b8", "#59a14f", "#c46a4a"])
    for row in table.itertuples():
        ax.annotate(row.backend_id, (row.query_p50_ms_median, row.ranking_recall_at_10_vs_flat_mean), xytext=(5, 4), textcoords="offset points")
    ax.set(xlabel="Median query p50 (ms)", ylabel="Top-10 ranking recall vs FlatIP", title="Dense backend quality–latency trade-off (E05)")
    save_figure(fig, output, "figure_e05_backends")


def save_frames(output: Path, values: dict[str, dict[str, Any]]) -> None:
    for experiment, collection in values.items():
        for name, value in collection.items():
            if isinstance(value, pd.DataFrame):
                value.to_csv(output / f"{experiment.lower()}_{name}.csv", index=False)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[1])
    args = parser.parse_args()
    root = args.root.resolve()
    revision = git(root, "rev-parse", "HEAD")
    dirty = bool(git(root, "status", "--porcelain"))
    if dirty:
        raise AnalysisFailure("analysis must run from a clean committed code revision")
    output = root / "results" / "derived" / f"confirmatory_analysis_{revision[:12]}"
    output.mkdir(parents=True, exist_ok=False)

    raw = {experiment: load_raw(root, experiment) for experiment in ("E01", "E02", "E03", "E04", "E05")}
    audit = audit_artifacts(root, raw)
    selected = audit.pop("selected")
    harnesses = load_harnesses(root)
    harness_names = {key: value.name for key, value in harnesses.items()}
    e01 = analyze_e01(selected["E01"], harness_names)
    e02 = analyze_e02(selected["E02"], harness_names)
    e03 = analyze_e03(selected["E03"], harness_names)
    e04 = analyze_e04(selected["E04"], harness_names)
    e05 = analyze_e05(selected["E05"])
    task_table = task_statistics(root)
    repo_stats = repository_statistics(root, task_table.iloc[0]["base_commit"])

    collections = {"E01": e01, "E02": e02, "E03": e03, "E04": e04, "E05": e05}
    save_frames(output, collections)
    task_table.to_csv(output / "task_characteristics.csv", index=False)
    make_figures(output, e01, e02, e03, e04, e05)

    statistical = {
        "schema_version": 1,
        "analysis_revision": revision,
        "analysis_seed": ANALYSIS_SEED,
        "bootstrap_repetitions": BOOTSTRAPS,
        "confidence_interval": "percentile task bootstrap",
        "continuous_p_values": "exact two-sided paired sign-flip randomization tests",
        "binary_p_values": "exact two-sided McNemar/binomial tests",
        "multiplicity": "Holm correction within each declared experiment family",
        "artifact_audit": audit,
        "repository": repo_stats,
        "E01": {"contrasts": e01["contrasts"], "factorial": e01["factorial"], "mixed_model": e01["mixed_model"]},
        "E02": {"contrasts": e02["contrasts"]},
        "E03": {"contrasts": e03["contrasts"], "localization_association": e03["association"]},
        "E04": {"tests": e04["tests"]},
        "E05": {"backend_summary": e05["summary"].to_dict(orient="records")},
    }
    write_json(output / "statistical_results.json", statistical)
    write_json(output / "artifact_audit.json", audit)
    write_json(output / "repository_statistics.json", repo_stats)
    paper_tables = {
        "task_characteristics": task_table.to_dict(orient="records"),
        "E01_summary": e01["summary"].to_dict(orient="records"),
        "E01_contrasts": e01["contrasts"],
        "E02_summary": e02["summary"].to_dict(orient="records"),
        "E02_contrasts": e02["contrasts"],
        "E03_summary": e03["summary"].to_dict(orient="records"),
        "E03_failures": e03["failure"].to_dict(orient="records"),
        "E04_stale": e04["stale_summary"].to_dict(orient="records"),
        "E04_distractors": e04["distractor_summary"].to_dict(orient="records"),
        "E05_summary": e05["summary"].to_dict(orient="records"),
    }
    write_json(output / "paper_tables.json", paper_tables)
    summary = {
        "output": str(output),
        "confirmatory_cells": audit["confirmatory_cells"],
        "repository_text_tokens": repo_stats["text_tokens"],
        "e01_best_recall_harness": e01["summary"].sort_values("file_recall_at_10_mean").iloc[-1]["harness_id"],
        "e02_best_recall_harness": e02["summary"].sort_values("file_recall_at_10_mean").iloc[-1]["harness_id"],
        "e03_total_resolved": int(e03["runs"]["resolved_at_1"].sum()),
        "e04_cells": len(e04["runs"]),
        "e05_cells": len(e05["runs"]),
    }
    write_json(output / "analysis_summary.json", summary)
    print(json.dumps(summary, indent=2))
    return 0


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