"""스케줄러 사이클 heartbeat 레지스트리 (순수 인메모리, DB/네트워크 없음). 각 스케줄러는 사이클이 끝날 때 ``beat(name, summary)`` 를 호출한다. main.py 의 감시 루프와 /health 가 ``snapshot()`` 으로 마지막 사이클 경과시간을 읽는다. """ from __future__ import annotations import time from typing import Any, Dict, Optional _beats: Dict[str, Dict[str, Any]] = {} def beat(name: str, summary: Optional[str] = None) -> None: _beats[name] = {"at": time.monotonic(), "summary": summary} def age_seconds(name: str, now: Optional[float] = None) -> Optional[float]: rec = _beats.get(name) if rec is None: return None return (now if now is not None else time.monotonic()) - rec["at"] def snapshot(now: Optional[float] = None) -> Dict[str, Dict[str, Any]]: n = now if now is not None else time.monotonic() return {k: {"age_sec": round(n - v["at"], 1), "summary": v["summary"]} for k, v in _beats.items()} def reset() -> None: _beats.clear()