File size: 5,718 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 | #!/usr/bin/env python3
"""Report Meta3 runner lock, stop, heartbeat, and ledger status. No network calls."""
from __future__ import annotations
import argparse
import json
from collections import Counter
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 load_jsonl(path: Path) -> list[dict[str, Any]]:
if not path.exists():
return []
rows: list[dict[str, Any]] = []
with path.open("r", encoding="utf-8") as handle:
for line_no, line in enumerate(handle, start=1):
if not line.strip():
continue
data = json.loads(line)
if not isinstance(data, dict):
raise ValueError(f"Expected JSON object in {path}:{line_no}")
rows.append(data)
return rows
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 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 mode_status(run_dir: Path, mode: str) -> dict[str, Any]:
lock_path = run_dir / "state" / f"{mode}_runner.lock.json"
heartbeat_path = run_dir / "state" / f"{mode}_runner_heartbeat.json"
ledger_path = run_dir / "state" / f"{mode}_call_ledger.jsonl"
stop_candidates = [run_dir / "state" / "STOP", run_dir / "state" / f"STOP_{mode}"]
ledger_rows = load_jsonl(ledger_path)
statuses = Counter(str(row.get("status") or "unknown") for row in ledger_rows)
calls = sum(int(row.get("network_calls_made") or 0) for row in ledger_rows)
return {
"mode": mode,
"lock_path": str(lock_path),
"lock_exists": lock_path.exists(),
"lock": load_json(lock_path) if lock_path.exists() else None,
"heartbeat_path": str(heartbeat_path),
"heartbeat_exists": heartbeat_path.exists(),
"heartbeat": load_json(heartbeat_path) if heartbeat_path.exists() else None,
"stop_paths": [str(path) for path in stop_candidates],
"active_stop_paths": [str(path) for path in stop_candidates if path.exists()],
"ledger_path": str(ledger_path),
"ledger_rows": len(ledger_rows),
"ledger_statuses": dict(sorted(statuses.items())),
"ledger_network_calls_made": calls,
}
def markdown_report(report: dict[str, Any]) -> str:
lines = [
"# Meta3 Runner Status",
"",
f"Run: `{report['run_id']}`",
f"Created: `{report['created_at']}`",
f"Overall status: `{report['status']}`",
"",
]
for mode, status in sorted(report["modes"].items()):
heartbeat = status.get("heartbeat") or {}
lines.extend(
[
f"## {mode}",
"",
f"- Lock exists: {status['lock_exists']}",
f"- Active stop files: {len(status['active_stop_paths'])}",
f"- Heartbeat status: {heartbeat.get('status')}",
f"- Completed: {heartbeat.get('completed')}",
f"- Failed: {heartbeat.get('failed')}",
f"- Ledger rows: {status['ledger_rows']}",
f"- Ledger statuses: {status['ledger_statuses']}",
f"- Ledger network calls: {status['ledger_network_calls_made']}",
"",
]
)
return "\n".join(lines)
def main() -> int:
parser = argparse.ArgumentParser(description="Report Meta3 runner status without API calls.")
parser.add_argument("--run-id", default=None, help="Run id under runs/. Defaults to latest run.")
parser.add_argument(
"--mode",
action="append",
choices=["calibration", "claim_ledger"],
help="Mode to inspect. Can be repeated. Defaults to both.",
)
args = parser.parse_args()
run_dir = run_dir_for(args.run_id)
modes = args.mode or ["calibration", "claim_ledger"]
mode_reports = {mode: mode_status(run_dir, mode) for mode in modes}
has_lock = any(item["lock_exists"] for item in mode_reports.values())
has_stop = any(item["active_stop_paths"] for item in mode_reports.values())
status = "blocked" if has_lock or has_stop else "idle_or_ready"
report = {
"schema": "meta3.runner_status.v1",
"run_id": run_dir.name,
"created_at": datetime.now(timezone.utc).isoformat(),
"status": status,
"modes": mode_reports,
"network_calls_made": 0,
}
report_path = run_dir / "reports/runner_status.json"
report_md_path = run_dir / "reports/RUNNER_STATUS.md"
write_json(report_path, report)
report_md_path.write_text(markdown_report(report), encoding="utf-8")
print(f"runner_status={report_path}")
print(f"runner_status_md={report_md_path}")
print(f"status={status}")
print(f"lock_modes={sum(1 for item in mode_reports.values() if item['lock_exists'])}")
print(f"stop_modes={sum(1 for item in mode_reports.values() if item['active_stop_paths'])}")
print("network_calls_made=0")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|