AgentsMeeting/gateway/scripts/agents_health_check.py

"""
agents_health_check.py — Tier 1 快速健康检查
每 5 分钟（Task Scheduler）触发。对所有注册服务做三合一检查：
  PID存活 → 端口监听 → HTTP /health
异常时写 TODO（health_todos.jsonl），由 self_todo_executor 消费。
正常时不输出任何噪音（silent-by-default）。
"""
import json, os, sys, time, subprocess, urllib.request, urllib.error
from datetime import datetime

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")
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"],
    })

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")


# ── Helpers ──────────────────────────────────────────────
def log(msg):
    ts = datetime.now().strftime("%H:%M:%S")
    line = f"[{ts}] {msg}"
    with open(HEALTH_LOG, "a", encoding="utf-8") as f:
        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):
    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:
        return False


def port_open(port):
    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):
    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:
        return (False, str(e.reason)[:60])
    except Exception as e:
        return (False, str(e)[:60])


def write_todo(name, issue, fix_script, fix_args, fix_cwd):
    """写一条 TODO 给 self_todo_executor 消费。
    JSONL 每行一条，含 service/issue/时间/fix_action。
    """
    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}")


# ── 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}
        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
        port_ok = port_open(svc["port"])
        entry["port_ok"] = port_ok

        # 3. HTTP health check
        http_msg = ""
        if port_ok:
            ok, detail = http_check(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"
            entry["http_ok"] = ok
            entry["http_detail"] = http_msg
        else:
            entry["http_ok"] = False
            entry["http_detail"] = "port_closed"

        healthy = (pid_ok or not pid_file) and port_ok and entry.get("http_ok", False)

        if healthy:
            entry["status"] = "ok"
            report["summary"]["ok"] += 1
        else:
            entry["status"] = "fail"
            report["summary"]["fail"] += 1
            dirty = True
            # 收集具体失败原因
            reasons = []
            if not pid_ok: reasons.append("pid_dead")
            if not port_ok: reasons.append("port_closed")
            if not entry.get("http_ok") and port_ok: reasons.append(f"http_{http_msg}")
            issue = f"异常: {' + '.join(reasons)}"
            write_todo(name, issue, svc["fix_script"], svc.get("fix_args", []), svc["fix_cwd"])

    # 只在有异常时写完整报告 + 输出到 stdout（给 Task Scheduler 日志）
    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']}: PID={svc['pid_ok']} PORT={svc['port_ok']} HTTP={svc.get('http_detail','?')}")
    else:
        # 全正常→完全静默（silent-by-default）
        pass

    # 保留报告到文件（供 dashboard 读取）
    with open(os.path.join(TEMP, "last_health_check.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False)


if __name__ == "__main__":
    main()