feat(ops): add batch Agent remote diagnose script

SSH from central creds to inspect systemctl, redacted config, 401/heartbeat logs, and /health on sub-servers.

docs/audit/2026-06-09-agent-remote-diagnose.md

@@ -0,0 +1,27 @@
+# 审计 — 子机 Agent 批量诊断脚本
+
+**Changelog**: `docs/changelog/2026-06-09-agent-remote-diagnose-script.md`
+
+## 变更文件
+
+| 文件 | 说明 |
+|------|------|
+| `scripts/agent_remote_diagnose.py` | 中心侧批量 SSH 诊断 |
+| `scripts/agent_remote_diagnose.sh` | 包装入口 |
+| `tests/test_agent_remote_diagnose.py` | 解析/脱敏单测 |
+
+## Step 3
+
+| 规则 | 结论 |
+|------|------|
+| 安全 | PASS — 仅脱敏输出 api_key；SSH 凭据来自 DB 加密字段 |
+| 行为 | PASS — 只读诊断，无 CUD |
+
+## Closure
+
+- `pytest tests/test_agent_remote_diagnose.py -q` PASS
+
+## DoD
+
+- [x] changelog + audit
+- [ ] 生产 `/opt/nexus` 同步脚本并验证 `--id 95`

docs/changelog/2026-06-09-agent-remote-diagnose-script.md

@@ -0,0 +1,31 @@
+# 2026-06-09 — 子机 Agent 批量诊断脚本
+
+## 摘要
+
+新增 `scripts/agent_remote_diagnose.py`：从中心侧按服务器 ID/名称/离线筛选，SSH 拉取子机 `systemctl`、脱敏 `config.json`、401/心跳错误日志、主站 `/health` 探测；并对照 Redis/MySQL 心跳。
+
+## 动机
+
+排查「显示离线但 SSH 可连」类问题时，需登录每台子机看 `nexus-agent` 日志；武汉颜又文案需可重复工具。
+
+## 涉及文件
+
+- `scripts/agent_remote_diagnose.py`
+- `scripts/agent_remote_diagnose.sh`
+- `tests/test_agent_remote_diagnose.py`
+
+## 用法
+
+```bash
+bash scripts/agent_remote_diagnose.sh --id 95
+bash scripts/agent_remote_diagnose.sh --name 颜又文
+bash scripts/agent_remote_diagnose.sh --offline-only --limit 20 -o docs/reports/agent-offline-snapshot.md
+```
+
+需在含 `.env` 的目录执行（本地或生产 `/opt/nexus`）。
+
+## 验证
+
+```bash
+pytest tests/test_agent_remote_diagnose.py -q
+```

scripts/agent_remote_diagnose.py

