| |
| """Fail-closed analysis for Study 5 repository-scale harness experiments. |
| |
| The task is the independent sampling unit. All contrasts first average over |
| models within task, then use a task-cluster bootstrap and a Monte-Carlo exact |
| sign-flip reference distribution. Raw cells are accepted only when they match |
| the frozen cell manifests and final experiment reports exactly. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| from collections import Counter, defaultdict |
| import csv |
| from hashlib import sha256 |
| import json |
| import math |
| from pathlib import Path |
| import random |
| import statistics |
| from typing import Any, Callable, Iterable, Sequence |
|
|
| import matplotlib.pyplot as plt |
| import numpy as np |
|
|
| from agent_harness.specs import load_harnesses, load_tasks |
|
|
|
|
| EXPERIMENTS = {"E13": 1440, "E14": 540, "E15": 540, "E16": 306} |
| REVISIONS = { |
| "E13": "7f4de67853deab34aca5a9ceaaf7e9f85b901088", |
| "E14": "7f4de67853deab34aca5a9ceaaf7e9f85b901088", |
| "E15": "7f4de67853deab34aca5a9ceaaf7e9f85b901088", |
| "E16": "6f8c86a1a0737db56c119f90b7e9516344d13dc3", |
| } |
| MODELS = ("M002", "M003", "M004") |
| BOOTSTRAPS = 20_000 |
| BOOTSTRAP_SEED = 20260720 |
| SIGN_FLIPS = 20_000 |
| EPS = 1e-15 |
|
|
|
|
| class Study5AnalysisError(RuntimeError): |
| pass |
|
|
|
|
| def canonical_hash(value: Any) -> str: |
| return sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest() |
|
|
|
|
| def raw_digest(root: Path, experiment: str) -> str: |
| digest = sha256() |
| for path in sorted((root / "results" / "raw" / experiment).glob("*/*/*/final_metrics.json")): |
| digest.update(str(path.relative_to(root)).encode()) |
| digest.update(b"\0") |
| digest.update(path.read_bytes()) |
| digest.update(b"\0") |
| return digest.hexdigest() |
|
|
|
|
| def holm(values: Sequence[float]) -> list[float]: |
| order = sorted(range(len(values)), key=values.__getitem__) |
| adjusted = [1.0] * len(values) |
| running = 0.0 |
| for rank, index in enumerate(order): |
| running = max(running, min(1.0, (len(values) - rank) * values[index])) |
| adjusted[index] = running |
| return adjusted |
|
|
|
|
| def cluster_bootstrap(values: Sequence[float], seed_offset: int = 0) -> tuple[float, float]: |
| if not values: |
| return math.nan, math.nan |
| generator = random.Random(BOOTSTRAP_SEED + seed_offset) |
| n = len(values) |
| draws = sorted( |
| statistics.fmean(values[generator.randrange(n)] for _ in range(n)) |
| for _ in range(BOOTSTRAPS) |
| ) |
| return draws[int(0.025 * BOOTSTRAPS)], draws[int(0.975 * BOOTSTRAPS) - 1] |
|
|
|
|
| def sign_flip(values: Sequence[float], seed_offset: int = 0) -> float: |
| if not values or all(abs(value) < EPS for value in values): |
| return 1.0 |
| observed = abs(statistics.fmean(values)) |
| generator = random.Random(BOOTSTRAP_SEED + 100_000 + seed_offset) |
| extreme = 0 |
| for _ in range(SIGN_FLIPS): |
| value = abs(statistics.fmean( |
| item if generator.getrandbits(1) else -item for item in values |
| )) |
| extreme += value + EPS >= observed |
| return (extreme + 1) / (SIGN_FLIPS + 1) |
|
|
|
|
| def exact_mcnemar(left: Sequence[int], right: Sequence[int]) -> tuple[int, int, float]: |
| n10 = sum(a == 1 and b == 0 for a, b in zip(left, right)) |
| n01 = sum(a == 0 and b == 1 for a, b in zip(left, right)) |
| n = n10 + n01 |
| if not n: |
| return n10, n01, 1.0 |
| tail = sum(math.comb(n, k) for k in range(min(n10, n01) + 1)) / 2**n |
| return n10, n01, min(1.0, 2 * tail) |
|
|
|
|
| def contrast_record(name: str, task_effects: Sequence[float], seed_offset: int = 0) -> dict[str, Any]: |
| low, high = cluster_bootstrap(task_effects, seed_offset) |
| return { |
| "contrast": name, |
| "independent_tasks": len(task_effects), |
| "estimate": statistics.fmean(task_effects), |
| "cluster_bootstrap_ci_low": low, |
| "cluster_bootstrap_ci_high": high, |
| "sign_flip_p": sign_flip(task_effects, seed_offset), |
| } |
|
|
|
|
| def _mechanism(trajectory: Path, gold_files: set[str]) -> dict[str, int]: |
| retrieved: set[str] = set() |
| read: set[str] = set() |
| accepted = False |
| for line in trajectory.read_text(encoding="utf-8").splitlines(): |
| event = json.loads(line) |
| payload = event.get("payload", {}) |
| if event.get("event_type") == "edit" and payload.get("accepted"): |
| accepted = True |
| if accepted: |
| continue |
| if event.get("event_type") == "retrieval_candidate" and payload.get("path"): |
| retrieved.add(str(payload["path"])) |
| if event.get("event_type") == "file_read" and payload.get("path"): |
| read.add(str(payload["path"])) |
| return { |
| "gold_retrieved_before_edit": int(bool(retrieved & gold_files)), |
| "gold_read_before_edit": int(bool(read & gold_files)), |
| "unique_retrieved_before_edit": len(retrieved), |
| "unique_read_before_edit": len(read), |
| } |
|
|
|
|
| def _complete_report(root: Path, experiment: str, expected: int) -> tuple[Path, dict[str, Any]]: |
| candidates: list[tuple[Path, dict[str, Any]]] = [] |
| for path in sorted((root / "results" / "reports").glob(f"{experiment}_*.json")): |
| if path.name.endswith("_progress.json"): |
| continue |
| report = json.loads(path.read_text(encoding="utf-8")) |
| if report.get("run_count") == expected: |
| candidates.append((path, report)) |
| if len(candidates) != 1: |
| raise Study5AnalysisError( |
| f"{experiment} requires exactly one complete final report; found {len(candidates)}" |
| ) |
| return candidates[0] |
|
|
|
|
| def discover(root: Path) -> tuple[list[dict[str, Any]], dict[str, Any]]: |
| tasks = load_tasks(root) |
| all_rows: list[dict[str, Any]] = [] |
| audit: dict[str, Any] = {"schema_version": 1, "experiments": {}} |
| selection = json.loads((root / "configs" / "study5" / "E16_selection.json").read_text()) |
| selection_check = dict(selection) |
| expected_selection_hash = selection_check.pop("selection_sha256") |
| if canonical_hash(selection_check) != expected_selection_hash: |
| raise Study5AnalysisError("E16 selection hash mismatch") |
|
|
| for experiment, expected_count in EXPERIMENTS.items(): |
| manifest_path = root / "configs" / "study5" / f"{experiment}_cells.json" |
| frozen = json.loads(manifest_path.read_text(encoding="utf-8")) |
| frozen_check = dict(frozen) |
| expected_design_hash = frozen_check.pop("design_sha256") |
| if canonical_hash(frozen_check) != expected_design_hash: |
| raise Study5AnalysisError(f"{experiment} frozen design hash mismatch") |
| expected = { |
| (cell["task_id"], cell["harness_id"], cell["interface_id"], cell["model_id"]) |
| for cell in frozen["cells"] |
| } |
| if len(expected) != expected_count or frozen["planned_cells"] != expected_count: |
| raise Study5AnalysisError(f"{experiment} frozen grid count mismatch") |
| final_paths = sorted((root / "results" / "raw" / experiment).glob("*/*/*/final_metrics.json")) |
| if len(final_paths) != expected_count: |
| raise Study5AnalysisError( |
| f"{experiment} requires {expected_count} final metrics, observed {len(final_paths)}" |
| ) |
| seen: set[tuple[str, str, str, str]] = set() |
| revisions: Counter[str] = Counter() |
| for final_path in final_paths: |
| directory = final_path.parent |
| required = [ |
| directory / "run_manifest.json", directory / "trajectory.jsonl", |
| directory / "messages.json", directory / "model.patch", directory / "validation.json", |
| ] |
| if not all(path.is_file() for path in required): |
| raise Study5AnalysisError(f"incomplete raw cell: {directory}") |
| final = json.loads(final_path.read_text(encoding="utf-8")) |
| run_manifest = json.loads(required[0].read_text(encoding="utf-8")) |
| identity = run_manifest["identity"] |
| key = ( |
| identity["task_id"], final["retrieval_harness_id"], |
| final["edit_interface_id"], identity["model_id"], |
| ) |
| if key not in expected or key in seen: |
| raise Study5AnalysisError(f"unexpected or duplicate {experiment} identity: {key}") |
| if identity["harness_id"] != f"{key[1]}__{key[2]}": |
| raise Study5AnalysisError(f"composite treatment mismatch: {directory}") |
| if identity["context_budget"] != 65536 or identity["seed"] != 0: |
| raise Study5AnalysisError(f"inference identity drift: {directory}") |
| if any(len(item.get("after_instances", [])) != 1 for item in final["residency_transitions"]): |
| raise Study5AnalysisError(f"non-exclusive model residency: {directory}") |
| seen.add(key) |
| revisions[identity["code_revision"]] += 1 |
| task = tasks[identity["task_id"]] |
| mechanism = _mechanism(required[1], set(task.gold_files)) |
| all_rows.append({ |
| "experiment_id": experiment, |
| "run_id": run_manifest["run_id"], |
| "task_id": identity["task_id"], |
| "repository_sha": identity["repository_sha"], |
| "model_id": identity["model_id"], |
| "harness_id": final["retrieval_harness_id"], |
| "interface_id": final["edit_interface_id"], |
| "resolved_at_1": int(bool(final["resolved_at_1"])), |
| "accepted_edit_cell": int(bool(final["accepted_edit_cell"])), |
| "applicable_final_patch": int(bool(final["applicable_final_patch"])), |
| "exact_modified_file_match": int(bool(final["exact_modified_file_match"])), |
| "fail_to_pass": int(bool(final["fail_to_pass"])), |
| "search_gold_any": int(final["search_localization_metrics"]["file_recall_at_10"] > 0), |
| "read_gold_any": int(final["read_localization_metrics"]["file_recall_at_10"] > 0), |
| "search_mrr": float(final["search_localization_metrics"]["mrr"]), |
| "read_mrr": float(final["read_localization_metrics"]["mrr"]), |
| **mechanism, |
| "total_tokens": int(final["usage"]["total_tokens"]), |
| "prompt_tokens": int(final["usage"]["prompt_tokens"]), |
| "completion_tokens": int(final["usage"]["completion_tokens"]), |
| "elapsed_seconds": float(final["elapsed_seconds"]), |
| "model_elapsed_seconds": float(final["model_elapsed_seconds"]), |
| "model_switch_seconds": float(final["model_switch_seconds"]), |
| "model_switch_count": int(final["model_switch_count"]), |
| "tool_calls": int(final["tool_calls"]), |
| "protocol_violation_count": len(final["protocol_violations"]), |
| "failure_stage": str(final["failure_stage"]), |
| "finished_reason": str(final["finished_reason"]), |
| }) |
| if seen != expected or set(revisions) != {REVISIONS[experiment]}: |
| raise Study5AnalysisError(f"{experiment} grid/revision mismatch: {revisions}") |
| report_path, report = _complete_report(root, experiment, expected_count) |
| counts = { |
| "run_count": expected_count, |
| "accepted_edit_count": sum(row["accepted_edit_cell"] for row in all_rows if row["experiment_id"] == experiment), |
| "applicable_patch_count": sum(row["applicable_final_patch"] for row in all_rows if row["experiment_id"] == experiment), |
| "resolved_count": sum(row["resolved_at_1"] for row in all_rows if row["experiment_id"] == experiment), |
| } |
| if report.get("code_revision") != REVISIONS[experiment]: |
| raise Study5AnalysisError(f"{experiment} final report revision mismatch") |
| if any(report.get(key) != value for key, value in counts.items()): |
| raise Study5AnalysisError(f"{experiment} raw ledger/final report count mismatch") |
| if report.get("manifest_sha256") != expected_design_hash: |
| raise Study5AnalysisError(f"{experiment} report/design manifest hash mismatch") |
| digest = raw_digest(root, experiment) |
| if experiment in ("E13", "E15") and digest != selection["source_raw_sha256"][experiment]: |
| raise Study5AnalysisError(f"{experiment} raw hash no longer matches E16 screening freeze") |
| audit["experiments"][experiment] = { |
| "cells": expected_count, |
| "unique_identities": len(seen), |
| "code_revision": REVISIONS[experiment], |
| "design_sha256": expected_design_hash, |
| "raw_final_metrics_sha256": digest, |
| "final_report": str(report_path.relative_to(root)), |
| "final_report_sha256": sha256(report_path.read_bytes()).hexdigest(), |
| **counts, |
| } |
| audit["total_cells"] = len(all_rows) |
| audit["selection_sha256"] = expected_selection_hash |
| audit["passed"] = len(all_rows) == sum(EXPERIMENTS.values()) |
| audit["audit_sha256"] = canonical_hash(audit) |
| return all_rows, audit |
|
|
|
|
| def summarize(rows: Sequence[dict[str, Any]], keys: Sequence[str]) -> list[dict[str, Any]]: |
| grouped: dict[tuple[Any, ...], list[dict[str, Any]]] = defaultdict(list) |
| for row in rows: |
| grouped[tuple(row[key] for key in keys)].append(row) |
| output = [] |
| for group, values in sorted(grouped.items()): |
| record = dict(zip(keys, group)) |
| record.update({ |
| "cells": len(values), |
| "tasks": len({row["task_id"] for row in values}), |
| "search_gold_rate": statistics.fmean(row["search_gold_any"] for row in values), |
| "read_gold_rate": statistics.fmean(row["read_gold_any"] for row in values), |
| "accepted_edit_rate": statistics.fmean(row["accepted_edit_cell"] for row in values), |
| "applicable_patch_rate": statistics.fmean(row["applicable_final_patch"] for row in values), |
| "resolved_rate": statistics.fmean(row["resolved_at_1"] for row in values), |
| "mean_total_tokens": statistics.fmean(row["total_tokens"] for row in values), |
| "mean_wall_seconds": statistics.fmean(row["elapsed_seconds"] for row in values), |
| "mean_model_switch_seconds": statistics.fmean(row["model_switch_seconds"] for row in values), |
| "mean_protocol_violations": statistics.fmean(row["protocol_violation_count"] for row in values), |
| }) |
| output.append(record) |
| return output |
|
|
|
|
| def _task_factor_effects( |
| rows: Sequence[dict[str, Any]], endpoint: str, contrast: Callable[[dict[str, int]], int] |
| ) -> list[float]: |
| grouped: dict[str, dict[int, list[float]]] = defaultdict(lambda: defaultdict(list)) |
| for row in rows: |
| code = {"L": int(row["L"]), "S": int(row["S"]), "D": int(row["D"])} |
| grouped[row["task_id"]][contrast(code)].append(float(row[endpoint])) |
| effects = [] |
| for task in sorted(grouped): |
| if set(grouped[task]) != {-1, 1}: |
| raise Study5AnalysisError(f"incomplete factorial contrast for task {task}") |
| effects.append(statistics.fmean(grouped[task][1]) - statistics.fmean(grouped[task][-1])) |
| return effects |
|
|
|
|
| def e13_factorial(rows: Sequence[dict[str, Any]], harness_specs: dict[str, Any]) -> dict[str, Any]: |
| selected = [dict(row) for row in rows if row["experiment_id"] == "E13"] |
| for row in selected: |
| spec = harness_specs[row["harness_id"]] |
| row.update(L=int(spec.lexical), S=int(spec.syntax == "tree_sitter"), D=int(spec.dense)) |
| contrasts: dict[str, Callable[[dict[str, int]], int]] = { |
| "L": lambda x: 1 if x["L"] else -1, |
| "S": lambda x: 1 if x["S"] else -1, |
| "D": lambda x: 1 if x["D"] else -1, |
| "LxS": lambda x: (1 if x["L"] else -1) * (1 if x["S"] else -1), |
| "LxD": lambda x: (1 if x["L"] else -1) * (1 if x["D"] else -1), |
| "SxD": lambda x: (1 if x["S"] else -1) * (1 if x["D"] else -1), |
| "LxSxD": lambda x: (1 if x["L"] else -1) * (1 if x["S"] else -1) * (1 if x["D"] else -1), |
| } |
| endpoints = ("resolved_at_1", "accepted_edit_cell", "applicable_final_patch", "search_gold_any", "read_gold_any") |
| output: dict[str, Any] = {"estimand": "marginal risk difference averaged over models and remaining factors", "endpoints": {}} |
| offset = 0 |
| for endpoint in endpoints: |
| records = [] |
| for name, function in contrasts.items(): |
| effects = _task_factor_effects(selected, endpoint, function) |
| records.append(contrast_record(f"E13_{name}_{endpoint}", effects, offset)) |
| offset += 1 |
| adjusted = holm([record["sign_flip_p"] for record in records]) |
| for record, value in zip(records, adjusted): |
| record["sign_flip_p_holm_within_endpoint"] = value |
| output["endpoints"][endpoint] = records |
| model_interactions = [] |
| for endpoint in ("resolved_at_1", "accepted_edit_cell", "applicable_final_patch"): |
| for factor in ("L", "S", "D"): |
| function = contrasts[factor] |
| per_model: dict[str, list[float]] = {} |
| for model in MODELS: |
| per_model[model] = _task_factor_effects( |
| [row for row in selected if row["model_id"] == model], endpoint, function |
| ) |
| for left, right in (("M002", "M003"), ("M002", "M004"), ("M003", "M004")): |
| effects = [a - b for a, b in zip(per_model[left], per_model[right])] |
| model_interactions.append(contrast_record( |
| f"E13_{factor}_{endpoint}_{left}_minus_{right}", effects, offset |
| )) |
| offset += 1 |
| adjusted = holm([record["sign_flip_p"] for record in model_interactions]) |
| for record, value in zip(model_interactions, adjusted): |
| record["sign_flip_p_holm"] = value |
| output["model_interactions"] = model_interactions |
| return output |
|
|
|
|
| def _cell_lookup(rows: Sequence[dict[str, Any]], dimensions: Sequence[str]) -> dict[tuple[Any, ...], dict[str, Any]]: |
| lookup = {tuple(row[key] for key in dimensions): row for row in rows} |
| if len(lookup) != len(rows): |
| raise Study5AnalysisError(f"duplicate cells for lookup dimensions {dimensions}") |
| return lookup |
|
|
|
|
| def e14_interactions(rows: Sequence[dict[str, Any]]) -> dict[str, Any]: |
| selected = [row for row in rows if row["experiment_id"] == "E14"] |
| lookup = _cell_lookup(selected, ("task_id", "model_id", "harness_id", "interface_id")) |
| tasks = sorted({row["task_id"] for row in selected}) |
| retrieval_pairs = (("H006", "H000"), ("H007", "H000"), ("H007", "H006")) |
| action_pairs = (("P002", "P001"), ("P003", "P001"), ("P003", "P002")) |
| output: dict[str, Any] = {"estimand": "task-level difference-in-differences averaged over models", "endpoints": {}} |
| offset = 100 |
| for endpoint in ("accepted_edit_cell", "applicable_final_patch", "resolved_at_1"): |
| records = [] |
| for high_h, low_h in retrieval_pairs: |
| for high_p, low_p in action_pairs: |
| effects = [] |
| for task in tasks: |
| model_effects = [] |
| for model in MODELS: |
| def value(h: str, p: str) -> float: |
| return float(lookup[(task, model, h, p)][endpoint]) |
| model_effects.append((value(high_h, high_p) - value(low_h, high_p)) - |
| (value(high_h, low_p) - value(low_h, low_p))) |
| effects.append(statistics.fmean(model_effects)) |
| records.append(contrast_record( |
| f"E14_{high_h}-{low_h}_x_{high_p}-{low_p}_{endpoint}", effects, offset |
| )) |
| offset += 1 |
| adjusted = holm([record["sign_flip_p"] for record in records]) |
| for record, value in zip(records, adjusted): |
| record["sign_flip_p_holm_within_endpoint"] = value |
| output["endpoints"][endpoint] = records |
| return output |
|
|
|
|
| def _paired_harness_effects( |
| selected: Sequence[dict[str, Any]], left: str, right: str, endpoint: str |
| ) -> list[float]: |
| lookup = _cell_lookup(selected, ("task_id", "model_id", "harness_id")) |
| tasks = sorted({row["task_id"] for row in selected}) |
| return [statistics.fmean( |
| float(lookup[(task, model, left)][endpoint]) - float(lookup[(task, model, right)][endpoint]) |
| for model in MODELS |
| ) for task in tasks] |
|
|
|
|
| def e15_blocks(rows: Sequence[dict[str, Any]]) -> dict[str, Any]: |
| selected = [row for row in rows if row["experiment_id"] == "E15"] |
| blocks = { |
| "graph": (("H007", "H008"), ("H009", "H008")), |
| "query_policy": (("H010", "H008"),), |
| "search_interface": (("H011", "H008"),), |
| "query_by_interface": (("H010", "H008"), ("H011", "H008"), ("H012", "H008")), |
| "packing": (("H013", "H008"), ("H014", "H008"), ("H015", "H008")), |
| } |
| endpoints = ( |
| "resolved_at_1", "accepted_edit_cell", "applicable_final_patch", |
| "gold_retrieved_before_edit", "gold_read_before_edit", "total_tokens", |
| "elapsed_seconds", "model_switch_seconds", "protocol_violation_count", |
| ) |
| output: dict[str, Any] = {} |
| offset = 200 |
| for block, pairs in blocks.items(): |
| block_records = [] |
| for endpoint in endpoints: |
| endpoint_records = [] |
| for left, right in pairs: |
| effects = _paired_harness_effects(selected, left, right, endpoint) |
| endpoint_records.append(contrast_record( |
| f"E15_{block}_{left}_minus_{right}_{endpoint}", effects, offset |
| )) |
| offset += 1 |
| adjusted = holm([record["sign_flip_p"] for record in endpoint_records]) |
| for record, value in zip(endpoint_records, adjusted): |
| record["sign_flip_p_holm_within_block_endpoint"] = value |
| block_records.extend(endpoint_records) |
| output[block] = block_records |
| return output |
|
|
|
|
| def rankdata(values: dict[str, float], higher_better: bool = True) -> dict[str, float]: |
| ordered = sorted(values, key=lambda key: (-values[key] if higher_better else values[key], key)) |
| output: dict[str, float] = {} |
| index = 0 |
| while index < len(ordered): |
| end = index + 1 |
| while end < len(ordered) and abs(values[ordered[end]] - values[ordered[index]]) < EPS: |
| end += 1 |
| average_rank = (index + 1 + end) / 2 |
| for key in ordered[index:end]: |
| output[key] = average_rank |
| index = end |
| return output |
|
|
|
|
| def spearman(left: dict[str, float], right: dict[str, float], higher_better: bool = True) -> float: |
| keys = sorted(set(left) & set(right)) |
| a = rankdata({key: left[key] for key in keys}, higher_better) |
| b = rankdata({key: right[key] for key in keys}, higher_better) |
| av = statistics.fmean(a.values()); bv = statistics.fmean(b.values()) |
| numerator = sum((a[key] - av) * (b[key] - bv) for key in keys) |
| denominator = math.sqrt(sum((a[key] - av) ** 2 for key in keys) * sum((b[key] - bv) ** 2 for key in keys)) |
| return numerator / denominator if denominator else math.nan |
|
|
|
|
| def pareto(summary: Sequence[dict[str, Any]]) -> list[str]: |
| def dominates(a: dict[str, Any], b: dict[str, Any]) -> bool: |
| av = (a["resolved_rate"], a["applicable_patch_rate"], a["accepted_edit_rate"], |
| -a["mean_total_tokens"], -a["mean_wall_seconds"]) |
| bv = (b["resolved_rate"], b["applicable_patch_rate"], b["accepted_edit_rate"], |
| -b["mean_total_tokens"], -b["mean_wall_seconds"]) |
| return all(x >= y for x, y in zip(av, bv)) and any(x > y for x, y in zip(av, bv)) |
| return sorted(row["harness_id"] for row in summary if not any( |
| other["harness_id"] != row["harness_id"] and dominates(other, row) for other in summary |
| )) |
|
|
|
|
| def e16_validation(rows: Sequence[dict[str, Any]], selection: dict[str, Any]) -> dict[str, Any]: |
| selected = [row for row in rows if row["experiment_id"] == "E16"] |
| summary = summarize(selected, ("harness_id",)) |
| by_harness = {row["harness_id"]: row for row in summary} |
| screening = selection["metrics"] |
| metric_pairs = { |
| "resolved": ("resolved_rate", "resolved_rate", True), |
| "accepted": ("accepted_edit_rate", "accepted_edit_rate", True), |
| "applicable": ("applicable_patch_rate", "applicable_patch_rate", True), |
| "tokens": ("mean_total_tokens", "mean_total_tokens", False), |
| "wall_seconds": ("mean_wall_seconds", "mean_wall_seconds", False), |
| } |
| rank_stability = {} |
| for name, (screen_key, held_key, high) in metric_pairs.items(): |
| left = {key: float(screening[key][screen_key]) for key in by_harness} |
| right = {key: float(by_harness[key][held_key]) for key in by_harness} |
| rank_stability[name] = { |
| "screening": left, "held_out": right, |
| "spearman_rank_correlation": spearman(left, right, high), |
| } |
| paired = [] |
| offset = 400 |
| for harness in sorted(by_harness): |
| if harness == "H000": |
| continue |
| for endpoint in ("resolved_at_1", "accepted_edit_cell", "applicable_final_patch", "total_tokens", "elapsed_seconds"): |
| effects = _paired_harness_effects(selected, harness, "H000", endpoint) |
| paired.append(contrast_record(f"E16_{harness}_minus_H000_{endpoint}", effects, offset)) |
| offset += 1 |
| adjusted = holm([row["sign_flip_p"] for row in paired if row["contrast"].endswith("accepted_edit_cell")]) |
| for row, value in zip([r for r in paired if r["contrast"].endswith("accepted_edit_cell")], adjusted): |
| row["sign_flip_p_holm_accepted_family"] = value |
| return { |
| "harness_summary": summary, |
| "rank_stability": rank_stability, |
| "paired_vs_exact_baseline": paired, |
| "held_out_pareto_frontier": pareto(summary), |
| "resolution_event_warning": "Only one E16 cell resolved; resolution ranks are descriptive and not evidence of equivalence.", |
| } |
|
|
|
|
| def failure_taxonomy(rows: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: |
| grouped: Counter[tuple[str, str, str]] = Counter( |
| (row["experiment_id"], row["model_id"], row["failure_stage"]) for row in rows |
| ) |
| return [ |
| {"experiment_id": experiment, "model_id": model, "failure_stage": stage, "cells": count} |
| for (experiment, model, stage), count in sorted(grouped.items()) |
| ] |
|
|
|
|
| def write_csv(path: Path, rows: Sequence[dict[str, Any]]) -> None: |
| if not rows: |
| return |
| keys = list(rows[0]) |
| with path.open("w", encoding="utf-8", newline="") as handle: |
| writer = csv.DictWriter( |
| handle, fieldnames=keys, extrasaction="ignore", lineterminator="\n" |
| ) |
| writer.writeheader(); writer.writerows(rows) |
|
|
|
|
| def figures(output: Path, summaries: Sequence[dict[str, Any]], e16: dict[str, Any]) -> list[str]: |
| files: list[str] = [] |
| e13 = [row for row in summaries if row["experiment_id"] == "E13"] |
| harnesses = [row["harness_id"] for row in e13] |
| x = np.arange(len(harnesses)) |
| fig, ax = plt.subplots(figsize=(10, 5.4)) |
| ax.plot(x, [row["search_gold_rate"] for row in e13], "o-", label="Gold retrieved") |
| ax.plot(x, [row["accepted_edit_rate"] for row in e13], "s-", label="Accepted edit") |
| ax.plot(x, [row["applicable_patch_rate"] for row in e13], "^-", label="Applicable patch") |
| ax.plot(x, [row["resolved_rate"] for row in e13], "D-", label="Resolved@1") |
| ax.set_xticks(x, harnesses, rotation=45); ax.set_ylim(-.02, 1.02) |
| ax.set_ylabel("Cell rate"); ax.set_title("E13 stage-aware outcome funnel by retrieval harness") |
| ax.grid(axis="y", alpha=.25); ax.legend(ncol=2); fig.tight_layout() |
| for suffix in ("png", "pdf"): |
| path = output / f"figure_e13_funnel.{suffix}" |
| metadata = {"CreationDate": None, "ModDate": None} if suffix == "pdf" else None |
| fig.savefig(path, dpi=220, metadata=metadata); files.append(path.name) |
| plt.close(fig) |
|
|
| held = e16["harness_summary"] |
| fig, ax = plt.subplots(figsize=(7.5, 5.5)) |
| for row in held: |
| ax.scatter(row["mean_total_tokens"], row["applicable_patch_rate"], s=70) |
| ax.annotate(row["harness_id"], (row["mean_total_tokens"], row["applicable_patch_rate"]), xytext=(4, 4), textcoords="offset points") |
| ax.set_xlabel("Mean total tokens (lower is better)"); ax.set_ylabel("Applicable-patch rate") |
| ax.set_title("E16 held-out quality–cost behavior"); ax.grid(alpha=.25); fig.tight_layout() |
| for suffix in ("png", "pdf"): |
| path = output / f"figure_e16_quality_cost.{suffix}" |
| metadata = {"CreationDate": None, "ModDate": None} if suffix == "pdf" else None |
| fig.savefig(path, dpi=220, metadata=metadata); files.append(path.name) |
| plt.close(fig) |
| return files |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser() |
| parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parents[1]) |
| args = parser.parse_args() |
| root = args.root.resolve() |
| rows, audit = discover(root) |
| harness_specs = load_harnesses(root) |
| selection = json.loads((root / "configs" / "study5" / "E16_selection.json").read_text()) |
| summaries = summarize(rows, ("experiment_id", "harness_id")) |
| model_summaries = summarize(rows, ("experiment_id", "model_id")) |
| analysis = { |
| "schema_version": 1, |
| "bootstrap_draws": BOOTSTRAPS, |
| "sign_flip_draws": SIGN_FLIPS, |
| "seed": BOOTSTRAP_SEED, |
| "independent_unit": "task", |
| "e13_factorial": e13_factorial(rows, harness_specs), |
| "e14_retrieval_action": e14_interactions(rows), |
| "e15_navigation_packing": e15_blocks(rows), |
| "e16_held_out": e16_validation(rows, selection), |
| "failure_taxonomy": failure_taxonomy(rows), |
| "claim_boundary": "Zero or sparse resolution differences are no evidence of improvement, not evidence of equivalence.", |
| } |
| output = root / "results" / "derived" / "study5" |
| output.mkdir(parents=True, exist_ok=True) |
| write_csv(output / "all_cells.csv", rows) |
| write_csv(output / "harness_summary.csv", summaries) |
| write_csv(output / "model_summary.csv", model_summaries) |
| write_csv(output / "failure_taxonomy.csv", analysis["failure_taxonomy"]) |
| (output / "artifact_audit.json").write_text(json.dumps(audit, indent=2, sort_keys=True) + "\n") |
| (output / "statistical_results.json").write_text(json.dumps(analysis, indent=2, sort_keys=True, allow_nan=False) + "\n") |
| figure_files = figures(output, summaries, analysis["e16_held_out"]) |
| manifest = { |
| "schema_version": 1, |
| "input_audit_sha256": audit["audit_sha256"], |
| "analysis_sha256": sha256((output / "statistical_results.json").read_bytes()).hexdigest(), |
| "cells_csv_sha256": sha256((output / "all_cells.csv").read_bytes()).hexdigest(), |
| "figure_files": figure_files, |
| "files": {}, |
| } |
| for path in sorted(output.iterdir()): |
| if path.is_file() and path.name != "analysis_manifest.json": |
| manifest["files"][path.name] = sha256(path.read_bytes()).hexdigest() |
| manifest["manifest_sha256"] = canonical_hash(manifest) |
| (output / "analysis_manifest.json").write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n") |
| print(json.dumps({ |
| "passed": audit["passed"], "cells": len(rows), "output": str(output), |
| "audit_sha256": audit["audit_sha256"], "analysis_sha256": manifest["analysis_sha256"], |
| }, indent=2)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
