#!/usr/bin/env python3 """Summarize whether a Meta3 overnight run is ready to start. No network calls.""" from __future__ import annotations import argparse import json import os 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 maybe_load(path: Path) -> dict[str, Any] | None: return load_json(path) if path.exists() else None def latest_run_dir() -> Path: runs_root = ROOT / "runs" candidates = sorted(p for p in runs_root.iterdir() if p.is_dir()) if not candidates: raise FileNotFoundError("No run dirs found. Run scripts/meta3_prepare.py first.") return candidates[-1] def run_dir_for(run_id: str | None) -> Path: return ROOT / "runs" / run_id if run_id else latest_run_dir() def launcher_markers(path: Path) -> dict[str, Any]: exists = path.exists() text = path.read_text(encoding="utf-8") if exists else "" return { "exists": exists, "path": str(path), "has_placeholders": any(token in text for token in ["TODO", "TBD", "YYYY-MM-DDTHH:MM:SS+02:00"]), "has_status_locked": "Status: locked" in text, "has_sections": all( marker in text for marker in ["## Ziel", "## Fragen", "## Quellen", "## Budget", "## Outputs", "## Abbruchbedingungen", "## Morgen-Erfolg"] ), } def ledger_rows(path: Path) -> int: if not path.exists(): return 0 rows = 0 with path.open("r", encoding="utf-8") as handle: for line in handle: if line.strip(): rows += 1 return rows def source_snapshot_summary(run_dir: Path) -> dict[str, Any]: path = run_dir / "manifests/source_snapshot_manifest.json" manifest = maybe_load(path) or {} summary = manifest.get("summary") if isinstance(manifest, dict) else {} if not isinstance(summary, dict): summary = {} return { "path": str(path), "exists": path.exists(), "schema": manifest.get("schema"), "source_drift_from_audit": manifest.get("source_drift_from_audit"), "frozen_rows": int(summary.get("frozen_rows") or 0), "raw_rows": int(summary.get("raw_rows") or 0), "frozen_digest_path": manifest.get("frozen_digest_path"), } def markdown_report(report: dict[str, Any]) -> str: lines = [ "# Meta3 Launch Gate", "", f"Run: `{report['run_id']}`", f"Created: `{report['created_at']}`", f"Status: `{report['status']}`", "", "## Ready", "", ] for item in report["ready"]: lines.append(f"- `{item}`") lines.extend(["", "## Blockers", ""]) for item in report["blockers"]: lines.append(f"- {item}") lines.extend(["", "## Details", ""]) lines.append(f"- source_snapshot: {report['source_snapshot']}") lines.append(f"- requirements_lock: {report['requirements_lock']}") lines.append(f"- runner_status: {report['runner_status']}") lines.append(f"- readiness: {report['readiness_status']}") lines.append(f"- preflight: {report['network_preflight_status']}") lines.append(f"- morning: {report['morning_status']}") lines.append(f"- api_env: {report['api_env']}") lines.append(f"- ledger_rows: {report['ledger_rows']}") return "\n".join(lines) + "\n" def main() -> int: parser = argparse.ArgumentParser(description="Summarize Meta3 overnight start readiness.") parser.add_argument("--run-id", default=None, help="Run id under runs/. Defaults to latest run.") args = parser.parse_args() run_dir = run_dir_for(args.run_id) readiness = maybe_load(run_dir / "reports/run_readiness_report.json") or {} preflight = maybe_load(run_dir / "reports/network_preflight.json") or {} morning = maybe_load(run_dir / "reports/morning_report.json") or {} runner_status = maybe_load(run_dir / "reports/runner_status.json") or {} source_snapshot = source_snapshot_summary(run_dir) requirements_lock = launcher_markers(ROOT / "requirements/LOCKED.md") api_env = { "base_url": bool(os.environ.get("LITELLM_BASE_URL") or os.environ.get("OPENAI_BASE_URL")), "api_key": bool(os.environ.get("LITELLM_API_KEYS") or os.environ.get("OPENAI_API_KEY")), } source_ready = source_snapshot["exists"] and source_snapshot["schema"] == "meta3.source_snapshot_manifest.v1" requirements_ready = ( requirements_lock["exists"] and requirements_lock["has_status_locked"] and requirements_lock["has_sections"] and not requirements_lock["has_placeholders"] ) launchable = all( [ source_ready, requirements_ready, readiness.get("status") == "pass", runner_status.get("status") == "idle_or_ready", preflight.get("status") == "pass", api_env["base_url"], api_env["api_key"], ] ) blockers: list[str] = [] if not source_ready: blockers.append("source snapshot missing or invalid") if not requirements_ready: blockers.append("requirements lock missing or incomplete") if readiness.get("status") != "pass": blockers.append("run readiness is not pass") if runner_status.get("status") != "idle_or_ready": blockers.append("runner status is not idle_or_ready") if preflight.get("status") != "pass": blockers.append("network preflight is not pass") if not api_env["base_url"]: blockers.append("API base URL env missing") if not api_env["api_key"]: blockers.append("API key env missing") ready: list[str] = [] if source_ready: ready.append("frozen source snapshot exists") if source_snapshot["frozen_digest_path"]: ready.append(f"frozen digest path: {source_snapshot['frozen_digest_path']}") if readiness.get("status") == "pass": ready.append("run readiness passes") if runner_status.get("status") == "idle_or_ready": ready.append("runner is idle") if preflight.get("accepted_call_budget") is not None: ready.append(f"accepted budget planned: {preflight.get('accepted_call_budget')}") if preflight.get("planned_calls_total") is not None: ready.append(f"planned calls: {preflight.get('planned_calls_total')}") report = { "schema": "meta3.launch_gate.v1", "run_id": run_dir.name, "created_at": datetime.now(timezone.utc).isoformat(), "status": "ready_to_launch" if launchable else "blocked", "ready": ready, "blockers": blockers, "source_snapshot": source_snapshot, "requirements_lock": requirements_lock, "readiness_status": readiness.get("status"), "network_preflight_status": preflight.get("status"), "morning_status": morning.get("status"), "runner_status": runner_status.get("status"), "api_env": api_env, "ledger_rows": { "calibration": ledger_rows(run_dir / "state/calibration_call_ledger.jsonl"), "claim_ledger": ledger_rows(run_dir / "state/claim_ledger_call_ledger.jsonl"), }, "network_calls_made": 0, } report_path = run_dir / "reports/launch_gate.json" report_md_path = run_dir / "reports/LAUNCH_GATE.md" report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") report_md_path.write_text(markdown_report(report), encoding="utf-8") print(f"launch_gate={report_path}") print(f"launch_gate_md={report_md_path}") print(f"status={report['status']}") print(f"ready={len(ready)}") print(f"blockers={len(blockers)}") print("network_calls_made=0") return 0 if launchable else 1 if __name__ == "__main__": raise SystemExit(main())