From 28ae15d49e13c6d31e5eb5752ef2cdab54a6e8c0 Mon Sep 17 00:00:00 2001 From: hmo Date: Mon, 13 Jul 2026 21:40:49 +0800 Subject: [PATCH] =?UTF-8?q?Phase=201-2:=20=E7=9B=91=E6=8E=A7=E4=BD=93?= =?UTF-8?q?=E7=B3=BB=E5=85=A8=E9=9D=A2=E5=B0=B1=E7=BB=AA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 多进程看门狗: xmpp_watchdog.py (守护 bot + article_processor) - Tier 1 快速健康检查: agents_health_check.py (5min) - Tier 2 全面健康检查: agents_daily_health.py (每日08:00) - TODO 自修复执行器: self_todo_executor.py (10min) - 共享服务注册表: service_registry.py - xmpp_agent_core: /health 免认证 + /send 端点 - Dashboard: SSH检测修复 + kanban版块 + 路径跨平台兼容 - 修复kanban-api-reference.md误导内容 - 修复R02: frp隧道heartbeat保活 --- docs/kanban-api-reference.md | 17 +- gateway/scripts/agents_daily_health.py | 255 +++++++++++++++++ gateway/scripts/agents_health_check.py | 182 ++++++++++++ gateway/scripts/dashboard.py | 128 +++++++-- gateway/scripts/self_todo_executor.py | 180 ++++++++++++ gateway/scripts/service_registry.py | 56 ++++ gateway/scripts/templates/dashboard.html | 109 ++++++- gateway/scripts/xmpp_watchdog.py | 343 ++++++++++++++--------- xmpp_agent_core.py | 210 ++++++++++++-- 9 files changed, 1287 insertions(+), 193 deletions(-) create mode 100644 gateway/scripts/agents_daily_health.py create mode 100644 gateway/scripts/agents_health_check.py create mode 100644 gateway/scripts/self_todo_executor.py create mode 100644 gateway/scripts/service_registry.py diff --git a/docs/kanban-api-reference.md b/docs/kanban-api-reference.md index 1d90399..1b6b158 100644 --- a/docs/kanban-api-reference.md +++ b/docs/kanban-api-reference.md @@ -1,17 +1,16 @@ # Kanban HTTP API — 跨机器任务协作 -看板已合入 AgentsMeeting Dashboard（5803），不再独立于 9580 端口。 +看板数据库使用 `~/.hermes/kanban.db`（Hermes CLI 同一份）。 -## 基础 URL +## 访问方式 -``` -http://192.168.1.246:5803 -``` +| 方式 | 命令 / URL | 状态 | +|------|-----------|------| +| Hermes CLI | `hermes kanban list` | ✅ 可用（任意机器） | +| 独立 API | `scripts/kanban_api.py` → `:9580` | ❌ 需启动 | +| Hermes Dashboard | `hermes dashboard` → Kanban 插件 | ✅ 246 上可用 | -## 浏览器查看 - -直接打开 `http://192.168.1.246:5803/` → 点顶栏 **📋 Kanban** tab。 -支持按状态和指派者筛选。 +> **注**：看板**暂未**合入 AgentsMeeting Flask Dashboard（:5803）。该文档是过时设计意图，待 Phase 3 开发时实现合并。 ## REST API diff --git a/gateway/scripts/agents_daily_health.py b/gateway/scripts/agents_daily_health.py new file mode 100644 index 0000000..7f7ed4e --- /dev/null +++ b/gateway/scripts/agents_daily_health.py @@ -0,0 +1,255 @@ +""" +agents_daily_health.py — Tier 2 全面健康检查（每日 08:00 触发） +在 Tier 1（进程/端口/HTTP）基础上增加： + - 磁盘空间检查 + - 定时任务存活检查 + - 数据新鲜度（各服务最后响应时间） + - 跨服务依赖链检查 + - XMPP 异常报告 + +基于 MoFin morning_health_check.py 的模式。 +""" +import json, os, sys, time, subprocess, urllib.request, urllib.error +from datetime import datetime + +# 强制 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" + +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 → 严重 + +# ── 服务注册表（从共享 registry 导入）───────────── +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(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", []), + }) + + +def log(msg): + ts = datetime.now().strftime("%H:%M:%S") + line = f"[{ts}] {msg}" + print(line) + with open(REPORT_LOG, "a", encoding="utf-8") as f: + 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): + try: + r = subprocess.run(["tasklist", "/FI", f"PID eq {pid}", "/NH"], + capture_output=True, text=True, timeout=5) + 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) + 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]) + except Exception as e: + return (False, str(e)[:60]) + + +def check_disk(path="C:"): + """Windows 磁盘剩余空间检查。""" + 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") + + +def check_cron_tasks(): + """检查关键定时任务是否启用。""" + expected = ["agents-health-check"] + 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}) + else: + results.append({"name": name, "status": "missing"}) + except: + results.append({"name": "check_failed", "status": "error"}) + return results + + +# ── 主检查 ──────────────────────────────────── +def main(): + report = { + "time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "services": [], + "disk": {}, + "cron_tasks": [], + "summary": {"total": 0, "ok": 0, "warn": 0, "fail": 0}, + "recommendations": [], + } + + # 0. 磁盘 + disk_level, disk_msg = check_disk("C:") + 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} + 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"]) + http_ok = False + http_msg = "" + if port_ok: + http_ok, http_msg = http_check(svc["health_url"], accept_401=svc.get("accept_401", False)) + + 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: + entry["status"] = "ok" + report["summary"]["ok"] += 1 + elif pid_ok and port_ok and not http_ok: + entry["status"] = "degraded" + report["summary"]["warn"] += 1 + report["recommendations"].append(f"{name}: PID/Port 正常但 /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}）") + + # 2. 定时任务检查 + 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) + 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} 小时未更新，可能看门狗已死") + else: + report["watchdog_log_age_hours"] = None + report["recommendations"].append("看门狗日志不存在") + + # 4. 输出报告 + summary = report["summary"] + log(f"=== 每日健康检查: {summary['ok']}/{summary['total']} 正常, {summary['warn']} 降级, {summary['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" [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") + + 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) + + +if __name__ == "__main__": + main() diff --git a/gateway/scripts/agents_health_check.py b/gateway/scripts/agents_health_check.py new file mode 100644 index 0000000..34ad1ac --- /dev/null +++ b/gateway/scripts/agents_health_check.py @@ -0,0 +1,182 @@ +""" +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) + 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) + 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() diff --git a/gateway/scripts/dashboard.py b/gateway/scripts/dashboard.py index 172ad0c..18237fb 100644 --- a/gateway/scripts/dashboard.py +++ b/gateway/scripts/dashboard.py @@ -9,22 +9,35 @@ Flask app on :5803. Monitors agents across platforms via: Auto-recovery: restarts local Windows agents after 3 consecutive offline checks. """ -import os, sys, re, json, time, subprocess, logging, urllib.request +import os, sys, re, json, time, subprocess, logging, urllib.request, sqlite3 from pathlib import Path from datetime import datetime, timedelta from flask import Flask, jsonify, request, send_from_directory -# ---- Paths ---- -PROJECT_ROOT = Path("/home/hmo/agentsmeeting-venv") -GATEWAY_ROOT = Path("/home/hmo/agentsmeeting-venv") -CONFIG_DIR = Path("/home/hmo/agentsmeeting-venv/config") -LOGS_DIR = Path("/home/hmo/agentsmeeting-venv/logs") -TEMPLATES_DIR = Path("/home/hmo/agentsmeeting-venv/templates") +# ---- Paths (platform-agnostic: auto-detect from script location) ---- +_SCRIPT_DIR = Path(__file__).resolve().parent # gateway/scripts/ +_GATEWAY_DIR = _SCRIPT_DIR.parent # gateway/ +_PROJECT_DIR = _GATEWAY_DIR.parent # AgentsMeeting/ +TEMPLATES_DIR = _SCRIPT_DIR / "templates" +CONFIG_DIR = _PROJECT_DIR / "config" +LOGS_DIR = _GATEWAY_DIR / "logs" +TEMP_DIR = _GATEWAY_DIR / "temp" +XMPP_BRIDGE_URL = os.environ.get("XMPP_BRIDGE_URL", "http://127.0.0.1:5802") +EJABBERD_HOST = os.environ.get("EJABBERD_HOST", "192.168.1.246") -sys.path.insert(0, str(GATEWAY_ROOT / "scripts")) +# On Linux (mohe), dashboard runs from systemd with different paths. Support explicit override. +_PROJECT_OVERRIDE = os.environ.get("AGENTSMEETING_ROOT") +if _PROJECT_OVERRIDE: + _PROJECT_DIR = Path(_PROJECT_OVERRIDE) + _GATEWAY_DIR = _PROJECT_DIR / "gateway" + TEMPLATES_DIR = _GATEWAY_DIR / "scripts" / "templates" + CONFIG_DIR = _PROJECT_DIR / "config" + LOGS_DIR = _GATEWAY_DIR / "logs" + TEMP_DIR = _GATEWAY_DIR / "temp" + +sys.path.insert(0, str(_GATEWAY_DIR / "scripts")) from proc_guard import guard -# ---- Flask ---- app = Flask(__name__, template_folder=str(TEMPLATES_DIR)) # ---- Logging ---- @@ -39,18 +52,8 @@ log = logging.getLogger("dashboard") # ---- Constants ---- AGENTS_YAML = CONFIG_DIR / "agents.yaml" -SCRIPT_NAMES = { - "xmpp_bot": "xmpp_bot.py", - "wechat_bridge": "wechat_agent.py", - "api_proxy": "api_proxy.py", - "health_check": "health_check_xxm.py", - "mohe_watcher": "mohe_watcher.py", - "watchdog": "xmpp_watchdog.py", -} -PYTHON = "/home/hmo/agentsmeeting-venv/bin/python3" -SCRIPTS_DIR = Path("/home/hmo/agentsmeeting-venv") -XMPP_BRIDGE_URL = "http://192.168.1.16:5802" -EJABBERD_HOST = "192.168.1.246" +PYTHON = os.environ.get("PYTHON", sys.executable) +SCRIPTS_DIR = _GATEWAY_DIR / "scripts" # Auto-recovery: restart after this many consecutive offline checks AUTO_RECOVER_THRESHOLD = 3 @@ -113,13 +116,24 @@ def _xmpp_health(): def _ejabberd_online_jids(): - """SSH to Linux and run ejabberdctl connected_users. - Returns set of bare JIDs currently connected to ejabberd. - This is the authoritative cross-platform presence source.""" + """Query ejabberdctl connected_users. + On Linux (local): direct docker exec. + On Windows (remote): SSH to EJABBERD_HOST. + Returns set of bare JIDs currently connected to ejabberd.""" try: - cmd = ["docker", "exec", "ejabberd", "ejabberdctl", "connected_users"] - result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) + if sys.platform == "win32": + # Windows: SSH to Linux server + ssh_cmd = ["ssh", "-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=5", + "-o", "BatchMode=yes", f"hmo@{EJABBERD_HOST}", + "docker exec ejabberd ejabberdctl connected_users"] + result = subprocess.run(ssh_cmd, capture_output=True, text=True, timeout=15) + else: + # Linux: direct docker exec + result = subprocess.run( + ["docker", "exec", "ejabberd", "ejabberdctl", "connected_users"], + capture_output=True, text=True, timeout=10) if result.returncode != 0: + log.debug(f"ejabberdctl failed (exit={result.returncode}): {result.stderr[:100]}") return set() jids = set() for line in result.stdout.strip().split("\n"): @@ -128,7 +142,7 @@ def _ejabberd_online_jids(): jids.add(line.split("/")[0]) return jids except Exception as e: - log.debug(f"ejabberd SSH query failed: {e}") + log.debug(f"ejabberd query failed: {e}") return set() @@ -454,6 +468,9 @@ PLATFORM_SERVICES = [ {"id": "api_proxy", "name": "API Proxy", "type": "APIRouter", "desc": "proxies volcengine API with retry/fallback", "host": "192.168.1.16", "port": 8787}, + {"id": "article_processor", "name": "文章抓取服务 (5810)", "type": "wechat-fetch", + "desc": "fetches wechat article content + OCR", + "host": "192.168.1.16", "health_url": "http://192.168.1.16:5810/health"}, ] @app.route("/api/platform") @@ -463,12 +480,18 @@ def api_platform(): result = [] for ps in PLATFORM_SERVICES: status = "stopped" + health_data = None health_url = ps.get("health_url", "") if health_url: try: req = _ur.Request(health_url) - _ur.urlopen(req, timeout=3) - status = "running" + with _ur.urlopen(req, timeout=3) as resp: + body = resp.read() + status = "running" + try: + health_data = json.loads(body) + except Exception: + pass except Exception: status = "stopped" elif "port" in ps: @@ -481,16 +504,59 @@ def api_platform(): status = "running" except Exception: status = "stopped" - result.append({ + entry = { "id": ps["id"], "name": ps["name"], "type": ps["type"], "desc": ps["desc"], "status": status, - }) + } + if health_data: + entry["health_data"] = health_data + result.append(entry) return jsonify(result) +@app.route("/api/service/5810/logs") +def api_article_processor_logs(): + """Proxy logs from article-processor service on Windows.""" + try: + req = urllib.request.Request("http://192.168.1.16:5810/logs?lines=200") + with urllib.request.urlopen(req, timeout=5) as resp: + data = json.loads(resp.read()) + lines = data.get("lines", []) + response = jsonify({"lines": lines}) + response.headers["Access-Control-Allow-Origin"] = "*" + return response + except Exception as e: + response = jsonify({"lines": ["".format(str(e))], "error": True}) + response.headers["Access-Control-Allow-Origin"] = "*" + return response + + +@app.route("/api/kanban") +def api_kanban(): + """Return kanban tasks from shared Hermes kanban.db.""" + kanban_db = Path(os.path.expanduser("~/.hermes/kanban.db")) + if not kanban_db.exists(): + return jsonify({"tasks": [], "db_exists": False}) + try: + db = sqlite3.connect(str(kanban_db)) + rows = db.execute( + "SELECT id, title, body, status, assignee, created_by, created_at FROM tasks ORDER BY created_at DESC" + ).fetchall() + db.close() + tasks = [] + for r in rows: + tasks.append({ + "id": r[0], "title": r[1], "body": r[2], "status": r[3], + "assignee": r[4], "created_by": r[5], "created_at": r[6] + }) + return jsonify({"tasks": tasks, "db_exists": True, "count": len(tasks)}) + except Exception as e: + return jsonify({"error": str(e), "db_exists": True}) + + @app.route("/api/platform") @app.route("/api/health") def api_health(): diff --git a/gateway/scripts/self_todo_executor.py b/gateway/scripts/self_todo_executor.py new file mode 100644 index 0000000..a793f05 --- /dev/null +++ b/gateway/scripts/self_todo_executor.py @@ -0,0 +1,180 @@ +""" +self_todo_executor.py — TODO 自修复执行器 + - 轮询 health_todos.jsonl 中 status=pending 的条目 + - 执行 fix_action（启动脚本） + - 成功→标记 completed + - 失败→ escalation 到 LLM（通过 xmpp_bot 通知） + +MoFin self_todo_executor.py 的 AgentsMeeting 适配版。 +""" +import json, os, sys, time, subprocess, shlex, urllib.request +from datetime import datetime + +try: + sys.stdout.reconfigure(encoding="utf-8") + sys.stderr.reconfigure(encoding="utf-8") +except Exception: + 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" + +TODO_FILE = os.path.join(TEMP, "health_todos.jsonl") +EXECUTOR_LOG = os.path.join(LOGS, "todo_executor.log") + + +def log(msg): + ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + line = f"{ts} [executor] {msg}" + print(line, flush=True) + with open(EXECUTOR_LOG, "a", encoding="utf-8") as f: + f.write(line + "\n") + + +def read_pending_todos(): + """读取 health_todos.jsonl 中所有 status=pending 的条目。""" + todos = [] + if not os.path.exists(TODO_FILE): + return todos + try: + with open(TODO_FILE, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + if entry.get("status") == "pending": + todos.append(entry) + except json.JSONDecodeError: + continue + except Exception as 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 + entry["resolved_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + if result: + entry["result"] = result + + # 重写整个文件，标记匹配条目 + lines = [] + try: + with open(TODO_FILE, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + 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)) + else: + lines.append(line) + except json.JSONDecodeError: + lines.append(line) + except Exception as e: + log(f"重写 TODO 文件失败: {e}") + return + + try: + with open(TODO_FILE, "w", encoding="utf-8") as f: + f.write("\n".join(lines) + "\n") + except Exception as e: + 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:])}") + + try: + r = subprocess.run( + cmd, + cwd=cwd, + capture_output=True, + text=True, + timeout=30, + creationflags=subprocess.CREATE_NO_WINDOW, + ) + if r.returncode == 0: + return (True, f"exit=0") + else: + detail = r.stderr[:200] if r.stderr else r.stdout[:200] + return (False, f"exit={r.returncode}: {detail}") + except subprocess.TimeoutExpired: + return (False, "timeout (30s)") + except Exception as e: + 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() + log(f"待处理 TODO: {len(todos)} 条") + + for entry in todos: + service = entry.get("service", "?") + issue = entry.get("issue", "?") + log(f"处理: {service} — {issue}") + + success, detail = execute_fix(entry) + if success: + mark_todo(entry, "completed", detail) + log(f" ✅ {service}: 修复成功 ({detail})") + 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"=== TODO Executor 完成 ({len(todos)} 条处理) ===") + + +if __name__ == "__main__": + main() diff --git a/gateway/scripts/service_registry.py b/gateway/scripts/service_registry.py new file mode 100644 index 0000000..e225339 --- /dev/null +++ b/gateway/scripts/service_registry.py @@ -0,0 +1,56 @@ +""" +service_registry.py — 共享服务注册表 +所有监控组件（watchdog、Tier1、Tier2）统一从此文件读取服务定义。 +新增服务只需在此添加一条记录，三处监控自动生效。 +""" +import os + +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(GATEWAY_DIR, "temp") +PYTHON = r"C:\Users\hmo\AppData\Local\Programs\Python\Python310\python.exe" + +SERVICES = [ + { + "name": "xmpp_bot", + "script": os.path.join(BASE, "xmpp_agent_core.py"), + "args": ["--agent", "xxm"], + "workdir": BASE, + "pid_file": os.path.join(TEMP, ".xmpp_bot.pid"), + "port": 5802, + "health_url": "http://127.0.0.1:5802/health", + "accept_401": False, # /health 现在返回 200（免认证） + "depends_on": [], + "log_files": lambda: [os.path.join(GATEWAY_DIR, "logs", "xmpp_bot.log"), + os.path.join(GATEWAY_DIR, "logs", "bridge.log")], + "description": "XMPP Bot (xxm@yoin.fun)", + }, + { + "name": "article_processor", + "script": os.path.join(os.path.dirname(BASE), "self-growing-knowledge", + "scripts", "article_processor.py"), + "args": [], + "workdir": os.path.join(os.path.dirname(BASE), "self-growing-knowledge"), + "pid_file": os.path.join(TEMP, ".article_processor.pid"), + "port": 5810, + "health_url": "http://127.0.0.1:5810/health", + "accept_401": False, + "depends_on": [], + "log_files": lambda: [os.path.join(GATEWAY_DIR, "logs", "article_processor.log")], + "description": "微信文章全文抓取服务", + }, + { + "name": "dashboard", + "script": os.path.join(SCRIPT_DIR, "dashboard.py"), + "args": [], + "workdir": SCRIPT_DIR, + "pid_file": None, # dashboard 管理自身进程 + "port": 5803, + "health_url": "http://127.0.0.1:5803/api/health", + "accept_401": False, + "depends_on": ["xmpp_bot"], + "log_files": lambda: [], + "description": "AgentsMeeting 管理门户", + }, +] diff --git a/gateway/scripts/templates/dashboard.html b/gateway/scripts/templates/dashboard.html index 7ca928a..8e32277 100644 --- a/gateway/scripts/templates/dashboard.html +++ b/gateway/scripts/templates/dashboard.html @@ -102,6 +102,22 @@ h1 { font-size:20px; font-weight:600; color:var(--accent); margin-bottom:4px; } .platform-svc .dot.stopped { background:var(--dim); } .platform-svc .type { color:var(--accent); font-size:10px; text-transform:uppercase; } +/* 5810 article-processor card */ +.ap-svc-card { background:var(--card); border:1px solid var(--border); border-radius:8px; padding:12px 16px; margin-top:12px; } +.ap-svc-card .ap-header { display:flex; align-items:center; gap:8px; margin-bottom:6px; } +.ap-svc-card .dot { width:8px; height:8px; border-radius:50%; flex-shrink:0; } +.ap-svc-card .dot.running { background:var(--green); } +.ap-svc-card .dot.stopped { background:var(--dim); } +.ap-svc-card .fetch-info { font-size:11px; color:var(--dim); margin-top:4px; line-height:1.6; } +.ap-svc-card .fetch-info .ok { color:var(--green); } +.ap-svc-card .fetch-info .error { color:var(--red); } +.ap-svc-card .fetch-info .none { color:var(--yellow); } +.ap-svc-card .ap-log-panel { display:none; margin-top:10px; border-top:1px solid var(--border); padding-top:10px; } +.ap-svc-card .ap-log-panel.open { display:block; } +.ap-svc-card .ap-log-content { font:11px/1.6 'Cascadia Code','Consolas',monospace; color:var(--dim); max-height:300px; overflow-y:auto; background:#06080c; border:1px solid var(--border); border-radius:6px; padding:10px; white-space:pre-wrap; word-break:break-all; } +.ap-svc-card .ap-log-content::-webkit-scrollbar { width:6px; } +.ap-svc-card .ap-log-content::-webkit-scrollbar-thumb { background:var(--border); border-radius:3px; } + @@ -116,6 +132,12 @@ h1 { font-size:20px; font-weight:600; color:var(--accent); margin-bottom:4px; }

Infrastructure

+
+ + +
+

Kanban Board

+
Loading...
@@ -124,6 +146,7 @@ h1 { font-size:20px; font-weight:600; color:var(--accent); margin-bottom:4px; } const API = '/api/agents'; let agentsData = []; const openLogs = new Set(); +let apLogsOpen = false; function toast(msg, type) { const t = document.getElementById('toast'); @@ -163,6 +186,26 @@ async function fetchLogs(id) { el.scrollTop = el.scrollHeight; } +function toggleAPLogs() { + apLogsOpen = !apLogsOpen; + const panel = document.getElementById('ap-log-panel'); + if (panel) panel.className = 'ap-log-panel' + (apLogsOpen ? ' open' : ''); + if (apLogsOpen) fetchAPLogs(); +} + +async function fetchAPLogs() { + const el = document.getElementById('ap-log-content'); + if (!el || !apLogsOpen) return; + try { + const r = await fetch('/api/service/5810/logs'); + const d = await r.json(); + el.innerHTML = (d.lines || []).map(l => '
' + esc(l) + '
').join('') || 'no log data'; + } catch(e) { + el.innerHTML = 'Failed: ' + esc(e.message) + ''; + } + el.scrollTop = el.scrollHeight; +} + function esc(s) { return String(s).replace(/&/g,'&').replace(//g,'>'); } function render() { @@ -237,7 +280,42 @@ async function fetchPlatform() { try { const r = await fetch('/api/platform'); const data = await r.json(); - document.getElementById('platform').innerHTML = data.map(s => '
'+esc(s.name)+''+s.type+''+(s.pid?'PID:'+s.pid:'')+'
').join(''); + const others = data.filter(s => s.id !== 'article_processor'); + const ap = data.find(s => s.id === 'article_processor'); + + document.getElementById('platform').innerHTML = others.map(s => '
'+esc(s.name)+''+s.type+'
').join(''); + + if (ap) { + const hd = ap.health_data || {}; + const lf = hd.last_fetch || {}; + const lfStatus = lf.status || 'none'; + const lfTime = lf.timestamp || ''; + const lfErr = lf.error || ''; + const lfUrl = lf.url || ''; + const lfMethod = lf.method || ''; + + document.getElementById('ap-service').innerHTML = + '
' + + '
' + + '' + + ''+esc(ap.name)+'' + + ''+esc(ap.type)+'' + + '' + + '
' + + '
' + + 'Last fetch: '+esc(lfStatus)+'' + + (lfTime ? ' · ' + esc(lfTime) : '') + + (lfMethod ? ' ['+esc(lfMethod)+']' : '') + + (lfUrl ? '
URL: '+esc(lfUrl.substring(0,80))+(lfUrl.length>80?'...':'')+'' : '') + + (lfErr ? '
Error: '+esc(lfErr)+'' : '') + + '
' + + '
' + + '
Service Logs' + + '
' + + '
Click \u65e5\u5fd7 to load
' + + '
' + + '
'; + } } catch(e) {} } @@ -250,13 +328,42 @@ async function fetchEjabberd() { } catch(e) {} } +async function fetchKanban() { + try { + const r = await fetch('/api/kanban'); + const d = await r.json(); + const el = document.getElementById('kanban-list'); + const ct = document.getElementById('kanban-count'); + if (!d.tasks || d.tasks.length === 0) { + el.innerHTML = 'No tasks yet'; + ct.textContent = ''; + return; + } + ct.textContent = '(' + d.count + ' tasks)'; + el.innerHTML = d.tasks.map(t => { + const statusColors = {ready:'var(--yellow)',progress:'var(--accent)',done:'var(--green)',blocked:'var(--red)'}; + const color = statusColors[t.status] || 'var(--dim)'; + return '
' + + '' + + '' + esc(t.title) + '' + + '' + esc(t.status||'') + '' + + (t.assignee ? '' + esc(t.assignee) + '' : '') + + '' + esc((t.created_at||'').substring(0,10)) + '' + + '
'; + }).join(''); + } catch(e) { /* kanban unavailable */ } +} + fetchAgents(); fetchPlatform(); fetchEjabberd(); +fetchKanban(); setInterval(fetchAgents, 5000); setInterval(fetchPlatform, 10000); setInterval(fetchEjabberd, 10000); +setInterval(fetchKanban, 15000); setInterval(() => openLogs.forEach(id => fetchLogs(id)), 3000); +setInterval(() => { if (apLogsOpen) fetchAPLogs(); }, 3000); diff --git a/gateway/scripts/xmpp_watchdog.py b/gateway/scripts/xmpp_watchdog.py index 677dc82..f4d56f6 100644 --- a/gateway/scripts/xmpp_watchdog.py +++ b/gateway/scripts/xmpp_watchdog.py @@ -1,174 +1,261 @@ """ -xmpp_watchdog.py — monitors xmpp_bot, auto-restarts on crash, reports status. -Runs alongside xmpp_bot.py as a separate process. +xmpp_watchdog.py — monitors xmpp_agent_core + article_processor, +auto-restarts on crash, reports status. +Runs alongside gateway processes as a separate 30s-loop daemon. """ -import os, sys, time, subprocess, json, threading - -PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) -BOT_SCRIPT = os.path.join(PROJECT_ROOT, "xmpp_bot.py") -LOG_DIR = os.path.join(os.path.dirname(PROJECT_ROOT), "logs") -WATCHDOG_LOG = os.path.join(LOG_DIR, "watchdog.log") -PID_FILE = os.path.join(os.path.dirname(PROJECT_ROOT), "temp", ".xmpp_watchdog.pid") -BOT_PID_FILE = os.path.join(os.path.dirname(PROJECT_ROOT), "temp", ".xmpp_bot.pid") -PYTHON = r"C:\Users\hmo\AppData\Local\Programs\Python\Python310\python.exe" -CHECK_INTERVAL = 30 # seconds between health checks +import os, sys, time, subprocess, json, urllib.request, urllib.error +# ── Paths ──────────────────────────────────────────────────── +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) # gateway/scripts/ +GATEWAY_DIR = os.path.dirname(PROJECT_ROOT) # gateway/ +TEMP_DIR = os.path.join(GATEWAY_DIR, "temp") +LOG_DIR = os.path.join(GATEWAY_DIR, "logs") os.makedirs(LOG_DIR, exist_ok=True) +os.makedirs(TEMP_DIR, exist_ok=True) +PYTHON = r"C:\Users\hmo\AppData\Local\Programs\Python\Python310\python.exe" +CHECK_INTERVAL = 30 # seconds between health checks +HEALTH_TIMEOUT = 5 # HTTP health check timeout + +WATCHDOG_LOG = os.path.join(LOG_DIR, "watchdog.log") +PID_FILE = os.path.join(TEMP_DIR, ".multi_watchdog.pid") + + +# ── Service registry（从共享 registry 导入）─────────────────── +sys.path.insert(0, PROJECT_ROOT) +from service_registry import SERVICES as _SR + +SERVICES = {} +for svc in _SR: + name = svc["name"] + SERVICES[name] = { + "script": svc["script"], + "args": svc["args"], + "workdir": svc["workdir"], + "pid_file": svc["pid_file"], + "port": svc["port"], + "health_url": svc["health_url"], + "log_files": svc.get("log_files", lambda: [])(), + } + + +# ── Logging ─────────────────────────────────────────────────── def wlog(msg: str): ts = time.strftime("%Y-%m-%d %H:%M:%S") + line = f"{ts} [watchdog] {msg}\n" with open(WATCHDOG_LOG, "a", encoding="utf-8") as f: - f.write(f"{ts} [watchdog] {msg}\n") - print(f"[watchdog] {msg}", flush=True) + f.write(line) + print(line, end="", flush=True) +# ── Log rotation ────────────────────────────────────────────── def rotate_log(path: str, max_bytes: int = 5 * 1024 * 1024): - """Rotate log file if it exceeds max_bytes. Keeps last 3 backups.""" try: + if not os.path.exists(path): + return if os.path.getsize(path) > max_bytes: - # shift .2→.tmp, .1→.2, file→.1 bak2 = f"{path}.2" bak1 = f"{path}.1" - if os.path.exists(bak2): os.remove(bak2) - if os.path.exists(bak1): os.rename(bak1, bak2) + if os.path.exists(bak2): + os.remove(bak2) + if os.path.exists(bak1): + os.rename(bak1, bak2) os.rename(path, bak1) wlog(f"Rotated: {os.path.basename(path)}") - except: + except Exception: pass +# ── Process helpers ─────────────────────────────────────────── def is_process_alive(pid: int) -> bool: - """Check if a process with given PID is alive.""" try: - proc = subprocess.run( - ['tasklist', '/FI', f'PID eq {pid}', '/NH'], + r = subprocess.run( + ["tasklist", "/FI", f"PID eq {pid}", "/NH"], capture_output=True, text=True, timeout=5 ) - return str(pid) in proc.stdout + return str(pid) in r.stdout except: return False -def kill_bot(): - """Kill ALL existing xmpp_bot.py processes before starting a new one.""" - killed = 0 +def get_pid_from_file(path: str) -> int: try: - r = subprocess.run( - ['tasklist', '/FO', 'CSV', '/NH', '/FI', 'IMAGENAME eq python.exe'], - capture_output=True, text=True, timeout=10 - ) - for line in r.stdout.splitlines(): - parts = line.strip('"').split('","') - if len(parts) >= 2 and parts[0] == 'python.exe': - pid_str = parts[1].strip() - try: - wmi = subprocess.run( - ['wmic', 'process', 'where', f'ProcessId={pid_str}', - 'get', 'CommandLine', '/format:list'], - capture_output=True, text=True, timeout=5 - ) - if 'xmpp_bot' in wmi.stdout and 'watchdog' not in wmi.stdout: - subprocess.run(['taskkill', '/f', '/pid', pid_str], - capture_output=True, timeout=5) - killed += 1 - wlog(f"Killed old bot (PID {pid_str})") - except: - pass - except: - pass - if killed > 0: - time.sleep(3) # wait for process cleanup - -def start_bot() -> int: - """Start xmpp_bot.py and return its PID. Kills old instances first.""" - kill_bot() - wlog("Starting xmpp_bot...") - proc = subprocess.Popen( - [PYTHON, BOT_SCRIPT], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - creationflags=subprocess.CREATE_NO_WINDOW - ) - pid = proc.pid - with open(BOT_PID_FILE, "w") as f: - f.write(str(pid)) - wlog(f"xmpp_bot started (PID {pid})") - return pid - - -def get_last_log_activity() -> float: - """Get timestamp of last xmpp_bot.log modification.""" - log_file = os.path.join(LOG_DIR, "xmpp_bot.log") - try: - return os.path.getmtime(log_file) + with open(path) as f: + return int(f.read().strip()) except: return 0 -def health_check(bot_pid: int, last_activity: float) -> tuple[bool, int, float]: +def is_port_listening(port: int) -> bool: + """Quick port liveness check via netstat.""" + try: + r = subprocess.run( + ["netstat", "-ano"], + capture_output=True, text=True, timeout=5 + ) + return any(f":{port} " in line and "LISTENING" in line + for line in r.stdout.splitlines()) + except: + return False + + +def check_http_health(url: str, timeout: int = HEALTH_TIMEOUT) -> tuple[bool, str]: + """Returns (ok, detail).""" + try: + r = urllib.request.urlopen(url, timeout=timeout) + body = r.read().decode("utf-8", errors="replace")[:200] + return (r.status == 200, body) + 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]) + + +# ── Process lifecycle ──────────────────────────────────────── +def kill_service(name: str, script_match: str): + """Kill ALL processes whose command line contains script_match.""" + killed = 0 + try: + r = subprocess.run( + ["tasklist", "/FO", "CSV", "/NH", "/FI", "IMAGENAME eq python.exe"], + capture_output=True, text=True, timeout=10 + ) + for line in r.stdout.splitlines(): + parts = line.strip('"').split('","') + if len(parts) < 2 or parts[0] != "python.exe": + continue + pid_str = parts[1].strip() + try: + wmi = subprocess.run( + ["wmic", "process", "where", f"ProcessId={pid_str}", + "get", "CommandLine", "/format:list"], + capture_output=True, text=True, timeout=5 + ) + if script_match in wmi.stdout: + subprocess.run(["taskkill", "/f", "/pid", pid_str], + capture_output=True, timeout=5) + killed += 1 + wlog(f"Killed old {name} (PID {pid_str})") + except: + pass + if killed > 0: + time.sleep(3) + except: + pass + return killed + + +def start_service(name: str, cfg: dict) -> int: + """Start service and return PID. Kills old instances first.""" + kill_service(name, os.path.basename(cfg["script"])) + wlog(f"Starting {name}...") + proc = subprocess.Popen( + [PYTHON, cfg["script"]] + cfg["args"], + cwd=cfg["workdir"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + creationflags=subprocess.CREATE_NO_WINDOW, + ) + pid = proc.pid + with open(cfg["pid_file"], "w") as f: + f.write(str(pid)) + wlog(f"{name} started (PID {pid})") + return pid + + +# ── Service health check ───────────────────────────────────── +def check_service(name: str, cfg: dict, pid: int) -> tuple[bool, int, str]: """ - Check bot health. - Returns (is_alive, pid, last_activity). - If dead, restarts bot. + Returns (is_healthy, new_pid, status_msg). + If dead, auto-restarts. """ - alive = is_process_alive(bot_pid) - + alive = is_process_alive(pid) + if not alive: - wlog(f"Bot PID {bot_pid} is DEAD. Restarting...") - bot_pid = start_bot() + wlog(f"{name} PID {pid} is DEAD. Restarting...") + new_pid = start_service(name, cfg) time.sleep(5) - last_activity = get_last_log_activity() - return (True, bot_pid, last_activity) - - # Check if bot has been active recently (last 5 minutes) - current_activity = get_last_log_activity() - if current_activity > last_activity: - last_activity = current_activity - - # If no activity for 5 minutes but bot is alive, warn - if time.time() - last_activity > 300: - wlog(f"WARNING: Bot PID {bot_pid} alive but no activity for 5+ min") - - return (True, bot_pid, last_activity) + return (True, new_pid, "restarted") + + # Port-level check + port_ok = is_port_listening(cfg["port"]) + if not port_ok: + wlog(f"{name} PID {pid} alive but port {cfg['port']} not listening. Restarting...") + new_pid = start_service(name, cfg) + time.sleep(5) + return (True, new_pid, "restarted (port dead)") + + # HTTP health check + http_ok, detail = check_http_health(cfg["health_url"]) + if not http_ok: + # 401/403 = service alive but requires auth → treat as healthy + if "HTTP 401" in detail or "HTTP 403" in detail: + wlog(f"{name} PID {pid} /health {detail} (auth required, service alive)") + else: + wlog(f"{name} PID {pid} /health FAILED: {detail}. Restarting...") + new_pid = start_service(name, cfg) + time.sleep(5) + return (True, new_pid, f"restarted (/health: {detail})") + + return (True, pid, "ok") +# ── Main ────────────────────────────────────────────────────── if __name__ == "__main__": - wlog("Watchdog started") - - # Start bot if not already running - bot_pid = 0 - if os.path.exists(BOT_PID_FILE): - try: - with open(BOT_PID_FILE) as f: - bot_pid = int(f.read().strip()) - if not is_process_alive(bot_pid): - bot_pid = 0 - except: - bot_pid = 0 - - if bot_pid == 0: - bot_pid = start_bot() - - last_activity = get_last_log_activity() - wlog(f"Initial: bot PID {bot_pid}, log last activity: {time.ctime(last_activity)}") - + wlog("=== Multi-Watchdog started ===") + + # Discover / start all services + pids = {} + statuses = {} + for name, cfg in SERVICES.items(): + pid = get_pid_from_file(cfg["pid_file"]) + if pid and is_process_alive(pid) and is_port_listening(cfg["port"]): + wlog(f"{name}: found running (PID {pid})") + else: + pid = start_service(name, cfg) + pids[name] = pid + statuses[name] = "starting" + time.sleep(2) + + # Track "last good" per-service for activity monitoring + last_http_ok = {name: time.time() for name in SERVICES} + log_rotate_counter = 0 - - # Main monitoring loop + cycle = 0 + while True: time.sleep(CHECK_INTERVAL) - alive, bot_pid, last_activity = health_check(bot_pid, last_activity) - - # Log rotation (every 30 checks ≈ 15 min) + cycle += 1 log_rotate_counter += 1 + + any_restart = False + + for name, cfg in SERVICES.items(): + ok, new_pid, msg = check_service(name, cfg, pids[name]) + if new_pid != pids[name]: + any_restart = True + pids[name] = new_pid + statuses[name] = msg + else: + statuses[name] = msg + if ok: + last_http_ok[name] = time.time() + + # Log rotation every 30 checks ≈ 15 min if log_rotate_counter >= 30: log_rotate_counter = 0 - bot_log = os.path.join(LOG_DIR, "xmpp_bot.log") - bridge_log = os.path.join(LOG_DIR, "bridge.log") - rotate_log(bot_log) - rotate_log(bridge_log) - - # Every 5 minutes, report status - if int(time.time()) % 300 < CHECK_INTERVAL: - alive_str = "ALIVE" if alive else "RESTARTED" - wlog(f"Status: bot PID {bot_pid} [{alive_str}]") \ No newline at end of file + for name, cfg in SERVICES.items(): + for log_path in cfg["log_files"]: + rotate_log(log_path) + rotate_log(WATCHDOG_LOG) + + # Status report every 10 cycles (5 min) + if cycle % 10 == 0 or any_restart: + parts = [] + for name in SERVICES: + p = pids[name] + s = statuses[name] + parts.append(f"{name} PID {p} [{s}]") + wlog(" | ".join(parts)) diff --git a/xmpp_agent_core.py b/xmpp_agent_core.py index d94c54b..cafef66 100644 --- a/xmpp_agent_core.py +++ b/xmpp_agent_core.py @@ -77,6 +77,7 @@ AGENTS = { "nick": "xxm", "name_cn": "笑笑", "http_port": 5802, + "bridge_api_key": "xxm_bridge_8f3a2c", "bridge": "chat_bridge", # use local chat_bridge instead of Hermes API "session_id": "ses_xxm_xmpp", "kanban_session_id": "xmpp-xxm-kanban", @@ -216,28 +217,135 @@ _EASYTIER_PID_FILE = os.path.join(_EASYTIER_DIR, "easytier.pid") _EASYTIER_NET = "--network-name mynet --network-secret ce75d0a5" _EASYTIER_RELAY = "--peers tcp://47.115.32.206:11010" _EASYTIER_IP = "--ipv4 10.144.144.3" +_EASYTIER_FLAGS = "--disable-encryption" # TEMP: match relay+246; TODO remove after relay/246 also remove it + +# ============================================================ +# RDP Toggle - remote desktop via SSH reverse tunnel (port 8080) +# ============================================================ + +_RDP_SSH_HOST = 'root@47.115.32.206' +_RDP_SSH_PORT = '8080' +_RDP_PID_FILE = os.path.join(os.path.dirname(__file__), 'gateway', 'scripts', 'rdp_tunnel.pid') + +def _rdp_tunnel_pid(): + try: + if os.path.exists(_RDP_PID_FILE): + with open(_RDP_PID_FILE) as f: + return int(f.read().strip()) + except Exception: + pass + return None + +def _rdp_enable(): + import subprocess as _sp, winreg as _wr, time as _t + try: + k = _wr.OpenKey(_wr.HKEY_LOCAL_MACHINE, r'SYSTEM\CurrentControlSet\Control\Terminal Server', 0, _wr.KEY_SET_VALUE) + _wr.SetValueEx(k, 'fDenyTSConnections', 0, _wr.REG_DWORD, 0) + _wr.CloseKey(k) + except Exception as e: + log(f'RDP enable registry error: {e}') + try: + _sp.run(['net', 'localgroup', 'Remote Desktop Users', 'hmo', '/add'], capture_output=True, timeout=5) + except Exception: + pass + _rdp_kill_tunnel() + try: + cmd = ['ssh.exe', '-o', 'StrictHostKeyChecking=no', '-o', 'ServerAliveInterval=30', '-N', '-R', '0.0.0.0:8080:localhost:3389', 'root@47.115.32.206'] + si = _sp.STARTUPINFO() + si.dwFlags |= _sp.STARTF_USESHOWWINDOW + p = _sp.Popen(cmd, startupinfo=si, stdout=_sp.DEVNULL, stderr=_sp.DEVNULL) + _t.sleep(2) + if p.poll() is not None: + return False, 'SSH tunnel exited immediately (code ' + str(p.poll()) + ')' + with open(_RDP_PID_FILE, 'w') as f: + f.write(str(p.pid)) + log('RDP tunnel enabled (SSH reverse :8080, PID ' + str(p.pid) + ')') + return True, 'RDP access enabled' + except Exception as e: + return False, 'Failed: ' + str(e) + +def _rdp_disable(): + import subprocess as _sp, winreg as _wr + _rdp_kill_tunnel() + try: + k = _wr.OpenKey(_wr.HKEY_LOCAL_MACHINE, r'SYSTEM\CurrentControlSet\Control\Terminal Server', 0, _wr.KEY_SET_VALUE) + _wr.SetValueEx(k, 'fDenyTSConnections', 0, _wr.REG_DWORD, 1) + _wr.CloseKey(k) + log('RDP access disabled') + return True, 'RDP access disabled' + except Exception as e: + return False, 'Failed to disable RDP: ' + str(e) + +def _rdp_kill_tunnel(): + import subprocess as _sp + pid = _rdp_tunnel_pid() + if pid: + try: + _sp.run(['taskkill', '/f', '/pid', str(pid)], capture_output=True, timeout=5) + log('Killed RDP tunnel (PID ' + str(pid) + ')') + except Exception: + pass + try: + if os.path.exists(_RDP_PID_FILE): + os.remove(_RDP_PID_FILE) + except Exception: + pass + +def _rdp_status(): + import subprocess as _sp, winreg as _wr + tunnel_on = False + pid = _rdp_tunnel_pid() + if pid: + try: + r = _sp.run(['tasklist', '/fi', 'PID eq ' + str(pid), '/fi', 'imagename eq ssh.exe'], capture_output=True, text=True, timeout=5) + tunnel_on = 'ssh.exe' in r.stdout + except Exception: + pass + rdp_on = False + try: + k = _wr.OpenKey(_wr.HKEY_LOCAL_MACHINE, r'SYSTEM\CurrentControlSet\Control\Terminal Server', 0, _wr.KEY_READ) + v, _ = _wr.QueryValueEx(k, 'fDenyTSConnections') + _wr.CloseKey(k) + rdp_on = (v == 0) + except Exception: + pass + return {'ok': True, 'tunnel_running': tunnel_on, 'rdp_enabled': rdp_on, 'rdp_port': 3389, 'tunnel_host': 'root@47.115.32.206', 'tunnel_port': 8080} + def _start_easytier(): """Start EasyTier on Windows.""" import subprocess as _sp + import time as _time if not os.path.exists(_EASYTIER_CORE): log(f"EasyTier binary not found: {_EASYTIER_CORE}") return - # Check if already running - if os.path.exists(_EASYTIER_PID_FILE): - try: - with open(_EASYTIER_PID_FILE) as f: - old_pid = int(f.read().strip()) - _sp.run(["taskkill", "/f", "/pid", str(old_pid)], capture_output=True, timeout=5) - log(f"Killed old EasyTier (PID {old_pid})") - except Exception: - pass + # Kill ALL existing easytier-core.exe processes FIRST (prevents duplicate conflicts) + _sp.run(["taskkill", "/f", "/im", "easytier-core.exe"], capture_output=True, timeout=5) + _time.sleep(1) + # Clean stale PID file + try: + if os.path.exists(_EASYTIER_PID_FILE): + os.remove(_EASYTIER_PID_FILE) + except Exception: + pass # Start - cmd = f'start /b "" "{_EASYTIER_CORE}" {_EASYTIER_NET} {_EASYTIER_RELAY} {_EASYTIER_IP} --disable-encryption --no-listener' + cmd = f'start /b "" "{_EASYTIER_CORE}" {_EASYTIER_NET} {_EASYTIER_RELAY} {_EASYTIER_IP} {_EASYTIER_FLAGS} --no-listener' try: _sp.run(cmd, shell=True, timeout=5) log("EasyTier start command issued") + # Write PID file so future calls can track this instance + _time.sleep(2) + r = _sp.run(["tasklist", "/fi", "imagename eq easytier-core.exe", "/fo", "csv"], + capture_output=True, text=True, timeout=5) + import csv as _csv + for row in _csv.reader(r.stdout.splitlines()): + if len(row) >= 2 and row[0].strip('"') == "easytier-core.exe": + pid = int(row[1]) + with open(_EASYTIER_PID_FILE, "w") as f: + f.write(str(pid)) + log(f"EasyTier started (PID {pid})") + break except Exception as e: log(f"EasyTier start error: {e}") @@ -592,9 +700,42 @@ def _record_group_msg(nickname: str, body: str): _MSG_BUF[:] = _MSG_BUF[-150:] + +_BRIDGE_API_KEY = cfg.get('bridge_api_key', '') + +def _bridge_auth(self) -> bool: + if not _BRIDGE_API_KEY: + return True + hdr = self.headers.get('X-Api-Key', '') + if hdr == _BRIDGE_API_KEY: + return True + parsed = urllib.parse.urlparse(self.path) + qs = urllib.parse.parse_qs(parsed.query) + if qs.get('key', [None])[0] == _BRIDGE_API_KEY: + return True + return False + class _BridgeHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): parsed = urllib.parse.urlparse(self.path) + # /health 免认证，供监控系统使用 + if parsed.path == "/health": + try: + bot = _xmpp_ref + session_ok = bot.session_started_event.is_set() if (bot and hasattr(bot, 'session_started_event')) else False + socket_ok = bot.is_connected() if (bot and hasattr(bot, 'is_connected')) else False + self._reply(200, { + "ok": True, "xmpp_connected": session_ok or socket_ok, + "agent": _agent_name, "jid": cfg["jid"], + "uptime_sec": int(time.time() - _START_TIME), + "muc_rooms": cfg["muc_rooms"], + }) + except Exception as e: + self._reply(500, {"ok": False, "error": str(e)}) + return + if not _bridge_auth(self): + self._reply(401, {'ok': False, 'error': 'unauthorized'}) + return if parsed.path == "/muc": try: muc_info = {"rooms": {}} @@ -620,20 +761,6 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler): except Exception as e: self._reply(500, {"ok": False, "error": str(e)}) return - if parsed.path == "/health": - try: - bot = _xmpp_ref - session_ok = bot.session_started_event.is_set() if (bot and hasattr(bot, 'session_started_event')) else False - socket_ok = bot.is_connected() if (bot and hasattr(bot, 'is_connected')) else False - self._reply(200, { - "ok": True, "xmpp_connected": session_ok or socket_ok, - "agent": _agent_name, "jid": cfg["jid"], - "uptime_sec": int(time.time() - _START_TIME), - "muc_rooms": cfg["muc_rooms"], - }) - except Exception as e: - self._reply(500, {"ok": False, "error": str(e)}) - return if parsed.path.startswith("/presence"): jid_to_check = parsed.path[len("/presence/"):].strip() if not jid_to_check: @@ -667,11 +794,28 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler): self._reply(404, {"ok": False, "error": "not found"}) def do_POST(self): + if not _bridge_auth(self): + self._reply(401, {'ok': False, 'error': 'unauthorized'}) + return try: length = int(self.headers.get('Content-Length', 0)) body = json.loads(self.rfile.read(length)) path = urllib.parse.urlparse(self.path).path.rstrip('/') + + if path == '/rdp': + action = body.get('action', '') + if action == 'start': + ok, msg = _rdp_enable() + self._reply(200, {'ok': ok, 'message': msg}) + elif action == 'stop': + ok, msg = _rdp_disable() + self._reply(200, {'ok': ok, 'message': msg}) + elif action == 'status': + self._reply(200, _rdp_status()) + else: + self._reply(400, {'ok': False, 'error': 'action must be start|stop|status'}) + return # /easytier endpoint — execute EasyTier action locally (no XMPP DM) if path == "/easytier": action = body.get("action", "") @@ -687,6 +831,24 @@ class _BridgeHandler(http.server.BaseHTTPRequestHandler): else: self._reply(400, {"ok": False, "error": "action must be start|stop|status"}) return + # /send — 发送消息到 MUC 群聊 + if path == "/send": + to = body.get('to', '') + if not to: + rooms = cfg.get("muc_rooms", []) + to = rooms[0] if rooms else "coregroup@conference.yoin.fun" + msg = body.get('message', '') or body.get('body', '') + msg_type = body.get('type', 'groupchat') + if not msg: + self._reply(400, {"ok": False, "error": "empty message"}) + return + bot = _xmpp_ref + if bot: + bot.send_message(mto=to, mbody=msg.strip(), mtype=msg_type) + _record_group_msg(cfg["nick"], msg) + log(f"[http] → [{to.split('@')[0]}]: {msg[:80]} (type={msg_type})") + self._reply(200, {"ok": True}) + return to = body.get('to', cfg["muc_rooms"][0]) msg = body.get('message', '') or body.get('body', '')