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