hyowons
292 lines
12 KiB
Python
292 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""폴백 감지 → 클로(main) 텔레그램 알림 (LLM 미경유).
|
|
|
|
주 모델이 죽으면 게이트웨이가 조용히 폴백해서, 관리자님이 그 사실을 모르는 채로
|
|
품질·비용이 달라지는 문제 대응. 각 에이전트 DB(openclaw-agent.sqlite)의
|
|
trajectory_runtime_events 테이블에 쌓이는 `model.completed` 이벤트(실제 턴을 완료한
|
|
승자 모델)를 스캔한다. 텔레그램은 Bot API 에 HTTP POST 만 하는 직접 발송이라 LLM 미경유.
|
|
|
|
⚠️ **판정 기준이 2026-09-10 에 바뀌었다: provider → primary 와 다른 모델인가.**
|
|
그전엔 `provider==openrouter` 를 폴백으로 봤는데, 폴백 체인이 전부 무과금
|
|
(구독 gpt-5.6-terra/luna → openrouter/free)이 되면서 sol→terra 같은 **구독 내 폴백을
|
|
통째로 놓치게** 됐다. 과금이 0이어도 '주 모델이 죽었다'는 사실은 알아야 한다.
|
|
`_primary_models()` 로 에이전트별 primary 를 읽어 비교하고, primary 를 못 읽은
|
|
경우에만 옛 규칙으로 후퇴한다.
|
|
|
|
알림은 과금 여부로 문구가 갈린다(`_is_billed`) — 유료 경로면 🚨 즉시 점검,
|
|
무과금 폴백이면 ⚠️ 과금 없음. ⚠️ 무료 판정은 `_model_key` 정규화를 거쳐야 한다
|
|
(openrouter 는 modelId 에 provider 가 이미 붙어 와서, 안 거치면 무료가 유료로 오판된다).
|
|
|
|
- 신호원: agents/*/agent/openclaw-agent.sqlite → trajectory_runtime_events (on-disk, 무비용)
|
|
⚠️ 2026-09-10 OpenClaw 2026.9.3 업그레이드로 trajectory 가 파일(.trajectory.jsonl)에서 이 테이블로
|
|
이관됐다. 파일 glob 을 계속 보면 매치 0건이라 '폴백 없음'만 반환하며 조용히 무력화된다.
|
|
- 정상: model.completed provider=openai model=gpt-5.6-sol (codex 하네스)
|
|
- dedupe: last_ts 워터마크 이후 이벤트만, 알림은 COOLDOWN당 1회(장기 다운 스팸 방지)
|
|
- 결정론적, 실패해도 절대 raise 안 함(모니터가 워크플로 안 깨게)
|
|
|
|
CLI: python3 codex_fallback_monitor.py [--dry-run] [--status]
|
|
launchd: ai.openclaw.codex-fallback-monitor (5분 주기)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import glob
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import datetime, timezone, timedelta
|
|
|
|
KST = timezone(timedelta(hours=9))
|
|
ROOT = os.path.expanduser("~/.openclaw")
|
|
STATE_PATH = os.path.join(ROOT, "workspace/state/codex_fallback_monitor.json")
|
|
CONFIG_PATH = os.path.join(ROOT, "openclaw.json")
|
|
TELEGRAM_ACCOUNT = "default" # 클로(main) 봇
|
|
AGENT_DB_GLOB = os.path.join(ROOT, "agents/*/agent/openclaw-agent.sqlite")
|
|
|
|
LOOKBACK_SEC = 3 * 3600 # created_at 이 구간 내 이벤트만 스캔
|
|
COOLDOWN_SEC = 30 * 60 # 알림 최소 간격 (장기 다운 스팸 방지)
|
|
FALLBACK_PROVIDERS = {"openrouter"} # 정상 primary=openai(codex). openrouter 승자 = 폴백
|
|
|
|
|
|
def _now() -> float:
|
|
return datetime.now(tz=KST).timestamp()
|
|
|
|
|
|
def _parse_ts(ts: str) -> float:
|
|
"""trajectory ts (UTC ISO, e.g. 2026-07-20T01:34:11.915Z) → epoch."""
|
|
try:
|
|
return datetime.fromisoformat(ts.replace("Z", "+00:00")).timestamp()
|
|
except Exception:
|
|
return 0.0
|
|
|
|
|
|
def _load_state() -> dict:
|
|
try:
|
|
with open(STATE_PATH, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def _save_state(st: dict) -> None:
|
|
try:
|
|
os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True)
|
|
tmp = STATE_PATH + ".tmp"
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
json.dump(st, f, ensure_ascii=False, indent=2)
|
|
os.replace(tmp, STATE_PATH)
|
|
except Exception as e:
|
|
print(f"[state] save 실패: {e}", file=sys.stderr)
|
|
|
|
|
|
def _telegram_targets() -> tuple[str | None, list[str]]:
|
|
"""openclaw.json에서 클로(main) 봇 토큰 + chat_id 목록 (LLM 미경유 직접 발송용)."""
|
|
try:
|
|
with open(CONFIG_PATH, encoding="utf-8") as f:
|
|
acct = json.load(f)["channels"]["telegram"]["accounts"][TELEGRAM_ACCOUNT]
|
|
return acct.get("botToken") or None, acct.get("allowFrom") or []
|
|
except Exception as e:
|
|
print(f"[tg] config 읽기 실패: {e}", file=sys.stderr)
|
|
return None, []
|
|
|
|
|
|
def _primary_models() -> dict[str, str]:
|
|
"""에이전트별 primary 모델 id(소문자). 못 읽으면 빈 dict.
|
|
|
|
폴백 판정 기준이 2026-09-10 에 provider 에서 **primary 와 다른 모델인가**로 바뀌었다.
|
|
폴백 체인이 전부 무과금(구독 gpt-5.6-terra/luna + openrouter/free)이 되면서
|
|
provider 만 보면 sol→terra 같은 구독 내 폴백을 통째로 놓치기 때문이다.
|
|
과금은 0이어도 '주 모델이 죽었다'는 사실은 알아야 한다.
|
|
"""
|
|
try:
|
|
with open(CONFIG_PATH, encoding="utf-8") as f:
|
|
cfg = json.load(f)
|
|
except Exception as e:
|
|
print(f"[cfg] primary 조회 실패: {e}", file=sys.stderr)
|
|
return {}
|
|
ag = cfg.get("agents") or {}
|
|
fallback_primary = ((ag.get("defaults") or {}).get("model") or {}).get("primary")
|
|
out: dict[str, str] = {}
|
|
for aid, entry in (ag.get("entries") or {}).items():
|
|
p = ((entry or {}).get("model") or {}).get("primary") or fallback_primary
|
|
if p:
|
|
out[aid] = p.lower()
|
|
return out
|
|
|
|
|
|
def _model_key(provider: str | None, model_id: str | None) -> str:
|
|
"""`provider/model` 정규화(소문자).
|
|
|
|
⚠️ openrouter 는 modelId 에 이미 provider 가 붙어 온다(`openrouter/free`).
|
|
그대로 이어붙이면 `openrouter/openrouter/free` 가 되고, 무료 모델이
|
|
`_is_billed` 의 무료 목록과 안 맞아 **유료로 오판**된다(2026-09-10 실측).
|
|
"""
|
|
p = (provider or "").lower()
|
|
m = (model_id or "").lower()
|
|
if not p:
|
|
return m
|
|
return m if m.startswith(f"{p}/") else f"{p}/{m}"
|
|
|
|
|
|
def _is_billed(model_id: str) -> bool:
|
|
"""유료 과금 경로인가. openai/*=ChatGPT 구독, openrouter/free·*:free=무료."""
|
|
if model_id.startswith("openai/"):
|
|
return False
|
|
if model_id == "openrouter/free" or model_id.endswith(":free"):
|
|
return False
|
|
return True
|
|
|
|
|
|
def scan_fallbacks(since_ts: float) -> tuple[list[dict], float]:
|
|
"""since_ts 이후 model.completed 이벤트에서 폴백 건 추출.
|
|
|
|
반환: (폴백 이벤트 리스트, 관찰한 최대 이벤트 ts)
|
|
"""
|
|
events: list[dict] = []
|
|
max_ts = since_ts
|
|
primaries = _primary_models()
|
|
cutoff_ms = int((_now() - LOOKBACK_SEC) * 1000)
|
|
for path in glob.glob(AGENT_DB_GLOB):
|
|
agent = path.split("/agents/", 1)[-1].split("/", 1)[0]
|
|
try:
|
|
# 게이트웨이가 쓰는 중이므로 read-only + busy timeout
|
|
con = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=5)
|
|
try:
|
|
rows = con.execute(
|
|
"SELECT event_json FROM trajectory_runtime_events "
|
|
"WHERE created_at >= ? ORDER BY created_at",
|
|
(cutoff_ms,),
|
|
).fetchall()
|
|
finally:
|
|
con.close()
|
|
for (line,) in rows:
|
|
if '"model.completed"' not in line:
|
|
continue
|
|
try:
|
|
d = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
if d.get("type") != "model.completed":
|
|
continue
|
|
ets = _parse_ts(d.get("ts", ""))
|
|
if ets <= since_ts:
|
|
continue
|
|
if ets > max_ts:
|
|
max_ts = ets
|
|
provider = (d.get("provider") or "").lower()
|
|
model_id = _model_key(provider, d.get('modelId'))
|
|
primary = primaries.get(agent)
|
|
# primary 를 못 읽었을 때만 옛 규칙(openrouter=폴백)으로 후퇴한다.
|
|
fell_back = (model_id != primary if primary
|
|
else provider in FALLBACK_PROVIDERS)
|
|
if fell_back:
|
|
events.append({
|
|
"agent": agent,
|
|
"ts": ets,
|
|
"provider": d.get("provider"),
|
|
"model": d.get("modelId"),
|
|
"runId": d.get("runId", ""),
|
|
})
|
|
except Exception as e:
|
|
print(f"[scan] {path} 조회 실패: {e}", file=sys.stderr)
|
|
return events, max_ts
|
|
|
|
|
|
def _compose(events: list[dict]) -> str:
|
|
latest = max(events, key=lambda e: e["ts"])
|
|
when = datetime.fromtimestamp(latest["ts"], tz=KST).strftime("%m-%d %H:%M")
|
|
agents = sorted({e["agent"] for e in events})
|
|
models = sorted({_model_key(e['provider'], e['model']) for e in events})
|
|
billed = [m for m in models if _is_billed(m)]
|
|
if billed:
|
|
head = "🚨 유료 폴백 감지"
|
|
note = (f"주 모델 실패 → 유료 경로로 대체 중입니다 ({', '.join(billed)}).\n"
|
|
"→ 즉시 codex 점검 (토큰비 누수)")
|
|
else:
|
|
head = "⚠️ 폴백 감지 (과금 없음)"
|
|
note = ("주 모델 실패 → 무과금 폴백으로 대체 중입니다.\n"
|
|
"→ 과금은 없지만 codex 점검 필요")
|
|
return (
|
|
f"{head}\n"
|
|
f"최근: {when} KST\n"
|
|
f"대체모델: {', '.join(models)}\n"
|
|
f"에이전트: {', '.join(agents)} (총 {len(events)}건)\n"
|
|
f"{note}"
|
|
)
|
|
|
|
|
|
def _send_telegram(text: str) -> bool:
|
|
token, chat_ids = _telegram_targets()
|
|
if not token or not chat_ids:
|
|
print("[tg] 봇 토큰/chat_id 없음 — 발송 skip", file=sys.stderr)
|
|
return False
|
|
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
|
ok = True
|
|
for chat_id in chat_ids:
|
|
data = urllib.parse.urlencode({
|
|
"chat_id": chat_id,
|
|
"text": text[:4000],
|
|
"disable_web_page_preview": "true",
|
|
}).encode()
|
|
try:
|
|
req = urllib.request.Request(url, data=data, method="POST")
|
|
with urllib.request.urlopen(req, timeout=15) as r:
|
|
if r.status != 200:
|
|
ok = False
|
|
print(f"[tg] HTTP {r.status}", file=sys.stderr)
|
|
except Exception as e:
|
|
ok = False
|
|
print(f"[tg] 발송 오류: {e}", file=sys.stderr)
|
|
return ok
|
|
|
|
|
|
def main() -> None:
|
|
dry = "--dry-run" in sys.argv
|
|
st = _load_state()
|
|
|
|
if "--status" in sys.argv:
|
|
print(json.dumps(st, ensure_ascii=False, indent=2))
|
|
return
|
|
|
|
# 첫 실행: 과거(이미 해결된 건) 알림 방지 — 워터마크만 now로 세팅
|
|
if "last_ts" not in st:
|
|
st["last_ts"] = _now()
|
|
st["last_alert_ts"] = 0
|
|
_save_state(st)
|
|
print("[init] 첫 실행 — 워터마크 설정, 과거 건 무시")
|
|
return
|
|
|
|
since = float(st.get("last_ts", 0))
|
|
events, max_ts = scan_fallbacks(since)
|
|
st["last_ts"] = max(since, max_ts)
|
|
|
|
if not events:
|
|
_save_state(st)
|
|
print(f"[ok] 폴백 없음 (since {datetime.fromtimestamp(since, tz=KST):%m-%d %H:%M})")
|
|
return
|
|
|
|
text = _compose(events)
|
|
print(text)
|
|
|
|
now = _now()
|
|
last_alert = float(st.get("last_alert_ts", 0))
|
|
if dry:
|
|
print("[dry-run] 발송 안 함")
|
|
elif now - last_alert < COOLDOWN_SEC:
|
|
print(f"[cooldown] {int((now-last_alert)/60)}분 전 알림 — 이번엔 발송 skip")
|
|
else:
|
|
if _send_telegram(text):
|
|
st["last_alert_ts"] = now
|
|
print("[tg] 발송 완료")
|
|
_save_state(st)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception as e:
|
|
print(f"[fatal] {e}", file=sys.stderr)
|
|
sys.exit(0) # 모니터는 절대 비정상 종료로 시끄럽게 안 함
|