From d6950b098285769194ded166488cd429d02c5565 Mon Sep 17 00:00:00 2001 From: hmo Date: Sun, 19 Jul 2026 09:21:26 +0800 Subject: [PATCH] rewrite: health check, daily health, todo executor for Linux 246 deployment (native tools, no Windows deps) --- gateway/scripts/agents_daily_health.py | 234 +++++++++-------------- gateway/scripts/agents_health_check.py | 250 +++++++++++++------------ gateway/scripts/self_todo_executor.py | 106 ++++------- 3 files changed, 251 insertions(+), 339 deletions(-) diff --git a/gateway/scripts/agents_daily_health.py b/gateway/scripts/agents_daily_health.py index 4ae4140..7551941 100644 --- a/gateway/scripts/agents_daily_health.py +++ b/gateway/scripts/agents_daily_health.py @@ -1,61 +1,51 @@ -""" -agents_daily_health.py — Tier 2 全面健康检查（每日 08:00 触发） -在 Tier 1（进程/端口/HTTP）基础上增加： - - 磁盘空间检查 - - 定时任务存活检查 - - 数据新鲜度（各服务最后响应时间） - - 跨服务依赖链检查 - - XMPP 异常报告 - -基于 MoFin morning_health_check.py 的模式。 +#!/usr/bin/env python3 """ -import json, os, sys, time, subprocess, urllib.request, urllib.error -from datetime import datetime +agents_daily_health.py — Tier 2 每日全面健康检查（Linux/246 部署版） +================================================================== +每天 08:00 触发。检查范围：服务状态、磁盘空间、crontab 存活、看门狗日志新鲜度。 + +部署：Linux crontab: 0 8 * * * cd /home/hmo/AgentsMeeting/gateway/scripts && /usr/bin/python3 agents_daily_health.py >> ../logs/daily_health.log 2>&1 +""" +import json, os, sys, time, subprocess, shutil +from datetime import datetime +from urllib.request import urlopen, Request +from urllib.error import HTTPError, URLError -# 强制 stdout/stderr 为 UTF-8，避免 Windows GBK 控制台对中文/emoji 编码失败 try: sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") except Exception: - # Python <3.7 不支持 reconfigure，忽略 pass SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) GATEWAY_DIR = os.path.dirname(SCRIPT_DIR) -BASE = os.path.dirname(GATEWAY_DIR) -TEMP = os.path.join(BASE, "gateway", "temp") -LOGS = os.path.join(BASE, "gateway", "logs") -PYTHON = r"C:\Users\hmo\AppData\Local\Programs\Python\Python310\python.exe" - +TEMP = os.path.join(GATEWAY_DIR, "temp") +LOGS = os.path.join(GATEWAY_DIR, "logs") os.makedirs(LOGS, exist_ok=True) os.makedirs(TEMP, exist_ok=True) REPORT_LOG = os.path.join(LOGS, "daily_health_report.log") REPORT_FILE = os.path.join(TEMP, "last_daily_health.json") -# ── 磁盘阈值 ───────────────────────────────── -DISK_WARN_GB = 10 # 剩余 <10G → 警告 -DISK_CRIT_GB = 2 # 剩余 <2G → 严重 +# ── 阈值 ────────────────────────────────────────────── +DISK_WARN_GB = 10 +DISK_CRIT_GB = 2 -# ── 服务注册表（从共享 registry 导入）───────────── -sys.path.insert(0, SCRIPT_DIR) -from service_registry import SERVICES as _SR, TEMP as _TEMP, GATEWAY_DIR as _GD +# ── 服务列表（同 Tier1，但增加 depends_on / description）── +SERVICES = [ + {"name": "dashboard", "host": "127.0.0.1", "port": 5803, "health_url": "http://127.0.0.1:5803/api/health", "depends_on": [], "desc": "管理门户"}, + {"name": "hermes_gateway_mohe", "host": "127.0.0.1", "port": 8642, "health_url": "http://127.0.0.1:8642/v1/health", "depends_on": [], "desc": "莫荷 AI Gateway"}, + {"name": "hermes_gateway_zhiwei","host": "127.0.0.1", "port": 8643, "health_url": "http://127.0.0.1:8643/v1/health", "depends_on": [], "desc": "知微 AI Gateway"}, + {"name": "wechat_bridge", "host": "127.0.0.1", "port": 3001, "health_url": None, "depends_on": [], "desc": "微信桥接 Docker"}, + {"name": "xmpp_bot_xxm", "host": "192.168.1.16", "port": 5802, "health_url": "http://192.168.1.16:5802/health", "depends_on": ["ejabberd"],"desc": "小小莫 XMPP Bot"}, + {"name": "article_processor", "host": "192.168.1.16", "port": 5810, "health_url": "http://192.168.1.16:5810/health", "depends_on": [], "desc": "文章抓取服务"}, +] -TEMP = _TEMP -LOGS = os.path.join(_GD, "logs") -os.makedirs(LOGS, exist_ok=True) -os.makedirs(TEMP, exist_ok=True) - -SERVICES = [] -for svc in _SR: - SERVICES.append({ - "name": svc["name"], - "pid_file": svc["pid_file"], - "port": svc["port"], - "health_url": svc["health_url"], - "accept_401": svc.get("accept_401", False), - "depends_on": svc.get("depends_on", []), - }) +# ── 期望的定时任务（crontab 条目关键字）────────────────── +EXPECTED_CRON = [ + "auto_heal.py", + "agents_health_check.py", +] def log(msg): @@ -66,95 +56,60 @@ def log(msg): f.write(line + "\n") -def sizeof_fmt(n): - for unit in ("B", "K", "M", "G"): - if n < 1024: - return f"{n:.1f}{unit}" - n /= 1024 - return f"{n:.1f}T" - - -# ── 检查基元 ────────────────────────────────── -def pid_alive(pid): +def port_open(host, port, timeout=3): + import socket try: - r = subprocess.run(["tasklist", "/FI", f"PID eq {pid}", "/NH"], - capture_output=True, text=True, timeout=5, - creationflags=subprocess.CREATE_NO_WINDOW) - return str(pid) in r.stdout + s = socket.create_connection((host, port), timeout=timeout) + s.close() + return True except: return False -def port_open(port): +def http_check(url, timeout=8): + if not url: + return (True, "no_health_url") try: - r = subprocess.run(["netstat", "-ano"], capture_output=True, - text=True, timeout=5, - creationflags=subprocess.CREATE_NO_WINDOW) - return any(f":{port} " in line and "LISTENING" in line - for line in r.stdout.splitlines()) - except: - return False - - -def http_check(url, timeout=5, accept_401=False): - try: - r = urllib.request.urlopen(url, timeout=timeout) - return (r.status == 200, f"HTTP {r.status}") - except urllib.error.HTTPError as e: - if accept_401 and e.code in (401, 403): - return (True, "auth") - return (False, f"HTTP {e.code}") - except urllib.error.URLError as e: - return (False, str(e.reason)[:60]) + req = Request(url) + with urlopen(req, timeout=timeout) as r: + return (r.status == 200, f"HTTP {r.status}") + except HTTPError as e: + return (e.code in (401, 403), f"HTTP {e.code}") except Exception as e: return (False, str(e)[:60]) -def check_disk(path="C:"): - """Windows 磁盘剩余空间检查。""" +def check_disk(path="/"): + """Linux df 磁盘检查。""" try: - r = subprocess.run( - ["wmic", "LogicalDisk", "where", f"DeviceID='{path}'", - "get", "FreeSpace,Size", "/format:csv"], - capture_output=True, text=True, timeout=10 - ) - for line in r.stdout.splitlines(): - parts = [p.strip() for p in line.split(",")] - if len(parts) >= 3 and parts[1] == path: - free = int(parts[2]) - free_gb = free / (1024**3) - if free_gb < DISK_CRIT_GB: - return ("critical", f"{free_gb:.1f}GB free (threshold: {DISK_CRIT_GB}GB)") - elif free_gb < DISK_WARN_GB: - return ("warn", f"{free_gb:.1f}GB free (threshold: {DISK_WARN_GB}GB)") - else: - return ("ok", f"{free_gb:.1f}GB free") - except: - pass - return ("unknown", "disk check failed") + usage = shutil.disk_usage(path) + free_gb = usage.free / (1024 ** 3) + if free_gb < DISK_CRIT_GB: + return ("critical", f"{free_gb:.1f}GB free (threshold: {DISK_CRIT_GB}GB)") + elif free_gb < DISK_WARN_GB: + return ("warn", f"{free_gb:.1f}GB free (threshold: {DISK_WARN_GB}GB)") + else: + return ("ok", f"{free_gb:.1f}GB free") + except Exception as e: + return ("unknown", f"disk check failed: {e}") def check_cron_tasks(): - """检查关键定时任务是否启用。""" - expected = ["agents-health-check"] + """检查 crontab 中是否包含期望的定时任务。""" results = [] try: - for name in expected: - r = subprocess.run( - ["schtasks", "/Query", "/TN", name, "/FO", "CSV", "/NH"], - capture_output=True, text=True, timeout=10 - ) - if name in r.stdout: - status = "ok" if "Ready" in r.stdout or "Running" in r.stdout else "warn" - results.append({"name": name, "status": status}) + r = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=5) + cron_text = r.stdout + for name in EXPECTED_CRON: + if name in cron_text: + results.append({"name": name, "status": "ok"}) else: results.append({"name": name, "status": "missing"}) - except: - results.append({"name": "check_failed", "status": "error"}) + except Exception as e: + results.append({"name": "check_failed", "status": f"error: {e}"}) return results -# ── 主检查 ──────────────────────────────────── def main(): report = { "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), @@ -166,89 +121,72 @@ def main(): } # 0. 磁盘 - disk_level, disk_msg = check_disk("C:") + disk_level, disk_msg = check_disk("/") report["disk"] = {"level": disk_level, "message": disk_msg} if disk_level in ("warn", "critical"): - report["summary"][disk_level] += 1 report["recommendations"].append(f"磁盘空间不足: {disk_msg}") # 1. 服务检查 for svc in SERVICES: name = svc["name"] - entry = {"name": name} + host = svc["host"] + port = svc["port"] + url = svc.get("health_url") + entry = {"name": name, "desc": svc.get("desc", "")} report["services"].append(entry) report["summary"]["total"] += 1 - pid = 0 - pid_file = svc.get("pid_file") - if pid_file: - try: - with open(pid_file) as f: - pid = int(f.read().strip()) - except: - pid = 0 - - pid_ok = (pid > 0 and pid_alive(pid)) if pid_file else True # 无 PID 文件不视为异常 - port_ok = port_open(svc["port"]) + port_ok = port_open(host, port) http_ok = False http_msg = "" if port_ok: - http_ok, http_msg = http_check(svc["health_url"], accept_401=svc.get("accept_401", False)) + http_ok, http_msg = http_check(url, timeout=8) - entry["pid"] = pid if pid_ok else 0 - entry["pid_ok"] = pid_ok entry["port_ok"] = port_ok entry["http_ok"] = http_ok entry["http_detail"] = http_msg - if pid_ok and port_ok and http_ok: + if port_ok and http_ok: entry["status"] = "ok" report["summary"]["ok"] += 1 - elif pid_ok and port_ok and not http_ok: + elif port_ok and not http_ok: entry["status"] = "degraded" report["summary"]["warn"] += 1 - report["recommendations"].append(f"{name}: PID/Port 正常但 /health 异常（{http_msg}）") + report["recommendations"].append(f"{name}: 端口通但 /health 异常（{http_msg}）") else: entry["status"] = "fail" report["summary"]["fail"] += 1 - report["recommendations"].append(f"{name}: 离线（PID={pid_ok}, Port={port_ok}, HTTP={http_msg}）") + report["recommendations"].append(f"{name}: 离线（Port={port_ok}, HTTP={http_msg}）") - # 2. 定时任务检查 + # 2. Crontab 检查 report["cron_tasks"] = check_cron_tasks() for t in report["cron_tasks"]: if t["status"] == "missing": report["recommendations"].append(f"定时任务 {t['name']} 缺失") - # 3. 看门狗日志新鲜度（最后修改时间） - watchdog_log = os.path.join(LOGS, "watchdog.log") - if os.path.exists(watchdog_log): - mtime = os.path.getmtime(watchdog_log) + # 3. 看门狗日志新鲜度 + wd_log = os.path.join(LOGS, "watchdog.log") + if os.path.exists(wd_log): + mtime = os.path.getmtime(wd_log) age_hours = (time.time() - mtime) / 3600 report["watchdog_log_age_hours"] = round(age_hours, 1) if age_hours > 1: - report["recommendations"].append(f"看门狗日志 {age_hours:.1f} 小时未更新，可能看门狗已死") + report["recommendations"].append(f"看门狗日志 {age_hours:.1f}h 未更新") else: report["watchdog_log_age_hours"] = None - report["recommendations"].append("看门狗日志不存在") - # 4. 输出报告 - summary = report["summary"] - log(f"=== 每日健康检查: {summary['ok']}/{summary['total']} 正常, {summary['warn']} 降级, {summary['fail']} 离线 ===") + # 4. 输出 + s = report["summary"] + log(f"=== 每日健康: {s['ok']}/{s['total']} OK, {s['warn']} degraded, {s['fail']} fail ===") for svc in report["services"]: emoji = {"ok": "[OK]", "degraded": "[WARN]", "fail": "[DOWN]"}.get(svc["status"], "[?]") - detail = svc.get("http_detail", "") - log(f" {emoji} {svc['name']} | PID={svc['pid_ok']} Port={svc['port_ok']} HTTP={detail}") + log(f" {emoji} {svc['name']} | Port={svc['port_ok']} HTTP={svc.get('http_detail','')}") log(f" [DISK] {report['disk']['message']}") for t in report["cron_tasks"]: log(f" [CRON] {t['name']}: {t['status']}") - log(f" [WATCHDOG] 日志: {report.get('watchdog_log_age_hours', 'N/A')}h old") + for r in report["recommendations"]: + log(f" -> {r}") - if report["recommendations"]: - log("=== 建议动作 ===") - for r in report["recommendations"]: - log(f" - {r}") - - # 保存 JSON 报告 with open(REPORT_FILE, "w", encoding="utf-8") as f: json.dump(report, f, ensure_ascii=False, indent=2) diff --git a/gateway/scripts/agents_health_check.py b/gateway/scripts/agents_health_check.py index 491dd65..83c5564 100644 --- a/gateway/scripts/agents_health_check.py +++ b/gateway/scripts/agents_health_check.py @@ -1,44 +1,93 @@ +#!/usr/bin/env python3 """ -agents_health_check.py — Tier 1 快速健康检查 -每 5 分钟（Task Scheduler）触发。对所有注册服务做三合一检查： - PID存活 → 端口监听 → HTTP /health -异常时写 TODO（health_todos.jsonl），由 self_todo_executor 消费。 -正常时不输出任何噪音（silent-by-default）。 +agents_health_check.py — Tier 1 快速健康检查（Linux/246 部署版） +============================================================ +每 5 分钟触发。对所有注册服务做检查：PID存活 → 端口监听 → HTTP /health。 +异常时写 TODO 给 self_todo_executor 消费。正常时静默。 + +部署：Linux crontab: */5 * * * * cd /home/hmo/AgentsMeeting/gateway/scripts && /usr/bin/python3 agents_health_check.py >> ../logs/health_check.log 2>&1 """ -import json, os, sys, time, subprocess, urllib.request, urllib.error +import json, os, sys, socket, subprocess from datetime import datetime +from urllib.request import urlopen, Request +from urllib.error import HTTPError, URLError SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) -# 从共享注册表导入服务定义 -sys.path.insert(0, SCRIPT_DIR) -from service_registry import SERVICES as _SR, TEMP as _TEMP, GATEWAY_DIR as _GD - -TEMP = _TEMP -LOGS = os.path.join(_GD, "logs") +GATEWAY_DIR = os.path.dirname(SCRIPT_DIR) +BASE = os.path.dirname(GATEWAY_DIR) +TEMP = os.path.join(GATEWAY_DIR, "temp") +LOGS = os.path.join(GATEWAY_DIR, "logs") os.makedirs(TEMP, exist_ok=True) os.makedirs(LOGS, exist_ok=True) -# 转成 Tier 1 需要的格式（补充 fix_* 字段） -SERVICES = [] -for svc in _SR: - SERVICES.append({ - "name": svc["name"], - "pid_file": svc["pid_file"], - "port": svc["port"], - "health_url": svc["health_url"], - "accept_401": svc.get("accept_401", False), - "fix_script": svc["script"], - "fix_args": svc["args"], - "fix_cwd": svc["workdir"], - "remote": svc.get("remote", False), - }) - -TODO_FILE = os.path.join(TEMP, "health_todos.jsonl") -HEALTH_LOG = os.path.join(LOGS, "health_check_report.log") NOW = datetime.now().strftime("%Y-%m-%d %H:%M:%S") +TODO_FILE = os.path.join(TEMP, "health_todos.jsonl") +REPORT_FILE = os.path.join(TEMP, "last_health_check.json") +HEALTH_LOG = os.path.join(LOGS, "health_check_report.log") + +# ── 服务注册表（Linux 246 视角）────────────────────────────── +# 本机服务用 127.0.0.1，远程服务用实际 IP + HTTP only +SERVICES = [ + # === 本机服务（246）=== + { + "name": "dashboard", + "host": "127.0.0.1", + "port": 5803, + "health_url": "http://127.0.0.1:5803/api/health", + "fix_cmd": ["sudo", "systemctl", "restart", "agentsmeeting-dashboard"], + "remote": False, + "critical": True, + }, + { + "name": "hermes_gateway_mohe", + "host": "127.0.0.1", + "port": 8642, + "health_url": "http://127.0.0.1:8642/v1/health", + "fix_cmd": ["sudo", "systemctl", "restart", "hermes-gateway@mohe"], + "remote": False, + "critical": True, + }, + { + "name": "hermes_gateway_zhiwei", + "host": "127.0.0.1", + "port": 8643, + "health_url": "http://127.0.0.1:8643/v1/health", + "fix_cmd": ["sudo", "systemctl", "restart", "hermes-gateway@zhiwei"], + "remote": False, + "critical": True, + }, + { + "name": "wechat_bridge", + "host": "127.0.0.1", + "port": 3001, + "health_url": None, # TCP only + "fix_cmd": ["docker", "restart", "wxBotWebhook"], + "remote": False, + "critical": True, + }, + # === 远程服务（跨网 HTTP 检查）=== + { + "name": "xmpp_bot_xxm", + "host": "192.168.1.16", + "port": 5802, + "health_url": "http://192.168.1.16:5802/health", + "fix_cmd": None, # 远程服务无法自动修复 + "remote": True, + "critical": True, + }, + { + "name": "article_processor", + "host": "192.168.1.16", + "port": 5810, + "health_url": "http://192.168.1.16:5810/health", + "fix_cmd": None, + "remote": True, + "critical": False, + }, +] -# ── Helpers ────────────────────────────────────────────── +# ── 检查基元（Linux 原生）──────────────────────────────── def log(msg): ts = datetime.now().strftime("%H:%M:%S") line = f"[{ts}] {msg}" @@ -46,152 +95,113 @@ def log(msg): f.write(line + "\n") -def read_pid(path): - try: - with open(path) as f: - return int(f.read().strip()) - except: - return 0 - - def pid_alive(pid): + """发送信号 0 测试进程存活。""" try: - r = subprocess.run(["tasklist", "/FI", f"PID eq {pid}", "/NH"], - capture_output=True, text=True, timeout=5, - creationflags=subprocess.CREATE_NO_WINDOW) - return str(pid) in r.stdout - except: + os.kill(pid, 0) + return True + except (ProcessLookupError, PermissionError): return False -def port_open(port): +def port_open(host, port, timeout=3): + """socket connect 测试端口是否监听。""" try: - r = subprocess.run(["netstat", "-ano"], capture_output=True, - text=True, timeout=5, - creationflags=subprocess.CREATE_NO_WINDOW) - return any(f":{port} " in line and "LISTENING" in line - for line in r.stdout.splitlines()) - except: + s = socket.create_connection((host, port), timeout=timeout) + s.close() + return True + except (socket.timeout, ConnectionRefusedError, OSError): return False def http_check(url, timeout=5): + """HTTP /health 检查。返回 (ok, detail)。""" + if not url: + return (True, "no_health_url") try: - r = urllib.request.urlopen(url, timeout=timeout) - return (r.status == 200, f"HTTP {r.status}") - except urllib.error.HTTPError as e: - return (False, f"HTTP {e.code}") - except urllib.error.URLError as e: + req = Request(url) + with urlopen(req, timeout=timeout) as r: + return (r.status == 200, f"HTTP {r.status}") + except HTTPError as e: + return (e.code in (401, 403), f"HTTP {e.code}") + except URLError as e: return (False, str(e.reason)[:60]) except Exception as e: return (False, str(e)[:60]) -def http_check_remote(url, timeout=10): - """远程 HTTP 检查，超时更长（跨网）。""" - return http_check(url, timeout=timeout) - - -def write_todo(name, issue, fix_script, fix_args, fix_cwd): - """写一条 TODO 给 self_todo_executor 消费。 - JSONL 每行一条，含 service/issue/时间/fix_action。 - """ +def write_todo(name, issue): + """写 TODO 供 executor 消费。""" entry = { "created": NOW, "service": name, "issue": issue, - "fix_script": fix_script, - "fix_args": fix_args, - "fix_cwd": fix_cwd, "status": "pending", } with open(TODO_FILE, "a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") - log(f"TODO 已写入: {name} — {issue}") + log(f"TODO: {name} — {issue}") -# ── Main check ─────────────────────────────────────────── +# ── 主检查 ──────────────────────────────────────────── def main(): report = {"time": NOW, "services": [], "summary": {"total": 0, "ok": 0, "fail": 0}} dirty = False for svc in SERVICES: name = svc["name"] - entry = {"name": name} + host = svc["host"] + port = svc["port"] + url = svc.get("health_url") + is_remote = svc.get("remote", False) + critical = svc.get("critical", True) + + entry = {"name": name, "critical": critical} report["services"].append(entry) report["summary"]["total"] += 1 - # 1. PID check（允许无 PID 文件的服务，仅靠 port+HTTP 判断） - pid_file = svc.get("pid_file") - if pid_file: - pid = read_pid(pid_file) - pid_ok = pid > 0 and pid_alive(pid) - else: - pid = 0 - pid_ok = True # 无 PID 文件不视为异常 - entry["pid"] = pid if pid_ok else 0 - entry["pid_ok"] = pid_ok - - # 2. Port check（远程服务跳过本地端口检测） - is_remote = svc.get("remote", False) - if is_remote: - port_ok = True # 远程服务不检查本地端口 - else: - port_ok = port_open(svc["port"]) + # 1. Port check（socket connect，跨平台） + port_ok = port_open(host, port) entry["port_ok"] = port_ok - # 3. HTTP health check（远程服务直接用 HTTP 判断） - http_msg = "" - if port_ok or is_remote: - checker = http_check_remote if is_remote else http_check - ok, detail = checker(svc["health_url"]) - # accept 401/403 as healthy if configured - if not ok and svc.get("accept_401") and ("401" in detail or "403" in detail): - http_msg = "auth (alive)" - ok = True - else: - http_msg = detail if not ok else "ok" + # 2. HTTP check + if port_ok: + ok, detail = http_check(url, timeout=5 if is_remote else 8) entry["http_ok"] = ok - entry["http_detail"] = http_msg + entry["http_detail"] = detail else: entry["http_ok"] = False entry["http_detail"] = "port_closed" - # 健康判定：port+HTTP 都 OK → 健康；PID 仅作为辅助信号，不阻塞判定 + # 3. 判定：Port+HTTP OK = 健康 primary_ok = port_ok and entry.get("http_ok", False) entry["status"] = "ok" if primary_ok else "fail" if primary_ok: - if not pid_ok: - entry["pid_warning"] = True # PID 异常但服务在正常运行 report["summary"]["ok"] += 1 else: report["summary"]["fail"] += 1 dirty = True - # 收集具体失败原因 reasons = [] - if not port_ok and not is_remote: reasons.append("port_closed") - if not entry.get("http_ok"): reasons.append(f"http_{http_msg}") + if not port_ok: + reasons.append("port_closed") + if port_ok and not entry.get("http_ok"): + reasons.append(f"http_{entry.get('http_detail','?')}") issue = f"异常: {' + '.join(reasons)}" - write_todo(name, issue, svc["fix_script"], svc.get("fix_args", []), svc["fix_cwd"]) + write_todo(name, issue) - # 只在有异常时写完整报告 + 输出到 stdout（给 Task Scheduler 日志） + # 输出报告 + summary = report["summary"] if dirty: - summary = report["summary"] - log(f"=== 健康检查: {summary['ok']}/{summary['total']} 正常, {summary['fail']} 异常 ===") - # 只输出异常服务到 stdout（silent-by-default） - for svc in report["services"]: - if svc["status"] != "ok": - print(f"[FAIL] {svc['name']}: PORT={svc['port_ok']} HTTP={svc.get('http_detail','?')}") - elif svc.get("pid_warning"): - print(f"[WARN] {svc['name']}: PID异常但服务正常") + log(f"=== 健康检查: {summary['ok']}/{summary['total']} OK, {summary['fail']} FAIL ===") + for s in report["services"]: + if s["status"] == "fail": + print(f"[FAIL] {s['name']}: PORT={s['port_ok']} HTTP={s.get('http_detail','?')}") else: - # 全正常→完全静默（silent-by-default） - pass + log(f"=== 健康检查: {summary['ok']}/{summary['total']} 全部正常 ===") - # 保留报告到文件（供 dashboard 读取） - with open(os.path.join(TEMP, "last_health_check.json"), "w", encoding="utf-8") as f: - json.dump(report, f, ensure_ascii=False) + with open(REPORT_FILE, "w", encoding="utf-8") as f: + json.dump(report, f, ensure_ascii=False, indent=2) if __name__ == "__main__": diff --git a/gateway/scripts/self_todo_executor.py b/gateway/scripts/self_todo_executor.py index f44d863..6aa528d 100644 --- a/gateway/scripts/self_todo_executor.py +++ b/gateway/scripts/self_todo_executor.py @@ -1,13 +1,12 @@ +#!/usr/bin/env python3 """ -self_todo_executor.py — TODO 自修复执行器 - - 轮询 health_todos.jsonl 中 status=pending 的条目 - - 执行 fix_action（启动脚本） - - 成功→标记 completed - - 失败→ escalation 到 LLM（通过 xmpp_bot 通知） +self_todo_executor.py — TODO 自修复执行器（Linux/246 部署版） +============================================================= +每 10 分钟触发。轮询 health_todos.jsonl 中 pending 条目，执行修复，标记结果。 -MoFin self_todo_executor.py 的 AgentsMeeting 适配版。 +部署：Linux crontab: */10 * * * * cd /home/hmo/AgentsMeeting/gateway/scripts && /usr/bin/python3 self_todo_executor.py >> ../logs/todo_executor.log 2>&1 """ -import json, os, sys, time, subprocess, shlex, urllib.request +import json, os, sys, subprocess from datetime import datetime try: @@ -18,14 +17,24 @@ except Exception: SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) GATEWAY_DIR = os.path.dirname(SCRIPT_DIR) -BASE = os.path.dirname(GATEWAY_DIR) -TEMP = os.path.join(BASE, "gateway", "temp") -LOGS = os.path.join(BASE, "gateway", "logs") -PYTHON = r"C:\Users\hmo\AppData\Local\Programs\Python\Python310\python.exe" +TEMP = os.path.join(GATEWAY_DIR, "temp") +LOGS = os.path.join(GATEWAY_DIR, "logs") +os.makedirs(LOGS, exist_ok=True) TODO_FILE = os.path.join(TEMP, "health_todos.jsonl") EXECUTOR_LOG = os.path.join(LOGS, "todo_executor.log") +# ── 服务修复命令映射（key = service name）───────────────── +FIX_MAP = { + "dashboard": ["sudo", "systemctl", "restart", "agentsmeeting-dashboard"], + "hermes_gateway_mohe": ["sudo", "systemctl", "restart", "hermes-gateway@mohe"], + "hermes_gateway_zhiwei": ["sudo", "systemctl", "restart", "hermes-gateway@zhiwei"], + "wechat_bridge": ["docker", "restart", "wxBotWebhook"], + # 远程服务无本地修复命令 + "xmpp_bot_xxm": None, + "article_processor": None, +} + def log(msg): ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") @@ -36,7 +45,6 @@ def log(msg): def read_pending_todos(): - """读取 health_todos.jsonl 中所有 status=pending 的条目。""" todos = [] if not os.path.exists(TODO_FILE): return todos @@ -53,12 +61,11 @@ def read_pending_todos(): except json.JSONDecodeError: continue except Exception as e: - log(f"读取 TODO 文件失败: {e}") + log(f"读取 TODO 失败: {e}") return todos def mark_todo(entry, status, result=""): - """在 JSONL 中将条目标记为 completed 或 failed。""" if not os.path.exists(TODO_FILE): return entry["status"] = status @@ -66,7 +73,6 @@ def mark_todo(entry, status, result=""): if result: entry["result"] = result - # 重写整个文件，标记匹配条目 lines = [] try: with open(TODO_FILE, "r", encoding="utf-8") as f: @@ -76,7 +82,6 @@ def mark_todo(entry, status, result=""): continue try: existing = json.loads(line) - # 按 created + service 匹配 if (existing.get("created") == entry.get("created") and existing.get("service") == entry.get("service")): lines.append(json.dumps(entry, ensure_ascii=False)) @@ -85,7 +90,7 @@ def mark_todo(entry, status, result=""): except json.JSONDecodeError: lines.append(line) except Exception as e: - log(f"重写 TODO 文件失败: {e}") + log(f"读取 TODO 文件失败: {e}") return try: @@ -95,30 +100,17 @@ def mark_todo(entry, status, result=""): log(f"写入 TODO 文件失败: {e}") -def execute_fix(entry): - """执行修复操作。返回 (success, detail)。""" - script = entry.get("fix_script", "") - args = entry.get("fix_args", []) - cwd = entry.get("fix_cwd", None) - service = entry.get("service", "unknown") - - if not script or not os.path.exists(script): - return (False, f"修复脚本不存在: {script}") - - cmd = [PYTHON, script] + list(args) - log(f"执行修复: {service} → {' '.join(cmd[-3:])}") +def execute_fix(service_name): + """执行修复命令。返回 (success, detail)。""" + cmd = FIX_MAP.get(service_name) + if not cmd: + return (False, "无本地修复命令（远程服务）") + log(f"执行修复: {service_name} → {' '.join(cmd)}") try: - r = subprocess.run( - cmd, - cwd=cwd, - capture_output=True, - text=True, - timeout=30, - creationflags=subprocess.CREATE_NO_WINDOW, - ) + r = subprocess.run(cmd, capture_output=True, text=True, timeout=30) if r.returncode == 0: - return (True, f"exit=0") + return (True, "exit=0") else: detail = r.stderr[:200] if r.stderr else r.stdout[:200] return (False, f"exit={r.returncode}: {detail}") @@ -128,29 +120,6 @@ def execute_fix(entry): return (False, str(e)[:100]) -def escalate_to_xmpp(entry, result): - """通过 xmpp_bot 的 HTTP 桥发送告警到群聊（失败 escalation）。""" - service = entry.get("service", "?") - issue = entry.get("issue", "?") - detail = result[:200] - payload = json.dumps({ - "message": f"[executor] 修复失败: {service}\n问题: {issue}\n结果: {detail}\n需要人工介入" - }).encode("utf-8") - try: - req = urllib.request.Request( - "http://127.0.0.1:5802/send", - data=payload, - headers={ - "Content-Type": "application/json", - "X-Api-Key": "xxm_bridge_8f3a2c", - }, - ) - urllib.request.urlopen(req, timeout=5) - log(f"已通过 XMPP 发送 escalation: {service}") - except Exception as e: - log(f"XMPP escalation 失败: {e}") - - def main(): log("=== TODO Executor 启动 ===") todos = read_pending_todos() @@ -161,21 +130,16 @@ def main(): issue = entry.get("issue", "?") log(f"处理: {service} — {issue}") - success, detail = execute_fix(entry) - # "another instance is already running" = 服务已在正常运行，不算失败 - if not success and "another instance is already running" in detail: + success, detail = execute_fix(service) + if not success and "another instance" in detail.lower(): mark_todo(entry, "completed", "already_running") - log(f" ✅ {service}: 已在运行，无需修复 ({detail[:60]})") + log(f" ✅ {service}: 已在运行") elif success: mark_todo(entry, "completed", detail) - log(f" ✅ {service}: 修复成功 ({detail})") + log(f" ✅ {service}: 修复成功") else: mark_todo(entry, "failed", detail) - log(f" ❌ {service}: 修复失败 ({detail}) → escalation") - try: - escalate_to_xmpp(entry, detail) - except Exception as e: - log(f" escalation 异常: {e}") + log(f" ❌ {service}: 修复失败 ({detail})") log(f"=== TODO Executor 完成 ({len(todos)} 条处理) ===")