#!/usr/bin/env python3 """Create a local Meta3 run manifest without making network calls.""" from __future__ import annotations import argparse import json import os import re from datetime import datetime, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] def load_json(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8") as handle: data = json.load(handle) if not isinstance(data, dict): raise ValueError(f"Expected JSON object: {path}") return data def write_json(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def utc_stamp() -> str: return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") def slugify(value: str) -> str: slug = re.sub(r"[^a-zA-Z0-9]+", "-", value.strip().lower()).strip("-") return slug or "run" def path_status(paths: dict[str, str]) -> dict[str, dict[str, Any]]: status: dict[str, dict[str, Any]] = {} for key, value in paths.items(): p = Path(value).expanduser() status[key] = { "path": str(p), "exists": p.exists(), "is_file": p.is_file(), "is_dir": p.is_dir(), "size_bytes": p.stat().st_size if p.exists() and p.is_file() else None, } return status def env_status(profile: dict[str, Any]) -> dict[str, Any]: api_profile = profile.get("api_profile", {}) key_envs = list(api_profile.get("api_key_envs", [])) base_envs = list(api_profile.get("base_url_envs", [])) model_env = api_profile.get("model_env") return { "api_key_envs_present": {name: bool(os.environ.get(name)) for name in key_envs}, "base_url_envs_present": {name: bool(os.environ.get(name)) for name in base_envs}, "model_env_present": bool(model_env and os.environ.get(model_env)), "secrets_recorded": False, } def validate_questions(questions_doc: dict[str, Any]) -> list[dict[str, Any]]: questions = questions_doc.get("questions") if not isinstance(questions, list) or not questions: raise ValueError("config/questions.json must contain a non-empty questions list") seen: set[str] = set() for item in questions: if not isinstance(item, dict): raise ValueError("Each question must be an object") qid = item.get("id") prompt = item.get("prompt") if not isinstance(qid, str) or not qid.strip(): raise ValueError("Each question needs a non-empty id") if qid in seen: raise ValueError(f"Duplicate question id: {qid}") if not isinstance(prompt, str) or not prompt.strip(): raise ValueError(f"Question {qid} needs a non-empty prompt") seen.add(qid) return questions def validate_process_model(process_doc: dict[str, Any]) -> list[str]: if process_doc.get("schema") != "meta3.process_model.v1": raise ValueError("config/process_model.json schema must be meta3.process_model.v1") phases = process_doc.get("phases") if not isinstance(phases, list) or not phases: raise ValueError("config/process_model.json must contain phases") phase_ids: list[str] = [] for phase in phases: if not isinstance(phase, dict): raise ValueError("Each process phase must be an object") phase_id = phase.get("id") if not isinstance(phase_id, str) or not phase_id.strip(): raise ValueError("Each process phase needs an id") phase_ids.append(phase_id) if len(phase_ids) != len(set(phase_ids)): raise ValueError("Duplicate process phase ids") return phase_ids def main() -> int: parser = argparse.ArgumentParser(description="Prepare a Meta3 overnight run manifest.") parser.add_argument("--profile", type=Path, default=ROOT / "config/run_profile.example.json") parser.add_argument("--questions", type=Path, default=ROOT / "config/questions.json") parser.add_argument("--process-model", type=Path, default=ROOT / "config/process_model.json") parser.add_argument("--experiment-design", type=Path, default=ROOT / "config/experiment_design.json") parser.add_argument("--run-name", default="overnight_prepare") args = parser.parse_args() profile = load_json(args.profile) questions_doc = load_json(args.questions) process_doc = load_json(args.process_model) experiment_doc = load_json(args.experiment_design) questions = validate_questions(questions_doc) process_phase_ids = validate_process_model(process_doc) if experiment_doc.get("schema") != "meta3.experiment_design.v1": raise ValueError("config/experiment_design.json schema must be meta3.experiment_design.v1") run_id = f"{utc_stamp()}_{slugify(args.run_name)}" run_dir = ROOT / "runs" / run_id for rel in ["manifests", "state", "outputs", "logs", "tmp"]: (run_dir / rel).mkdir(parents=True, exist_ok=False) source_status = path_status(profile.get("source_artifacts", {})) manifest = { "schema": "meta3.run_manifest.v1", "run_id": run_id, "created_at": datetime.now(timezone.utc).isoformat(), "root": str(ROOT), "mode": profile.get("mode"), "profile_path": str(args.profile), "questions_path": str(args.questions), "process_model_path": str(args.process_model), "experiment_design_path": str(args.experiment_design), "question_count": len(questions), "question_ids": [q["id"] for q in questions], "process_phase_count": len(process_phase_ids), "process_phase_ids": process_phase_ids, "profile": profile, "process_model_schema": process_doc.get("schema"), "experiment_design_schema": experiment_doc.get("schema"), "questions_version": questions_doc.get("version"), "source_status": source_status, "env_status": env_status(profile), "network_calls_made": 0, "next_step": "Run scripts/meta3_plan_batches.py --run-id " + run_id, } write_json(run_dir / "manifests/run_manifest.json", manifest) print(f"run_id={run_id}") print(f"run_dir={run_dir}") print(f"questions={len(questions)}") print(f"process_phases={len(process_phase_ids)}") missing = [key for key, value in source_status.items() if not value["exists"]] print(f"missing_sources={len(missing)}") if missing: for key in missing: print(f" missing: {key} -> {source_status[key]['path']}") return 0 if __name__ == "__main__": raise SystemExit(main())