"""Outcome-blind preflight for the frozen Study 2 execution revision.""" from __future__ import annotations from dataclasses import asdict from hashlib import sha256 import importlib.metadata import json import platform from pathlib import Path import shutil import subprocess import sys import time from typing import Any 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.specs import ( load_embeddings, load_models, load_repositories, validate_configuration_tree, ) from agent_harness.study2_experiment import tokenizer_for PACKAGES = ( "faiss-cpu", "numpy", "pytest", "scipy", "statsmodels", "tokenizers", "tree-sitter", "tree-sitter-go", "tree-sitter-python", ) def _command(arguments: list[str], cwd: Path | None = None) -> dict[str, Any]: result = subprocess.run( arguments, cwd=cwd, text=True, capture_output=True, check=False, timeout=120, ) return { "command": arguments, "returncode": result.returncode, "stdout": result.stdout, "stderr": result.stderr, } def _tool_probe(client: LMStudioClient, model_key: str) -> dict[str, Any]: response = client.chat_completions( model_key, [ { "role": "system", "content": "This is a runtime preflight. Use the required tool exactly once.", }, { "role": "user", "content": "Call preflight_echo with marker MODEL_TOOL_OK. Do not answer in prose.", }, ], tools=[ { "type": "function", "function": { "name": "preflight_echo", "description": "Return the requested preflight marker.", "parameters": { "type": "object", "properties": {"marker": {"type": "string"}}, "required": ["marker"], "additionalProperties": False, }, }, } ], max_tokens=2_048, seed=0, ) try: message = response["choices"][0]["message"] calls = message["tool_calls"] function = calls[0]["function"] arguments = function["arguments"] decoded = arguments if isinstance(arguments, dict) else json.loads(arguments) except (KeyError, IndexError, TypeError, json.JSONDecodeError) as exc: raise RuntimeError("model did not return a valid preflight tool call") from exc if function.get("name") != "preflight_echo" or decoded != {"marker": "MODEL_TOOL_OK"}: raise RuntimeError(f"unexpected tool preflight payload: {function}") serialized = json.dumps(response, sort_keys=True, separators=(",", ":")) return { "tool_name": function["name"], "arguments": decoded, "finish_reason": response["choices"][0].get("finish_reason"), "usage": response.get("usage", {}), "response_sha256": sha256(serialized.encode()).hexdigest(), } 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}") disk = shutil.disk_usage(root) if disk.free < 50 * 1024**3: raise RuntimeError(f"less than 50 GiB free before Study 2: {disk.free} bytes") models = load_models(root) embedding = load_embeddings(root)["EMB002"] repositories = load_repositories(root) 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 2", "outcome_blind": True, "research_code_revision": revision, "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 }, "repositories": {}, "models": {}, "embedding": {}, } 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(), } for model_id in ("M002", "M003"): model = models[model_id] transition = residency.ensure_exclusive( model.expected_inference_key, model.context_length ) client = LMStudioClient(model, timeout_seconds=1_800) discovery, resolved = 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(client, resolved.inference_key), } report["models"][model_id]["unload"] = residency.unload_all().to_dict() transition = residency.ensure_exclusive( embedding.model_key, embedding.loaded_context_length ) embedding_client = LMStudioEmbeddingClient(embedding, timeout_seconds=1_800) record = embedding_client.resolve() probe = embedding_client.probe() report["embedding"] = { "spec": asdict(embedding), "config_hash": embedding.config_hash, "transition": transition.to_dict(), "resolved": record, "probe": probe.to_dict(), "unload": residency.unload_all().to_dict(), } report["passed"] = True return report finally: cleanup_errors: list[str] = [] try: report["final_unload"] = residency.unload_all().to_dict() except Exception as cleanup_error: cleanup_errors.append(f"unload_all: {cleanup_error}") try: status = server.status() report["server_stop"] = ( server.stop() if status["running"] else {"action": "already_stopped", "status": status} ) except Exception as cleanup_error: cleanup_errors.append(f"server_stop: {cleanup_error}") report["cleanup_errors"] = cleanup_errors report["finished_unix"] = time.time() def main() -> None: root = Path(__file__).resolve().parents[1] report: dict[str, Any] = {} output = root / "results" / "reports" / "study2_preflight.json" output.parent.mkdir(parents=True, exist_ok=True) 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", encoding="utf-8", ) raise output.write_text( json.dumps(report, indent=2, sort_keys=True, default=str) + "\n", encoding="utf-8", ) print(json.dumps({"passed": True, "report": str(output)}, indent=2)) if __name__ == "__main__": main()