@@ -0,0 +1,303 @@
+#!/usr/bin/env python3
+"""Batch SSH diagnostics for Nexus Agent on sub-servers (run from central / dev with .env).
+
+Collects per server:
+  - Central: Redis heartbeat, MySQL last_heartbeat, agent_api_key_set
+  - Remote (SSH): systemctl, redacted config.json, recent 401/heartbeat errors, /health curl
+
+Usage:
+  python scripts/agent_remote_diagnose.py --id 95
+  python scripts/agent_remote_diagnose.py --name 颜又文
+  python scripts/agent_remote_diagnose.py --offline-only --limit 20
+  python scripts/agent_remote_diagnose.py --ids 95,12,33
+
+Requires .env with DATABASE_URL + REDIS_URL (or NEXUS_*). SSH creds from servers table.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+REDIS_KEY_PREFIX = "heartbeat:"
+
+REMOTE_AGENT_DIAG_SHELL = r"""
+INSTALL_DIR=/opt/nexus-agent
+echo "### systemctl ###"
+(systemctl is-active nexus-agent 2>/dev/null || echo inactive)
+systemctl show nexus-agent -p ActiveState,SubState,ActiveEnterTimestamp --value 2>/dev/null | tr '\n' ' '
+echo
+echo "### config_redacted ###"
+if [ -f "$INSTALL_DIR/config.json" ]; then
+  python3 -c "import json; c=json.load(open('$INSTALL_DIR/config.json')); k=c.get('api_key',''); c['api_key']=((k[:8]+'...') if k else ''); print(json.dumps(c, ensure_ascii=False))"
+else
+  echo missing_config
+fi
+echo "### log_errors ###"
+for f in /var/log/nexus-agent.log "$INSTALL_DIR/agent.log"; do
+  if [ -f "$f" ]; then
+    echo "--- $f ---"
+    tail -40 "$f" | grep -E '401|Heartbeat rejected|Stopping heartbeat|Heartbeat error|discarded' || echo "(no matching lines)"
+  fi
+done
+if command -v journalctl >/dev/null 2>&1; then
+  echo "--- journalctl ---"
+  journalctl -u nexus-agent --no-pager -n 30 2>/dev/null | grep -E '401|Heartbeat rejected|Stopping heartbeat|Heartbeat error|discarded' || echo "(no matching lines)"
+fi
+echo "### central_health ###"
+CU=$(python3 -c "import json; print(json.load(open('$INSTALL_DIR/config.json')).get('central_url',''))" 2>/dev/null || true)
+if [ -n "$CU" ]; then
+  curl -fsS -o /dev/null -w "http_code=%{http_code}\n" "${CU%/}/health" --connect-timeout 8 2>&1 || echo curl_failed
+else
+  echo no_central_url
+fi
+""".strip()
+
+
+def _load_dotenv() -> None:
+    env_path = ROOT / ".env"
+    if not env_path.is_file():
+        return
+    for line in env_path.read_text(encoding="utf-8", errors="replace").splitlines():
+        line = line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, _, value = line.partition("=")
+        key, value = key.strip(), value.strip().strip("\"'")
+        if key.startswith("NEXUS_") and key[6:] not in os.environ:
+            os.environ[key[6:]] = value
+        elif key and key not in os.environ:
+            os.environ[key] = value
+
+
+def _has_agent_monitoring(server: Any) -> bool:
+    return bool((server.agent_api_key or "").strip() or (server.agent_version or "").strip())
+
+
+def _central_status(has_agent: bool, heartbeat: dict[str, str] | None) -> str:
+    if heartbeat:
+        return "online" if heartbeat.get("is_online") == "True" else "offline"
+    if has_agent:
+        return "offline"
+    return "unknown"
+
+
+async def _fetch_servers(
+    session: Any,
+    *,
+    server_id: int | None,
+    server_ids: list[int] | None,
+    name_like: str | None,
+    offline_only: bool,
+    limit: int,
+) -> list[Any]:
+    from sqlalchemy import or_, select
+
+    from server.domain.models import Server
+    from server.infrastructure.redis.client import get_redis
+
+    q = select(Server).order_by(Server.id)
+    if server_id is not None:
+        q = q.where(Server.id == server_id)
+    elif server_ids:
+        q = q.where(Server.id.in_(server_ids))
+    if name_like:
+        q = q.where(or_(Server.name.like(f"%{name_like}%"), Server.domain.like(f"%{name_like}%")))
+
+    result = await session.execute(q)
+    servers = list(result.scalars().all())
+
+    if offline_only:
+        redis = get_redis()
+        filtered = []
+        for s in servers:
+            if not _has_agent_monitoring(s):
+                continue
+            hb = await redis.hgetall(f"{REDIS_KEY_PREFIX}{s.id}")
+            if _central_status(True, hb or None) == "offline":
+                filtered.append(s)
+        servers = filtered
+
+    if limit > 0:
+        servers = servers[:limit]
+    return servers
+
+
+async def _probe_remote(server: Any, *, timeout: int) -> dict[str, Any]:
+    from server.application.server_batch_common import sudo_wrap
+    from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
+
+    ssh_user = (server.username or "root").strip()
+    cmd = sudo_wrap(REMOTE_AGENT_DIAG_SHELL, ssh_user)
+    try:
+        r = await exec_ssh_command(server, cmd, timeout=timeout)
+        return {
+            "ssh_ok": r.get("exit_code") == 0,
+            "exit_code": r.get("exit_code"),
+            "stdout": (r.get("stdout") or "").strip(),
+            "stderr": (r.get("stderr") or "").strip(),
+            "error": "",
+        }
+    except Exception as exc:
+        return {
+            "ssh_ok": False,
+            "exit_code": -1,
+            "stdout": "",
+            "stderr": "",
+            "error": str(exc)[:500],
+        }
+
+
+async def _diagnose_one(
+    server: Any,
+    heartbeat: dict[str, str] | None,
+    *,
+    ssh: bool,
+    timeout: int,
+) -> dict[str, Any]:
+    has_agent = _has_agent_monitoring(server)
+    status = _central_status(has_agent, heartbeat)
+    row: dict[str, Any] = {
+        "id": server.id,
+        "name": server.name,
+        "domain": server.domain,
+        "central_status": status,
+        "has_agent": has_agent,
+        "agent_version_db": (server.agent_version or "").strip(),
+        "last_heartbeat_db": str(server.last_heartbeat or ""),
+        "last_heartbeat_redis": (heartbeat or {}).get("last_heartbeat", ""),
+        "redis_ttl_hint": "fresh" if heartbeat else "missing",
+    }
+    if ssh:
+        remote = await _probe_remote(server, timeout=timeout)
+        row["remote"] = remote
+    return row
+
+
+async def run_diagnose(args: argparse.Namespace) -> list[dict[str, Any]]:
+    from server.config import settings
+    from server.infrastructure.database.session import AsyncSessionLocal, init_db
+    from server.infrastructure.redis.client import close_redis, get_redis, init_redis
+
+    await init_db()
+    await init_redis(settings.REDIS_URL)
+    redis = get_redis()
+
+    ids: list[int] | None = None
+    if args.ids:
+        ids = [int(x.strip()) for x in args.ids.split(",") if x.strip()]
+
+    async with AsyncSessionLocal() as session:
+        servers = await _fetch_servers(
+            session,
+            server_id=args.id,
+            server_ids=ids,
+            name_like=args.name,
+            offline_only=args.offline_only,
+            limit=args.limit,
+        )
+
+    sem = asyncio.Semaphore(max(1, args.concurrency))
+
+    async def one(server: Any) -> dict[str, Any]:
+        async with sem:
+            hb = await redis.hgetall(f"{REDIS_KEY_PREFIX}{server.id}")
+            return await _diagnose_one(
+                server,
+                hb or None,
+                ssh=not args.central_only,
+                timeout=args.timeout,
+            )
+
+    try:
+        return list(await asyncio.gather(*(one(s) for s in servers)))
+    finally:
+        await close_redis()
+
+
+def _format_report(rows: list[dict[str, Any]], *, verbose: bool) -> str:
+    lines: list[str] = []
+    ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
+    lines.append(f"# Nexus Agent remote diagnose @ {ts}")
+    lines.append(f"servers: {len(rows)}")
+    lines.append("")
+
+    for row in rows:
+        lines.append(f"## [{row['id']}] {row['name']} ({row['domain']})")
+        lines.append(
+            f"- central: **{row['central_status']}** | "
+            f"redis_hb={row['last_heartbeat_redis'] or '—'} | "
+            f"mysql_hb={row['last_heartbeat_db'] or '—'} | "
+            f"agent_db={row['agent_version_db'] or '—'}"
+        )
+        remote = row.get("remote")
+        if remote:
+            if remote.get("error"):
+                lines.append(f"- SSH: **FAIL** — {remote['error']}")
+            elif not remote.get("ssh_ok"):
+                lines.append(
+                    f"- SSH: exit {remote.get('exit_code')} stderr={remote.get('stderr', '')[:200]}"
+                )
+            if verbose or remote.get("stdout"):
+                body = remote.get("stdout") or "(empty)"
+                lines.append("```")
+                lines.append(body[:4000])
+                lines.append("```")
+        lines.append("")
+
+    return "\n".join(lines).rstrip() + "\n"
+
+
+def build_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(description="SSH batch diagnostics for Nexus Agent on sub-servers")
+    p.add_argument("--id", type=int, help="Single server id")
+    p.add_argument("--ids", help="Comma-separated server ids")
+    p.add_argument("--name", help="Filter name/domain substring (SQL LIKE)")
+    p.add_argument("--offline-only", action="store_true", help="Only servers with Agent but no Redis heartbeat")
+    p.add_argument("--limit", type=int, default=50, help="Max servers (default 50)")
+    p.add_argument("--concurrency", type=int, default=5, help="Parallel SSH probes (default 5)")
+    p.add_argument("--timeout", type=int, default=45, help="SSH command timeout seconds")
+    p.add_argument("--central-only", action="store_true", help="Skip SSH; only Redis/MySQL snapshot")
+    p.add_argument("-o", "--output", help="Write report markdown to file")
+    p.add_argument("-q", "--quiet", action="store_true", help="Only print summary lines")
+    return p
+
+
+def main() -> int:
+    _load_dotenv()
+    args = build_parser().parse_args()
+    if not any([args.id, args.ids, args.name, args.offline_only]):
+        print("Specify --id, --ids, --name, or --offline-only", file=sys.stderr)
+        return 2
+
+    rows = asyncio.run(run_diagnose(args))
+    if not rows:
+        print("No matching servers.")
+        return 0
+
+    report = _format_report(rows, verbose=not args.quiet)
+    if args.output:
+        Path(args.output).write_text(report, encoding="utf-8")
+        print(f"Wrote {args.output} ({len(rows)} servers)")
+    else:
+        print(report)
+
+    offline = sum(1 for r in rows if r["central_status"] == "offline")
+    ssh_fail = sum(
+        1 for r in rows if r.get("remote") and (r["remote"].get("error") or not r["remote"].get("ssh_ok"))
+    )
+    print(f"Summary: total={len(rows)} central_offline={offline} ssh_fail={ssh_fail}", file=sys.stderr)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/agent_remote_diagnose.sh

@@ -0,0 +1,22 @@
+#!/usr/bin/env bash
+# Batch Agent diagnostics on sub-servers (SSH from central DB creds).
+#
+# Usage:
+#   bash scripts/agent_remote_diagnose.sh --id 95
+#   bash scripts/agent_remote_diagnose.sh --name 颜又文
+#   bash scripts/agent_remote_diagnose.sh --offline-only --limit 20
+#   bash scripts/agent_remote_diagnose.sh --offline-only -o /tmp/agent-offline.md
+#
+# Run on machine with .env (local dev or ssh nexus + cd /opt/nexus).
+set -euo pipefail
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+SECRETS="${ROOT}/deploy/nexus-1panel.secrets.sh"
+if [[ -f "$SECRETS" ]]; then
+  # shellcheck disable=SC1090
+  source "$SECRETS"
+fi
+PYTHON="${ROOT}/.venv/bin/python"
+if [[ ! -x "$PYTHON" ]]; then
+  PYTHON=python3
+fi
+exec "$PYTHON" "${ROOT}/scripts/agent_remote_diagnose.py" "$@"

tests/test_agent_remote_diagnose.py

@@ -0,0 +1,30 @@
+"""Unit tests for agent_remote_diagnose helpers."""
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from scripts.agent_remote_diagnose import _central_status, _has_agent_monitoring
+
+
+class _Srv:
+    def __init__(self, key: str = "", ver: str = "") -> None:
+        self.agent_api_key = key
+        self.agent_version = ver
+
+
+def test_has_agent_monitoring():
+    assert _has_agent_monitoring(_Srv("nxs-x", "")) is True
+    assert _has_agent_monitoring(_Srv("", "2.0.0")) is True
+    assert _has_agent_monitoring(_Srv("", "")) is False
+
+
+def test_central_status():
+    hb = {"is_online": "True", "last_heartbeat": "2026-06-09T01:00:00+00:00"}
+    assert _central_status(True, hb) == "online"
+    hb["is_online"] = "False"
+    assert _central_status(True, hb) == "offline"
+    assert _central_status(True, None) == "offline"
+    assert _central_status(False, None) == "unknown"