#!/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())