hmo/AgentsMeeting
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b11b88887cd7d1d662a0b3e5af61128fd8229922
AgentsMeeting/gateway/scripts/dashboard.py
T
hmo b11b88887c merge: 恢复077c649内容(开发原则/K tests/spec/prd paths) + 保留flicker-free EasyTier/RDP + Linux兼容改善 
- dashboard.html: 应用fI()闪烁修复到077c649版本(create-once/update-state pattern)
- dashboard.py: 新增 /api/tests 端点 (import tests_api)
- dashboard.py: /api/git 改用 git log 命令优先, 降级到 reflog
- dashboard.py: /api/monitor + /api/expected 增加 Linux 支持(systemd timer/crontab)
- dashboard.py: /api/spec + /api/prd 增加 246 venv 路径候选
- dashboard.py: /api/metagrowth git路径探测改善
- 恢复: meta_growth.py, tests_api.py, checklist_audit.py, service_registry.py
- 恢复: sync-venv.sh, post-deploy-check.sh, meta-growth-design.md
2026-07-15 10:10:25 +08:00

1135 lines
47 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
"""
dashboard.py - AgentsMeeting management dashboard backend
=========================================================
Flask app on :5803. Monitors agents across platforms via:
- SSH + ejabberdctl connected_users (cross-platform, authoritative)
- xmpp_bot HTTP API :5802 (/health, /muc - fallback)
- Local process/port checks (Windows only)
Auto-recovery: restarts local Windows agents after 3 consecutive offline checks.
"""
import os, sys, re, json, time, subprocess, logging, urllib.request, sqlite3, shutil
from pathlib import Path
from datetime import datetime, timedelta
from flask import Flask, jsonify, request, send_from_directory
# ---- 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://192.168.1.16:5802")
EJABBERD_HOST = os.environ.get("EJABBERD_HOST", "192.168.1.246")
# 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"
# Auto-recovery script name map (used by /api/agents)
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",
}
sys.path.insert(0, str(_GATEWAY_DIR / "scripts"))
from proc_guard import guard
app = Flask(__name__, template_folder=str(TEMPLATES_DIR))
# ---- Logging ----
LOG_FILE = LOGS_DIR / "dashboard.log"
LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
logging.basicConfig(
filename=str(LOG_FILE),
level=logging.INFO,
format="%(asctime)s [dashboard] %(message)s",
)
log = logging.getLogger("dashboard")
# ---- Constants ----
AGENTS_YAML = CONFIG_DIR / "agents.yaml"
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
_offline_counter: dict[str, int] = {}
# ============================================================
# Config
# ============================================================
def load_agents_config():
if AGENTS_YAML.exists():
import yaml
with open(AGENTS_YAML, "r", encoding="utf-8") as f:
return yaml.safe_load(f).get("agents", [])
return _default_agents()
def _default_agents():
return [
{
"id": "agent-001", "name": "R&D Assistant", "display_name": "xxm",
"jid": "xxm@yoin.fun", "platform": "windows", "host": "192.168.1.16",
"bot_type": "xmpp", "provider": "volcengine",
"services": [{"type": "xmpp_bot", "port": 5802}],
},
{
"id": "agent-002", "name": "Automation Manager", "display_name": "mohe",
"jid": "mohe@yoin.fun", "platform": "linux", "host": "192.168.1.246",
"bot_type": "hermes", "provider": "ocg-new",
"services": [{"type": "hermes_gateway", "port": 8642}, {"type": "xmpp_bot"}],
},
{
"id": "agent-003", "name": "Local Inference", "display_name": "xiaoguo",
"jid": "xiaoguo@yoin.fun", "platform": "mac", "host": "192.168.1.122",
"bot_type": "xmpp", "provider": "ocg-old",
"services": [{"type": "xmpp_bot"}, {"type": "omlx_server", "port": 18003}],
},
{
"id": "agent-004", "name": "Position Analyst", "display_name": "zhiwei",
"jid": "zhiwei@yoin.fun", "platform": "linux", "host": "192.168.1.246",
"bot_type": "hermes", "provider": "ocg-old",
"services": [{"type": "hermes_gateway", "port": 8643}, {"type": "xmpp_bot"}],
},
]
# ============================================================
# Cross-platform monitoring (XMPP + SSH)
# ============================================================
def _xmpp_health():
"""Query xmpp_bot /health for XMPP connection and ejabberd status."""
try:
req = urllib.request.Request(f"{XMPP_BRIDGE_URL}/health")
with urllib.request.urlopen(req, timeout=5) as resp:
return json.loads(resp.read())
except Exception as e:
return {"ok": False, "error": str(e), "xmpp_connected": False, "ejabberd_alive": False}
def _ejabberd_online_jids():
"""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:
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"):
line = line.strip()
if line and "@" in line:
jids.add(line.split("/")[0])
return jids
except Exception as e:
log.debug(f"ejabberd query failed: {e}")
return set()
def _muc_participants():
"""Fallback: query xmpp_bot /muc for room participants.
Currently unreliable due to MUC join timeout (R01)."""
try:
req = urllib.request.Request(f"{XMPP_BRIDGE_URL}/muc")
with urllib.request.urlopen(req, timeout=5) as resp:
data = json.loads(resp.read())
except Exception:
return set()
participants = set()
for room_data in data.get("rooms", {}).values():
for p in room_data.get("participants", []):
jid = p.get("jid", "")
if jid:
participants.add(jid)
nick = p.get("nick", "")
if nick and "@" in nick:
participants.add(nick)
return participants
# ============================================================
# Local process detection (Windows only)
# ============================================================
def _get_local_processes():
processes = []
try:
result = subprocess.run(["ps", "aux"], capture_output=True, text=True, timeout=5)
for line in result.stdout.split("\n"):
if "xmpp_bot" in line or "wechat_agent" in line or "dashboard" in line:
parts = line.split()
if len(parts) >= 11:
processes.append({"pid": int(parts[1]), "cmdline": " ".join(parts[10:])})
except Exception as e:
log.error(f"Process scan failed: {e}")
return processes
# ============================================================
# Log helpers
# ============================================================
def _tail_logs(max_lines=50):
all_lines = []
log_files = sorted(LOGS_DIR.glob("*.log"), key=lambda p: p.stat().st_mtime, reverse=True)
for lf in log_files:
try:
with open(lf, "r", encoding="utf-8", errors="replace") as f:
f.seek(0, os.SEEK_END)
size = f.tell()
read_size = min(size, max_lines * 500)
if read_size > 0:
f.seek(max(0, size - read_size))
for line in f.read().strip().split("\n"):
if line.strip():
all_lines.append(f"[{lf.name}] {line}")
except Exception:
pass
return all_lines[-max_lines:] if len(all_lines) > max_lines else all_lines
def _count_recent_messages(minutes=5):
log_path = LOGS_DIR / "xmpp_bot.log"
if not log_path.exists():
return 0
try:
with open(log_path, "r", encoding="utf-8", errors="replace") as f:
lines = f.readlines()
cutoff = datetime.now() - timedelta(minutes=minutes)
count = 0
pattern = re.compile(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})")
for line in reversed(lines):
m = pattern.search(line)
if m:
try:
if datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S") < cutoff:
break
count += 1
except ValueError:
pass
return count
except Exception:
return 0
# ============================================================
# Routes
# ============================================================
@app.route("/")
def index():
return send_from_directory(str(TEMPLATES_DIR), "dashboard.html")
@app.route("/api/agents")
def api_agents():
agents_config = load_agents_config()
local_procs = _get_local_processes()
message_count = _count_recent_messages(5)
# Primary: SSH ejabberdctl for cross-platform presence
online_jids = _ejabberd_online_jids()
if not online_jids:
online_jids = _muc_participants() # fallback
result = []
for agent in agents_config:
agent_id = agent["id"]
jid = agent.get("jid", "")
platform = agent.get("platform", "")
host = agent.get("host", "")
# --- Presence ---
xmpp_in_ejabberd = jid in online_jids if online_jids else None
# --- Local process (Windows only) ---
local_pid = None
xmpp_connected = False
if platform == "windows" and host in ("192.168.1.16", "127.0.0.1", "localhost"):
for proc in local_procs:
if "xmpp_bot.py" in proc.get("cmdline", ""):
local_pid = proc["pid"]
break
health = _xmpp_health()
xmpp_connected = health.get("xmpp_connected", False)
# --- Service status ---
services = []
for svc in agent.get("services", []):
svc_type = svc.get("type", "")
svc_port = svc.get("port")
if svc_type == "xmpp_bot":
if platform == "windows" and local_pid:
svc_status = "running" if xmpp_connected else "degraded"
svc_pid = local_pid
elif xmpp_in_ejabberd is True:
svc_status = "running"
svc_pid = None
elif xmpp_in_ejabberd is False:
svc_status = "stopped"
svc_pid = None
else:
svc_status = "unknown"
svc_pid = None
elif svc_type in ("hermes_gateway", "omlx_server"):
svc_status = "unknown"
svc_pid = None
else:
svc_status = "stopped"
svc_pid = None
for proc in local_procs:
script_name = SCRIPT_NAMES.get(svc_type, "")
if script_name and script_name in proc.get("cmdline", ""):
svc_status = "running"
svc_pid = proc["pid"]
break
services.append({
"type": svc_type,
"port": svc_port,
"status": svc_status,
"pid": svc_pid,
})
# --- Overall status ---
if xmpp_in_ejabberd is True:
status = "online"
elif xmpp_in_ejabberd is False:
status = "offline"
elif platform == "windows" and host in ("192.168.1.16", "127.0.0.1", "localhost"):
if xmpp_connected:
status = "online"
elif local_pid:
status = "degraded"
else:
status = "offline"
else:
status = "unknown"
# --- Auto-recovery ---
if status == "offline" and platform == "windows":
_offline_counter[agent_id] = _offline_counter.get(agent_id, 0) + 1
if _offline_counter[agent_id] >= AUTO_RECOVER_THRESHOLD:
log.warning(f"Auto-recovery: restarting {agent_id}")
_try_auto_recover(agent)
else:
_offline_counter[agent_id] = 0
result.append({
"id": agent_id,
"name": agent.get("name", ""),
"display_name": agent.get("display_name", ""),
"jid": jid,
"platform": platform,
"host": host,
"status": status,
"xmpp_connected": xmpp_connected,
"pid": local_pid,
"last_message": None,
"message_count_5min": message_count,
"errors": 0,
"offline_checks": _offline_counter.get(agent_id, 0),
"restartable": platform == "windows" and host in ("192.168.1.16", "127.0.0.1", "localhost"),
"services": services,
})
return jsonify(result)
def _try_auto_recover(agent):
agent_id = agent["id"]
platform = agent.get("platform", "")
host = agent.get("host", "")
if platform != "windows" or host not in ("192.168.1.16", "127.0.0.1", "localhost"):
return
for svc in agent.get("services", []):
script_name = SCRIPT_NAMES.get(svc.get("type", ""))
if not script_name:
continue
script_path = SCRIPTS_DIR / script_name
if not script_path.exists():
continue
try:
subprocess.Popen(
[PYTHON, str(script_path)],
cwd=str(SCRIPTS_DIR),
creationflags=subprocess.CREATE_NO_WINDOW,
)
log.info(f"Auto-restarted {script_name} for {agent_id}")
except Exception as e:
log.error(f"Auto-restart failed: {e}")
@app.route("/api/ejabberd")
def api_ejabberd():
health = _xmpp_health()
online_jids = _ejabberd_online_jids()
return jsonify({
"alive": len(online_jids) > 0,
"xmpp_bot_connected": health.get("xmpp_connected", False),
"online_jids": sorted(list(online_jids)) if online_jids else [],
"bot_jid": health.get("bot_jid", ""),
})
@app.route("/api/agents/<agent_id>/logs")
def api_agent_logs(agent_id):
lines = request.args.get("lines", 50, type=int)
return jsonify({"lines": _tail_logs(lines)})
@app.route("/api/agents/<agent_id>/start", methods=["POST"])
def api_agent_start(agent_id):
agents_config = load_agents_config()
agent = next((a for a in agents_config if a["id"] == agent_id), None)
if not agent:
return jsonify({"ok": False, "error": "Agent not found"}), 404
if agent.get("platform") != "windows":
return jsonify({"ok": False, "error": "Remote restart not supported yet"}), 400
started = []
for svc in agent.get("services", []):
script_name = SCRIPT_NAMES.get(svc.get("type", ""))
if not script_name:
continue
script_path = SCRIPTS_DIR / script_name
if not script_path.exists():
continue
try:
subprocess.Popen(
[PYTHON, str(script_path)],
cwd=str(SCRIPTS_DIR),
creationflags=subprocess.CREATE_NO_WINDOW,
)
started.append(script_name)
log.info(f"Started {script_name} for {agent_id}")
except Exception as e:
log.error(f"Failed to start {script_name}: {e}")
_offline_counter[agent_id] = 0
return jsonify({"ok": True, "started": started})
@app.route("/api/agents/<agent_id>/stop", methods=["POST"])
def api_agent_stop(agent_id):
agents_config = load_agents_config()
agent = next((a for a in agents_config if a["id"] == agent_id), None)
if not agent:
return jsonify({"ok": False, "error": "Agent not found"}), 404
if agent.get("platform") != "windows":
return jsonify({"ok": False, "error": "Remote stop not supported yet"}), 400
processes = _get_local_processes()
stopped = []
for svc in agent.get("services", []):
script_name = SCRIPT_NAMES.get(svc.get("type", ""))
if not script_name:
continue
for proc in processes:
if script_name in proc.get("cmdline", ""):
pid = proc["pid"]
try:
subprocess.run(["taskkill", "/f", "/pid", str(pid)], capture_output=True)
stopped.append({"script": script_name, "pid": pid})
except Exception as e:
log.error(f"Failed to stop {script_name}: {e}")
return jsonify({"ok": True, "stopped": stopped})
@app.route("/api/agents/<agent_id>/restart", methods=["POST"])
def api_agent_restart(agent_id):
api_agent_stop(agent_id)
time.sleep(2)
return api_agent_start(agent_id)
PLATFORM_SERVICES = [
{"id": "wechat_bridge", "name": "WeChat Bridge", "type": "ChannelBridge",
"desc": "bridges WeChat to mohe's hermes gateway",
"health_url": "http://192.168.1.16:5801/health"},
{"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")
def api_platform():
"""Return platform services status by querying health endpoints."""
import urllib.request as _ur
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)
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:
import socket
try:
s = socket.socket()
s.settimeout(2)
s.connect((ps.get("host", "127.0.0.1"), ps["port"]))
s.close()
status = "running"
except Exception:
status = "stopped"
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})
# ════════════════════════════════════════════════════════════
# A — 源码管理可视化
# ════════════════════════════════════════════════════════════
@app.route("/api/git")
def api_git():
"""最近 git 提交历史 + 分支状态（从 .git 直接读取，不依赖 git 命令）"""
try:
# 先用 _PROJECT_DIR（已处理 AGENTSMEETING_ROOT 覆盖），不行再探测常见路径
candidates = [str(_PROJECT_DIR)]
if _PROJECT_OVERRIDE:
candidates.append(str(_PROJECT_DIR))
# 246 生产环境常见路径
if sys.platform != "win32":
candidates.extend([
"/home/hmo/projects/AgentsMeeting",
os.path.expanduser("~/projects/AgentsMeeting"),
])
git_dir = None
for c in candidates:
if os.path.exists(os.path.join(c, ".git")):
git_dir = c
break
if not git_dir:
return jsonify({"ok": True, "log": [], "dirty": False, "note": "未找到.git目录,请设置AGENTSMEETING_ROOT"})
dot_git = os.path.join(git_dir, ".git")
log_lines = []
# 优先用 git log 命令（真实提交信息），降级到解析 reflog
tried_git = False
# Linux nohup 环境 PATH 可能不含 /usr/bin，必须用绝对路径
GIT_CANDIDATES = [
os.environ.get("GIT_CMD", ""),
shutil.which("git") or "",
"/usr/bin/git", "/usr/local/bin/git",
]
git_bin = ""
for c in GIT_CANDIDATES:
if c and os.path.exists(c):
git_bin = c
break
if git_bin:
try:
r = subprocess.run(
[git_bin, "log", "--format=%h|%ad|%s", "--date=format:%m-%d %H:%M", "--no-decorate"],
cwd=git_dir, capture_output=True, text=True, timeout=10)
if r.returncode == 0 and r.stdout.strip():
tried_git = True
for line in r.stdout.strip().split("\n"):
line = line.strip()
if line:
parts = line.split("|", 2)
if len(parts) == 3:
log_lines.append(f"{parts[0]} {parts[1]} {parts[2]}")
else:
log_lines.append(line)
except:
pass
if not tried_git:
reflog = os.path.join(dot_git, "logs", "HEAD")
if os.path.exists(reflog):
with open(reflog, "r", encoding="utf-8", errors="replace") as f:
lines = f.readlines()
for line in reversed(lines):
line = line.strip()
if not line:
continue
parts = line.split("\t")
msg = parts[-1] if len(parts) > 1 else line
hash_part = line.split()[1][:7] if len(line.split()) > 1 else "?"
# 从 reflog 提取时间戳: 第3+4字段 = unix时间 + 时区
ts = ""
try:
cols = line.split()
if len(cols) >= 5:
ts = datetime.fromtimestamp(int(cols[3])).strftime("%m-%d %H:%M") + " "
except:
pass
log_lines.append(f"{ts}{hash_part} {msg}")
# 检查是否有未提交的改动
dirty = False
head_file = os.path.join(dot_git, "HEAD")
index_file = os.path.join(dot_git, "index")
if os.path.exists(head_file) and os.path.exists(index_file):
try:
dirty = os.path.getmtime(index_file) > os.path.getmtime(head_file)
except:
pass
# 分支名
branch = "master"
try:
with open(head_file, "r") as f:
ref = f.read().strip()
if ref.startswith("ref: refs/heads/"):
branch = ref[16:]
except:
pass
return jsonify({"ok": True, "log": log_lines, "dirty": dirty, "branch": branch, "source": git_dir, "method": ("git_log" if tried_git else "reflog")})
except Exception as e:
return jsonify({"ok": False, "error": str(e), "log": [], "dirty": False})
# ════════════════════════════════════════════════════════════
# B — 服务架构可视化
# ════════════════════════════════════════════════════════════
@app.route("/api/services")
def api_services():
"""服务功能树 + watchdog 覆盖矩阵"""
try:
services = []
# xmpp_bot
svc1 = {"name": "xmpp_bot", "port": 5802, "health": _health_status("http://127.0.0.1:5802/health"),
"watchdog": True, "pid_lock": True, "type": "core", "depends_on": ["ejabberd"]}
services.append(svc1)
# article_processor
svc2 = {"name": "article_processor", "port": 5810, "health": _health_status("http://127.0.0.1:5810/health"),
"watchdog": True, "pid_lock": False, "type": "bridge", "depends_on": []}
services.append(svc2)
# dashboard
svc3 = {"name": "dashboard", "port": 5803, "health": _health_status("http://127.0.0.1:5803/api/health"),
"watchdog": False, "pid_lock": True, "type": "mgmt", "depends_on": ["xmpp_bot"]}
services.append(svc3)
# ejabberd
online = _ejabberd_online_jids()
services.append({"name": "ejabberd", "port": 5222, "health": {"ok": len(online) > 0},
"watchdog": True, "pid_lock": False, "type": "infra", "depends_on": []})
# remote gateways (R01 monitoring)
g1 = {"name": "gateway_mohe", "port": 8642, "health": _health_status("http://192.168.1.246:8642/v1/health"),
"watchdog": False, "pid_lock": False, "type": "gateway", "depends_on": ["ejabberd"]}
services.append(g1)
g2 = {"name": "gateway_zhiwei", "port": 8643, "health": _health_status("http://192.168.1.246:8643/v1/health"),
"watchdog": False, "pid_lock": False, "type": "gateway", "depends_on": ["ejabberd"]}
services.append(g2)
# 统计
watched = sum(1 for s in services if s.get("watchdog"))
return jsonify({"ok": True, "services": services, "watched": watched, "total": len(services)})
except Exception as e:
return jsonify({"ok": False, "error": str(e)})
def _health_status(url, timeout=3):
try:
r = urllib.request.urlopen(url, timeout=timeout)
return {"ok": r.status == 200, "status": r.status, "detail": "ok"}
except urllib.error.HTTPError as e:
return {"ok": False, "status": e.code, "detail": f"HTTP {e.code}"}
except Exception as e:
return {"ok": False, "status": 0, "detail": str(e)[:50]}
# ════════════════════════════════════════════════════════════
# C — 监控结果可视化
# ════════════════════════════════════════════════════════════
@app.route("/api/monitor")
def api_monitor():
"""Tier1/Tier2 最近检查结果 + 定时任务状态"""
result = {"ok": True, "tier1": None, "tier2": None, "tasks": [], "platform": sys.platform}
# Tier1 最新报告
t1_file = TEMP_DIR / "last_health_check.json"
if t1_file.exists():
try:
with open(str(t1_file)) as f:
result["tier1"] = json.load(f)
except:
pass
else:
result["tier1"] = {"status": "no_data", "note": f"{t1_file} 不存在", "summary": {"ok": 0, "total": 0}}
# Tier2 最新报告
t2_file = TEMP_DIR / "last_daily_health.json"
if t2_file.exists():
try:
with open(str(t2_file)) as f:
result["tier2"] = json.load(f)
except:
pass
else:
result["tier2"] = {"status": "no_data", "note": f"{t2_file} 不存在", "summary": {"ok": 0, "total": 0}}
# 定时任务状态
expected_tasks = ["agents-health-check", "agents-daily-health", "agents-todo-executor"]
if sys.platform == "win32":
for name in expected_tasks:
try:
r = subprocess.run(["schtasks", "/Query", "/TN", name, "/FO", "CSV", "/NH"],
capture_output=True, text=True, timeout=5,
creationflags=subprocess.CREATE_NO_WINDOW)
status = "ok" if ("Ready" in r.stdout or "\u5c31\u7eea" in r.stdout) else ("missing" if name not in r.stdout else "other")
except:
status = "error"
result["tasks"].append({"name": name, "status": status})
else:
# Linux: 检查 systemd timer（如已部署）或 crontab
for name in expected_tasks:
try:
r = subprocess.run(["systemctl", "list-timers", "--all", "--no-pager"],
capture_output=True, text=True, timeout=5)
if name in r.stdout:
result["tasks"].append({"name": name, "status": "timer_ok"})
else:
# 查 crontab
r2 = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=5)
if name in r2.stdout:
result["tasks"].append({"name": name, "status": "cron_ok"})
else:
result["tasks"].append({"name": name, "status": "not_deployed"})
except:
result["tasks"].append({"name": name, "status": "not_deployed", "note": "无systemd/cron"})
return jsonify(result)
# ════════════════════════════════════════════════════════════
# E — 元成长回路可视化
# ════════════════════════════════════════════════════════════
@app.route("/api/metagrowth")
def api_metagrowth():
"""元成长回路（Phase 3 规划中）"""
try:
# 统计已完成的修复次数
todo_file = TEMP_DIR / "health_todos.jsonl"
total_fixes = 0
completed_fixes = 0
if todo_file.exists():
with open(str(todo_file), encoding="utf-8-sig") as f:
for line in f:
line = line.strip()
if line:
total_fixes += 1
try:
entry = json.loads(line)
except json.JSONDecodeError:
continue
if entry.get("status") == "completed":
completed_fixes += 1
# 统计 git 提交次数（从 reflog 行数计算，跟 api_git 同样路径探测逻辑）
commit_count = 0
git_candidates = [str(_PROJECT_DIR)]
if sys.platform != "win32":
git_candidates.extend(["/home/hmo/projects/AgentsMeeting",
os.path.expanduser("~/projects/AgentsMeeting")])
for c in git_candidates:
rfl = os.path.join(c, ".git", "logs", "HEAD")
if os.path.exists(rfl):
try:
with open(rfl, "r", encoding="utf-8", errors="replace") as f:
commit_count = sum(1 for line in f if line.strip())
except:
pass
break
return jsonify({
"ok": True,
"status": "planned", # Phase 3
"total_fixes": total_fixes,
"completed_fixes": completed_fixes,
"commit_count": commit_count,
"meta_growth_enabled": False,
"description": "E 元成长回路是 Phase 3 范围。启用后将自动分析修复模式并扩展检查规则。",
})
except Exception as e:
return jsonify({"ok": False, "error": str(e)})
# ════════════════════════════════════════════════════════════
# D — 自修复流水线可视化
# ════════════════════════════════════════════════════════════
@app.route("/api/todos")
def api_todos():
"""TODO 执行记录 + escalation 日志"""
todos = []
todo_file = TEMP_DIR / "health_todos.jsonl"
if todo_file.exists():
try:
with open(str(todo_file)) as f:
for line in f:
line = line.strip()
if line:
todos.append(json.loads(line))
except:
pass
# executor 日志最新行
executor_log = LOGS_DIR / "todo_executor.log"
executor_tail = ""
if executor_log.exists():
try:
lines = open(str(executor_log)).read().strip().split("\n")
executor_tail = "\n".join(lines[-5:])
except:
pass
return jsonify({"todos": todos, "executor_log_tail": executor_tail, "count": len(todos)})
# ════════════════════════════════════════════════════════════
# F — 期望状态 vs 实际状态
# ════════════════════════════════════════════════════════════
@app.route("/api/expected")
def api_expected():
"""期望状态矩阵：应该运行的 vs 实际运行的"""
expected = [
{"name": "xmpp_bot", "port": 5802, "expected": "running", "critical": True},
{"name": "article_processor", "port": 5810, "expected": "running", "critical": True},
{"name": "dashboard", "port": 5803, "expected": "running", "critical": True},
{"name": "watchdog", "expected": "running", "critical": True},
{"name": "agents-health-check", "expected": "scheduled (5min)", "critical": False},
{"name": "agents-daily-health", "expected": "scheduled (daily 08:00)", "critical": False},
{"name": "agents-todo-executor", "expected": "scheduled (10min)", "critical": False},
]
actual = {}
for e in expected:
name = e["name"]
port = e.get("port")
if port:
actual[name] = "running" if port_open(port) else "stopped"
elif name == "watchdog":
if sys.platform == "win32":
try:
r = subprocess.run(
["tasklist", "/FI", "IMAGENAME eq python.exe", "/NH"],
capture_output=True, text=True, timeout=5,
creationflags=subprocess.CREATE_NO_WINDOW,
)
actual[name] = "running" if "python" in r.stdout else "stopped"
except:
actual[name] = "unknown"
else:
try:
r = subprocess.run(
["pgrep", "-f", "xmpp_watchdog|health_check"],
capture_output=True, text=True, timeout=5)
actual[name] = "running" if r.stdout.strip() else "stopped"
except:
# pgrep not available
r = subprocess.run(
["ps", "aux"], capture_output=True, text=True, timeout=5)
has = "watchdog" in r.stdout or "health_check" in r.stdout
actual[name] = "running" if has else "stopped"
elif "scheduled" in e["expected"]:
task_name = name
if sys.platform == "win32":
try:
r = subprocess.run(
["schtasks", "/Query", "/TN", task_name, "/FO", "CSV", "/NH"],
capture_output=True, text=True, timeout=5,
creationflags=subprocess.CREATE_NO_WINDOW,
)
# 中文"就绪"、英文"Ready"
ready = "Ready" in r.stdout or "\u5c31\u7eea" in r.stdout
actual[name] = "scheduled" if ready else "missing"
except:
actual[name] = "error"
else:
# Linux: 检查 systemd timer 或 crontab
try:
r = subprocess.run(["systemctl", "list-timers", "--all", "--no-pager"],
capture_output=True, text=True, timeout=5)
if task_name in r.stdout:
actual[name] = "timer_ok"
else:
r2 = subprocess.run(["crontab", "-l"], capture_output=True,
text=True, timeout=5)
actual[name] = "cron_ok" if task_name in r2.stdout else "not_deployed"
except:
actual[name] = "not_deployed"
return jsonify({"expected": expected, "actual": actual})
def port_open(port):
"""Check if a port is listening. Works on both Linux and Windows."""
try:
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(2)
result = s.connect_ex(("127.0.0.1", port))
s.close()
return result == 0
except:
return False
# EasyTier/RDP API key for xmpp_bot on Windows
_BRIDGE_KEY = "xxm_bridge_8f3a2c"
def _bridge_post(path, payload, timeout=15):
"""POST to xmpp_bot HTTP bridge on Windows."""
req = urllib.request.Request(
f"{XMPP_BRIDGE_URL}{path}",
data=json.dumps(payload).encode(),
headers={"Content-Type": "application/json", "X-Api-Key": _BRIDGE_KEY},
method="POST")
resp = urllib.request.urlopen(req, timeout=timeout)
return json.loads(resp.read())
# ════════════════════════════════════════════════════════════
# EasyTier VPN + RDP 远程桌面控制
# ════════════════════════════════════════════════════════════
@app.route("/api/easytier")
def api_easytier_status():
"""EasyTier VPN 状态 — proxy to xmpp_bot on Windows."""
try:
windows_running = False
status_246 = "unknown"
try:
data = _bridge_post("/easytier", {"action": "status"}, timeout=5)
windows_running = data.get("running", False)
status_246 = "running" if data.get("running") else "stopped"
except:
pass
return jsonify({
"ok": True,
"status": {"windows": "running" if windows_running else "stopped", "246": status_246},
"virtual_ips": {"windows": "10.144.144.3", "246": "10.144.144.1"},
"windows_running": windows_running,
})
except Exception as e:
return jsonify({"ok": False, "error": str(e)})
@app.route("/api/easytier/toggle", methods=["POST"])
def api_easytier_toggle():
"""EasyTier VPN 开关 — proxy to xmpp_bot on Windows."""
try:
body = json.loads(request.data)
action = body.get("action", "")
if action not in ("start", "stop"):
return jsonify({"ok": False, "error": "action must be start|stop"})
data = _bridge_post("/easytier", {"action": action}, timeout=15)
return jsonify({"ok": data.get("ok", False), "message": data.get("message", "")})
except Exception as e:
return jsonify({"ok": False, "error": str(e)})
@app.route("/api/rdp")
def api_rdp_status():
"""RDP 远程桌面状态 — proxy to xmpp_bot on Windows."""
try:
rdp_enabled = False
tunnel_running = False
try:
data = _bridge_post("/rdp", {"action": "status"}, timeout=5)
rdp_enabled = data.get("rdp_enabled", False)
tunnel_running = data.get("tunnel_running", False)
except:
pass
return jsonify({
"ok": True,
"rdp_enabled": rdp_enabled,
"tunnel_running": tunnel_running,
"rdp_port": 3389,
"tunnel_port": 8080,
"public_endpoint": "47.115.32.206:8080",
})
except Exception as e:
return jsonify({"ok": False, "error": str(e)})
@app.route("/api/rdp/toggle", methods=["POST"])
def api_rdp_toggle():
"""RDP 远程桌面开关 — proxy to xmpp_bot on Windows."""
try:
body = json.loads(request.data)
action = body.get("action", "")
if action not in ("start", "stop"):
return jsonify({"ok": False, "error": "action must be start|stop"})
data = _bridge_post("/rdp", {"action": action}, timeout=15)
return jsonify({"ok": data.get("ok", False), "message": data.get("message", "")})
except Exception as e:
return jsonify({"ok": False, "error": str(e)})
# ════════════════════════════════════════════════════════════
# Spec 文档展示
# ════════════════════════════════════════════════════════════
@app.route("/api/spec")
def api_spec():
"""返回开发规范文档内容"""
# Try multiple possible locations
candidates = [
os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))), ".memory", "dev-spec.md"),
os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), ".memory", "dev-spec.md"),
os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "..", "..", ".memory", "dev-spec.md"),
# 246 venv: ~/ (5 dirname ups from venv/dashboard.py)
os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))), ".memory", "dev-spec.md"),
os.path.expanduser("~/.memory/dev-spec.md"),
]
spec_path = None
for c in candidates:
p = os.path.normpath(c)
if os.path.exists(p):
spec_path = p
break
if not spec_path:
return jsonify({"ok": False, "error": "dev-spec.md not found", "content": ""})
try:
with open(spec_path, "r", encoding="utf-8") as f:
content = f.read()
return jsonify({"ok": True, "content": content, "path": spec_path})
except Exception as e:
return jsonify({"ok": False, "error": str(e), "content": ""})
@app.route("/api/prd")
def api_prd():
"""返回产品需求文档 (PRD) 内容"""
candidates = [
os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "docs", "PRD.md"),
os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "docs", "PRD.md"),
# venv deployment: ~/projects/AgentsMeeting/docs/PRD.md
os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))), "projects", "AgentsMeeting", "docs", "PRD.md"),
os.path.expanduser("~/projects/AgentsMeeting/docs/PRD.md"),
]
prd_path = None
for c in candidates:
p = os.path.normpath(c)
if os.path.exists(p):
prd_path = p
break
if not prd_path:
return jsonify({"ok": False, "error": "PRD.md not found", "content": ""})
try:
with open(prd_path, "r", encoding="utf-8") as f:
content = f.read()
return jsonify({"ok": True, "content": content, "path": prd_path})
except Exception as e:
return jsonify({"ok": False, "error": str(e), "content": ""})
# ════════════════════════════════════════════════════════════
# K — 自动测试接口
# ════════════════════════════════════════════════════════════
@app.route("/api/tests")
def api_tests():
"""运行自动测试并返回结果"""
try:
from tests_api import run_tests
return jsonify(run_tests())
except Exception as e:
return jsonify({"ok": False, "error": str(e)})
@app.route("/api/platform")
@app.route("/api/health")
def api_health():
xmpp = _xmpp_health()
return jsonify({
"ok": True,
"time": datetime.now().isoformat(),
"xmpp_bot_alive": xmpp.get("xmpp_connected", False),
"ejabberd_alive": xmpp.get("ejabberd_alive", False),
})
# ============================================================
# Main
# ============================================================
def main():
lock = guard("dashboard")
if not lock.ok:
log.error(lock.message)
print(f"[dashboard] {lock.message}")
sys.exit(1)
port = int(os.environ.get("DASHBOARD_PORT", 5803))
log.info(f"Dashboard starting on :{port}")
print(f"[dashboard] Starting on http://127.0.0.1:{port}")
app.run(host="0.0.0.0", port=port, debug=False, use_reloader=False, threaded=True)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink