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