673 lines
27 KiB
Python
673 lines
27 KiB
Python
#!/usr/bin/env python3
|
||
"""xmpp_logger.py — XMPP 消息日志采集器
|
||
|
||
记录所有 XMPP 通信事件到 JSONL 日志文件。通过 hook 方式接入：
|
||
只需在消息发送点加一行 log_xmpp() 调用，不动业务逻辑。
|
||
|
||
日志文件: gateway/logs/xmpp_messages.jsonl
|
||
自动轮转: 保留最近 7 天
|
||
"""
|
||
import json
|
||
import time as _time
|
||
import subprocess as _sp
|
||
import urllib.request as _ur
|
||
import re as _re
|
||
from pathlib import Path
|
||
from datetime import datetime, timedelta
|
||
|
||
LOG_DIR = Path(__file__).resolve().parent / "gateway" / "logs"
|
||
LOG_FILE = LOG_DIR / "xmpp_messages.jsonl"
|
||
MAX_AGE_DAYS = 7
|
||
|
||
# 监控的 Hermes Gateway profiles
|
||
PROFILES = {
|
||
"zhiwei": {
|
||
"config": Path("/home/hmo/.hermes/profiles/position-analyst/config.yaml"),
|
||
"agent_log": Path("/home/hmo/.hermes/profiles/position-analyst/logs/agent.log"),
|
||
"service": "hermes-gateway-zhiwei.service",
|
||
"user_service": False,
|
||
"gateway_port": 8643,
|
||
},
|
||
"default": {
|
||
"config": Path("/home/hmo/.hermes/config.yaml"),
|
||
"agent_log": Path("/home/hmo/.hermes/logs/agent.log"),
|
||
"service": "hermes-gateway.service",
|
||
"user_service": True,
|
||
"gateway_port": 8642,
|
||
},
|
||
}
|
||
# 向后兼容（旧引用）
|
||
HERMES_CONFIG = PROFILES["zhiwei"]["config"]
|
||
GATEWAY_SERVICE = PROFILES["zhiwei"]["service"]
|
||
RESTART_COOLDOWN_FILE = LOG_DIR / "last_restart.txt"
|
||
RESTART_COOLDOWN_SEC = 180 # 3 min: don't trigger another restart within 3 min of last
|
||
|
||
# Map AgentsMeeting key_id -> Hermes provider name
|
||
# NOTE: key7 (kimi) is intentionally NOT in this map — different model provider.
|
||
# zhiwei uses deepseek-v4-flash via OCG; auto_heal must never switch to kimi.
|
||
KEY_TO_PROVIDER = {
|
||
"key1": "ocg-new",
|
||
"key2": "ocg-old",
|
||
"key3": "ocg-3",
|
||
"key4": "ocg-key4",
|
||
"key5": "ocg-key5",
|
||
"key6": "ocg-key6",
|
||
}
|
||
|
||
|
||
def current_provider(profile="zhiwei") -> str | None:
|
||
"""Read current model.provider from Hermes config.
|
||
|
||
Walks the file tracking the model: block to find its nested `provider:`
|
||
(avoiding other blocks like `agent.alerts[0].provider:`).
|
||
"""
|
||
cfg_path = PROFILES.get(profile, PROFILES["zhiwei"])["config"]
|
||
try:
|
||
in_model = False
|
||
for line in cfg_path.read_text().splitlines():
|
||
stripped = line.rstrip()
|
||
# Detect top-level "model:" at column 0
|
||
if stripped == "model:" or stripped.startswith("model:") and not line.startswith(" "):
|
||
in_model = True
|
||
continue
|
||
# If we're in the model: block and hit a new column-0 key, exit
|
||
if in_model and line and not line.startswith((" ", "\t")):
|
||
in_model = False
|
||
continue
|
||
if in_model:
|
||
m = _re.match(r"^ provider:\s*(\S+)\s*$", line)
|
||
if m:
|
||
return m.group(1)
|
||
except Exception:
|
||
return None
|
||
return None
|
||
|
||
|
||
def _ensure_provider_block(target_cfg: Path, provider: str) -> bool:
|
||
"""若目标 config 缺少 provider 凭据块，从 zhiwei config（凭据单一事实源）复制注入。"""
|
||
txt = target_cfg.read_text()
|
||
if _re.search(rf"^ {provider}:\s*$", txt, flags=_re.MULTILINE):
|
||
return True
|
||
src = PROFILES["zhiwei"]["config"].read_text()
|
||
m = _re.search(rf"(^ {provider}:\n(?: .+\n)+)", src, flags=_re.MULTILINE)
|
||
if not m:
|
||
return False
|
||
block = m.group(1)
|
||
# 插到 providers: 区块第一个 provider 之前
|
||
m2 = _re.search(r"^providers:\s*$", txt, flags=_re.MULTILINE)
|
||
if not m2:
|
||
return False
|
||
txt = txt[:m2.end()] + "\n" + block + txt[m2.end():]
|
||
target_cfg.write_text(txt)
|
||
return True
|
||
|
||
|
||
def _restart_gateway(profile: str) -> dict:
|
||
"""异步重启指定 profile 的 gateway。"""
|
||
p = PROFILES[profile]
|
||
try:
|
||
if p["user_service"]:
|
||
_sp.Popen(["systemctl", "--user", "restart", p["service"]],
|
||
stdout=_sp.DEVNULL, stderr=_sp.DEVNULL, start_new_session=True)
|
||
else:
|
||
_sp.Popen(["sudo", "-n", "systemctl", "restart", p["service"]],
|
||
stdout=_sp.DEVNULL, stderr=_sp.DEVNULL, start_new_session=True)
|
||
return {"success": True, "detail": "restart triggered (async)"}
|
||
except Exception as e:
|
||
return {"success": False, "detail": str(e)[:100]}
|
||
|
||
|
||
def switch_key(key_id: str, profile: str = "zhiwei") -> dict:
|
||
"""Switch Hermes config to use the given key_id's provider, then restart Gateway."""
|
||
provider = KEY_TO_PROVIDER.get(key_id)
|
||
if not provider:
|
||
return {"switched": False, "detail": f"unknown key_id {key_id}"}
|
||
|
||
cfg_path = PROFILES[profile]["config"]
|
||
old = current_provider(profile)
|
||
if old == provider:
|
||
return {"switched": False, "old": old, "new": provider, "detail": "already on this key"}
|
||
|
||
try:
|
||
if not _ensure_provider_block(cfg_path, provider):
|
||
return {"switched": False, "detail": f"provider {provider} 凭据块缺失且无法注入"}
|
||
txt = cfg_path.read_text()
|
||
new_txt, n = _re.subn(r"^ provider:\s*\S+\s*$",
|
||
f" provider: {provider}", txt, count=1, flags=_re.MULTILINE)
|
||
if n == 0:
|
||
return {"switched": False, "detail": "no provider: line found"}
|
||
cfg_path.write_text(new_txt)
|
||
except Exception as e:
|
||
return {"switched": False, "detail": f"config edit failed: {e}"}
|
||
|
||
r = _restart_gateway(profile)
|
||
RESTART_COOLDOWN_FILE.write_text(str(_time.time()))
|
||
|
||
return {
|
||
"switched": True,
|
||
"old": old,
|
||
"new": provider,
|
||
"key_id": key_id,
|
||
"profile": profile,
|
||
"restart": r["success"],
|
||
"detail": r["detail"],
|
||
}
|
||
|
||
|
||
def _verify_llm():
|
||
"""Quick LLM ping test through Gateway."""
|
||
try:
|
||
payload = json.dumps({"model": "deepseek-v4-flash",
|
||
"messages": [{"role": "user", "content": "ping"}],
|
||
"max_tokens": 5, "stream": False}).encode()
|
||
req = _ur.Request("http://127.0.0.1:8643/v1/chat/completions",
|
||
data=payload,
|
||
headers={"Content-Type": "application/json",
|
||
"Authorization": "Bearer hermes123"})
|
||
_ur.urlopen(req, timeout=90)
|
||
return {"status": "ok"}
|
||
except Exception as e:
|
||
msg = str(e)[:200]
|
||
is_429 = "429" in msg or "RateLimit" in msg or "Weekly usage" in msg
|
||
is_timeout = "timed out" in msg.lower() or "Timeout" in msg
|
||
return {"status": "rate_limited" if is_429 else ("timeout" if is_timeout else "error"),
|
||
"error": msg}
|
||
|
||
|
||
def log_xmpp(direction, from_jid, to_jid, body, status="ok", error=None, latency_ms=0):
|
||
"""记录一条 XMPP 消息事件。
|
||
|
||
Args:
|
||
direction: "out"(发送) 或 "in"(接收)
|
||
from_jid: 发送者 JID
|
||
to_jid: 接收者 JID
|
||
body: 消息体（自动截取前 200 字预览）
|
||
status: "ok" / "error" / "timeout"
|
||
error: 错误信息（仅 status!="ok" 时）
|
||
latency_ms: 延迟（毫秒）
|
||
"""
|
||
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
entry = {
|
||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"direction": direction,
|
||
"from": from_jid,
|
||
"to": to_jid,
|
||
"body_preview": (body or "")[:200].replace("\n", " "),
|
||
"status": status,
|
||
"error": str(error)[:200] if error else None,
|
||
"latency_ms": latency_ms,
|
||
"epoch": _time.time(),
|
||
}
|
||
|
||
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
||
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
|
||
|
||
# 每次写入后检查是否需要轮转（采样：每 20 条轮转一次）
|
||
if LOG_FILE.stat().st_size > 500 * 1024: # >500KB
|
||
_rotate()
|
||
|
||
|
||
def _rotate():
|
||
"""保留最近 MAX_AGE_DAYS 天，丢弃旧日志"""
|
||
if not LOG_FILE.exists():
|
||
return
|
||
cutoff = datetime.now() - timedelta(days=MAX_AGE_DAYS)
|
||
kept = []
|
||
try:
|
||
with open(LOG_FILE, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
try:
|
||
e = json.loads(line)
|
||
if e["timestamp"][:10] >= cutoff.strftime("%Y-%m-%d"):
|
||
kept.append(line)
|
||
except Exception:
|
||
continue
|
||
with open(LOG_FILE, "w", encoding="utf-8") as f:
|
||
f.writelines(kept)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def query(since=None, agent=None, status=None, limit=50):
|
||
"""查询消息日志
|
||
|
||
Args:
|
||
since: ISO datetime string, 只返回此时间之后的消息
|
||
agent: JID 片段，筛选发送或接收方包含此字符串的消息
|
||
status: 筛选状态 "ok"/"error"/"timeout"
|
||
limit: 最大返回条数（默认 50）
|
||
"""
|
||
if not LOG_FILE.exists():
|
||
return []
|
||
results = []
|
||
try:
|
||
with open(LOG_FILE, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
try:
|
||
e = json.loads(line)
|
||
if since and e["timestamp"] < since:
|
||
continue
|
||
if agent and agent not in e["from"] and agent not in e["to"]:
|
||
continue
|
||
if status and e["status"] != status:
|
||
continue
|
||
results.append(e)
|
||
except Exception:
|
||
continue
|
||
except Exception:
|
||
pass
|
||
results.sort(key=lambda x: x["timestamp"], reverse=True)
|
||
return results[:limit]
|
||
|
||
|
||
def stats():
|
||
"""获取消息统计：今日 + 本周"""
|
||
if not LOG_FILE.exists():
|
||
return {"today": {"sent": 0, "failed": 0, "latency_avg": 0}, "week": {"sent": 0, "failed": 0}}
|
||
|
||
now = datetime.now()
|
||
today = now.strftime("%Y-%m-%d")
|
||
week_ago = (now - timedelta(days=7)).strftime("%Y-%m-%d")
|
||
|
||
latencies = []
|
||
s = {"today": {"sent": 0, "failed": 0}, "week": {"sent": 0, "failed": 0}}
|
||
|
||
try:
|
||
with open(LOG_FILE, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
try:
|
||
e = json.loads(line)
|
||
d = e["timestamp"][:10]
|
||
if d >= week_ago:
|
||
s["week"]["sent"] += 1
|
||
if e["status"] != "ok":
|
||
s["week"]["failed"] += 1
|
||
if d == today:
|
||
s["today"]["sent"] += 1
|
||
if e["status"] != "ok":
|
||
s["today"]["failed"] += 1
|
||
if e.get("latency_ms"):
|
||
latencies.append(e["latency_ms"])
|
||
except Exception:
|
||
continue
|
||
except Exception:
|
||
pass
|
||
|
||
s["today"]["latency_avg"] = round(sum(latencies) / len(latencies)) if latencies else 0
|
||
return s
|
||
|
||
|
||
def health():
|
||
"""快速健康检查：返回最后消息年龄 + 最近1h错误率 + bot 日志状态 + Hermes Gateway 状态"""
|
||
result = {"status": "no_data", "last_message_age_sec": -1, "error_rate_1h": 0,
|
||
"bot_activity": {}, "gateways": {}, "llm_provider": {}}
|
||
|
||
# 1. 检查 xmpp_messages.jsonl
|
||
now_epoch = _time.time()
|
||
if LOG_FILE.exists():
|
||
last_epoch = 0
|
||
recent_total = 0
|
||
recent_errors = 0
|
||
one_hour_ago = now_epoch - 3600
|
||
try:
|
||
with open(LOG_FILE, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
try:
|
||
e = json.loads(line)
|
||
ep = e.get("epoch", 0)
|
||
if ep > last_epoch: last_epoch = ep
|
||
if ep > one_hour_ago:
|
||
recent_total += 1
|
||
if e["status"] != "ok": recent_errors += 1
|
||
except Exception: continue
|
||
except Exception: pass
|
||
result["last_message_age_sec"] = int(now_epoch - last_epoch) if last_epoch else -1
|
||
result["error_rate_1h"] = round(recent_errors / recent_total * 100, 1) if recent_total else 0
|
||
|
||
# 2. 检查知微 Bot systemd journal + 持久化状态
|
||
try:
|
||
import subprocess
|
||
r = subprocess.run(["journalctl", "-u", "xmpp-zhiwei", "--no-pager", "--since", "30 min ago", "-o", "cat"],
|
||
capture_output=True, timeout=5, text=True)
|
||
lines = [l for l in r.stdout.split("\n") if l.strip()]
|
||
inbound = [l for l in lines if "📩 收到" in l]
|
||
outbound = [l for l in lines if "📤 发送" in l or "📤" in l]
|
||
errors = [l for l in lines if "ERROR" in l or "TimeoutError" in l or "timed out" in l]
|
||
|
||
# 解析最后一条 inbound 的时间（行首 python logging 时间戳 "2026-07-19 23:31:07,342"）
|
||
import re as _re2
|
||
def _line_epoch(line):
|
||
m = _re2.search(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})", line)
|
||
if not m:
|
||
return 0
|
||
try:
|
||
return datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S").timestamp()
|
||
except Exception:
|
||
return 0
|
||
|
||
last_inbound_ts = max((_line_epoch(l) for l in inbound), default=0)
|
||
last_outbound_ts = max((_line_epoch(l) for l in outbound), default=0)
|
||
last_error_ts = max((_line_epoch(l) for l in errors), default=0)
|
||
result["bot_activity"] = {
|
||
"inbound": len(inbound), "outbound": len(outbound), "errors": len(errors),
|
||
"last_error": errors[-1][:250] if errors else None,
|
||
"last_inbound": inbound[-1][:250] if inbound else None,
|
||
"last_outbound": outbound[-1][:200] if outbound else None,
|
||
"last_inbound_age_sec": int(now_epoch - last_inbound_ts) if last_inbound_ts else -1,
|
||
"last_outbound_age_sec": int(now_epoch - last_outbound_ts) if last_outbound_ts else -1,
|
||
"last_error_age_sec": int(now_epoch - last_error_ts) if last_error_ts else -1,
|
||
# 错误之后已有成功出站 → 该错误已被覆盖，不算当前问题
|
||
"last_error_resolved": bool(last_error_ts and last_outbound_ts > last_error_ts),
|
||
}
|
||
# 只有未覆盖的错误才降级状态
|
||
unresolved_error = last_error_ts and not (last_outbound_ts > last_error_ts)
|
||
if errors and not outbound and unresolved_error: result["status"] = "degraded"
|
||
if inbound and not outbound and unresolved_error: result["status"] = "degraded"
|
||
except Exception:
|
||
result["bot_activity"] = {"error": "journalctl failed"}
|
||
|
||
# 3. 检查 Hermes Gateway 各 profile 状态（端口检测）
|
||
try:
|
||
import socket
|
||
profiles = {"zhiwei": 8643, "mohe": 8642, "xiaoguo": 8645}
|
||
for name, port in profiles.items():
|
||
ok = False
|
||
try:
|
||
s = socket.create_connection(("127.0.0.1", port), timeout=2)
|
||
s.close()
|
||
ok = True
|
||
except Exception: pass
|
||
result["gateways"][name] = {"port": port, "alive": ok}
|
||
|
||
# LLM 健康：扫 gateway agent.log 最近一次真实调用结果（零成本，不发 LLM 请求）
|
||
# agent.log 里每次真实调用都有记录：成功 "API call #N: ... latency=Xs" / 失败 "HTTP 429..."
|
||
result["llm_provider"] = _scan_agent_log(now_epoch, "zhiwei")
|
||
# default profile（莫荷 8642）同样监控——它的 429 曾导致整组 cron 失败
|
||
result["llm_provider_default"] = _scan_agent_log(now_epoch, "default")
|
||
except Exception as e:
|
||
result["gateways"] = {"error": str(e)[:100]}
|
||
|
||
# 4. 综合判定
|
||
if result.get("status") != "degraded":
|
||
la = result.get("last_message_age_sec", -1)
|
||
er = result.get("error_rate_1h", 0)
|
||
llm = result.get("llm_provider", {}).get("status", "")
|
||
if llm in ("timeout", "error"): result["status"] = "degraded"
|
||
elif la < 0: result["status"] = "no_data"
|
||
elif la > 600: result["status"] = "critical"
|
||
elif er > 50: result["status"] = "degraded"
|
||
elif la > 0: result["status"] = "ok"
|
||
|
||
# 5. API Key 可用性
|
||
bk = best_key()
|
||
if bk:
|
||
result["best_key"] = bk
|
||
if bk["issues"]:
|
||
result["status"] = "degraded"
|
||
|
||
return result
|
||
|
||
|
||
def _scan_agent_log(now_epoch, profile="zhiwei", tail_lines=300):
|
||
"""扫 gateway agent.log 尾部，取最近一次真实 LLM 调用的结果。
|
||
|
||
成功行: "2026-07-19 20:04:42,880 INFO ... API call #4: ... latency=16.0s ..."
|
||
失败行: "2026-07-19 19:35:59 WARNING ... API call failed ... HTTP 429: Weekly usage limit reached ..."
|
||
返回 {"status": "ok"|"error"|"unknown", "latency"/"error", "age_sec"}
|
||
"""
|
||
log_path = PROFILES.get(profile, PROFILES["zhiwei"])["agent_log"]
|
||
if not log_path.exists():
|
||
return {"status": "unknown", "error": "agent.log not found"}
|
||
try:
|
||
# 只读尾部（大文件不全读）
|
||
size = log_path.stat().st_size
|
||
with open(log_path, "rb") as f:
|
||
f.seek(max(0, size - 64 * 1024))
|
||
chunk = f.read().decode("utf-8", errors="replace")
|
||
lines = [l for l in chunk.splitlines() if l.strip()][-tail_lines:]
|
||
except Exception as e:
|
||
return {"status": "unknown", "error": str(e)[:80]}
|
||
|
||
last_ok = None # (epoch, latency_str)
|
||
last_fail = None # (epoch, summary)
|
||
for line in lines:
|
||
m_ts = _re.search(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})", line)
|
||
if not m_ts:
|
||
continue
|
||
try:
|
||
ts = datetime.strptime(m_ts.group(1), "%Y-%m-%d %H:%M:%S").timestamp()
|
||
except Exception:
|
||
continue
|
||
m_ok = _re.search(r"API call #\d+.*latency=([\d.]+)s", line)
|
||
if m_ok:
|
||
last_ok = (ts, m_ok.group(1))
|
||
continue
|
||
if "API call failed" in line and ("summary=" in line or "after" in line):
|
||
m_err = _re.search(r"summary=(.+?)$", line)
|
||
summary = (m_err.group(1) if m_err else line)[:150]
|
||
last_fail = (ts, summary)
|
||
|
||
if not last_ok and not last_fail:
|
||
return {"status": "unknown", "error": "no LLM calls in log"}
|
||
|
||
if last_fail and (not last_ok or last_fail[0] > last_ok[0]):
|
||
return {"status": "error",
|
||
"error": last_fail[1],
|
||
"age_sec": int(now_epoch - last_fail[0]),
|
||
"source": "agent.log"}
|
||
return {"status": "ok",
|
||
"latency": f"{last_ok[1]}s",
|
||
"age_sec": int(now_epoch - last_ok[0]),
|
||
"source": "agent.log"}
|
||
|
||
|
||
def _fetch_keys():
|
||
"""从 AgentsMeeting Dashboard 获取可用 API Key 列表"""
|
||
try:
|
||
req = _ur.Request("http://127.0.0.1:5803/api/keys")
|
||
resp = _ur.urlopen(req, timeout=5)
|
||
data = json.loads(resp.read())
|
||
return data.get("keys", []) if data.get("ok") else []
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def best_key():
|
||
"""选择最佳可用 API Key。
|
||
|
||
优先级: rolling ok > weekly ok > monthly ok > 最低 usage
|
||
返回: {"key_id": "key6", "label": "...", "usage": {...}, "reason": "..."}
|
||
"""
|
||
keys = _fetch_keys()
|
||
if not keys:
|
||
return None
|
||
|
||
def score(k):
|
||
"""分数越低越好"""
|
||
s = 0
|
||
r = k.get("rolling", {})
|
||
w = k.get("weekly", {})
|
||
m = k.get("monthly", {})
|
||
# rate-limited 惩罚
|
||
if r.get("status") != "ok": s += 1000
|
||
if w.get("status") != "ok": s += 100
|
||
if m.get("status") != "ok": s += 10
|
||
# usage 越高越差
|
||
s += (r.get("usage_percent") or 0) * 0.01
|
||
s += (w.get("usage_percent") or 0) * 0.001
|
||
s += (m.get("usage_percent") or 0) * 0.0001
|
||
# session_expired 惩罚
|
||
if k.get("session_expired"): s += 500
|
||
return s
|
||
|
||
best = min(keys, key=score)
|
||
reasons = []
|
||
if best["rolling"]["status"] != "ok": reasons.append(f"rolling {best['rolling']['usage_percent']}%")
|
||
if best["weekly"]["status"] != "ok": reasons.append(f"weekly {best['weekly']['usage_percent']}%")
|
||
if best["monthly"]["status"] != "ok": reasons.append(f"monthly {best['monthly']['usage_percent']}%")
|
||
if best.get("session_expired"): reasons.append("session_expired")
|
||
|
||
return {
|
||
"key_id": best["key_id"],
|
||
"label": best["label"],
|
||
"masked": best.get("key_masked", ""),
|
||
"rolling": best["rolling"],
|
||
"weekly": best["weekly"],
|
||
"monthly": best["monthly"],
|
||
"session_expired": best.get("session_expired", False),
|
||
"issues": reasons,
|
||
"total_keys": len(keys),
|
||
}
|
||
|
||
|
||
def auto_heal():
|
||
"""自愈：检测并尝试修复 XMPP 通信问题。"""
|
||
h = health()
|
||
actions = []
|
||
|
||
# 0. 检查每个 profile 的 LLM 状态（zhiwei + default），429 → 切 key
|
||
bk_cache = None
|
||
for profile in ("zhiwei", "default"):
|
||
llm = h.get("llm_provider" if profile == "zhiwei" else "llm_provider_default", {})
|
||
llm_status = llm.get("status")
|
||
llm_error = llm.get("error", "") or ""
|
||
|
||
is_rate_limited = (llm_status == "rate_limited" or
|
||
"429" in llm_error or "Weekly usage" in llm_error or
|
||
"RateLimit" in llm_error)
|
||
if not is_rate_limited:
|
||
continue
|
||
|
||
if bk_cache is None:
|
||
bk_cache = best_key()
|
||
bk = bk_cache
|
||
if not bk:
|
||
continue
|
||
actions.append({
|
||
"action": "check_keys",
|
||
"profile": profile,
|
||
"best_key": bk["key_id"],
|
||
"current_provider": current_provider(profile),
|
||
"best_provider": KEY_TO_PROVIDER.get(bk["key_id"]),
|
||
"weekly_pct": bk["weekly"]["usage_percent"],
|
||
})
|
||
|
||
target_provider = KEY_TO_PROVIDER.get(bk["key_id"])
|
||
current = current_provider(profile)
|
||
if target_provider is None:
|
||
actions.append({
|
||
"action": "skip_switch",
|
||
"profile": profile,
|
||
"reason": f"best key {bk['key_id']} is non-OCG (kimi), stay on deepseek",
|
||
})
|
||
elif target_provider != current:
|
||
if bk["weekly"]["status"] == "ok":
|
||
action = switch_key(bk["key_id"], profile=profile)
|
||
action["action_group"] = "switch_key"
|
||
action["profile"] = profile
|
||
actions.append(action)
|
||
else:
|
||
actions.append({
|
||
"action": "skip_switch",
|
||
"profile": profile,
|
||
"reason": f"best key {bk['key_id']} weekly status {bk['weekly']['status']}, no improvement",
|
||
})
|
||
|
||
# zhiwei LLM timeout/error（非429）→ 重启 Gateway
|
||
llm_status = h.get("llm_provider", {}).get("status")
|
||
llm_error = h.get("llm_provider", {}).get("error", "") or ""
|
||
is_rate_limited_zw = ("429" in llm_error or "Weekly usage" in llm_error or "RateLimit" in llm_error)
|
||
is_timeout_or_error = llm_status in ("timeout", "error")
|
||
|
||
# 1. LLM Provider timeout/error（非429）→ 重启 Gateway（不是切 key）
|
||
if is_timeout_or_error and not is_rate_limited_zw:
|
||
gw = h.get("gateways", {}).get("zhiwei", {})
|
||
if gw.get("alive"):
|
||
# Check cooldown to avoid restart loops
|
||
try:
|
||
last_restart = 0
|
||
if RESTART_COOLDOWN_FILE.exists():
|
||
last_restart = float(RESTART_COOLDOWN_FILE.read_text().strip() or 0)
|
||
except Exception:
|
||
last_restart = 0
|
||
elapsed = _time.time() - last_restart
|
||
if elapsed < RESTART_COOLDOWN_SEC:
|
||
actions.append({"action": "skip_restart",
|
||
"reason": f"cooldown: last restart {int(elapsed)}s ago (need >{RESTART_COOLDOWN_SEC}s)",
|
||
"last_restart_ts": last_restart})
|
||
else:
|
||
try:
|
||
_sp.Popen(["sudo", "-n", "systemctl", "restart", GATEWAY_SERVICE],
|
||
stdout=_sp.DEVNULL, stderr=_sp.DEVNULL, start_new_session=True)
|
||
RESTART_COOLDOWN_FILE.write_text(str(_time.time()))
|
||
actions.append({"action": "restart_hermes_gateway",
|
||
"target": "position-analyst",
|
||
"success": True,
|
||
"detail": "restart triggered (async), cooldown set"})
|
||
except Exception as e:
|
||
actions.append({"action": "restart_hermes_gateway",
|
||
"target": "position-analyst",
|
||
"success": False,
|
||
"detail": str(e)[:100]})
|
||
|
||
# 2. Bot 真卡死才重启：最后一条 inbound 超过 10 分钟仍无出站，才算死（不是 LLM 慢）
|
||
# LLM 正常冷启动/工具调用需要 1-10 分钟，inbound 后立即重启会杀掉进行中的调用。
|
||
ba = h.get("bot_activity", {})
|
||
inbound_age = ba.get("last_inbound_age_sec", -1)
|
||
outbound_age = ba.get("last_outbound_age_sec", -1)
|
||
bot_truly_stuck = (
|
||
inbound_age > 600 # 最后一条用户消息已超过10分钟
|
||
and (outbound_age < 0 or outbound_age > inbound_age - 600) # 且其间没有任何出站
|
||
and ba.get("inbound", 0) > 0
|
||
)
|
||
if bot_truly_stuck:
|
||
# cooldown: bot 重启不频繁于10分钟一次
|
||
try:
|
||
last_bot_restart = 0
|
||
bf = LOG_DIR / "last_bot_restart.txt"
|
||
if bf.exists():
|
||
last_bot_restart = float(bf.read_text().strip() or 0)
|
||
except Exception:
|
||
last_bot_restart = 0
|
||
bot_elapsed = _time.time() - last_bot_restart
|
||
if bot_elapsed < 600:
|
||
actions.append({"action": "skip_bot_restart",
|
||
"reason": f"bot restart cooldown {int(bot_elapsed)}s < 600s"})
|
||
else:
|
||
try:
|
||
_sp.Popen(["sudo", "-n", "systemctl", "restart", "xmpp-zhiwei"],
|
||
stdout=_sp.DEVNULL, stderr=_sp.DEVNULL, start_new_session=True)
|
||
(LOG_DIR / "last_bot_restart.txt").write_text(str(_time.time()))
|
||
actions.append({"action": "restart_zhiwei_bot", "target": "xmpp-zhiwei",
|
||
"success": True, "detail": "restart triggered (async)"})
|
||
except Exception as e:
|
||
actions.append({"action": "restart_zhiwei_bot", "target": "xmpp-zhiwei",
|
||
"success": False, "detail": str(e)[:100]})
|
||
elif ba.get("inbound", 0) > 0 and ba.get("outbound", 0) == 0 and ba.get("errors", 0) > 0:
|
||
# inbound 后仍在处理窗口内（LLM 慢但正常），不动 bot
|
||
actions.append({"action": "bot_busy_not_stuck",
|
||
"inbound_age_sec": inbound_age,
|
||
"reason": "inbound < 10min ago — LLM still processing, no restart"})
|
||
|
||
# 3. ejabberd → 重启容器
|
||
if h.get("ejabberd", "") and "Up" not in str(h["ejabberd"]):
|
||
try:
|
||
r = _sp.run(["sudo", "-n", "docker", "restart", "ejabberd"], capture_output=True, timeout=30, text=True)
|
||
actions.append({"action": "restart_ejabberd", "target": "ejabberd",
|
||
"success": r.returncode == 0, "detail": "restarted" if r.returncode == 0 else r.stderr[:100]})
|
||
except Exception as e:
|
||
actions.append({"action": "restart_ejabberd", "target": "ejabberd",
|
||
"success": False, "detail": str(e)[:100]})
|
||
|
||
return {"actions": actions, "status": h.get("status", "unknown")}
|
||
|
||
|
||
def _verify_heal():
|
||
"""自愈后验证：等 Gateway 启动完成，检测 LLM 是否恢复"""
|
||
_time.sleep(8) # 等 Gateway 完全启动
|
||
h = health()
|
||
llm_ok = h.get("llm_provider", {}).get("status") == "ok"
|
||
gw_ok = h.get("gateways", {}).get("zhiwei", {}).get("alive", False)
|
||
return {"llm_recovered": llm_ok, "gateway_alive": gw_ok, "status": h.get("status")}
|