feat(xmpp_monitor): dynamic key switching + coalesce auto-heal

- xmpp_logger.py:
  - Add switch_key(key_id): switch Hermes model.provider via sed + systemctl restart
  - Add current_provider() reading model: block correctly (not just first '  provider:' line)
  - Add KEY_TO_PROVIDER mapping (AgentsMeeting key_id -> Hermes provider name)
  - Add RESTART_COOLDOWN_FILE/SEC = 180s to prevent restart loops
  - auto_heal(): detect HTTP 429 / Weekly usage limit -> call best_key() -> switch_key()
  - auto_heal(): detect timeout/error -> async systemctl restart (Popen, not blocking)
  - _verify_llm(): bump timeout 25s -> 90s (cold-start gateway takes 20-40s)
  - health(): urlopen timeout 8s -> 90s (match verify window)
  - Use sudo NOPASSWD (hmo ALL=(ALL) NOPASSWD: ALL already configured)

- agents_health_check.py:
  - Replace inline pkill+Popen restart logic (caused multiple instances) with systemctl
  - Add RESTART_COOLDOWN_FILE state to skip restart within 3 min of last
  - Call xmpp_logger.auto_heal() at end of every cron cycle
  - Both inline restart and auto_heal restarts share cooldown file

- scripts/key_status.py: Reports weekly/monthly/rolling status of all 6 OCG keys
- scripts/test_llm.py: 90s timeout test (was 15s, gateway cold-start >= 30s)
- scripts/test_production.py: smoke test on /home/hmo/MoFin/ (hardlinked to web-dashboard)

Fixes:
- Old assumption 'Cloudflare blocks Python User-Agent' was WRONG.
  True cause was HTTP 429 Weekly usage limit on key5 (12hr reset window).
  Hermes silently ignored providers.X.headers config keys; only model.default_headers works.
  Config already has model.default_headers: User-Agent: curl/8.5.0 (defense in depth).
- Multiple gateway instances were caused by 3 competing systemd units
  (hermes-gateway@.service template + hermes-gateway-zhiwei.service named).
  Masked the template unit @position-analyst and @zhiwei so only the named one wins.
