From 08eef1e181f1dca9a9be8e134f61a0914cd9247d Mon Sep 17 00:00:00 2001 From: hmo Date: Mon, 20 Jul 2026 19:39:58 +0800 Subject: [PATCH] feat(self-check): L0-L4 layered self-check architecture with LLM auto-repair MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit User directive: daily not weekly; clear responsibilities per layer with no overlap; functional criteria (does the function WORK) not process liveness; problems get FIXED via LLM with file-and-report discipline (act first, report after); plus a meta-layer watching the watchers; deeply integrated into F健康. Architecture (responsibility matrix in dev-spec.md): - L0 agents_health_check (5min): port/HTTP/DB liveness + auto_heal executor - L1 functional_health_check (15min trading): per-module FUNCTIONAL criteria — output freshness/validity per REGISTRY (live_prices/market_ snapshots/mtf_cache/macro_context/bot/LLM/cron engine), not process alive - L2 system_hygiene_audit (daily 08:20, was weekly): divergence/hardlink/ zombie/orphan/dead-cron/db-freshness - L3 self_repair (30min): reads L1/L2 failures -> LLM diagnoses -> executes WHITELISTED repair actions directly (rerun_script/restart_service/ sync_links/switch_llm_key/none) -> repair_log.jsonl + XMPP report. Max 2 repairs/module/day anti-loop. LLM unavailable -> rule fallback. - L4 meta_watchdog (hourly): checks L0-L3 output freshness + L3 cron registration + XMPP bridge; direct XMPP alert as last resort Retired (overlap): Cron监护-高频 (cron_watchdog -> L3), 全局cron健康监控 (cron_health_monitor -> L1). Dashboard: mofin_health.py now emits self_check section (functional/meta/ hygiene/recent_repairs); mofin_health.html new '🩺 自检体系' tab rendering L4 layers, L1 module checks, L2 issues, L3 repair history. E2E verified: stopped xmpp bot -> L1 flagged fail -> systemd recovered -> L3 LLM correctly diagnosed 'none needed' and logged; rerun_script whitelist path executes real scripts successfully; meta_watchdog all-green after fix. --- .../functional_health_check.py | 216 +++++++++++++++ deploy/profile-scripts/meta_watchdog.py | 123 +++++++++ deploy/profile-scripts/mofin_health.py | 42 +++ deploy/profile-scripts/self_repair.py | 255 ++++++++++++++++++ docs/dev-spec.md | 22 +- scripts/notify_user6.py | 65 +++++ scripts/register_l134_jobs.py | 59 ++++ scripts/remove_cron_watchdog.py | 15 ++ scripts/test_rerun_action.py | 6 + scripts/verify_self_check_section.py | 10 + static/mofin_health.html | 62 +++++ 11 files changed, 874 insertions(+), 1 deletion(-) create mode 100644 deploy/profile-scripts/functional_health_check.py create mode 100644 deploy/profile-scripts/meta_watchdog.py create mode 100644 deploy/profile-scripts/self_repair.py create mode 100644 scripts/notify_user6.py create mode 100644 scripts/register_l134_jobs.py create mode 100644 scripts/remove_cron_watchdog.py create mode 100644 scripts/test_rerun_action.py create mode 100644 scripts/verify_self_check_section.py diff --git a/deploy/profile-scripts/functional_health_check.py b/deploy/profile-scripts/functional_health_check.py new file mode 100644 index 00000000..e47315ed --- /dev/null +++ b/deploy/profile-scripts/functional_health_check.py @@ -0,0 +1,216 @@ +#!/usr/bin/env python3 +"""functional_health_check.py — L1 功能健康检查 + +判据不是"进程活着"，而是"功能是否达成"：每个核心模块检查其 +**输出物的新鲜度和有效性**。输出物不新鲜 = 功能未达成 = 异常。 + +频率：交易时段每 15 分钟（cron），输出 gateway/logs/functional_health.json。 +责任矩阵 L1。修复由 L3 self_repair 读取本报告执行。 +""" +import os, sys, json, sqlite3, subprocess +from datetime import datetime, timedelta + +sys.path.insert(0, '/home/hmo/MoFin') + +DB = '/home/hmo/MoFin/data/mofin.db' +OUT = '/home/hmo/MoFin/gateway/logs/functional_health.json' + +# ── 功能注册表：模块 → 功能判据 ────────────────────────────── +# type: +# db_freshness: DB 表 MAX(col) 距今不超过 max_age（trading=交易时段才检查, daily=每日, always=总是） +# file_freshness: 文件 mtime 距今不超过 max_age +# bot_activity: XMPP bot 活动检查（journal） +# agent_log: gateway agent.log 最近有成功 LLM 调用 +# soft=True: 无输出可能属正常（如区间未触发），降级为 warn 而非 fail +REGISTRY = [ + {"module": "price_monitor", "function": "实时价格写入DB(live_prices)", + "check": {"type": "db_freshness", "table": "live_prices", "col": "updated_at", + "max_age_min": 6, "when": "trading"}, + "repair": {"action": "rerun_script", "script": "price_monitor.py"}}, + {"module": "market_watch", "function": "市场快照采集(market_snapshots)", + "check": {"type": "db_freshness", "table": "market_snapshots", "col": "created_at", + "max_age_min": 40, "when": "trading"}, + "repair": {"action": "rerun_script", "script": "market_watch.py"}}, + {"module": "mtf_cache", "function": "多周期均线缓存刷新(mtf_cache)", + "check": {"type": "db_freshness", "table": "mtf_cache", "col": "updated_at", + "max_age_min": 75, "when": "trading"}, + "repair": {"action": "rerun_script", "script": "refresh_mtf_cache.py"}}, + {"module": "macro_context", "function": "宏观上下文刷新(macro_context_log)", + "check": {"type": "db_freshness", "table": "macro_context_log", "col": "created_at", + "max_age_min": 45, "when": "trading"}, + "repair": {"action": "rerun_script", "script": "refresh_macro_context.py"}}, + {"module": "health_collector", "function": "健康数据采集(mofin_health.json)", + "check": {"type": "file_freshness", "path": "/home/hmo/web-dashboard/static/mofin_health.json", + "max_age_min": 20, "when": "trading"}, + "repair": {"action": "rerun_script", "script": "mofin_health.py"}}, + {"module": "premarket", "function": "盘前全量重评(premarket summary)", + "check": {"type": "file_freshness", "path": "/tmp/mofin_premarket/summary.json", + "max_age_h": 26, "when": "daily"}, + "repair": {"action": "rerun_script", "script": "premarket_full_review.py"}}, + {"module": "gateway_llm", "function": "LLM调用链可用(gateway agent.log)", + "check": {"type": "agent_log", "max_age_min": 30, "when": "always"}, + "repair": {"action": "llm_diagnose"}}, + {"module": "xmpp_bot", "function": "XMPP消息收发(bot journal)", + "check": {"type": "bot_activity", "when": "always"}, + "repair": {"action": "llm_diagnose"}}, + {"module": "cron_engine", "function": "cron调度引擎本身(有job在最近10min运行)", + "check": {"type": "cron_engine", "max_age_min": 12, "when": "trading"}, + "repair": {"action": "llm_diagnose"}}, +] + + +def is_trading_now(now): + return now.weekday() < 5 and 9 <= now.hour <= 16 + + +def check_db_freshness(conn, chk, now): + try: + row = conn.execute(f"SELECT MAX({chk['col']}) FROM {chk['table']}").fetchone() + if not row or not row[0]: + return "fail", f"表 {chk['table']} 无数据" + last = datetime.fromisoformat(str(row[0]).replace("Z", "")) + age_min = (now - last).total_seconds() / 60 + limit = chk.get("max_age_min", chk.get("max_age_h", 24) * 60) + if age_min > limit: + return ("warn" if chk.get("soft") else "fail"), \ + f"{chk['table']} 最新记录 {last.strftime('%m-%d %H:%M')}（{age_min/60:.1f}h前，阈值{limit}min）" + return "ok", f"{age_min:.0f}min前" + except Exception as e: + return "fail", f"查询失败: {str(e)[:80]}" + + +def check_file_freshness(chk, now): + p = chk["path"] + if not os.path.exists(p): + return "fail", f"文件不存在: {p}" + age_min = (now.timestamp() - os.path.getmtime(p)) / 60 + limit = chk.get("max_age_min", chk.get("max_age_h", 24) * 60) + if age_min > limit: + return ("warn" if chk.get("soft") else "fail"), \ + f"{os.path.basename(p)} {age_min/60:.1f}h 未更新（阈值{limit}min）" + return "ok", f"{age_min:.0f}min前" + + +def check_agent_log(chk, now): + try: + sys.path.insert(0, '/home/hmo/MoFin') + from xmpp_logger import _scan_agent_log + r = _scan_agent_log(now.timestamp(), "zhiwei") + if r["status"] == "ok": + age = r.get("age_sec", -1) + if age > chk["max_age_min"] * 60: + return "warn", f"最近成功LLM调用在 {age//60}min 前（无新流量，可能正常）" + return "ok", f"latency={r.get('latency')}" + return "fail", f"agent.log 最近调用失败: {r.get('error','?')[:100]}" + except Exception as e: + return "fail", f"检查失败: {e}" + + +def check_bot_activity(chk, now): + try: + # 第一判据：服务当前是否 active（权威"现在在不在跑"） + r2 = subprocess.run(["systemctl", "is-active", "xmpp-zhiwei"], + capture_output=True, text=True, timeout=5) + if r2.stdout.strip() != "active": + return "fail", "bot 服务非 active（已停止）" + # 第二判据：最近 journal 是否处于断线循环 + r = subprocess.run( + ["journalctl", "-u", "xmpp-zhiwei", "--no-pager", "--since", "30 min ago", "-o", "cat"], + capture_output=True, timeout=8, text=True) + lines = r.stdout + if "连接超时" in lines and "就绪" not in lines: + return "fail", "bot 处于断线重连循环" + if "XMPP 就绪" in lines or "已发送" in lines or "收到" in lines: + return "ok", "bot 活动正常" + return "ok", "bot 在线空闲（无新消息）" + except Exception as e: + return "fail", f"检查失败: {e}" + + +def check_cron_engine(chk, now): + """cron 引擎：最近 max_age_min 内是否有任何 job 运行过""" + try: + d = json.load(open('/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json')) + jobs = d if isinstance(d, list) else d.get('jobs', []) + latest = None + for j in jobs: + lr = j.get('last_run_at') + if lr: + try: + t = datetime.fromisoformat(lr.replace('Z', '+00:00')).replace(tzinfo=None) + if latest is None or t > latest: + latest = t + except Exception: + pass + if not latest: + return "fail", "所有 job 均无运行记录" + age_min = (now - latest).total_seconds() / 60 + if age_min > chk["max_age_min"]: + return "fail", f"最近 job 运行在 {age_min:.0f}min 前，调度引擎疑似停摆" + return "ok", f"最近 job 运行于 {age_min:.0f}min 前" + except Exception as e: + return "fail", f"检查失败: {e}" + + +def main(): + now = datetime.now() + trading = is_trading_now(now) + results = [] + conn = sqlite3.connect(DB, timeout=10) + + for item in REGISTRY: + chk = item["check"] + when = chk.get("when", "always") + if when == "trading" and not trading: + results.append({"module": item["module"], "function": item["function"], + "status": "skip", "reason": "非交易时段"}) + continue + if when == "daily": + # 每日类：只 fail 不 warn，且非交易日放宽到 72h + if now.weekday() >= 5: + chk = dict(chk) + chk["max_age_h"] = 72 + + t = chk["type"] + if t == "db_freshness": + status, reason = check_db_freshness(conn, chk, now) + elif t == "file_freshness": + status, reason = check_file_freshness(chk, now) + elif t == "agent_log": + status, reason = check_agent_log(chk, now) + elif t == "bot_activity": + status, reason = check_bot_activity(chk, now) + elif t == "cron_engine": + status, reason = check_cron_engine(chk, now) + else: + status, reason = "fail", f"未知检查类型 {t}" + + results.append({"module": item["module"], "function": item["function"], + "status": status, "reason": reason, + "repair": item.get("repair")}) + + conn.close() + + fails = [r for r in results if r["status"] == "fail"] + warns = [r for r in results if r["status"] == "warn"] + report = { + "generated_at": now.strftime("%Y-%m-%d %H:%M:%S"), + "trading_hours": trading, + "summary": {"total": len(results), "ok": sum(1 for r in results if r["status"] == "ok"), + "warn": len(warns), "fail": len(fails), + "skip": sum(1 for r in results if r["status"] == "skip")}, + "status": "fail" if fails else ("warn" if warns else "ok"), + "checks": results, + } + os.makedirs(os.path.dirname(OUT), exist_ok=True) + with open(OUT, "w", encoding="utf-8") as f: + json.dump(report, f, ensure_ascii=False, indent=2) + + print(f"功能健康: {report['summary']} status={report['status']}") + for r in results: + if r["status"] in ("fail", "warn"): + print(f" {r['status'].upper()} {r['module']}: {r['reason']}") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/deploy/profile-scripts/meta_watchdog.py b/deploy/profile-scripts/meta_watchdog.py new file mode 100644 index 00000000..2e0c24de --- /dev/null +++ b/deploy/profile-scripts/meta_watchdog.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +"""meta_watchdog.py — L4 自检系统的自检（看门狗的看门狗） + +检查 L0-L3 各自检组件本身是否在正常运转： +- L0 agents_health_check: last_health_check.json 是否 <10min +- L1 functional_health_check: functional_health.json 是否 <20min（交易时段） +- L2 system_hygiene_audit: hygiene_report.json 是否 <26h（每日） +- L3 self_repair: repair_state.json 存在性 + cron 是否注册 +- mofin_health 采集: mofin_health.json 是否 <20min（交易时段） +- XMPP 桥: :5805 是否可发（self_repair 的报备通道） + +任何一层死了 → 推 XMPP 点名（这是最后的兜底，必须直达用户）。 +频率：每小时（cron）。输出 gateway/logs/meta_watchdog.json。 +""" +import os, sys, json, subprocess +from datetime import datetime + +OUT = '/home/hmo/MoFin/gateway/logs/meta_watchdog.json' + +LAYERS = [ + {"layer": "L0 agents_health_check", "file": "/home/hmo/MoFin/gateway/temp/last_health_check.json", + "max_age_min": 10, "when": "always", + "repair": "crontab */5 agents_health_check.py 停摆，检查系统 crontab"}, + {"layer": "L1 functional_health", "file": "/home/hmo/MoFin/gateway/logs/functional_health.json", + "max_age_min": 25, "when": "trading", + "repair": "L1 cron 停摆，检查 hermes cron 引擎"}, + {"layer": "L2 hygiene_audit", "file": "/home/hmo/MoFin/gateway/logs/hygiene_report.json", + "max_age_min": 26 * 60, "when": "always", + "repair": "L2 每日审计未跑，检查 hermes cron"}, + {"layer": "L1.5 mofin_health采集", "file": "/home/hmo/web-dashboard/static/mofin_health.json", + "max_age_min": 25, "when": "trading", + "repair": "mofin_health.py 采集停摆"}, + {"layer": "L3 self_repair", "file": "/home/hmo/MoFin/gateway/logs/repair_log.jsonl", + "max_age_min": None, "when": "meta", + "repair": "self_repair cron 未注册"}, +] + + +def is_trading(now): + return now.weekday() < 5 and 9 <= now.hour <= 16 + + +def main(): + now = datetime.now() + trading = is_trading(now) + results = [] + + for L in LAYERS: + if L["when"] == "trading" and not trading: + results.append({"layer": L["layer"], "status": "skip", "reason": "非交易时段"}) + continue + if L["when"] == "meta": + # 检查 self_repair 是否注册在 cron + try: + d = json.load(open('/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json')) + jobs = d if isinstance(d, list) else d.get('jobs', []) + registered = any(j.get('script') == 'self_repair.py' and j.get('enabled', True) for j in jobs) + results.append({"layer": L["layer"], + "status": "ok" if registered else "fail", + "reason": "已注册" if registered else "未在 cron 注册"}) + except Exception as e: + results.append({"layer": L["layer"], "status": "fail", "reason": str(e)[:60]}) + continue + + f = L["file"] + if not os.path.exists(f): + results.append({"layer": L["layer"], "status": "fail", + "reason": f"输出物不存在", "repair": L["repair"]}) + continue + age_min = (now.timestamp() - os.path.getmtime(f)) / 60 + if L["max_age_min"] and age_min > L["max_age_min"]: + results.append({"layer": L["layer"], "status": "fail", + "reason": f"输出物 {age_min/60:.1f}h 未更新（阈值 {L['max_age_min']}min）", + "repair": L["repair"]}) + else: + results.append({"layer": L["layer"], "status": "ok", + "reason": f"{age_min:.0f}min 前"}) + + # XMPP 桥（报备通道）：只收 POST，GET 会 501，但任何 HTTP 响应都说明进程活着 + try: + import urllib.request + urllib.request.urlopen('http://127.0.0.1:5805/', timeout=3) + results.append({"layer": "XMPP桥 :5805", "status": "ok", "reason": "可达"}) + except urllib.error.HTTPError as e: + results.append({"layer": "XMPP桥 :5805", "status": "ok", "reason": f"可达(HTTP {e.code})"}) + except Exception: + results.append({"layer": "XMPP桥 :5805", "status": "fail", + "reason": "不可达", "repair": "重启 xmpp-zhiwei"}) + + fails = [r for r in results if r["status"] == "fail"] + report = { + "generated_at": now.strftime("%Y-%m-%d %H:%M:%S"), + "status": "fail" if fails else "ok", + "layers": results, + } + os.makedirs(os.path.dirname(OUT), exist_ok=True) + with open(OUT, 'w', encoding='utf-8') as f: + json.dump(report, f, ensure_ascii=False, indent=2) + + print(f"meta_watchdog: {report['status']}") + for r in results: + icon = {"ok": "✅", "fail": "❌", "skip": "⏭"}[r["status"]] + print(f" {icon} {r['layer']}: {r['reason']}") + + if fails: + try: + import urllib.request + lines = [f"🚨 自检系统自检（L4兜底）发现 {len(fails)} 层异常："] + for r in fails: + lines.append(f"❌ {r['layer']}: {r['reason']}") + if r.get('repair'): + lines.append(f" → 处置建议: {r['repair']}") + payload = json.dumps({'to': 'hmo@yoin.fun', 'body': '\n'.join(lines), 'type': 'chat'}).encode() + req = urllib.request.Request('http://127.0.0.1:5805/', data=payload, + headers={'Content-Type': 'application/json'}) + urllib.request.urlopen(req, timeout=5) + print(' 📨 已推 XMPP（兜底直达）') + except Exception as e: + print(f' XMPP 失败: {e}') + + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/deploy/profile-scripts/mofin_health.py b/deploy/profile-scripts/mofin_health.py index efdfaf8e..542353bf 100644 --- a/deploy/profile-scripts/mofin_health.py +++ b/deploy/profile-scripts/mofin_health.py @@ -955,6 +955,47 @@ def build_report(): "profile": j.get("profile", "?"), }) + # ── 自检体系状态（L1功能健康/L3修复记录/L4元监控/L2卫生）── + self_check = {} + LOGS = Path('/home/hmo/MoFin/gateway/logs') + try: + fh = json.loads((LOGS / 'functional_health.json').read_text(encoding='utf-8')) + self_check['functional'] = { + 'generated_at': fh.get('generated_at'), 'status': fh.get('status'), + 'summary': fh.get('summary'), 'checks': fh.get('checks', []), + } + except Exception: + self_check['functional'] = None + try: + mw = json.loads((LOGS / 'meta_watchdog.json').read_text(encoding='utf-8')) + self_check['meta_watchdog'] = { + 'generated_at': mw.get('generated_at'), 'status': mw.get('status'), + 'layers': mw.get('layers', []), + } + except Exception: + self_check['meta_watchdog'] = None + try: + hy = json.loads((LOGS / 'hygiene_report.json').read_text(encoding='utf-8')) + self_check['hygiene'] = { + 'generated_at': hy.get('generated_at'), 'status': hy.get('status'), + 'issue_count': hy.get('issue_count'), 'issues': hy.get('issues', [])[:10], + } + except Exception: + self_check['hygiene'] = None + try: + repairs = [] + rp = LOGS / 'repair_log.jsonl' + if rp.exists(): + for line in rp.read_text(encoding='utf-8').splitlines()[-10:]: + try: + repairs.append(json.loads(line)) + except Exception: + pass + repairs.reverse() + self_check['recent_repairs'] = repairs + except Exception: + self_check['recent_repairs'] = [] + # ── 写JSON ── report = { "generated_at": now.strftime("%Y-%m-%d %H:%M:%S"), @@ -963,6 +1004,7 @@ def build_report(): "json_files": json_entities, "pipelines": pipelines, "db_freshness": db_freshness, + "self_check": self_check, } out_path = WEB_DATA / "mofin_health.json" with open(out_path, "w") as f: diff --git a/deploy/profile-scripts/self_repair.py b/deploy/profile-scripts/self_repair.py new file mode 100644 index 00000000..50299da5 --- /dev/null +++ b/deploy/profile-scripts/self_repair.py @@ -0,0 +1,255 @@ +#!/usr/bin/env python3 +"""self_repair.py — L3 LLM 修复循环（报备制） + +读取 L1 functional_health.json + L2 hygiene_report.json 的失败项， +调用 LLM 诊断并**直接执行修复**（先斩后奏），然后记录日志并 XMPP 报备。 + +安全设计： +- LLM 只能从白名单动作中选择（不许任意执行代码） +- 每个模块每天最多自动修复 2 次（防修复循环） +- LLM 不可用时降级为规则默认动作（rerun_script） +- 所有动作记录 repair_log.jsonl + +频率：每 30 分钟（cron）。 +""" +import os, sys, json, subprocess, sqlite3, time +from datetime import datetime + +sys.path.insert(0, '/home/hmo/MoFin') + +FUNCTIONAL_REPORT = '/home/hmo/MoFin/gateway/logs/functional_health.json' +HYGIENE_REPORT = '/home/hmo/MoFin/gateway/logs/hygiene_report.json' +REPAIR_LOG = '/home/hmo/MoFin/gateway/logs/repair_log.jsonl' +REPAIR_STATE = '/home/hmo/MoFin/gateway/logs/repair_state.json' +GATEWAY = 'http://127.0.0.1:8643/v1/chat/completions' +SCRIPTS_DIR = '/home/hmo/.hermes/profiles/position-analyst/scripts' +MAX_REPAIRS_PER_MODULE_PER_DAY = 2 + +WHITELIST_ACTIONS = """ +可执行的白名单动作（只能选其一，不许自创）： +1. {"action": "rerun_script", "script": "<脚本名>"} — 立即重跑指定 cron 脚本（限注册表内的脚本） +2. {"action": "restart_service", "service": "<服务名>"} — systemctl 重启（限: hermes-gateway-zhiwei, xmpp-zhiwei, mofin-dashboard） +3. {"action": "sync_links"} — 重建脚本硬链接（修复 cron 跑旧代码） +4. {"action": "switch_llm_key"} — 切换 LLM key（429 限流时） +5. {"action": "none", "reason": "<原因>"} — 判断为无需动作（如非交易时段的正常空闲） +""" + + +def log_repair(entry): + os.makedirs(os.path.dirname(REPAIR_LOG), exist_ok=True) + with open(REPAIR_LOG, 'a', encoding='utf-8') as f: + f.write(json.dumps(entry, ensure_ascii=False) + '\n') + + +def load_state(): + try: + return json.load(open(REPAIR_STATE)) + except Exception: + return {} + + +def save_state(s): + with open(REPAIR_STATE, 'w') as f: + json.dump(s, f) + + +def repairs_today(state, module): + today = datetime.now().strftime('%Y-%m-%d') + return state.get(today, {}).get(module, 0) + + +def bump_repair(state, module): + today = datetime.now().strftime('%Y-%m-%d') + state.setdefault(today, {}) + state[today][module] = state[today].get(module, 0) + 1 + # 只保留最近 3 天 + for k in list(state.keys()): + if k < (datetime.now().strftime('%Y-%m-%d'))[:8] + '17': + del state[k] + save_state(state) + + +def collect_failures(): + failures = [] + try: + fh = json.load(open(FUNCTIONAL_REPORT)) + for c in fh.get('checks', []): + if c.get('status') == 'fail': + failures.append({ + 'source': 'functional', 'module': c['module'], + 'function': c['function'], 'reason': c['reason'], + 'repair_hint': c.get('repair'), + }) + except Exception: + pass + try: + hy = json.load(open(HYGIENE_REPORT)) + for i in hy.get('issues', []): + failures.append({ + 'source': 'hygiene', 'module': i.get('type'), + 'function': i.get('file') or i.get('job') or i.get('table', '?'), + 'reason': json.dumps(i, ensure_ascii=False)[:200], + 'repair_hint': {'action': 'llm_diagnose'}, + }) + except Exception: + pass + return failures + + +def run_action(action, module): + """执行白名单动作，返回 (ok, detail)""" + act = action.get('action') + try: + if act == 'rerun_script': + script = action.get('script', '') + # 只允许注册表内脚本 + if not script.endswith('.py') or '/' in script or '..' in script: + return False, f'非法脚本名: {script}' + path = os.path.join(SCRIPTS_DIR, script) + if not os.path.exists(path): + return False, f'脚本不存在: {script}' + r = subprocess.run(['python3', path], capture_output=True, text=True, + timeout=600, cwd=SCRIPTS_DIR) + tail = (r.stdout or r.stderr)[-150:] + return r.returncode == 0, f'exit={r.returncode} {tail}' + + elif act == 'restart_service': + svc = action.get('service', '') + allowed = {'hermes-gateway-zhiwei': ['sudo', '-n', 'systemctl', 'restart', svc], + 'xmpp-zhiwei': ['sudo', '-n', 'systemctl', 'restart', svc], + 'mofin-dashboard': ['sudo', '-n', 'systemctl', 'restart', svc]} + if svc not in allowed: + return False, f'服务不在白名单: {svc}' + subprocess.Popen(allowed[svc], stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, start_new_session=True) + return True, f'{svc} 重启已触发(异步)' + + elif act == 'sync_links': + r = subprocess.run(['bash', '/home/hmo/MoFin/deploy/profile-scripts/sync_profile_scripts.sh'], + capture_output=True, text=True, timeout=60) + return True, r.stdout.strip() + + elif act == 'switch_llm_key': + sys.path.insert(0, '/home/hmo/MoFin') + from xmpp_logger import best_key, switch_key, KEY_TO_PROVIDER, current_provider + bk = best_key() + if not bk: + return False, '无可用 key' + target = KEY_TO_PROVIDER.get(bk['key_id']) + if not target or target == current_provider(): + return False, f'已在最优 key ({bk["key_id"]})' + r = switch_key(bk['key_id']) + return r.get('switched', False), json.dumps(r, ensure_ascii=False)[:150] + + elif act == 'none': + return True, f'无需动作: {action.get("reason", "")}' + + return False, f'未知动作: {act}' + except subprocess.TimeoutExpired: + return False, '动作超时(600s)' + except Exception as e: + return False, f'动作异常: {str(e)[:120]}' + + +def llm_diagnose(failure): + """调 LLM 诊断并返回白名单动作。LLM 不可用 → 返回 None（走规则默认）""" + hint = failure.get('repair_hint') or {} + default_action = hint if hint.get('action') and hint['action'] != 'llm_diagnose' else None + + prompt = f"""你是 MoFin 系统的自动修复工程师。一个健康检查发现了功能异常，请诊断并给出一个修复动作。 + +【异常信息】 +模块: {failure['module']} +功能: {failure['function']} +异常表现: {failure['reason']} +来源: {failure['source']} + +【背景】 +- 系统: Linux 246, MoFin 股票分析系统 +- cron 脚本目录: {SCRIPTS_DIR} +- 数据库: /home/hmo/MoFin/data/mofin.db +- 当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}（注意是否交易时段，非交易时段部分管道停跑属正常） + +{WHITELIST_ACTIONS} + +请只输出一个 JSON 动作（不要其他文字）。如果是非交易时段的正常空闲，选 none。""" + + try: + payload = json.dumps({ + 'model': 'deepseek-v4-flash', + 'messages': [{'role': 'user', 'content': prompt}], + 'max_tokens': 300, 'stream': False, + }).encode() + import urllib.request + req = urllib.request.Request(GATEWAY, data=payload, headers={ + 'Content-Type': 'application/json', 'Authorization': 'Bearer hermes123', + 'X-Hermes-Session-Id': 'self-repair'}) + resp = urllib.request.urlopen(req, timeout=120) + text = json.loads(resp.read())['choices'][0]['message']['content'] + # 提取 JSON + import re + m = re.search(r'\{[^{}]*"action"[^{}]*\}', text) + if m: + action = json.loads(m.group(0)) + if action.get('action') in ('rerun_script', 'restart_service', 'sync_links', + 'switch_llm_key', 'none'): + return action, text[:200] + except Exception as e: + print(f' LLM 诊断不可用: {str(e)[:80]}') + return (default_action, None) if default_action else (None, None) + + +def xmpp_report(lines): + try: + import urllib.request + payload = json.dumps({'to': 'hmo@yoin.fun', 'body': '\n'.join(lines), 'type': 'chat'}).encode() + req = urllib.request.Request('http://127.0.0.1:5805/', data=payload, + headers={'Content-Type': 'application/json'}) + urllib.request.urlopen(req, timeout=5) + except Exception as e: + print(f'XMPP 失败: {e}') + + +def main(): + now = datetime.now() + print('🔧 self_repair', now.strftime('%H:%M')) + failures = collect_failures() + if not failures: + print(' ✅ 无失败项') + return + + state = load_state() + reports = [] + for f in failures: + module = f['module'] + if repairs_today(state, module) >= MAX_REPAIRS_PER_MODULE_PER_DAY: + print(f' ⏭ {module}: 今日已修复 {MAX_REPAIRS_PER_MODULE_PER_DAY} 次，跳过（防循环）') + continue + + print(f' 🔍 {module}: {f["reason"][:60]}') + action, llm_note = llm_diagnose(f) + if action is None: + print(f' 无可用动作，跳过') + log_repair({'ts': now.isoformat(), 'module': module, 'reason': f['reason'], + 'action': 'skip_no_action', 'ok': None}) + continue + + ok, detail = run_action(action, module) + bump_repair(state, module) + entry = { + 'ts': now.isoformat(), 'module': module, 'function': f['function'], + 'reason': f['reason'], 'action': action, 'ok': ok, 'detail': detail[:300], + 'llm_note': llm_note, + } + log_repair(entry) + icon = '✅' if ok else '❌' + print(f' {icon} {action.get("action")} -> {detail[:80]}') + reports.append(f'{icon} [{module}] {action.get("action")}: {detail[:80]}') + + if reports: + xmpp_report(['🔧 自检系统自动修复报备：', f'发现 {len(failures)} 个功能异常，已直接处理：'] + reports + + ['', '详情: gateway/logs/repair_log.jsonl']) + + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/docs/dev-spec.md b/docs/dev-spec.md index c986f5f6..065f36cd 100644 --- a/docs/dev-spec.md +++ b/docs/dev-spec.md @@ -156,7 +156,27 @@ specs/{module}.json --- -## 三、开发流程 +## 三、自检体系责任矩阵（L0-L4） + +> 2026-07-20 确立。原则：**每层职责单一，不重叠不疏漏；判据是"功能是否达成"，不是"进程是否活着"**。 + +| 层 | 组件 | 频率 | 职责（唯一） | 判据 | +|---|------|------|-------------|------| +| L0 执行心跳 | `agents_health_check.py`（系统 crontab） | 5min | 端口/HTTP/DB 存活 + auto_heal 执行（gateway 重启/key 切换/bot 重启） | 端口通 + HTTP 200 + DB 可写 | +| L1 功能健康 | `functional_health_check.py`（hermes cron） | 交易时段 15min | 核心模块**功能是否达成**：检查输出物新鲜度/有效性（live_prices/market_snapshots/mtf_cache/macro_context/bot/LLM/cron引擎） | 每模块注册表判据（REGISTRY），输出 `functional_health.json` | +| L2 系统卫生 | `system_hygiene_audit.py`（hermes cron） | 每日 08:20 | 分叉副本/断裂硬链接/僵尸进程/孤儿文件/死cron/DB新鲜度（红线6-10 enforcement） | 输出 `hygiene_report.json` | +| L3 修复循环 | `self_repair.py`（hermes cron） | 30min | 读 L1/L2 失败项 → LLM 诊断 → **白名单动作直接修复（报备制）** → `repair_log.jsonl` + XMPP | LLM 只能选白名单动作；每模块每天≤2次防循环 | +| L4 元监控 | `meta_watchdog.py`（hermes cron） | 每小时 | **自检系统的自检**：L0-L3 输出物新鲜度 + L3 注册状态 + XMPP 桥 | 任何一层死亡直接 XMPP 点名（最后兜底） | + +**已退休（职责被合并）**：`Cron监护-高频`（cron_watchdog，并入 L3）、`全局cron健康监控-每10分`（cron_health_monitor，并入 L1）。 + +### 报备制（L3 的核心纪律） + +发现问题 → **直接修复** → 记录日志 → XMPP 报备。不是"发现问题→报告→等指示"。LLM 介入诊断但**只能执行白名单动作**（rerun_script/restart_service/sync_links/switch_llm_key/none），不允许任意代码执行。白名单兜不住的，在 XMPP 里明确说"需要人工"。 + +--- + +## 四、开发流程 ### 新增功能流程 diff --git a/scripts/notify_user6.py b/scripts/notify_user6.py new file mode 100644 index 00000000..3deeb7ff --- /dev/null +++ b/scripts/notify_user6.py @@ -0,0 +1,65 @@ +import json, urllib.request + +msg = """[笑笑] 系统大扫除 + 防复发机制 — 全部完成 + +■ 清理执行结果 + +【小果生态 — 全灭】 +- xmpp_xiaoguo_bot(root, 8天2.5GB) 已杀，3个systemd服务已删除 +- 4个 cron job 已删（quick-scan/tunnel-watchdog/情感分析/独立扫描） +- 7个脚本 + 数据文件 + profile目录 全部归档 archive/xiaoguo-retired-20260720/ + +【第三库 — 已合并】 +- sector_snapshots +4978行、market_snapshots +187行、todos +2 并入主库 +- 第三库及其数据目录已移 trashbox + +【文件源统一 — 31个分叉全处理】 +- 库文件(strategy_lifecycle等5个)：统一到 MoFin root（cron实际import的版本） +- cron脚本(13个)：deploy为唯一源，旧副本全归档 +- 新增 sync_profile_scripts.sh 防硬链接断裂 +- crontab 三处改指 deploy 路径（cron_to_xmpp/market_watch/market_screener），已验证可跑 + +【其余清理】 +- 21个遗留JSON + 3个废弃小库(market.db/market_data.db/stock_analysis.db) 归档 +- 8个禁用cron任务删除（pa 52 jobs / default 17 jobs，禁用项清零） +- 9个旧日志 + 旧bot/bak文件归档 +- projects/ 两个旧克隆(1.4G+2.2M) 归档 + +■ 问题是怎么产生的（根因） + +1. 没有单一事实源规则——同一文件可以"合法"住在4个地方，悄悄分叉 +2. 相对路径解析——mofin_db.py 用 __file__.parent 找数据目录，每个硬链接副本解析出不同的库（三库事件根源） +3. "备份留生产目录"的习惯——.bak/迁移残留留在数据目录，被监控扫描后变成假警报 +4. 迁移只做一半——DB建了表，旧JSON的读写着还留着（price_events 就是这么分裂的） +5. 模块死了没人收尸——小果宣布"已废弃"，但bot/gateway/tunnel/cron照跑 +6. 监控查"在不在"不查"活不活"——拿遗留文件mtime当管道健康，报出"停滞14天"假警报 +7. 压根没有"系统卫生"这个检查类别——分叉/僵尸/孤儿文件没人看 + +■ 防复发机制（已落地） + +【规范层】dev-spec.md 升级 v2.0：五条红线 → 十条红线 +#6 单一事实源（只许硬链接，禁止独立副本） +#7 数据路径必须绝对（禁止__file__相对解析数据路径） +#8 备份/遗留物禁止留生产数据目录 +#9 死模块收尸六步清单（杀进程/停服务/删cron/归档脚本/归档数据/摘监控） +#10 监控查活不查在（DB表新鲜度才算健康证据） ++ 文件位置宪法：每类内容的唯一权威位置表 + +【监控层】新增 system_hygiene_audit.py +- 每周一 07:30 自动跑：分叉副本/断裂硬链接/僵尸进程/孤儿文件/死cron/DB新鲜度 六项检查 +- 结果写 hygiene_report.json，有问题自动推 XMPP +- 已注册进 cron（pa jobs.json） +- specs/hygiene.json 模块 spec 已就位 +- 首轮实测：发现5个真问题（1个漏网分叉+2个我清理时误删的default脚本+2个合法容器进程误报），已全部修复，复检六项全绿 + +这套机制保证：下次再出现"这一行那也可以"的冗余，周一早上 07:30 你就会收到 XMPP 点名。 + +代码已提交合并（246: c545a820）。""" + +payload = json.dumps({"to": "hmo@yoin.fun", "body": msg, "type": "chat"}).encode() +req = urllib.request.Request("http://127.0.0.1:5805/", data=payload, + headers={"Content-Type": "application/json"}) +try: + print("XMPP:", urllib.request.urlopen(req, timeout=10).read().decode()[:80]) +except Exception as e: + print("XMPP fail:", e) \ No newline at end of file diff --git a/scripts/register_l134_jobs.py b/scripts/register_l134_jobs.py new file mode 100644 index 00000000..bc2cb1fa --- /dev/null +++ b/scripts/register_l134_jobs.py @@ -0,0 +1,59 @@ +import json, shutil, uuid +from datetime import datetime + +jf = '/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json' +shutil.copy(jf, jf + '.bak-20260720-l134') +d = json.load(open(jf)) +is_list = isinstance(d, list) +jobs = d if is_list else d.get('jobs', []) + +def has_script(s): + return any(j.get('script') == s for j in jobs) + +def mkjob(name, script, schedule, display, next_run): + return { + "id": uuid.uuid4().hex[:12], "name": name, "prompt": "", "skills": [], "skill": None, + "model": None, "provider": None, "base_url": None, "script": script, + "no_agent": True, "context_from": None, + "schedule": {"kind": "cron", "expr": schedule, "display": display}, + "schedule_display": display, + "repeat": {"times": None, "completed": 0}, "enabled": True, "state": "scheduled", + "paused_at": None, "paused_reason": None, "created_at": datetime.now().isoformat(), + "next_run_at": next_run, "last_run_at": None, "last_status": None, + } + +# 1. 新增 L1/L3/L4 jobs +added = [] +if not has_script('functional_health_check.py'): + jobs.append(mkjob('功能健康检查-L1', 'functional_health_check.py', '*/15 9-16,20-22 * * 1-5', '*/15 9-16,20-22 * * 1-5', '2026-07-20T20:30:00+08:00')) + added.append('功能健康检查-L1(15min)') +if not has_script('self_repair.py'): + jobs.append(mkjob('LLM修复循环-L3', 'self_repair.py', '*/30 9-16,20-22 * * 1-5', '*/30 9-16,20-22 * * 1-5', '2026-07-20T20:30:00+08:00')) + added.append('LLM修复循环-L3(30min)') +if not has_script('meta_watchdog.py'): + jobs.append(mkjob('元监控-自检系统的自检-L4', 'meta_watchdog.py', '5 * * * *', '5 * * * *', '2026-07-20T20:05:00+08:00')) + added.append('元监控-L4(每小时)') + +# 2. hygiene 改每日（原每周一 07:30 → 每日 08:20） +for j in jobs: + if j.get('script') == 'system_hygiene_audit.py': + j['schedule'] = {"kind": "cron", "expr": "20 8 * * *", "display": "20 8 * * *"} + j['schedule_display'] = "20 8 * * *" + j['name'] = '系统卫生审计-每日' + j['next_run_at'] = '2026-07-21T08:20:00+08:00' + added.append('hygiene改每日08:20') + +# 3. 退休重叠组件：Cron监护-高频(cron_watchdog) + 全局cron健康监控(cron_health_monitor) +REMOVE = {'Cron监护-高频', '全局cron健康监控-每10分'} +removed = [j.get('name') for j in jobs if j.get('name') in REMOVE] +jobs = [j for j in jobs if j.get('name') not in REMOVE] + +if is_list: + json.dump(jobs, open(jf, 'w'), ensure_ascii=False, indent=2) +else: + d['jobs'] = jobs + json.dump(d, open(jf, 'w'), ensure_ascii=False, indent=2) + +print('added:', added) +print('removed:', removed) +print('total jobs:', len(jobs)) \ No newline at end of file diff --git a/scripts/remove_cron_watchdog.py b/scripts/remove_cron_watchdog.py new file mode 100644 index 00000000..ee345e16 --- /dev/null +++ b/scripts/remove_cron_watchdog.py @@ -0,0 +1,15 @@ +import json, shutil +jf = '/home/hmo/.hermes/cron/jobs.json' +shutil.copy(jf, jf + '.bak-20260720-l134') +d = json.load(open(jf)) +is_list = isinstance(d, list) +jobs = d if is_list else d.get('jobs', []) +REMOVE = {'Cron监护-高频'} +removed = [j.get('name') for j in jobs if j.get('name') in REMOVE] +jobs = [j for j in jobs if j.get('name') not in REMOVE] +if is_list: + json.dump(jobs, open(jf, 'w'), ensure_ascii=False, indent=2) +else: + d['jobs'] = jobs + json.dump(d, open(jf, 'w'), ensure_ascii=False, indent=2) +print('removed:', removed, '| total:', len(jobs)) \ No newline at end of file diff --git a/scripts/test_rerun_action.py b/scripts/test_rerun_action.py new file mode 100644 index 00000000..b7314f6a --- /dev/null +++ b/scripts/test_rerun_action.py @@ -0,0 +1,6 @@ +import sys +sys.path.insert(0, '/home/hmo/.hermes/profiles/position-analyst/scripts') +from self_repair import run_action +ok, detail = run_action({'action': 'rerun_script', 'script': 'mofin_health.py'}, 'health_collector') +print('ok:', ok) +print('detail:', detail[:200]) \ No newline at end of file diff --git a/scripts/verify_self_check_section.py b/scripts/verify_self_check_section.py new file mode 100644 index 00000000..ebb51148 --- /dev/null +++ b/scripts/verify_self_check_section.py @@ -0,0 +1,10 @@ +import json +d = json.load(open('/home/hmo/web-dashboard/static/mofin_health.json')) +sc = d.get('self_check', {}) +print('functional:', bool(sc.get('functional')), '| meta:', bool(sc.get('meta_watchdog')), + '| hygiene:', bool(sc.get('hygiene')), '| repairs:', len(sc.get('recent_repairs', []))) +if sc.get('functional'): + fh = sc['functional'] + print('functional status:', fh.get('status'), fh.get('summary')) +if sc.get('meta_watchdog'): + print('meta status:', sc['meta_watchdog'].get('status')) \ No newline at end of file diff --git a/static/mofin_health.html b/static/mofin_health.html index 05c39322..866aa75d 100644 --- a/static/mofin_health.html +++ b/static/mofin_health.html @@ -51,16 +51,75 @@ td.wrap{white-space:normal;font-size:11px;max-width:200px}
🔧 全部流程/Cron
🗃️ 数据实体
🔀 数据流
+
🩺 自检体系
+