hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: capture live morning state before session merge

Browse Source
This commit is contained in:
知微
2026-07-20 08:50:29 +08:00
parent b5dade3b65
commit 0190c02064
21 changed files with 2409 additions and 964 deletions
Download Patch File Download Diff File Expand all files Collapse all files
xmpp_logger.py
+241 -87
View File
@@ -19,13 +19,32 @@ LOG_DIR = Path(__file__).resolve().parent / "gateway" / "logs"
LOG_FILE = LOG_DIR / "xmpp_messages.jsonl"
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"
# 监控的 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",
@@ -36,15 +55,16 @@ KEY_TO_PROVIDER = {
}
def current_provider() -> str | None:
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 HERMES_CONFIG.read_text().splitlines():
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(" "):
@@ -63,50 +83,74 @@ def current_provider() -> str | None:
return None
def switch_key(key_id: str) -> dict:
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}"}
old = current_provider()
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:
txt = HERMES_CONFIG.read_text()
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"}
HERMES_CONFIG.write_text(new_txt)
cfg_path.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()
r = _restart_gateway(profile)
RESTART_COOLDOWN_FILE.write_text(str(_time.time()))
return {
"switched": True,
"old": old,
"new": provider,
"key_id": key_id,
"restart": ok,
"verify": verify,
"detail": detail,
"profile": profile,
"restart": r["success"],
"detail": r["detail"],
}
@@ -290,14 +334,36 @@ def health():
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),
}
if errors and not outbound: result["status"] = "degraded"
if inbound and not outbound: result["status"] = "degraded"
# 只有未覆盖的错误才降级状态
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"}
@@ -314,20 +380,11 @@ def health():
except Exception: pass
result["gateways"][name] = {"port": port, "alive": ok}
# 测试知微的 LLM 调用是否可达
if result["gateways"].get("zhiwei", {}).get("alive"):
try:
import urllib.request
payload = json.dumps({"model": "deepseek-v4-flash", "messages": [{"role": "user", "content": "ping"}],
"max_tokens": 5, "stream": False}).encode()
req = urllib.request.Request("http://127.0.0.1:8643/v1/chat/completions",
data=payload, headers={"Content-Type": "application/json",
"Authorization": "Bearer hermes123"})
urllib.request.urlopen(req, timeout=90)
result["llm_provider"] = {"status": "ok", "latency": "fast"}
except Exception as e:
result["llm_provider"] = {"status": "timeout" if "timeout" in str(e).lower() else "error",
"error": str(e)[:150]}
# 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]}
@@ -352,6 +409,59 @@ def health():
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:
@@ -416,46 +526,62 @@ def auto_heal():
h = health()
actions = []
# 0. LLM Provider 异常 → 先查是否有更好的 Key 可切换
llm_status = h.get("llm_provider", {}).get("status")
llm_error = h.get("llm_provider", {}).get("error", "")
# 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 ""
# 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_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")
if is_rate_limited or is_timeout_or_error:
bk = best_key()
if bk:
actions.append({
"action": "check_keys",
"best_key": bk["key_id"],
"current_provider": current_provider(),
"best_provider": KEY_TO_PROVIDER.get(bk["key_id"]),
"rolling_pct": bk["rolling"]["usage_percent"],
"weekly_pct": bk["weekly"]["usage_percent"],
"issues": bk["issues"],
})
# 如果 best key 对应的 provider 跟当前不同 → 切换
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:
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
@@ -485,16 +611,44 @@ def auto_heal():
"success": False,
"detail": str(e)[:100]})
# 2. Bot 无出站 → 重启知微 Bot
# 2. Bot 真卡死才重启：最后一条 inbound 超过 10 分钟仍无出站，才算死（不是 LLM 慢）
# LLM 正常冷启动/工具调用需要 1-10 分钟，inbound 后立即重启会杀掉进行中的调用。
ba = h.get("bot_activity", {})
if ba.get("inbound", 0) > 0 and ba.get("outbound", 0) == 0 and ba.get("errors", 0) > 0:
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:
r = _sp.run(["sudo", "-n", "systemctl", "restart", "xmpp-zhiwei"], capture_output=True, timeout=30, text=True)
actions.append({"action": "restart_zhiwei_bot", "target": "xmpp-zhiwei",
"success": r.returncode == 0, "detail": "restarted" if r.returncode == 0 else r.stderr[:100]})
except Exception as e:
actions.append({"action": "restart_zhiwei_bot", "target": "xmpp-zhiwei",
"success": False, "detail": str(e)[:100]})
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"]):