diff --git a/deploy/profile-scripts/agent_spiral_watchdog.py b/deploy/profile-scripts/agent_spiral_watchdog.py new file mode 100644 index 00000000..ae18f1c3 --- /dev/null +++ b/deploy/profile-scripts/agent_spiral_watchdog.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +"""agent_spiral_watchdog.py — agent 会话螺旋检测（每 10 分钟 cron） + +盲区①的系统性补丁（2026-07-21 603288 事件：一次重评请求在 hermes agent +运行时里螺旋 35 分钟/37 轮工具/153k token，无任何组件察觉）。 + +检测：各 profile state.db 中仍开放的 api_server session， +启动超 15 分钟且 message_count > 80（正常调用 <20 条）→ 螺旋嫌疑。 + +v1 动作：XMPP 告警 + 日志（不自动 kill——kill 需要 hermes 会话取消 API， +待确认安全机制后升级 v2）。同一 session 只告警一次（state 文件去重）。 +""" +import json, os, sqlite3, glob +from datetime import datetime + +STATE_FILE = "/home/hmo/MoFin/gateway/logs/spiral_watchdog_state.json" +LOG = "/home/hmo/MoFin/gateway/logs/spiral_watchdog.log" +MIN_AGE_SEC = 15 * 60 +MIN_MESSAGES = 80 + + +def log(msg): + line = f"[{datetime.now().isoformat(timespec='seconds')}] {msg}" + print(line, flush=True) + try: + os.makedirs(os.path.dirname(LOG), exist_ok=True) + with open(LOG, "a", encoding="utf-8") as f: + f.write(line + "\n") + except Exception: + pass + + +def xmpp(msg): + try: + import urllib.request + req = urllib.request.Request( + "http://127.0.0.1:5805/", + data=json.dumps({"body": msg, "to": "hmo@yoin.fun", "type": "chat"}).encode(), + headers={"Content-Type": "application/json"}) + urllib.request.urlopen(req, timeout=5) + except Exception as e: + log(f"XMPP发送失败: {e}") + + +def load_state(): + try: + with open(STATE_FILE, encoding="utf-8") as f: + return json.load(f) + except Exception: + return {"alerted": []} + + +def save_state(st): + try: + st["alerted"] = st.get("alerted", [])[-50:] + with open(STATE_FILE, "w", encoding="utf-8") as f: + json.dump(st, f) + except Exception: + pass + + +def main(): + st = load_state() + alerted = set(st.get("alerted", [])) + now_ms = datetime.now().timestamp() * 1000 + found = 0 + + for db_path in glob.glob("/home/hmo/.hermes/profiles/*/state.db"): + profile = db_path.split("/")[-2] + try: + conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=10) + conn.row_factory = sqlite3.Row + rows = conn.execute(""" + SELECT id, started_at, message_count, input_tokens, tool_call_count + FROM sessions + WHERE source = 'api_server' + AND (ended_at IS NULL OR ended_at = 0) + ORDER BY started_at DESC LIMIT 30 + """).fetchall() + conn.close() + except Exception as e: + log(f"[{profile}] state.db 读取失败: {str(e)[:80]}") + continue + + for r in rows: + sa = r["started_at"] or 0 + age_sec = (now_ms - sa) / 1000 + if age_sec < MIN_AGE_SEC: + continue + if (r["message_count"] or 0) < MIN_MESSAGES: + continue + sid = r["id"] + if sid in alerted: + continue + found += 1 + alerted.add(sid) + msg = (f"🌀 agent 螺旋嫌疑\n" + f"profile: {profile}\n" + f"session: {sid}\n" + f"已运行: {age_sec/60:.0f} 分钟\n" + f"消息数: {r['message_count']} | 工具调用: {r['tool_call_count']} | " + f"输入token: {r['input_tokens']}\n" + f"特征类似 603288 事件（gateway agent 运行时螺旋）。" + f"如是正常长任务可忽略；否则需人工检查 gateway。") + log(f"SPIRAL: [{profile}] {sid} age={age_sec/60:.0f}min msgs={r['message_count']} tools={r['tool_call_count']}") + xmpp(msg) + + st["alerted"] = sorted(alerted) + save_state(st) + log(f"── 结束: 新告警 {found} ──") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/deploy/profile-scripts/deploy_guard.py b/deploy/profile-scripts/deploy_guard.py index a41d610c..3a12e681 100644 --- a/deploy/profile-scripts/deploy_guard.py +++ b/deploy/profile-scripts/deploy_guard.py @@ -129,6 +129,24 @@ def check_upstream_deploy(): restarted = " + dashboard已重启" if rr.returncode == 0 else " (dashboard重启失败,需人工)" actions.append(f"自动部署 session-work→master ({ahead}个提交){restarted}") log(f"UPSTREAM_DEPLOYED: {merge_out[:200]} | sync: {sync_out.stdout.strip()[-100:]}") + # ── 部署后自动验证（盲区②：提交级回归检测——知微 stale 提交事件的系统补丁）── + try: + vr = subprocess.run(["python3", f"{REPO}/scripts/verify_deployment.py"], + capture_output=True, text=True, timeout=300) + vout = (vr.stdout or "") + (vr.stderr or "") + import re as _re + m = _re.search(r"汇总:\s*(\d+)\s*项通过,\s*(\d+)\s*项失败", vout) + if m and int(m.group(2)) > 0: + fm = _re.search(r"失败项:\s*(\[.*?\])", vout) + problems.append(f"部署后验证发现 {m.group(2)} 项失败: {fm.group(1) if fm else '详见日志'}" + f"（刚合并的 {ahead} 个提交可能引入回归，需人工核查）") + log(f"POST_DEPLOY_VERIFY_FAIL: {m.group(2)} failures") + elif m: + log(f"POST_DEPLOY_VERIFY_OK: {m.group(1)} passed") + else: + log("POST_DEPLOY_VERIFY: 无法解析验证输出") + except Exception as e: + log(f"POST_DEPLOY_VERIFY_ERROR: {str(e)[:120]}") # ── 3. 硬链接一致性 ── diff --git a/deploy/profile-scripts/system_hygiene_audit.py b/deploy/profile-scripts/system_hygiene_audit.py index 05ba20f8..c21216f9 100644 --- a/deploy/profile-scripts/system_hygiene_audit.py +++ b/deploy/profile-scripts/system_hygiene_audit.py @@ -190,6 +190,57 @@ def check_dead_cron(): return issues +def check_stale_sessions(): + """盲区③：常驻 agent session 指令冻结检测（2026-07-21 发现：常驻 session 的 + system_prompt 冻结于创建时，SOUL.md 更新后旧 session 仍按旧指令行动）。 + 判定：未关闭的 api_server/cli session（msgs>50）启动时间 < 对应 profile SOUL.md + 的 mtime → 指令过期，建议 bump session id / 重启对应 bot。""" + issues = [] + import sqlite3 as _sq + for soul in glob.glob('/home/hmo/.hermes/profiles/*/SOUL.md'): + profile = soul.split('/')[-2] + state_db = os.path.join(os.path.dirname(soul), 'state.db') + if not os.path.exists(state_db): + continue + soul_mtime = os.path.getmtime(soul) + try: + conn = _sq.connect(f"file:{state_db}?mode=ro", uri=True, timeout=10) + conn.row_factory = _sq.Row + rows = conn.execute(""" + SELECT id, source, started_at, message_count FROM sessions + WHERE (ended_at IS NULL OR ended_at = 0) + AND message_count > 50 + ORDER BY message_count DESC LIMIT 10 + """).fetchall() + conn.close() + except Exception: + continue + for r in rows: + sa = (r['started_at'] or 0) / 1000 + if not (0 < sa < soul_mtime): + continue + # 只报"仍活跃"的：最近 6 小时内有消息（已被 bump 遗弃的旧 session 不报） + try: + conn2 = _sq.connect(f"file:{state_db}?mode=ro", uri=True, timeout=10) + last_msg = conn2.execute( + "SELECT MAX(created_at) FROM messages WHERE session_id=?", + (r['id'],)).fetchone()[0] + conn2.close() + last_ts = (last_msg or 0) / 1000 + if last_ts and (datetime.now().timestamp() - last_ts) > 6 * 3600: + continue + except Exception: + pass # 无法判活跃度时宁报不漏 + issues.append({ + 'type': '指令冻结session', 'file': f"[{profile}] {r['id']}", + 'action': f"常驻session({r['message_count']}条消息)启动于" + f"{datetime.fromtimestamp(sa).strftime('%m-%d %H:%M')}，" + f"早于SOUL.md最后修改({datetime.fromtimestamp(soul_mtime).strftime('%m-%d %H:%M')})" + f"→按旧指令运行中，需bump session id/重启bot", + }) + return issues + + def check_db_freshness(): """核心表新鲜度（红线10）""" issues = [] @@ -226,7 +277,8 @@ def main(): all_issues = [] for name, fn in [('分叉副本', check_diverged), ('断裂硬链接', check_broken_hardlinks), ('僵尸进程', check_zombies), ('孤儿文件', check_orphan_files), - ('死cron', check_dead_cron), ('DB新鲜度', check_db_freshness)]: + ('死cron', check_dead_cron), ('DB新鲜度', check_db_freshness), + ('指令冻结session', check_stale_sessions)]: found = fn() status = f'❌ {len(found)}' if found else '✅' print(f' {status} {name}')