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