#!/usr/bin/env python3 """Outcome-blind runtime, design, repository, and executor preflight for Study 5.""" from __future__ import annotations from dataclasses import asdict import importlib.metadata import json from pathlib import Path import platform import shutil import sys import time from typing import Any from preflight_study3 import PACKAGES, _command, _executor_conformance, _tool_probe from agent_harness.lm_studio import LMStudioClient from agent_harness.lm_studio_embeddings import LMStudioEmbeddingClient from agent_harness.lm_studio_management import LMStudioResidencyManager, LMStudioServer from agent_harness.pilot import research_code_revision from agent_harness.protocol_experiment import protocol_tool_definitions from agent_harness.specs import ( load_edit_interfaces, load_embeddings, load_experiments, load_harnesses, load_models, load_repositories, load_task_split, load_tasks, validate_configuration_tree, ) from agent_harness.study2_experiment import tokenizer_for EXPERIMENTS = ("E13", "E14", "E15") EXPECTED = {"E13": 1440, "E14": 540, "E15": 540} def run(root: Path) -> dict[str, Any]: revision = research_code_revision(root) errors, warnings = validate_configuration_tree(root) if errors or warnings: raise RuntimeError(f"configuration failed: errors={errors}, warnings={warnings}") for experiment_id in EXPERIMENTS: raw = root / "results" / "raw" / experiment_id if raw.exists() and any(raw.rglob("*")): raise RuntimeError(f"{experiment_id} raw outcomes exist before preflight") audit = json.loads((root / "docs" / "STUDY5_DESIGN_AUDIT.json").read_text()) if not audit.get("outcome_blind"): raise RuntimeError("Study 5 design audit is not outcome blind") manifests: dict[str, Any] = {} for experiment_id in EXPERIMENTS: manifest = json.loads( (root / "configs" / "study5" / f"{experiment_id}_cells.json").read_text() ) if manifest.get("planned_cells") != EXPECTED[experiment_id]: raise RuntimeError(f"{experiment_id} manifest count mismatch") if len(manifest.get("cells", [])) != EXPECTED[experiment_id]: raise RuntimeError(f"{experiment_id} cell array mismatch") manifests[experiment_id] = manifest tasks = load_tasks(root) fresh = load_task_split(root / "tasks" / "splits" / "study5_fresh.txt") if len(fresh) != 17 or len({tasks[item].gold_commit for item in fresh}) != 17: raise RuntimeError("Study 5 fresh split must contain 17 unique gold commits") prior = {task.gold_commit for task_id, task in tasks.items() if task_id not in fresh} if prior & {tasks[item].gold_commit for item in fresh}: raise RuntimeError("Study 5 fresh split overlaps a prior task commit") disk = shutil.disk_usage(root) if disk.free < 20 * 1024**3: raise RuntimeError(f"less than 20 GiB free before Study 5: {disk.free} bytes") models = load_models(root) experiments = load_experiments(root) embedding = load_embeddings(root)[experiments["E13"].embedding_id] repositories = load_repositories(root) harnesses = load_harnesses(root) interfaces = load_edit_interfaces(root) first_task = tasks[manifests["E15"]["cells"][0]["task_id"]] tool_signatures = { harness_id: [ item["function"]["name"] for item in protocol_tool_definitions(interfaces["P002"], first_task, harnesses[harness_id]) ] for harness_id in experiments["E15"].harness_ids } if tool_signatures["H008"] == tool_signatures["H011"]: raise RuntimeError("unified and specialized search signatures did not separate") server = LMStudioServer(port=1234) server_state = server.ensure_running() residency = LMStudioResidencyManager( models["M002"].base_url, models["M002"].api_token_env, timeout_seconds=1_800 ) report: dict[str, Any] = { "schema_version": 1, "study": "Study 5 / E13--E15", "outcome_blind": True, "research_code_revision": revision, "planned_cells": sum(EXPECTED.values()), "design_audit": audit, "fresh_task_count": len(fresh), "started_unix": time.time(), "platform": platform.platform(), "python": sys.version, "torch_used": False, "disk": {"total": disk.total, "used": disk.used, "free": disk.free}, "lms_version": _command([str(server.cli_path), "--version"]), "server_start": server_state, "dependencies": {package: importlib.metadata.version(package) for package in PACKAGES}, "executor_conformance": _executor_conformance(root), "tool_signatures": tool_signatures, "repositories": {}, "embedding": {}, "models": {}, } try: residency.unload_all() for repository_id, repository in sorted(repositories.items()): observed = _command(["git", "rev-parse", "HEAD"], cwd=root / repository.local_path) if observed["returncode"] or observed["stdout"].strip() != repository.pinned_head: raise RuntimeError(f"repository head mismatch: {repository_id}: {observed}") report["repositories"][repository_id] = { "spec": asdict(repository), "observed_head": observed["stdout"].strip(), } transition = residency.ensure_exclusive( embedding.model_key, embedding.loaded_context_length ) client = LMStudioEmbeddingClient(embedding, timeout_seconds=1_800) report["embedding"] = { "spec": asdict(embedding), "config_hash": embedding.config_hash, "transition": transition.to_dict(), "resolved": client.resolve(), "probe": client.probe().to_dict(), "unload": residency.unload_all().to_dict(), } for model_id in experiments["E13"].model_ids: model = models[model_id] transition = residency.ensure_exclusive(model.expected_inference_key, model.context_length) model_client = LMStudioClient(model, timeout_seconds=1_800) discovery, resolved = model_client.resolve() tokenizer = tokenizer_for(model) report["models"][model_id] = { "spec": asdict(model), "config_hash": model.config_hash, "transition": transition.to_dict(), "resolved": resolved.to_dict(), "discovery_errors": discovery.endpoint_errors, "tokenizer_path": str(tokenizer.path), "tokenizer_sha256": tokenizer.sha256, "tool_probe": _tool_probe(model_client, resolved.inference_key), "unload": residency.unload_all().to_dict(), } report["passed"] = True return report finally: cleanup_errors = [] try: report["final_unload"] = residency.unload_all().to_dict() except Exception as exc: cleanup_errors.append(f"unload_all: {exc}") try: status = server.status() report["server_stop"] = ( server.stop() if status["running"] else {"action": "already_stopped", "status": status} ) except Exception as exc: cleanup_errors.append(f"server_stop: {exc}") report["cleanup_errors"] = cleanup_errors report["finished_unix"] = time.time() def main() -> None: root = Path(__file__).resolve().parents[1] output = root / "results" / "reports" / "study5_preflight.json" output.parent.mkdir(parents=True, exist_ok=True) report: dict[str, Any] = {} try: report = run(root) except Exception as exc: report = {**report, "passed": False, "error": repr(exc)} output.write_text(json.dumps(report, indent=2, sort_keys=True, default=str) + "\n") raise output.write_text(json.dumps(report, indent=2, sort_keys=True, default=str) + "\n") print(json.dumps({"passed": True, "report": str(output)}, indent=2)) if __name__ == "__main__": main()