"""utils/schedule_id_freshness.py — task-2535 P0 (신호등 sync fix C). 회장 명시 (2026-05-10): 자동머지 시그널 (특히 schedule_id) 신선도 검증 부재. 오래된 cron schedule_id로 잘못된 활성 판정 가능. ★ Fix C: schedule_id freshness validator — heartbeat / staleness threshold 명시. 3 freshness states: FRESH — schedule_id 응답 ts가 SCHEDULE_FRESHNESS_THRESHOLD_MIN 이내 STALE — 마지막 응답 ts가 임계 초과 (cron schedule가 살아있는 것처럼 보이지만 무응답) MISSING — schedule_history/{ID}.log 부재 또는 chat=6937032012 record 0건 회장 §명시 격리: schedule_history 폴더는 모든 chat 공유. 회장 chat (CHAIRMAN_CHAT_ID = 6937032012) 레코드만 사용. 다른 chat record는 silently skip (cokacdir docs § isolation 정합). 회장 §금지: - ❌ 다른 chat record를 freshness 판정에 사용 - ❌ token / raw key / response 본문을 dict에 그대로 노출 - ❌ schedule_history 폴더 외부 IO (파일 쓰기 / 다른 path 읽기) - ❌ utils/lifecycle_reconciliation_manager.py 외 다른 영역 stuck case 수정 """ from __future__ import annotations import json import logging from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Literal, Optional logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Public constants # --------------------------------------------------------------------------- # 회장 §명시: 60분 이상 무응답 → stale (heartbeat threshold) SCHEDULE_FRESHNESS_THRESHOLD_MIN: int = 60 # 회장 chat — config/constants.json 의 chat_id 와 정합 (6937032012) CHAIRMAN_CHAT_ID: int = 6937032012 # schedule_history 기본 위치 (cokacdir 표준) — 사용자 홈 기준 (이식성) DEFAULT_SCHEDULE_HISTORY_DIR: Path = Path.home() / ".cokacdir" / "schedule_history" FreshnessState = Literal["FRESH", "STALE", "MISSING"] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _resolve_log_path(schedule_id: str, history_dir: Optional[Path]) -> Path: base = history_dir or DEFAULT_SCHEDULE_HISTORY_DIR return base / f"{schedule_id}.log" def _parse_ts(ts_value: object) -> Optional[datetime]: """schedule_history JSONL 의 'ts' 필드를 datetime(timezone-aware)으로 변환.""" if not isinstance(ts_value, str) or not ts_value: return None raw = ts_value.strip() # tolerate trailing Z (UTC) if raw.endswith("Z"): raw = raw[:-1] + "+00:00" try: dt = datetime.fromisoformat(raw) except ValueError: return None if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt def _coerce_now(now: Optional[datetime]) -> datetime: if now is None: return datetime.now(timezone.utc) if now.tzinfo is None: return now.replace(tzinfo=timezone.utc) return now @dataclass(frozen=True) class _LastRecord: ts: datetime chat_id: int def _read_last_chairman_record( log_path: Path, *, chairman_chat_id: int = CHAIRMAN_CHAT_ID, ) -> Optional[_LastRecord]: """schedule_history/{ID}.log 에서 chat=chairman 인 마지막 record 추출. 회장 §격리: chat_id 가 chairman 이 아닌 record 는 모두 skip (다른 chat 소유). 파일 부재 / 파싱 실패 / chairman record 0건 → None 반환. token / response 본문은 절대 dict 에 담아 반환하지 않는다 — ts / chat_id 만 노출. """ if not log_path.exists(): return None last: Optional[_LastRecord] = None try: # 스트리밍 방식 — 대용량 log 메모리 폭주 방지 (Gemini gate 권고) with log_path.open(encoding="utf-8", errors="replace") as fh: for raw_line in fh: line = raw_line.strip() if not line: continue try: record = json.loads(line) except json.JSONDecodeError: continue if not isinstance(record, dict): continue chat_value = record.get("chat_id") try: chat_int = int(chat_value) if chat_value is not None else None except (TypeError, ValueError): chat_int = None if chat_int is None or chat_int != chairman_chat_id: # 회장 §격리: 다른 chat record 는 silently skip continue ts_dt = _parse_ts(record.get("ts")) if ts_dt is None: continue # 마지막 (가장 최근 ts) record 채택. 파일은 시간순 append-only 이지만 안전을 위해 max 사용. if last is None or ts_dt > last.ts: last = _LastRecord(ts=ts_dt, chat_id=chat_int) except OSError as exc: logger.warning("schedule_history read failed: %s (%s)", log_path, exc) return None return last # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def is_schedule_id_fresh( schedule_id: str, now: Optional[datetime] = None, *, history_dir: Optional[Path] = None, threshold_min: int = SCHEDULE_FRESHNESS_THRESHOLD_MIN, chairman_chat_id: int = CHAIRMAN_CHAT_ID, ) -> tuple[bool, str]: """schedule_id 의 freshness 를 판정한다. return: (is_fresh, reason_token) - reason_token ∈ {"FRESH", "STALE", "MISSING"} (classify_freshness 와 정합) - is_fresh = (reason_token == "FRESH") 회장 §명시: ts_age = now - last_chairman_record.ts threshold = SCHEDULE_FRESHNESS_THRESHOLD_MIN (60분) ts_age < threshold → FRESH, 그 외 → STALE log 부재 / chairman record 0건 → MISSING """ state = classify_freshness( schedule_id, now=now, history_dir=history_dir, threshold_min=threshold_min, chairman_chat_id=chairman_chat_id, ) return (state == "FRESH"), state def classify_freshness( schedule_id: str, *, now: Optional[datetime] = None, history_dir: Optional[Path] = None, threshold_min: int = SCHEDULE_FRESHNESS_THRESHOLD_MIN, chairman_chat_id: int = CHAIRMAN_CHAT_ID, ) -> FreshnessState: """3 상태 분류 — FRESH / STALE / MISSING. 회장 §명시 freshness 판정 단일 진입점. lifecycle_reconciliation_manager 에서 STALE_SCHEDULE_ID stuck case 분류용으로 사용. """ if not schedule_id: return "MISSING" log_path = _resolve_log_path(schedule_id, history_dir) last = _read_last_chairman_record(log_path, chairman_chat_id=chairman_chat_id) if last is None: return "MISSING" now_dt = _coerce_now(now) age_seconds = (now_dt - last.ts).total_seconds() threshold_seconds = max(0, int(threshold_min)) * 60.0 if age_seconds < threshold_seconds: return "FRESH" return "STALE" def schedule_id_age_seconds( schedule_id: str, *, now: Optional[datetime] = None, history_dir: Optional[Path] = None, chairman_chat_id: int = CHAIRMAN_CHAT_ID, ) -> Optional[float]: """마지막 chairman record 로부터 경과 초 수. log 없으면 None. lifecycle_reconciliation_manager evidence 첨부용 (detail 메시지에서 사용). """ if not schedule_id: return None log_path = _resolve_log_path(schedule_id, history_dir) last = _read_last_chairman_record(log_path, chairman_chat_id=chairman_chat_id) if last is None: return None now_dt = _coerce_now(now) return (now_dt - last.ts).total_seconds() __all__ = [ "SCHEDULE_FRESHNESS_THRESHOLD_MIN", "CHAIRMAN_CHAT_ID", "DEFAULT_SCHEDULE_HISTORY_DIR", "FreshnessState", "is_schedule_id_fresh", "classify_freshness", "schedule_id_age_seconds", ]