File size: 8,070 Bytes
1219ee2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 | #!/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())
|