scripts/audit_study4_design.py

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