File size: 8,795 Bytes
d61821a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 | """Outcome-blind deterministic audit for the prospective E10 design freeze."""
from __future__ import annotations
from collections import Counter
from hashlib import sha256
import json
from pathlib import Path
import subprocess
from typing import Any
from agent_harness.specs import (
load_edit_interfaces,
load_experiments,
load_harnesses,
load_models,
load_repositories,
load_task_split,
load_tasks,
validate_configuration_tree,
)
def digest(path: Path) -> str:
return sha256(path.read_bytes()).hexdigest()
def canonical_digest(value: Any) -> str:
return sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
def run(root: Path) -> dict[str, Any]:
errors, warnings = validate_configuration_tree(root)
if errors or warnings:
raise RuntimeError(f"configuration failed: errors={errors}, warnings={warnings}")
raw = root / "results" / "raw" / "E10"
if raw.exists() and any(raw.rglob("*")):
raise RuntimeError("E10 outcome directory exists; design is no longer outcome-blind")
experiment = load_experiments(root)["E10"]
harnesses = load_harnesses(root)
interfaces = load_edit_interfaces(root)
models = load_models(root)
repositories = load_repositories(root)
tasks = load_tasks(root)
split = load_task_split(root / "tasks" / "splits" / "study4_fresh.txt")
prior_split = set(load_task_split(root / "tasks" / "splits" / "study2_confirmatory.txt"))
gate_path = root / "configs" / "gates" / "E09_model_interface_gate.json"
gate = json.loads(gate_path.read_text(encoding="utf-8"))
if gate["selected"] != {"M002": "P002", "M003": "P003", "M004": "P003"}:
raise RuntimeError("E09 compatibility gate drifted")
if len(split) != 20 or len(set(split)) != 20 or set(split) & prior_split:
raise RuntimeError("E10 split is not 20 unique fresh task IDs")
url_to_repository = {item.repository_url: item.repository_id for item in repositories.values()}
repository_counts: Counter[str] = Counter()
language_counts: Counter[str] = Counter()
prior_gold = {
task.gold_commit for task_id, task in tasks.items() if not task_id.startswith("TASK_S4_")
}
task_records: list[dict[str, Any]] = []
input_files: list[Path] = [
root / "configs" / "experiments" / "E10_fresh_retrieval.toml",
gate_path,
root / "tasks" / "splits" / "study4_fresh.txt",
root / "docs" / "STUDY4_PREREGISTRATION.md",
root / "docs" / "STUDY3_RESULTS.md",
root / "docs" / "PROTOCOL_AMENDMENTS.md",
root / "scripts" / "build_study4_tasks.py",
root / "scripts" / "audit_study4_design.py",
root / "scripts" / "analyze_study4.py",
root / "scripts" / "preflight_study4.py",
root / "src" / "agent_harness" / "protocol_experiment.py",
root / "src" / "agent_harness" / "live_agent_experiment.py",
root / "src" / "agent_harness" / "cli.py",
root / "tests" / "test_protocol_experiment.py",
root / "tests" / "test_specs.py",
root / "tests" / "test_study4_analysis.py",
]
input_files.extend(sorted((root / "tasks" / "selection" / "study4").glob("*.json")))
for task_id in split:
task = tasks[task_id]
if task.validation_status != "end_to_end_ready" or task.gold_commit in prior_gold:
raise RuntimeError(f"{task_id} is not fresh and end-to-end ready")
repository_id = url_to_repository[task.repository_url]
repository_counts[repository_id] += 1
language_counts[task.language] += 1
required = [
root / "tasks" / "manifests" / f"{task_id}.toml",
root / "tasks" / task.gold_patch,
root / "tasks" / task.test_patch,
root / "tasks" / "validation" / "study4" / f"{task_id}.json",
]
if not all(path.is_file() for path in required):
raise RuntimeError(f"{task_id} is missing a fresh-task artifact")
validation = json.loads(required[-1].read_text(encoding="utf-8"))
if not validation.get("valid_end_to_end"):
raise RuntimeError(f"{task_id} validation is not positive")
input_files.extend(required)
task_records.append(
{
"task_id": task_id,
"repository_id": repository_id,
"language": task.language,
"base_commit": task.base_commit,
"gold_commit": task.gold_commit,
"config_hash": task.config_hash,
"gold_files": list(task.gold_files),
"artifact_sha256": {
path.relative_to(root).as_posix(): digest(path) for path in required
},
}
)
if repository_counts != Counter({"R003": 10, "R002": 7, "R001": 3}):
raise RuntimeError(f"fresh repository counts drifted: {repository_counts}")
identities: set[tuple[str, str, str, str]] = set()
model_positions: Counter[tuple[str, int]] = Counter()
harness_positions: Counter[tuple[str, str, int]] = Counter()
model_ids = list(experiment.model_ids)
harness_ids = list(experiment.harness_ids)
for task_index, task_id in enumerate(split):
model_offset = task_index % len(model_ids)
model_order = model_ids[model_offset:] + model_ids[:model_offset]
for model_position, model_id in enumerate(model_order):
model_positions[(model_id, model_position)] += 1
interface_id = gate["selected"][model_id]
offset = (task_index + model_ids.index(model_id)) % len(harness_ids)
harness_order = harness_ids[offset:] + harness_ids[:offset]
for position, harness_id in enumerate(harness_order):
harness_positions[(model_id, harness_id, position)] += 1
identities.add((task_id, model_id, harness_id, interface_id))
if len(identities) != 180:
raise RuntimeError(f"E10 identity plan is not 180 unique cells: {len(identities)}")
for harness_id in experiment.harness_ids:
input_files.append(next((root / "configs" / "harnesses").glob(f"{harness_id}_*.toml")))
for interface_id in set(gate["selected"].values()):
input_files.append(next((root / "configs" / "edit_interfaces").glob(f"{interface_id}_*.toml")))
for model_id in experiment.model_ids:
input_files.append(next((root / "configs" / "models").glob(f"{model_id.lower()}_*.toml")))
input_files.extend(sorted((root / "configs" / "repositories").glob("R*.toml")))
missing_design_files = [path for path in input_files if not path.is_file()]
if missing_design_files:
raise RuntimeError(f"design files missing: {missing_design_files}")
revision = subprocess.run(
["git", "rev-parse", "HEAD"], cwd=root, check=True, capture_output=True, text=True
).stdout.strip()
unique_inputs = sorted(set(input_files))
report: dict[str, Any] = {
"schema_version": 1,
"study": "E10 protocol-normalized fresh-task retrieval",
"outcome_blind": True,
"prior_revision": revision,
"raw_e10_absent": True,
"tasks": len(split),
"models": model_ids,
"harnesses": harness_ids,
"gate_interfaces": gate["selected"],
"planned_cells": len(identities),
"repository_counts": dict(sorted(repository_counts.items())),
"language_counts": dict(sorted(language_counts.items())),
"model_position_counts": {
f"{model}@{position}": count for (model, position), count in sorted(model_positions.items())
},
"harness_position_counts": {
f"{model}/{harness}@{position}": count
for (model, harness, position), count in sorted(harness_positions.items())
},
"task_records": task_records,
"model_config_hashes": {item: models[item].config_hash for item in model_ids},
"harness_config_hashes": {item: harnesses[item].config_hash for item in harness_ids},
"interface_config_hashes": {
item: interfaces[item].config_hash for item in set(gate["selected"].values())
},
"input_sha256": {
path.relative_to(root).as_posix(): digest(path) for path in unique_inputs
},
}
report["design_sha256"] = canonical_digest(report)
return report
def main() -> None:
root = Path(__file__).resolve().parents[1]
report = run(root)
output = root / "docs" / "STUDY4_DESIGN_AUDIT.json"
output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(json.dumps({"output": str(output), "planned_cells": 180, "design_sha256": report["design_sha256"]}, indent=2))
if __name__ == "__main__":
main()
|