"""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()