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