rewrite: health check, daily health, todo executor for Linux 246 deployment (native tools, no Windows deps)

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)
 

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__":

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)} 条处理) ===")