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