Phase 1-2: 监控体系全面就绪

- 多进程看门狗: 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保活
This commit is contained in:
hmo
2026-07-14 02:49:07 +08:00
parent d53aa1e06b
commit 28ae15d49e
9 changed files with 1287 additions and 193 deletions
+8 -9
View File
@@ -1,17 +1,16 @@
# Kanban HTTP API — 跨机器任务协作
看板已合入 AgentsMeeting Dashboard5803),不再独立于 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
+255
View File
@@ -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()
+182
View File
@@ -0,0 +1,182 @@
"""
agents_health_check.py — Tier 1 快速健康检查
每 5 分钟(Task Scheduler)触发。对所有注册服务做三合一检查:
PID存活 → 端口监听 → HTTP /health
异常时写 TODOhealth_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']} 异常 ===")
# 只输出异常服务到 stdoutsilent-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()
+97 -31
View File
@@ -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": ["<error: {}>".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():
+180
View File
@@ -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()
+56
View File
@@ -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 管理门户",
},
]
+108 -1
View File
@@ -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; }
</style>
</head>
<body>
@@ -116,6 +132,12 @@ h1 { font-size:20px; font-weight:600; color:var(--accent); margin-bottom:4px; }
<h2>Infrastructure</h2>
<div class="platform-svcs" id="platform"></div>
<div class="platform-svcs" id="ejabberd-status" style="margin-top:8px;"></div>
<div id="ap-service"></div>
</div>
<div class="platform-section">
<h2>Kanban Board <span style="font-size:12px;color:var(--dim);font-weight:400;" id="kanban-count"></span></h2>
<div id="kanban-list" style="display:flex;flex-direction:column;gap:6px;"><span style="color:var(--dim);font-size:12px;">Loading...</span></div>
</div>
<div id="toast" class="toast"></div>
@@ -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 => '<div>' + esc(l) + '</div>').join('') || '<span style="color:var(--dim)">no log data</span>';
} catch(e) {
el.innerHTML = '<span style="color:var(--red)">Failed: ' + esc(e.message) + '</span>';
}
el.scrollTop = el.scrollHeight;
}
function esc(s) { return String(s).replace(/&/g,'&amp;').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 => '<div class="platform-svc"><span class="dot '+s.status+'"></span><strong>'+esc(s.name)+'</strong><span class="type">'+s.type+'</span><span style="margin-left:auto;color:var(--dim)">'+(s.pid?'PID:'+s.pid:'')+'</span></div>').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 => '<div class="platform-svc"><span class="dot '+s.status+'"></span><strong>'+esc(s.name)+'</strong><span class="type">'+s.type+'</span></div>').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 =
'<div class="ap-svc-card">'
+ '<div class="ap-header">'
+ '<span class="dot '+ap.status+'"></span>'
+ '<strong>'+esc(ap.name)+'</strong>'
+ '<span class="type">'+esc(ap.type)+'</span>'
+ '<button class="btn log" onclick="toggleAPLogs();event.stopPropagation()" style="margin-left:auto;">日志</button>'
+ '</div>'
+ '<div class="fetch-info">'
+ 'Last fetch: <span class="'+esc(lfStatus)+'">'+esc(lfStatus)+'</span>'
+ (lfTime ? ' &middot; ' + esc(lfTime) : '')
+ (lfMethod ? ' ['+esc(lfMethod)+']' : '')
+ (lfUrl ? '<br>URL: <a href="'+esc(lfUrl)+'" target="_blank" style="color:var(--accent)" title="'+esc(lfUrl)+'">'+esc(lfUrl.substring(0,80))+(lfUrl.length>80?'...':'')+'</a>' : '')
+ (lfErr ? '<br><span class="error">Error: '+esc(lfErr)+'</span>' : '')
+ '</div>'
+ '<div class="ap-log-panel'+(apLogsOpen?' open':'')+'" id="ap-log-panel">'
+ '<div class="log-header"><span>Service Logs</span>'
+ '<button class="btn log" onclick="fetchAPLogs();event.stopPropagation()">Refresh</button></div>'
+ '<div class="ap-log-content" id="ap-log-content">Click \u65e5\u5fd7 to load</div>'
+ '</div>'
+ '</div>';
}
} 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 = '<span style="color:var(--dim);font-size:12px;">No tasks yet</span>';
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 '<div style="background:var(--card);border:1px solid var(--border);border-radius:6px;padding:8px 12px;display:flex;align-items:center;gap:8px;">'
+ '<span style="width:8px;height:8px;border-radius:50%;background:'+color+';flex-shrink:0;"></span>'
+ '<span style="flex:1;font-size:13px;">' + esc(t.title) + '</span>'
+ '<span style="font-size:11px;color:var(--dim);">' + esc(t.status||'') + '</span>'
+ (t.assignee ? '<span class="badge" style="font-size:10px;">' + esc(t.assignee) + '</span>' : '')
+ '<span style="font-size:10px;color:var(--dim);">' + esc((t.created_at||'').substring(0,10)) + '</span>'
+ '</div>';
}).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);
</script>
</body>
</html>
+215 -128
View File
@@ -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}]")
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))
+186 -24
View File
@@ -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', '')