This commit is contained in:
hmo
2026-07-19 20:12:47 +08:00
parent 6d66cc6cf2
commit 9509b80d8b
10 changed files with 282 additions and 29 deletions
+42 -15
View File
@@ -117,26 +117,53 @@ def run():
with open(REPORT_FILE, "w", encoding="utf-8") as f: with open(REPORT_FILE, "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2) json.dump(report, f, ensure_ascii=False, indent=2)
# ── XMPP 通道健康采集 ── # ── XMPP 通道健康采集 ──
_collect_xmpp_health(now) _collect_xmpp_health(now)
# ── 自愈:Gateway 宕机自动重启 ── # ── 自愈:检测 rate-limit → 切 keyGateway 异常 → systemctl restart ──
try:
from xmpp_logger import auto_heal as _xmpp_auto_heal
heal_result = _xmpp_auto_heal()
actions = heal_result.get("actions", [])
if actions:
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] AUTO-HEAL actions:\n")
for a in actions:
f.write(f" {json.dumps(a, ensure_ascii=False)}\n")
print(f"[{now.strftime('%H:%M')}] Auto-heal: {len(actions)} action(s) — {heal_result.get('status')}")
for a in actions:
print(f"{a.get('action', '?')}: success={a.get('success', a.get('switched', '?'))}")
except Exception as e:
print(f"[{now.strftime('%H:%M')}] Auto-heal error: {e}")
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] AUTO-HEAL ERROR: {e}\n")
# ── Gateway 端口宕机兜底:异步 systemctl 重启(带 cooldown 防频繁触发)──
gateway_ok = any(r["name"] == "zhiwei_gateway" and r["health"]["ok"] for r in results) gateway_ok = any(r["name"] == "zhiwei_gateway" and r["health"]["ok"] for r in results)
if not gateway_ok: if not gateway_ok:
import time as _t
import subprocess as _sp
cooldown_file = LOGS_DIR / "last_restart.txt"
cooldown_sec = 180
try: try:
import subprocess as _sp last_ts = float(cooldown_file.read_text().strip()) if cooldown_file.exists() else 0
_sp.run(["sudo", "-n", "pkill", "-9", "-f", "position-analyst.*gateway"], except Exception:
capture_output=True, timeout=5) last_ts = 0
_sp.run(["rm", "-f", "/home/hmo/.hermes/profiles/position-analyst/gateway.lock", elapsed = _t.time() - last_ts
"/home/hmo/.hermes/profiles/position-analyst/gateway.pid"]) if elapsed < cooldown_sec:
_sp.Popen(["/home/hmo/hermes-agent/.venv/bin/python", "-m", "hermes_cli.main", with open(LOG_FILE, "a", encoding="utf-8") as f:
"-p", "position-analyst", "gateway", "run", "--replace"], f.write(f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] Gateway DOWN but cooldown {int(elapsed)}s < {cooldown_sec}s — skip restart\n")
stdout=open("/tmp/hermes_gw_auto.log", "a"), print(f"[{now.strftime('%H:%M')}] Gateway DOWN but in cooldown ({int(elapsed)}s)")
stderr=open("/tmp/hermes_gw_auto.log", "a"), else:
start_new_session=True) try:
print(f"[{now.strftime('%H:%M')}] Auto-heal: Gateway DOWN → restarted") _sp.Popen(["sudo", "-n", "systemctl", "restart", "hermes-gateway-zhiwei.service"],
except Exception as e: stdout=_sp.DEVNULL, stderr=_sp.DEVNULL, start_new_session=True)
print(f"[{now.strftime('%H:%M')}] Auto-heal error: {e}") cooldown_file.write_text(str(_t.time()))
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(f"[{now.strftime('%Y-%m-%d %H:%M:%S')}] Gateway DOWN → systemctl restart triggered (async, cooldown set)\n")
print(f"[{now.strftime('%H:%M')}] Gateway DOWN → systemctl restart triggered")
except Exception as e:
print(f"[{now.strftime('%H:%M')}] Gateway restart error: {e}")
# Handle issues # Handle issues
if issues: if issues:
+9
View File
@@ -0,0 +1,9 @@
import urllib.request, json
req = urllib.request.Request('http://127.0.0.1:8899/api/xmpp/health')
data = json.loads(urllib.request.urlopen(req, timeout=120).read())
print('status:', data.get('status'))
print('gateways:', list(k + '=' + str(v.get('alive')) for k, v in data.get('gateways', {}).items()))
print('llm:', data.get('llm_provider'))
print('best_key:', data.get('best_key', {}).get('key_id'),
'| weekly:', data.get('best_key', {}).get('weekly', {}).get('usage_percent'), '%',
'| rolling:', data.get('best_key', {}).get('rolling', {}).get('usage_percent'), '%')
+5
View File
@@ -0,0 +1,5 @@
import json, sys
line = sys.stdin.read().strip()
d = json.loads(line)
llm = d.get('llm_provider', {})
print(d['timestamp'], '| status:', d['status'], '| llm:', llm.get('status', '?'), '| best:', d.get('best_key', {}).get('key_id', '?'))
+20
View File
@@ -0,0 +1,20 @@
import urllib.request, json
req = urllib.request.Request('http://127.0.0.1:5803/api/keys')
data = json.loads(urllib.request.urlopen(req, timeout=10).read())
print(f"{'key':6} {'weekly':15} {'monthly':15} {'rolling':15} {'workspace':40}")
print('-'*95)
for k in data['keys']:
print(f"{k['key_id']:6} "
f"{k['weekly']['status']:15} "
f"{k['monthly']['status']:15} "
f"{k['rolling']['status']:15} "
f"{k['workspace_id'][:40]:40}")
best = None
for k in data['keys']:
if (k['weekly']['status'] == 'ok' and
k['monthly']['status'] == 'ok' and
k['rolling']['status'] == 'ok'):
if best is None or k['weekly']['usage_percent'] < best['weekly']['usage_percent']:
best = k
print('\nBEST:', best['key_id'] if best else 'NONE', '- all weekly rate-limited' if not best else '')
+3
View File
@@ -0,0 +1,3 @@
import sys, json
d = json.loads(sys.stdin.read())
print(d['timestamp'], d['status'], 'llm:', d['llm_provider']['status'], '| best_key:', d['best_key']['key_id'])
+7
View File
@@ -0,0 +1,7 @@
import sys
sys.path.insert(0, '/home/hmo/projects/MoFin')
import xmpp_logger as x
import json
print("=== AUTO HEAL RESULT ===")
result = x.auto_heal()
print(json.dumps(result, ensure_ascii=False, indent=2))
+9
View File
@@ -0,0 +1,9 @@
import urllib.request, json
data = json.dumps({'model':'deepseek-v4-flash','messages':[{'role':'user','content':'say hi'}],'max_tokens':10}).encode()
req = urllib.request.Request('http://127.0.0.1:8643/v1/chat/completions', data=data,
headers={'Content-Type':'application/json','Authorization':'Bearer hermes123'})
try:
resp = urllib.request.urlopen(req, timeout=90)
print('OK:', resp.read().decode()[:300])
except Exception as e:
print('FAIL:', e)
+8
View File
@@ -0,0 +1,8 @@
import sys, json
sys.path.insert(0, '/home/hmo/MoFin')
import xmpp_logger as x
print("current provider:", x.current_provider())
print("health status:", x.health().get("status"))
print("verify_llm:", x._verify_llm())
print("=== auto_heal ===")
print(json.dumps(x.auto_heal(), ensure_ascii=False, indent=2))
+11
View File
@@ -0,0 +1,11 @@
import sys
sys.path.insert(0, '/home/hmo/projects/MoFin')
import xmpp_logger as x
print("current provider:", x.current_provider())
h = x.health()
print("health status:", h.get("status"))
print("llm_provider:", h.get("llm_provider"))
print("=== best_key ===")
print(x.best_key())
print("=== verify_llm ===")
print(x._verify_llm())
+168 -14
View File
@@ -11,6 +11,7 @@ import json
import time as _time import time as _time
import subprocess as _sp import subprocess as _sp
import urllib.request as _ur import urllib.request as _ur
import re as _re
from pathlib import Path from pathlib import Path
from datetime import datetime, timedelta from datetime import datetime, timedelta
@@ -18,6 +19,116 @@ LOG_DIR = Path(__file__).resolve().parent / "gateway" / "logs"
LOG_FILE = LOG_DIR / "xmpp_messages.jsonl" LOG_FILE = LOG_DIR / "xmpp_messages.jsonl"
MAX_AGE_DAYS = 7 MAX_AGE_DAYS = 7
# Position-analyst profile config path (where model.provider is set)
HERMES_CONFIG = Path("/home/hmo/.hermes/profiles/position-analyst/config.yaml")
GATEWAY_SERVICE = "hermes-gateway-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
KEY_TO_PROVIDER = {
"key1": "ocg-new",
"key2": "ocg-old",
"key3": "ocg-3",
"key4": "ocg-key4",
"key5": "ocg-key5",
"key6": "ocg-key6",
}
def current_provider() -> 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:`).
"""
try:
in_model = False
for line in HERMES_CONFIG.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 switch_key(key_id: str) -> 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}"}
old = current_provider()
if old == provider:
return {"switched": False, "old": old, "new": provider, "detail": "already on this key"}
try:
txt = HERMES_CONFIG.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"}
HERMES_CONFIG.write_text(new_txt)
except Exception as e:
return {"switched": False, "detail": f"config edit failed: {e}"}
# systemctl restart blocks until gateway drain completes (60+s).
# Fire asynchronously via Popen to avoid blocking cron; verify later.
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()))
ok = True
detail = "restart triggered (async)"
except Exception as e:
ok = False
detail = f"restart trigger failed: {e}"
# Wait long enough for systemd to drain + restart the gateway (max 90s)
_time.sleep(45)
verify = _verify_llm()
return {
"switched": True,
"old": old,
"new": provider,
"key_id": key_id,
"restart": ok,
"verify": verify,
"detail": 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): def log_xmpp(direction, from_jid, to_jid, body, status="ok", error=None, latency_ms=0):
"""记录一条 XMPP 消息事件。 """记录一条 XMPP 消息事件。
@@ -207,12 +318,12 @@ def health():
if result["gateways"].get("zhiwei", {}).get("alive"): if result["gateways"].get("zhiwei", {}).get("alive"):
try: try:
import urllib.request import urllib.request
payload = json.dumps({"model": "gpt-4", "messages": [{"role": "user", "content": "ping"}], payload = json.dumps({"model": "deepseek-v4-flash", "messages": [{"role": "user", "content": "ping"}],
"max_tokens": 5, "stream": False}).encode() "max_tokens": 5, "stream": False}).encode()
req = urllib.request.Request("http://127.0.0.1:8643/v1/chat/completions", req = urllib.request.Request("http://127.0.0.1:8643/v1/chat/completions",
data=payload, headers={"Content-Type": "application/json", data=payload, headers={"Content-Type": "application/json",
"Authorization": "Bearer hermes123"}) "Authorization": "Bearer hermes123"})
urllib.request.urlopen(req, timeout=8) urllib.request.urlopen(req, timeout=90)
result["llm_provider"] = {"status": "ok", "latency": "fast"} result["llm_provider"] = {"status": "ok", "latency": "fast"}
except Exception as e: except Exception as e:
result["llm_provider"] = {"status": "timeout" if "timeout" in str(e).lower() else "error", result["llm_provider"] = {"status": "timeout" if "timeout" in str(e).lower() else "error",
@@ -305,31 +416,74 @@ def auto_heal():
h = health() h = health()
actions = [] actions = []
# 0. LLM Provider 异常 → 先查是否有更好的 Key # 0. LLM Provider 异常 → 先查是否有更好的 Key 可切换
if h.get("llm_provider", {}).get("status") in ("timeout", "error"): llm_status = h.get("llm_provider", {}).get("status")
llm_error = h.get("llm_provider", {}).get("error", "")
# Detect rate-limit / timeout / error
is_rate_limited = (llm_status == "rate_limited" or
"429" in llm_error or "Weekly usage" in llm_error or
"RateLimit" in llm_error)
is_timeout_or_error = llm_status in ("timeout", "error")
if is_rate_limited or is_timeout_or_error:
bk = best_key() bk = best_key()
if bk: if bk:
actions.append({ actions.append({
"action": "check_keys", "action": "check_keys",
"best_key": bk["key_id"], "best_key": bk["key_id"],
"current_provider": current_provider(),
"best_provider": KEY_TO_PROVIDER.get(bk["key_id"]),
"rolling_pct": bk["rolling"]["usage_percent"], "rolling_pct": bk["rolling"]["usage_percent"],
"weekly_pct": bk["weekly"]["usage_percent"], "weekly_pct": bk["weekly"]["usage_percent"],
"issues": bk["issues"], "issues": bk["issues"],
}) })
# 1. LLM Provider 超时 → 重启 Hermes Gateway # 如果 best key 对应的 provider 跟当前不同 → 切换
if h.get("llm_provider", {}).get("status") in ("timeout", "error"): target_provider = KEY_TO_PROVIDER.get(bk["key_id"])
current = current_provider()
if target_provider and target_provider != current:
# 只有当 best key 自身至少 weekly ok 才切,否则切过去也白搭
if bk["weekly"]["status"] == "ok":
action = switch_key(bk["key_id"])
action["action_group"] = "switch_key"
actions.append(action)
else:
actions.append({
"action": "skip_switch",
"reason": f"best key {bk['key_id']} weekly status {bk['weekly']['status']}, no improvement",
})
# 1. LLM Provider timeout/error(非429)→ 重启 Gateway(不是切 key
if is_timeout_or_error and not is_rate_limited:
gw = h.get("gateways", {}).get("zhiwei", {}) gw = h.get("gateways", {}).get("zhiwei", {})
if gw.get("alive"): if gw.get("alive"):
# Check cooldown to avoid restart loops
try: try:
r = _sp.run(["sudo", "-n", "systemctl", "restart", "hermes-gw-pa"], last_restart = 0
capture_output=True, timeout=30, text=True) if RESTART_COOLDOWN_FILE.exists():
ok = r.returncode == 0 last_restart = float(RESTART_COOLDOWN_FILE.read_text().strip() or 0)
actions.append({"action": "restart_hermes_gateway", "target": "position-analyst", except Exception:
"success": ok, "detail": "restarted" if ok else r.stderr[:100]}) last_restart = 0
except Exception as e: elapsed = _time.time() - last_restart
actions.append({"action": "restart_hermes_gateway", "target": "position-analyst", if elapsed < RESTART_COOLDOWN_SEC:
"success": False, "detail": str(e)[:100]}) 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 无出站 → 重启知微 Bot # 2. Bot 无出站 → 重启知微 Bot
ba = h.get("bot_activity", {}) ba = h.get("bot_activity", {})