UI布局改变

2026-04-23 01:19:02 +08:00 · 2026-04-23 01:19:02 +08:00 · e6d6bc4074
commit e6d6bc4074
parent 3144ec6013
17 changed files with 6777 additions and 1570 deletions
--- a/.env.example
+++ b/.env.example
@ -1,13 +1,27 @@
 APP_BIND=0.0.0.0
 APP_PORT=2026
+AUTH_DB_PATH=/opt/hy-app-monitor/run/auth.sqlite3
+AUTH_SESSION_TTL_HOURS=24
+AUTH_COOKIE_SECURE=0
 DEFAULT_REFRESH_INTERVAL_SECONDS=10
 REFRESH_INTERVAL_OPTIONS=5,10,15,30,60
 PROBE_TIMEOUT_SECONDS=3
 PROBE_CACHE_TTL_SECONDS=5
 PROBE_MAX_WORKERS=12
 HOST_METRIC_CACHE_TTL_SECONDS=25
+HOST_METRIC_TIMEOUT_SECONDS=30
+HOST_METRIC_CPU_SAMPLE_SECONDS=1
 TAT_METRIC_TIMEOUT_SECONDS=30
 TAT_CPU_SAMPLE_SECONDS=1
+SSH_USER=root
+SSH_USE_SUDO=0
+SSH_PORT=22
+SSH_CONNECT_TIMEOUT_SECONDS=10
+SSH_STRICT_HOST_KEY_CHECKING=accept-new
+SSH_PRIVATE_KEY_PATH=/opt/hy-app-monitor/run/ssh/id_ed25519
+SSH_KNOWN_HOSTS_PATH=/opt/hy-app-monitor/run/ssh/known_hosts
+SERVICE_LOG_PREVIEW_LINES=1000
+SERVICE_LOG_EXPORT_MAX_LINES=10000
 ROLLING_RESTART_TIMEOUT_SECONDS=180
 ROLLING_RESTART_POLL_SECONDS=2
 ROLLING_RESTART_STABILIZE_SECONDS=2
@ -29,8 +43,28 @@ HARBOR_PASSWORD=
 BUILD_DOCKER_PLATFORM=linux/amd64
 UPDATE_DEFAULT_JAVA_GIT_REF=main
 UPDATE_DEFAULT_GO_GIT_REF=main
+UPDATE_DEFAULT_ADMIN_GIT_REF=main
 JAVA_REPO_ROOT=/opt/deploy/sources/chatapp3-java
 GOLANG_REPO_ROOT=/opt/deploy/sources/chatapp3-golang
+ADMIN_FRONTEND_ENABLED=0
+ADMIN_FRONTEND_SERVICE_NAME=admin
+ADMIN_FRONTEND_REPO_URL=git@gitea.haiyihy.com:hy/chatapp3-admin.git
+ADMIN_FRONTEND_REPO_ROOT=/opt/deploy/sources/chatapp3-admin
+ADMIN_FRONTEND_HOST=
+ADMIN_FRONTEND_IP=
+ADMIN_FRONTEND_INSTANCE_ID=
+ADMIN_FRONTEND_SSH_HOST=
+ADMIN_FRONTEND_SSH_USER=root
+ADMIN_FRONTEND_SSH_PORT=22
+ADMIN_FRONTEND_SSH_USE_SUDO=0
+ADMIN_FRONTEND_REMOTE_DIR=/opt/yumi-admin/frontend-admin
+ADMIN_FRONTEND_SITE_DIR=/opt/yumi-admin/frontend-admin/site
+ADMIN_FRONTEND_COMPOSE_FILE=/opt/yumi-admin/frontend-admin/docker-compose.yml
+ADMIN_FRONTEND_COMPOSE_SERVICE=caddy
+ADMIN_FRONTEND_RELEASE_META_PATH=/opt/yumi-admin/frontend-admin/.hy-app-monitor-admin-release.json
+ADMIN_FRONTEND_BUILD_SCRIPT=build:antdv-next
+ADMIN_FRONTEND_DIST_DIR=apps/dist
+ADMIN_FRONTEND_PUBLIC_URL=
 MONGO_URI=
 MONGO_DB_NAME=tarab_all
 MONGO_INSTANCE_ID=ins-o9mmob66
@ -46,5 +80,11 @@ PROCESS_BAD_COUNT=260
 SERVICE_LATENCY_WARN_MS=1000
 SERVICE_LATENCY_BAD_MS=3000
 DEPLOY_REGION=me-saudi-arabia
+DEPLOY_INSTANCE_ID=
 TENCENT_SECRET_ID=
 TENCENT_SECRET_KEY=
+DEPLOY_SSH_HOST=
+DEPLOY_SSH_USER=root
+DEPLOY_SSH_PORT=22
+DEPLOY_SSH_IDENTITY_FILE=
+MONITOR_REMOTE_DIR=/opt/hy-app-monitor
--- a/.impeccable.md
+++ b/.impeccable.md
@ -0,0 +1,54 @@
+## Design Context
+
+### Users
+- 主要使用者是 deploy 机上的运维、后端开发和值班人员。
+- 他们会在同一页面里完成巡检、异常定位、配置核对、服务更新和滚动重启。
+- 这个面板的核心目标是 3 秒扫状态、10 秒定位异常、30 秒进入操作。
+
+### Brand Personality
+- 气质是冷静、可靠、克制。
+- 界面要传递“可控、可信、响应快”，不是展示型大屏，也不是炫技 demo。
+- 页面文案必须短，不写解释性口号，不加无意义描述。
+
+### Aesthetic Direction
+- 视觉方向固定为浅色工业控制台。
+- 主题采用浅底、钢蓝主强调、绿橙红只用于状态，不做装饰。
+- 结构采用硬质面板、清晰边线、小圆角、机架式主机条带。
+- 禁止继续使用玻璃拟态、霓虹渐变、厚重阴影、大 Hero、紫蓝科技感 demo 风。
+
+### Design Principles
+- 先扫读，再操作：总览在顶部，异常优先暴露，操作区保持直接。
+- 状态色只表达状态：正常、警告、异常必须一眼可分，但不要泛滥。
+- 同类控件同一语言：按钮、筛选、输入框、列表项统一边线和圆角系统。
+- 面板比卡片重要：减少漂浮卡片感，增强工作台和控制台感。
+- 文案最短化：标题、标签、动作名称直接，不补说明性废话。
+
+### Information Architecture
+- 首页默认视图必须先给“状态总览”，不能先给配置编辑器或发布表单。
+- 首页第一屏核心固定为“服务状态矩阵”，行是服务，列是部署主机/实例。
+- 主机区必须使用紧凑条带，一台主机一行；主机指标不能继续做四个大卡片。
+- `Nacos / Mongo / TAT` 在首页只做基础设施摘要，详细表格放到基础设施视图。
+- `配置` 和 `发布` 必须独立成单独视图，不再默认堆在首页长滚动流里。
+- 筛选优先按业务分组：`全部 / 异常 / 网关 / 应用 / 支付 / 基础设施`。
+
+### Density Rules
+- 首屏要尽量在不滚动的情况下同时看到异常摘要和大部分服务状态。
+- 首页首屏结构固定收敛为：顶部导航、工具条、紧凑摘要条、服务矩阵；不再插入宽而矮利用率低的大提示块。
+- 服务状态单元格只保留状态码或极简状态，不在矩阵里铺长 URL 或返回正文。
+- 最近异常并入可折叠右侧详情面板顶部，不单独占首页整行宽度。
+- 详情信息通过可折叠右侧详情面板或展开行承载，不通过继续增高卡片解决。
+- 异常项排序必须前置，正常项后置，保证值班扫屏优先看到问题。
+- 紧凑模式下隐藏分组、均值、URL、返回详情等次要信息，优先保留状态判断信息。
+- 正常态默认降饱和处理，灰绿或中性灰优先；红橙只服务于告警和异常。
+
+### Layout Rules
+- 服务矩阵按业务分组切成多个子矩阵，减少未部署空白格，不再把所有主机列硬拼成一张大表。
+- 服务矩阵只保留一层分组表达，禁止在行内和表头重复写相同分组信息。
+- 服务矩阵表头和第一列默认 sticky，横向或纵向滚动时不能丢服务名和主机名。
+- 首页主机区只展示业务主机条带，`Nacos / Mongo` 等基础设施主机放到基础设施或主机视图承接。
+- 主机指标必须收成一行短文本，不回退到四个大指标块或一排高饱和胶囊。
+- `总览 / 发布 / 配置 / 基础设施` 不共享同一套工具条；监控筛选只属于总览、服务、主机视图。
+- `发布` 页必须是专用工作台：顶部 sticky 操作栏、单独状态条、构建参数在上、更新目标和滚动重启都用紧凑表格。
+- `配置` 页必须拆成页内 tabs：`Nacos / Go Env / 历史版本`；编辑区使用 sticky 工具条、metadata strip、`源码 / Diff` 切换。
+- `基础设施` 页必须有二级导航：`状态总览 / Nacos 服务 / Mongo 指标 / TAT`；Nacos 服务默认折叠实例，Mongo 次要指标默认折叠。
+- 长文本默认缩写，完整值只通过悬停、复制或展开明细承接，不在主列表直接铺满一整行。
--- a/AGENTS.md
+++ b/AGENTS.md
@ -40,3 +40,22 @@
 - 不要删除页面里的主机指标区块。
 - Python 文件保持高注释密度，优先写清楚每个步骤在做什么。
 - Python 文件顶部的 `import` 区域不写包用途注释。
+
+## 视觉约定
+
+- 前端默认视觉风格固定为“浅色工业控制台”。
+- 设计上下文以项目根目录 `.impeccable.md` 为准，后续前端对话和改动先读取它。
+- 页面保持冷静、可靠、可操作，不回退到玻璃拟态、霓虹渐变、大 Hero 的 demo 风。
+- 页面文案只保留必要标签和动作，不添加无意义描述。
+- 首页默认是状态总览，不把配置编辑和发布表单继续堆在同一条长页面里。
+- 服务视图优先使用状态矩阵，主机视图优先使用紧凑条带，基础设施视图承接 Nacos / Mongo 详情。
+- 首页首屏默认收敛为顶部导航、工具条、紧凑摘要条、服务矩阵，不额外插入大块最近异常横幅。
+- 最近异常进入可折叠详情侧栏顶部，不独立占据首页主宽度。
+- 服务矩阵按业务分组切成子矩阵，异常排序前置，表头和第一列保持 sticky。
+- 首页主机区只展示业务主机；基础设施主机放到基础设施或主机视图承接。
+- 主机指标使用单行短文本，正常态降饱和，红橙色保留给异常和告警。
+- 监控筛选条只属于 `总览 / 服务 / 主机`；`发布 / 配置 / 基础设施` 必须使用各自专用工具条和页内导航。
+- `发布` 页按“构建参数 -> 更新目标 -> 执行日志 -> 滚动重启”组织，更新与重启目标都优先用紧凑表格，不再回退到大卡片流。
+- `配置` 页按 `Nacos / Go Env / 历史版本` tabs 组织，编辑器必须提供 sticky 工具条、metadata strip 和 `源码 / Diff` 切换。
+- `基础设施` 页按 `状态总览 / Nacos 服务 / Mongo 指标 / TAT` 二级导航组织，Nacos 服务默认折叠实例，Mongo 次要指标默认折叠。
+- 长文本默认缩写并提供复制或展开，不直接在主列表铺完整 digest、路径、长 ID 或长连接串。
--- a/README.md
+++ b/README.md
@ -4,10 +4,13 @@

 - 页面端口：`2026`
 - 页面入口：`http://43.164.75.199:2026/`
+- 登录入口：`http://43.164.75.199:2026/login`
 - 服务健康：直接探测各业务服务健康接口
- 主机资源：通过 `TAT` 批量采集 `CPU / 内存 / 磁盘 / 进程数`
+- 主机资源：通过 `SSH` 采集 `CPU / 内存 / 磁盘 / 进程数`
 - Nacos：直连 Nacos OpenAPI，列出命名空间、服务和注册实例
 - MongoDB：直连 Mongo，采集 `serverStatus / listDatabases / replSetGetStatus`
+- 宿主机运维：日志、滚动重启、compose 同步、服务更新统一走 `SSH`
+- 带外引导：只在打通 `deploy -> target:22` 时使用 `TAT`

 ## 目录

@ -17,7 +20,7 @@
 - `static/`：前端页面
 - `scripts/`：启停、自启安装、前台运行脚本
 - `systemd/`：systemd unit
- `ops/`：TAT 放行和远端部署脚本
+- `ops/`：SSH 引导、放行和部署脚本
 - `.env.example`：运行配置模板

 ## 运行配置
@ -29,14 +32,20 @@
 ```bash
 APP_BIND=0.0.0.0
 APP_PORT=2026
+AUTH_DB_PATH=/opt/hy-app-monitor/run/auth.sqlite3
+AUTH_SESSION_TTL_HOURS=24
 DEFAULT_REFRESH_INTERVAL_SECONDS=10
 REFRESH_INTERVAL_OPTIONS=5,10,15,30,60
 NACOS_BASE_URL=http://10.2.1.17:8848
 NACOS_NAMESPACE_ID=...
 MONGO_URI=mongodb://...
 DEPLOY_REGION=me-saudi-arabia
+DEPLOY_INSTANCE_ID=
 TENCENT_SECRET_ID=***
 TENCENT_SECRET_KEY=***
+SSH_USER=root
+SSH_PORT=22
+SSH_PRIVATE_KEY_PATH=/opt/hy-app-monitor/run/ssh/id_ed25519
 ```

 如果你要直接在页面里做“构建镜像 -> 推 Harbor -> 更新服务”，还需要补这几项：
@ -53,6 +62,33 @@ JAVA_MAVEN_THREADS=1C
 JAVA_IMAGE_BUILD_MAX_WORKERS=3
 UPDATE_DEFAULT_JAVA_GIT_REF=main
 UPDATE_DEFAULT_GO_GIT_REF=main
+UPDATE_DEFAULT_ADMIN_GIT_REF=main
+DEPLOY_SSH_HOST=10.2.x.x
+DEPLOY_SSH_USER=root
+DEPLOY_SSH_PORT=22
+DEPLOY_SSH_IDENTITY_FILE=~/.ssh/xxx
+```
+
+如果你还要直接发布 admin 前端静态站点，再补这几项：
+
+```bash
+ADMIN_FRONTEND_ENABLED=1
+ADMIN_FRONTEND_REPO_URL=git@gitea.haiyihy.com:hy/chatapp3-admin.git
+ADMIN_FRONTEND_REPO_ROOT=/opt/deploy/sources/chatapp3-admin
+ADMIN_FRONTEND_HOST=admin-platform
+ADMIN_FRONTEND_IP=10.2.1.2
+ADMIN_FRONTEND_INSTANCE_ID=ins-p11begwe
+ADMIN_FRONTEND_SSH_HOST=10.2.1.2
+ADMIN_FRONTEND_SSH_USER=root
+ADMIN_FRONTEND_SSH_PORT=22
+ADMIN_FRONTEND_REMOTE_DIR=/opt/yumi-admin/frontend-admin
+ADMIN_FRONTEND_SITE_DIR=/opt/yumi-admin/frontend-admin/site
+ADMIN_FRONTEND_COMPOSE_FILE=/opt/yumi-admin/frontend-admin/docker-compose.yml
+ADMIN_FRONTEND_COMPOSE_SERVICE=caddy
+ADMIN_FRONTEND_RELEASE_META_PATH=/opt/yumi-admin/frontend-admin/.hy-app-monitor-admin-release.json
+ADMIN_FRONTEND_BUILD_SCRIPT=build:antdv-next
+ADMIN_FRONTEND_DIST_DIR=apps/dist
+ADMIN_FRONTEND_PUBLIC_URL=https://yumi-admin.haiyihy.com
 ```

 可以直接复制模板：
@ -61,6 +97,12 @@ UPDATE_DEFAULT_GO_GIT_REF=main
 cp .env.example .env
 ```

+如需放到 HTTPS 反代后再启用 `Secure Cookie`，可在 `.env` 追加：
+
+```bash
+AUTH_COOKIE_SECURE=1
+```
+
 ## 本地运行

 ```bash
@ -74,6 +116,12 @@ bash scripts/status.sh
 http://127.0.0.1:2026/
 ```

+未登录会自动跳到：
+
+```bash
+http://127.0.0.1:2026/login
+```
+
 ## systemd 开机自启

 在远端项目目录执行：
@ -90,33 +138,44 @@ bash scripts/status.sh
 bash scripts/uninstall_systemd.sh
 ```

-## 通过 TAT 放行端口
+## 通过 TAT 引导 SSH
+
+```bash
+python3 ops/bootstrap_ssh_via_tat.py
+```
+
+默认会：
+
+1. 在 `deploy` 机生成专用 SSH 密钥对
+2. 通过 `TAT` 给目标机写入公钥并准备 `sshd`
+3. 只允许 `deploy -> target:22`
+4. 把目标机本机防火墙和安全组都限制到 `deploy` 私网 IP
+
+如果 `.env` 里开启了 `ADMIN_FRONTEND_ENABLED=1`，脚本也会把 admin 前端机器一起纳入这轮 SSH 引导，即使它不在 `config/targets.json` 里。
+
+如果你只需要放开监控面板端口，仍然可以使用：

 ```bash
 python3 ops/open_port_via_tat.py
 ```

-默认会：
-
-1. 解析 `deploy` 机器安全组
-2. 通过 `TAT` 放开机器内防火墙
-3. 通过腾讯云 API 放开安全组 TCP `2026`
-
-## 通过 TAT 远端部署
+## 通过 SSH 远端部署

 ```bash
-python3 ops/deploy_via_tat.py
+python3 ops/deploy_via_ssh.py
 ```

 默认会：

-1. 拉取或更新 `/opt/hy-app-monitor`
+1. 把当前工作区同步到远端 `/opt/hy-app-monitor`
 2. 把本地项目 `.env` 同步到远端项目 `.env`
 3. 创建远端 `.venv`
 4. 安装 `requirements-ops.txt`
 5. 安装并启用 `systemd`
 6. 重启服务并检查 `2026` 监听

+`ops/deploy_via_tat.py` 仍可保留成带外兜底，不建议继续作为日常主链路。
+
 ## 维护完整 Compose

 默认只刷新本地完整模板：
@ -134,6 +193,15 @@ python3 ops/sync_runtime_compose.py --sync-remote
 模板文件会写到本地 `run/compose_templates/`。  
 滚动重启在真正执行 `docker compose up -d --force-recreate --no-deps <service>` 之前，也会先把对应宿主机的完整 `docker-compose.yml` 同步到远端，避免线上 compose 被裁剪成子集。

+## 服务日志
+
+页面里的服务详情支持按实例查看容器日志：
+
+1. 预览固定最近 `1000` 条
+2. 导出支持自定义条数
+3. 导出条数会按 `SERVICE_LOG_EXPORT_MAX_LINES` 限制
+4. 日志命令走目标机 `docker logs --tail N`
+
 ## 服务更新

 页面里的“服务更新”会直接在当前机器上执行这条链路：
@ -142,15 +210,26 @@ python3 ops/sync_runtime_compose.py --sync-remote
 2. 本地构建镜像并推送到 Harbor
 3. 取回 Harbor `digest`
 4. 用新 `digest` 更新本地完整 compose 模板
-5. 把完整 compose 下发到目标宿主机
-6. 先在目标机 `docker pull` 新镜像
+5. 通过 `SSH` 把完整 compose 下发到目标宿主机
+6. 通过 `SSH` 先在目标机 `docker pull` 新镜像
 7. 最后只对你选中的服务执行 `docker compose up -d --force-recreate --no-deps <service>`

+如果选中 `admin` 前端目标，则会改走：
+
+1. 在 deploy 机拉取 / 更新 `chatapp3-admin`
+2. 执行 `corepack pnpm install --frozen-lockfile`
+3. 执行 `pnpm run build:antdv-next`
+4. 通过 `SSH + SCP` 把 `apps/dist` 同步到 admin 机 `site` 目录
+5. 备份旧站点文件并原地替换内容
+6. 执行 `docker compose up -d caddy`
+7. 把本次发布的 `gitRef / commit / 时间` 写到远端元数据文件，页面后续可直接看到当前已部署版本
+
 说明：

 - `other` 现在也按 Harbor 业务镜像更新，不再继续依赖本地 `./other/service.jar` 挂载
 - Java Maven 默认走增量 `package`，并支持通过 `JAVA_MAVEN_THREADS` 开启 reactor 并行
 - 多个 Java 服务会按 `JAVA_IMAGE_BUILD_MAX_WORKERS` 受限并行执行 `docker build + push`
+- `admin` 前端不会走 Harbor 镜像链路，而是直接发布静态文件到 `site`
 - 页面里的“滚动重启”仍然保留，适合只想让配置生效、不涉及代码构建的场景
 - 这条链路默认假设 `hy-app-monitor` 就运行在打包机 / deploy 机上，并且本机已经能访问业务仓、Docker 和 Harbor

@ -161,6 +240,7 @@ python3 ops/sync_runtime_compose.py --sync-remote
 - Nacos：命名空间、服务、注册实例、健康状态
 - Nacos 配置：读取当前 namespace 下的配置列表，支持格式校验、在线编辑保存
 - Golang 配置：读取 app 节点当前运行中的 `service.env`，展示 `chatapp3-golang` 实际消费的环境变量并支持保存
+- 服务日志：按实例查看最近日志并导出
 - 服务更新：本地构建选中服务、推送 Harbor、更新完整 compose，并只重建选中的服务
 - 滚动重启：可按服务选择 Java / Golang 节点，逐台重启并等待健康恢复
 - MongoDB：连接、内存、网络、操作计数、复制集、数据库统计
--- a/monitor/compose_templates.py
+++ b/monitor/compose_templates.py
@ -12,7 +12,7 @@ from typing import Any

 from .config import ROOT, RUNTIME_BASE_DIR, load_hosts, utc_now
 from .runtime_config import read_remote_text, write_remote_text
-from .tencent_tat import run_shell_command
+from .ssh_remote import copy_remote_file, run_host_script_checked


 # 本地完整 compose 模板统一放到 run 目录，避免把运行期产物写进 Git。
@ -49,7 +49,7 @@ def container_name(host_name: str, service_name: str) -> str:


 def docker_inspect_payload(host: dict[str, Any], service_names: list[str]) -> list[dict[str, Any]]:
-    # 这里只抓 compose 需要的关键字段，避免完整 inspect JSON 被 TAT 输出截断。
+    # 这里只抓 compose 需要的关键字段，避免远端 inspect 输出塞进一大坨无关 JSON。
    container_names = [container_name(host["host"], service_name) for service_name in service_names]
    if not container_names:
        raise RuntimeError(f"no compose services configured for host: {host['host']}")
@ -89,15 +89,12 @@ for item in payload:
    script = f"""set -eu
 docker inspect {quoted_names} | python3 -c {shlex.quote(python_code)}
 """
-    result_map = run_shell_command(
-        [host["instanceId"]],
+    result = run_host_script_checked(
+        host,
        script,
        command_name=f"inspect-compose-{host['host']}",
        timeout_seconds=90,
    )
-    result = result_map[host["instanceId"]]
-    if result.get("status") != "SUCCESS":
-        raise RuntimeError(f"docker inspect failed on {host['host']}: {result.get('status')}")

    payload: list[dict[str, Any]] = []
    for raw_line in str(result.get("output") or "").splitlines():
@ -466,15 +463,12 @@ def verify_remote_compose(host: dict[str, Any], remote_path: str) -> list[str]:
 FILE={shlex.quote(remote_path)}
 docker compose -f "$FILE" config --services
 """
-    result_map = run_shell_command(
-        [host["instanceId"]],
+    result = run_host_script_checked(
+        host,
        script,
        command_name=f"verify-compose-{host['host']}",
        timeout_seconds=60,
    )
-    result = result_map[host["instanceId"]]
-    if result.get("status") != "SUCCESS":
-        raise RuntimeError(f"compose verify failed on {host['host']}: {result.get('status')}")
    return [line.strip() for line in str(result.get("output") or "").splitlines() if line.strip()]


@ -506,7 +500,7 @@ def sync_remote_host_compose_template(host_name: str, *, refresh_from_remote: bo
    backup_path = ""
    if remote_current and remote_current != template_payload["content"]:
        backup_path = f"{remote_path}.hyapp.{utc_now().replace(':', '').replace('+', '_')}.bak"
-        write_remote_text(host, backup_path, remote_current)
+        copy_remote_file(host, remote_path, backup_path)

    changed = remote_current != template_payload["content"]
    if changed:
--- a/monitor/host_metrics.py
+++ b/monitor/host_metrics.py
@ -1,14 +1,16 @@
 from __future__ import annotations

+# SSH 主机指标改成并发采集，避免逐台串行拖慢首页。
+import concurrent.futures
 import json
 from typing import Any

 from .cache import TTLCache
-from .config import HOST_METRIC_CACHE_TTL_SECONDS, TAT_CPU_SAMPLE_SECONDS, TAT_METRIC_TIMEOUT_SECONDS, utc_now
-from .tencent_tat import run_shell_command
+from .config import HOST_METRIC_CACHE_TTL_SECONDS, HOST_METRIC_CPU_SAMPLE_SECONDS, HOST_METRIC_TIMEOUT_SECONDS, utc_now
+from .ssh_remote import run_host_script


-# 主机指标单独缓存，避免页面频繁刷新时不断调用 TAT。
+# 主机指标单独缓存，避免页面频繁刷新时不断压业务主机。
 HOST_METRIC_CACHE = TTLCache[dict[str, dict[str, Any]]](HOST_METRIC_CACHE_TTL_SECONDS)


@ -33,7 +35,7 @@ def read_cpu():
 # 第一次采样。
 idle_1, total_1 = read_cpu()
 # 等待一个很短的采样窗口。
-time.sleep({TAT_CPU_SAMPLE_SECONDS})
+time.sleep({HOST_METRIC_CPU_SAMPLE_SECONDS})
 # 第二次采样。
 idle_2, total_2 = read_cpu()

@ -91,57 +93,63 @@ def metrics_error_payload(error: str, instance_id: str = "") -> dict[str, Any]:
    }


-def collect_host_metrics(hosts: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
-    # 建立实例 ID 到 host 的映射。
-    host_by_instance = {host["instanceId"]: host for host in hosts if host.get("instanceId")}
-    # 去重后的实例 ID 列表。
-    instance_ids = list(host_by_instance.keys())
-
-    # 没有实例 ID 时直接返回错误。
-    if not instance_ids:
-        return {
-            host["host"]: metrics_error_payload("missing instanceId in config/targets.json")
-            for host in hosts
-        }
-
-    # 通过 TAT 批量执行主机指标采集脚本。
-    task_results = run_shell_command(
-        instance_ids,
+def collect_single_host_metrics(host: dict[str, Any]) -> tuple[str, dict[str, Any]]:
+    # SSH 远端执行失败时也按统一结构回传，避免单机问题拖垮整个汇总接口。
+    instance_id = str(host.get("instanceId") or "").strip()
+    result = run_host_script(
+        host,
        build_metrics_script(),
-        command_name="hy-app-monitor-host-metrics",
-        timeout_seconds=TAT_METRIC_TIMEOUT_SECONDS,
+        command_name=f"host-metrics-{host['host']}",
+        timeout_seconds=HOST_METRIC_TIMEOUT_SECONDS,
    )
+    status = str(result.get("status") or "").strip()
+    output = str(result.get("output") or "")

-    # 这里按 host name 回填最终结果。
+    if status == "SUCCESS":
+        try:
+            payload = json.loads(output or "{}")
+            payload["updatedAt"] = utc_now()
+            payload["instanceId"] = instance_id
+            return host["host"], payload
+        except Exception as exc:  # noqa: BLE001
+            return host["host"], metrics_error_payload(f"metrics parse failed: {exc}", instance_id)
+
+    return host["host"], metrics_error_payload(output.strip() or status or "metrics failed", instance_id)
+
+
+def collect_host_metrics(hosts: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
+    # 没有主机时直接返回空结构。
+    if not hosts:
+        return {}
+
+    # 主机没有基本 SSH 信息时先记错误，剩余主机继续采集。
    payload_by_host: dict[str, dict[str, Any]] = {}
-
-    # 逐台主机解析结果。
+    available_hosts: list[dict[str, Any]] = []
    for host in hosts:
-        # 读取实例 ID。
-        instance_id = host.get("instanceId", "")
-        # 读取该实例的远端执行结果。
-        task = task_results.get(instance_id, {})
-        # 读取执行状态。
-        status = task.get("status", "")
-        # 读取远端输出。
-        output = task.get("output", "")
+        if str(host.get("ip") or host.get("sshHost") or "").strip():
+            available_hosts.append(host)
+            continue
+        payload_by_host[host["host"]] = metrics_error_payload(
+            "missing ip/sshHost in config/targets.json",
+            host.get("instanceId", ""),
+        )

-        # 成功时尝试解析 JSON。
-        if status == "SUCCESS":
+    # SSH 无法像 TAT 那样一次多机下发，这里按主机并发执行。
+    workers = max(1, min(len(available_hosts) or 1, 8))
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
+        future_map = {executor.submit(collect_single_host_metrics, host): host for host in available_hosts}
+        for future in concurrent.futures.as_completed(future_map):
+            host = future_map[future]
            try:
-                # 把 JSON 输出反序列化成 Python 结构。
-                payload = json.loads(output or "{}")
-                # 补齐统一字段。
-                payload["updatedAt"] = utc_now()
-                payload["instanceId"] = instance_id
-                payload_by_host[host["host"]] = payload
-                continue
+                host_name, payload = future.result()
            except Exception as exc:  # noqa: BLE001
-                payload_by_host[host["host"]] = metrics_error_payload(f"metrics parse failed: {exc}", instance_id)
+                payload_by_host[host["host"]] = metrics_error_payload(
+                    f"metrics collection failed: {exc}",
+                    host.get("instanceId", ""),
+                )
                continue
-
-        # 失败时保留错误文本或状态码。
-        payload_by_host[host["host"]] = metrics_error_payload(output.strip() or status or "metrics failed", instance_id)
+            payload_by_host[host_name] = payload

    # 返回按 host 索引的主机指标。
    return payload_by_host
@ -160,5 +168,5 @@ def get_host_metrics(hosts: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
                for host in hosts
            }

-    # 通过缓存保护 TAT 调用。
+    # 通过缓存保护 SSH 调用。
    return HOST_METRIC_CACHE.get_or_build(builder)
--- a/monitor/http_app.py
+++ b/monitor/http_app.py
@ -572,6 +572,7 @@ class MonitorHandler(BaseHTTPRequestHandler):

        java_git_ref = str(payload.get("javaGitRef") or "").strip()
        go_git_ref = str(payload.get("goGitRef") or "").strip()
+        admin_git_ref = str(payload.get("adminGitRef") or "").strip()
        image_tag = str(payload.get("imageTag") or "").strip()
        skip_maven_build = bool(payload.get("skipMavenBuild"))

@ -580,6 +581,7 @@ class MonitorHandler(BaseHTTPRequestHandler):
                services,
                java_git_ref=java_git_ref,
                go_git_ref=go_git_ref,
+                admin_git_ref=admin_git_ref,
                image_tag=image_tag,
                skip_maven_build=skip_maven_build,
            )
--- a/monitor/runtime_config.py
+++ b/monitor/runtime_config.py
@ -1,17 +1,14 @@
 from __future__ import annotations

-# 远端写文件时需要 base64 传输，避免引号和换行转义问题。
-import base64
 # 返回给前端的摘要信息会带 sha256。
 import hashlib
 # env 值里有 JSON 字符串时直接复用标准库解析。
 import json
-# 远端路径和 shell 参数需要安全转义。
-import shlex
 from typing import Any

-from .config import RUNTIME_BASE_DIR, TAT_METRIC_TIMEOUT_SECONDS, load_hosts, utc_now
-from .tencent_tat import run_shell_command
+from .config import RUNTIME_BASE_DIR, load_hosts, utc_now
+from .ssh_remote import read_remote_text as ssh_read_remote_text
+from .ssh_remote import write_remote_text as ssh_write_remote_text


 # Go 配置按源码里的运行分段排序，页面上也按这个顺序展示。
@ -304,12 +301,6 @@ GO_ENV_PRIMARY_BY_KEY = {
 # 同时保留按主键索引的 spec，后续查类型更直接。
 GO_ENV_SPECS_BY_PRIMARY = {spec["primary"]: spec for spec in GO_ENV_SPECS}

-# 远端文件不存在时，用固定标记避免和 shell 错误混淆。
-REMOTE_FILE_MISSING_MARKER = "__HY_APP_MONITOR_FILE_MISSING__"
-# TAT 单次输出有长度上限，读取较大的 env / compose 文件时需要分块。
-REMOTE_TEXT_CHUNK_BYTES = 4096
-
-
 def go_service_hosts() -> list[dict[str, Any]]:
    # 只挑出当前监控目标里真正承载 golang 服务的主机。
    hosts = [
@ -379,90 +370,14 @@ def parse_env_text(raw: str) -> dict[str, str]:
    return values


-def remote_host_result(host: dict[str, Any], script: str, command_name: str, *, timeout_seconds: int = TAT_METRIC_TIMEOUT_SECONDS) -> dict[str, Any]:
-    # 统一执行单机 TAT 命令，并把错误折叠成 RuntimeError。
-    result_map = run_shell_command(
-        [host["instanceId"]],
-        script,
-        command_name=command_name,
-        timeout_seconds=timeout_seconds,
-    )
-    result = result_map[host["instanceId"]]
-    if result.get("status") != "SUCCESS":
-        raise RuntimeError(f"{host['host']} {command_name} failed: {result.get('status')}")
-    return result
-
-
-def remote_file_size(host: dict[str, Any], path: str) -> int:
-    quoted_path = shlex.quote(path)
-    script = f"""set -eu
-FILE={quoted_path}
-if [ ! -f "$FILE" ]; then
-  printf '%s' {shlex.quote(REMOTE_FILE_MISSING_MARKER)}
-  exit 0
-fi
-python3 - "$FILE" <<'PY'
-from pathlib import Path
-import sys
-
-print(Path(sys.argv[1]).stat().st_size, end="")
-PY
-"""
-    result = remote_host_result(host, script, f"stat-{host['host']}-remote-file")
-    output = str(result.get("output") or "")
-    if output.strip() == REMOTE_FILE_MISSING_MARKER:
-        raise RuntimeError(f"{host['host']} missing file: {path}")
-    return int(output.strip() or "0")
-
-
 def read_remote_text(host: dict[str, Any], path: str) -> str:
-    quoted_path = shlex.quote(path)
-    file_size = remote_file_size(host, path)
-    if file_size == 0:
-        return ""
-
-    # 通过多次小输出读取完整文件，避免 TAT 把大文本截断。
-    chunks: list[bytes] = []
-    for offset in range(0, file_size, REMOTE_TEXT_CHUNK_BYTES):
-        script = f"""set -eu
-FILE={quoted_path}
-if [ ! -f "$FILE" ]; then
-  printf '%s' {shlex.quote(REMOTE_FILE_MISSING_MARKER)}
-  exit 0
-fi
-python3 - "$FILE" {offset} {REMOTE_TEXT_CHUNK_BYTES} <<'PY'
-from pathlib import Path
-import base64
-import sys
-
-path = Path(sys.argv[1])
-offset = int(sys.argv[2])
-chunk_size = int(sys.argv[3])
-data = path.read_bytes()[offset : offset + chunk_size]
-print(base64.b64encode(data).decode("ascii"), end="")
-PY
-"""
-        result = remote_host_result(host, script, f"read-{host['host']}-remote-file-chunk-{offset}")
-        output = str(result.get("output") or "")
-        if output.strip() == REMOTE_FILE_MISSING_MARKER:
-            raise RuntimeError(f"{host['host']} missing file: {path}")
-        chunks.append(base64.b64decode(output.strip() or ""))
-
-    return b"".join(chunks).decode("utf-8")
+    # 远端 env / compose 文本现在统一走 SSH 直连读取。
+    return ssh_read_remote_text(host, path)


 def write_remote_text(host: dict[str, Any], path: str, content: str) -> None:
-    # 把文本整体 base64 编码后再下发，避免 shell here-doc 被内容污染。
-    encoded = base64.b64encode(content.encode("utf-8")).decode("ascii")
-    quoted_path = shlex.quote(path)
-    script = f"""set -eu
-FILE={quoted_path}
-TMP="${{FILE}}.hyapp.$$"
-mkdir -p "$(dirname "$FILE")"
-printf '%s' {shlex.quote(encoded)} | base64 -d > "$TMP"
-mv "$TMP" "$FILE"
-"""
-    remote_host_result(host, script, f"write-{host['host']}-service-env")
+    # 远端文本覆盖同样走 SSH，避免继续依赖 TAT 输出长度和轮询结果。
+    ssh_write_remote_text(host, path, content)


 def host_sha256(content: str) -> str:
--- a/monitor/service_ops.py
+++ b/monitor/service_ops.py
@ -27,7 +27,7 @@ from .nacos import (
    list_instances,
    nacos_truthy,
 )
-from .tencent_tat import run_shell_command
+from .ssh_remote import run_host_script


 def restartable_service_records() -> list[dict[str, Any]]:
@ -74,14 +74,13 @@ def service_records_by_name() -> dict[str, list[dict[str, Any]]]:


 def remote_command_result(service: dict[str, Any], script: str, command_name: str, timeout_seconds: int) -> dict[str, Any]:
-    # 单实例远端执行统一复用 TAT shell。
-    result_map = run_shell_command(
-        [service["instanceId"]],
+    # 单实例远端执行现在统一复用 SSH shell。
+    return run_host_script(
+        service,
        script,
        command_name=command_name,
        timeout_seconds=timeout_seconds,
    )
-    return result_map[service["instanceId"]]


 def restart_service_instance(service: dict[str, Any]) -> dict[str, Any]:
@ -225,8 +224,8 @@ def rolling_restart_payload(service_names: list[str]) -> dict[str, Any]:
                }

                restart_result = restart_service_instance(service)
-                step["tatStatus"] = restart_result.get("status")
-                step["tatOutput"] = str(restart_result.get("output") or "").strip()[:500]
+                step["remoteStatus"] = restart_result.get("status")
+                step["remoteOutput"] = str(restart_result.get("output") or "").strip()[:500]

                http_result = wait_http_ready(service, ROLLING_RESTART_TIMEOUT_SECONDS)
                step["http"] = {
--- a/monitor/service_release.py
+++ b/monitor/service_release.py
@ -4,6 +4,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
 # 镜像 tag 需要拼时间戳。
 from datetime import datetime
 # 本地构建和 git worktree 都依赖标准库 subprocess / tempfile。
+import json
 import os
 from pathlib import Path
 import re
@ -21,6 +22,25 @@ from .compose_templates import (
    update_local_service_image,
 )
 from .config import (
+    ADMIN_FRONTEND_BUILD_SCRIPT,
+    ADMIN_FRONTEND_COMPOSE_FILE,
+    ADMIN_FRONTEND_COMPOSE_SERVICE,
+    ADMIN_FRONTEND_DIST_DIR,
+    ADMIN_FRONTEND_ENABLED,
+    ADMIN_FRONTEND_HOST,
+    ADMIN_FRONTEND_INSTANCE_ID,
+    ADMIN_FRONTEND_IP,
+    ADMIN_FRONTEND_PUBLIC_URL,
+    ADMIN_FRONTEND_RELEASE_META_PATH,
+    ADMIN_FRONTEND_REMOTE_DIR,
+    ADMIN_FRONTEND_REPO_ROOT,
+    ADMIN_FRONTEND_REPO_URL,
+    ADMIN_FRONTEND_SERVICE_NAME,
+    ADMIN_FRONTEND_SITE_DIR,
+    ADMIN_FRONTEND_SSH_HOST,
+    ADMIN_FRONTEND_SSH_PORT,
+    ADMIN_FRONTEND_SSH_USE_SUDO,
+    ADMIN_FRONTEND_SSH_USER,
    BUILD_DOCKER_PLATFORM,
    GOLANG_REPO_ROOT,
    HARBOR_PASSWORD,
@ -36,6 +56,7 @@ from .config import (
    SERVICE_BUILD_CACHE_ROOT,
    SERVICE_BUILD_TIMEOUT_SECONDS,
    SERVICE_PULL_TIMEOUT_SECONDS,
+    UPDATE_DEFAULT_ADMIN_GIT_REF,
    UPDATE_DEFAULT_GO_GIT_REF,
    UPDATE_DEFAULT_JAVA_GIT_REF,
    utc_now,
@ -47,13 +68,15 @@ from .service_ops import (
    wait_http_ready,
    wait_java_registration_ready,
 )
-from .tencent_tat import run_shell_command
+from .ssh_remote import copy_local_path_to_host, read_remote_text, run_host_script_checked, target_display, write_remote_text


 # Java 仓里当前支持 Harbor 更新的服务集合。
 JAVA_BUILDABLE_SERVICES = ("auth", "gateway", "external", "console", "other", "live", "wallet", "order")
 # Go 仓目前只有主业务服务。
 GOLANG_BUILDABLE_SERVICES = ("golang",)
+# Admin 前端当前只接入一个发布目标。
+FRONTEND_BUILDABLE_SERVICES = (ADMIN_FRONTEND_SERVICE_NAME,)
 # Java 批量 Maven 构建时，需要映射回对应的启动模块。
 JAVA_SERVICE_MODULES = {
    "auth": "rc-auth",
@ -82,6 +105,8 @@ SERVICE_IMAGE_REPOSITORY_SUFFIXES = {
 ProgressLogger = Callable[[str, str, str], None]
 # 私有 Maven 安装缓存统一放到 run 目录，避免混进 Git 工作区。
 PRIVATE_MAVEN_CACHE_DIR = ROOT / "run" / "maven_private_cache"
+# Admin 前端站点备份最多保留最近若干份。
+FRONTEND_SITE_BACKUP_KEEP = 3


 def emit_progress(progress: ProgressLogger | None, stage: str, message: str, *, level: str = "info") -> None:
@ -137,7 +162,10 @@ def stream_log_file_lines(

 def buildable_service_names() -> list[str]:
    # 页面和接口统一按这里判定哪些服务允许“构建并更新”。
-    return [*JAVA_BUILDABLE_SERVICES, *GOLANG_BUILDABLE_SERVICES]
+    names = [*JAVA_BUILDABLE_SERVICES, *GOLANG_BUILDABLE_SERVICES]
+    if frontend_release_enabled():
+        names.extend(service_name for service_name in FRONTEND_BUILDABLE_SERVICES if str(service_name or "").strip())
+    return names


 def normalize_service_names(service_names: list[str]) -> list[str]:
@ -147,11 +175,13 @@ def normalize_service_names(service_names: list[str]) -> list[str]:


 def service_kind(service_name: str) -> str:
-    # 更新链路目前只分 Java / Golang 两类仓库。
+    # 更新链路目前分 Java / Golang / Frontend 三类仓库。
    if service_name in JAVA_BUILDABLE_SERVICES:
        return "java"
    if service_name in GOLANG_BUILDABLE_SERVICES:
        return "golang"
+    if frontend_release_enabled() and service_name in FRONTEND_BUILDABLE_SERVICES:
+        return "frontend"
    raise RuntimeError(f"unsupported update service: {service_name}")


@ -165,6 +195,27 @@ def ensure_repo_root(repo_root: Path, label: str) -> Path:
    return resolved


+def ensure_frontend_repo_root(*, progress: ProgressLogger | None = None) -> Path:
+    # Admin 前端允许第一次使用时按配置的 repo URL 自动 clone 到 deploy 机。
+    repo_root = Path(ADMIN_FRONTEND_REPO_ROOT).expanduser().resolve()
+    if (repo_root / ".git").exists():
+        return repo_root
+
+    repo_url = str(ADMIN_FRONTEND_REPO_URL or "").strip()
+    if not repo_url:
+        raise RuntimeError("missing ADMIN_FRONTEND_REPO_URL in .env")
+
+    repo_root.parent.mkdir(parents=True, exist_ok=True)
+    emit_progress(progress, "build.frontend.repo", f"初始化前端仓库：{repo_url} -> {repo_root}")
+    local_command_output(
+        ["git", "clone", repo_url, str(repo_root)],
+        cwd=repo_root.parent,
+        label="frontend clone repo",
+        timeout_seconds=900,
+    )
+    return ensure_repo_root(repo_root, "frontend")
+
+
 def local_command_output(
    command: list[str],
    *,
@ -207,6 +258,34 @@ def tail_text(path: Path, max_bytes: int = 8192) -> str:
        return handle.read().decode("utf-8", errors="replace").strip()


+def frontend_release_enabled() -> bool:
+    # Admin 前端需要仓库、主机和发布目录都齐备后才开放到页面。
+    return bool(
+        ADMIN_FRONTEND_ENABLED
+        and str(ADMIN_FRONTEND_SERVICE_NAME or "").strip()
+        and str(ADMIN_FRONTEND_HOST or "").strip()
+        and str(ADMIN_FRONTEND_SITE_DIR or "").strip()
+        and (str(ADMIN_FRONTEND_SSH_HOST or "").strip() or str(ADMIN_FRONTEND_IP or "").strip())
+        and (str(ADMIN_FRONTEND_REPO_URL or "").strip() or str(ADMIN_FRONTEND_REPO_ROOT or "").strip())
+    )
+
+
+def frontend_host_record(service_name: str) -> dict[str, Any]:
+    # 当前只有 admin 前端走这条链路，后续再扩展时继续在这里分流。
+    if not frontend_release_enabled() or service_name != ADMIN_FRONTEND_SERVICE_NAME:
+        raise RuntimeError(f"unsupported frontend service: {service_name}")
+
+    return {
+        "host": ADMIN_FRONTEND_HOST,
+        "ip": ADMIN_FRONTEND_IP,
+        "instanceId": ADMIN_FRONTEND_INSTANCE_ID,
+        "sshHost": ADMIN_FRONTEND_SSH_HOST or ADMIN_FRONTEND_IP,
+        "sshUser": ADMIN_FRONTEND_SSH_USER,
+        "sshUseSudo": ADMIN_FRONTEND_SSH_USE_SUDO,
+        "sshPort": ADMIN_FRONTEND_SSH_PORT,
+    }
+
+
 def run_logged_command(
    command: list[str],
    *,
@ -527,6 +606,105 @@ def derive_image_tag(git_ref: str, commit: str, image_tag: str) -> str:
    return f"{sanitized_ref_fragment(git_ref)}-{commit[:12]}-{timestamp}"


+def frontend_version_label(git_ref: str, commit: str) -> str:
+    # 前端发布结果统一显示成 ref@commit12，便于和镜像 tag 区分。
+    resolved_ref = str(git_ref or "").strip() or UPDATE_DEFAULT_ADMIN_GIT_REF
+    return f"{resolved_ref}@{commit[:12]}"
+
+
+def frontend_release_metadata(service_name: str) -> dict[str, Any]:
+    # 当前前端站点的已部署版本由远端元数据文件提供；缺失时允许返回空结构。
+    host = frontend_host_record(service_name)
+    try:
+        raw = read_remote_text(host, ADMIN_FRONTEND_RELEASE_META_PATH, timeout_seconds=30).strip()
+    except Exception as exc:  # noqa: BLE001
+        return {"error": str(exc)}
+
+    if not raw:
+        return {}
+
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError:
+        return {"error": f"invalid frontend release metadata on {target_display(host)}"}
+
+    if not isinstance(payload, dict):
+        return {"error": f"invalid frontend release metadata on {target_display(host)}"}
+    return payload
+
+
+def image_release_display(current_images: list[str]) -> dict[str, str]:
+    # 镜像型服务在表格里固定展示 tag / digest / 可复制的主镜像。
+    primary_image = str((current_images or [""])[0] or "").strip()
+    if not primary_image:
+        return {
+            "currentLabel": "-",
+            "currentTitle": "-",
+            "detailLabel": "-",
+            "detailTitle": "-",
+            "copyValue": "",
+        }
+
+    repository = image_repository_ref(primary_image)
+    meta = {
+        "repository": repository,
+        "tag": "",
+        "digest": "",
+    }
+    without_digest, _, digest_text = primary_image.partition("@")
+    meta["digest"] = digest_text or ""
+    last_slash = without_digest.rfind("/")
+    last_colon = without_digest.rfind(":")
+    if last_colon > last_slash:
+        meta["tag"] = without_digest[last_colon + 1 :]
+
+    return {
+        "currentLabel": meta["tag"] or "-",
+        "currentTitle": "\n".join(current_images) if len(current_images) > 1 else primary_image,
+        "detailLabel": (meta["digest"][:12] + "...") if len(meta["digest"]) > 15 else (meta["digest"] or "-"),
+        "detailTitle": primary_image,
+        "copyValue": primary_image,
+    }
+
+
+def frontend_release_display(service_name: str) -> dict[str, str]:
+    # 前端目标优先显示最近一次由监控面板写入的 gitRef / commit。
+    metadata = frontend_release_metadata(service_name)
+    version = str(metadata.get("version") or "").strip()
+    commit = str(metadata.get("commit") or "").strip()
+    git_ref = str(metadata.get("gitRef") or "").strip()
+    deployed_at = str(metadata.get("deployedAt") or "").strip()
+    error = str(metadata.get("error") or "").strip()
+
+    if version or commit or git_ref:
+        label = version or git_ref or commit[:12]
+        title_parts = [part for part in [version, git_ref, commit, deployed_at, ADMIN_FRONTEND_PUBLIC_URL] if part]
+        return {
+            "currentLabel": label,
+            "currentTitle": " | ".join(title_parts) if title_parts else label,
+            "detailLabel": commit[:12] if commit else "-",
+            "detailTitle": commit or deployed_at or label,
+            "copyValue": commit or git_ref,
+        }
+
+    if error:
+        return {
+            "currentLabel": "未记录",
+            "currentTitle": error,
+            "detailLabel": "SSH",
+            "detailTitle": error,
+            "copyValue": "",
+        }
+
+    return {
+        "currentLabel": "未记录",
+        "currentTitle": "当前站点还没有由 hy-app-monitor 写入过版本元数据",
+        "detailLabel": "-",
+        "detailTitle": "-",
+        "copyValue": "",
+    }
+
+
 def docker_digest(image_ref: str, *, cwd: Path) -> str:
    # push 完要拿 digest，后续 compose 一律按 digest 部署。
    return local_command_output(
@ -850,6 +1028,81 @@ def build_golang_images(
    }


+def build_frontend_bundle(
+    service_name: str,
+    git_ref: str,
+    *,
+    progress: ProgressLogger | None = None,
+) -> dict[str, Any]:
+    # Admin 前端在 deploy 机做 pnpm install + build，最后产出静态 dist 目录。
+    repo_root = ensure_frontend_repo_root(progress=progress)
+    resolved_ref = str(git_ref or "").strip() or UPDATE_DEFAULT_ADMIN_GIT_REF
+    emit_progress(progress, "build.frontend", f"刷新前端仓库 refs：{resolved_ref}")
+    fetch_repo_refs(repo_root, "frontend")
+    commit = resolve_repo_commit(repo_root, resolved_ref, "frontend")
+    emit_progress(progress, "build.frontend", f"Frontend ref 已解析：{resolved_ref} -> {commit[:12]}")
+    worktree_root = prepare_persistent_build_workspace(
+        repo_root,
+        resolved_ref,
+        commit,
+        "frontend",
+        progress=progress,
+    )
+
+    frontend_env = {
+        "CI": "true",
+        "NODE_OPTIONS": "--max-old-space-size=8192",
+    }
+    emit_progress(progress, "build.frontend", "准备 corepack / pnpm")
+    local_command_output(
+        ["corepack", "enable"],
+        cwd=worktree_root,
+        label="frontend corepack enable",
+        env=frontend_env,
+        timeout_seconds=120,
+    )
+    emit_progress(progress, "build.frontend.install", "开始安装前端依赖")
+    run_logged_command(
+        ["corepack", "pnpm", "install", "--frozen-lockfile"],
+        cwd=worktree_root,
+        label="frontend pnpm install",
+        env=frontend_env,
+        timeout_seconds=SERVICE_BUILD_TIMEOUT_SECONDS,
+        progress_stage="build.frontend.install",
+        progress=progress,
+    )
+    emit_progress(progress, "build.frontend.build", f"开始构建静态站点：{ADMIN_FRONTEND_BUILD_SCRIPT}")
+    run_logged_command(
+        ["corepack", "pnpm", "run", ADMIN_FRONTEND_BUILD_SCRIPT],
+        cwd=worktree_root,
+        label="frontend pnpm build",
+        env=frontend_env,
+        timeout_seconds=SERVICE_BUILD_TIMEOUT_SECONDS,
+        progress_stage="build.frontend.build",
+        progress=progress,
+    )
+
+    artifact_root = (worktree_root / ADMIN_FRONTEND_DIST_DIR).resolve()
+    if not artifact_root.exists():
+        raise RuntimeError(f"frontend dist missing after build: {artifact_root}")
+    if not artifact_root.is_dir():
+        raise RuntimeError(f"frontend dist is not a directory: {artifact_root}")
+
+    version = frontend_version_label(resolved_ref, commit)
+    emit_progress(progress, "build.frontend", f"前端静态产物已生成：{artifact_root}")
+    return {
+        "kind": "frontend",
+        "repoRoot": str(repo_root),
+        "repoUrl": str(ADMIN_FRONTEND_REPO_URL or "").strip(),
+        "gitRef": resolved_ref,
+        "commit": commit,
+        "imageTag": version,
+        "services": [service_name],
+        "artifactPath": str(artifact_root),
+        "serviceArtifacts": {service_name: str(artifact_root)},
+    }
+
+
 def build_service_images(
    service_names: list[str],
    *,
@ -937,16 +1190,12 @@ def render_remote_login_lines() -> list[str]:

 def remote_host_result(host_entry: dict[str, Any], script: str, command_name: str, timeout_seconds: int) -> dict[str, Any]:
    # 统一执行单宿主机命令，便于预拉镜像和其他操作复用。
-    result_map = run_shell_command(
-        [host_entry["instanceId"]],
+    return run_host_script_checked(
+        host_entry,
        script,
        command_name=command_name,
        timeout_seconds=timeout_seconds,
    )
-    result = result_map[host_entry["instanceId"]]
-    if result.get("status") != "SUCCESS":
-        raise RuntimeError(f"{host_entry['host']} {command_name} failed: {result.get('status')}")
-    return result


 def prepull_host_images(host_entry: dict[str, Any]) -> dict[str, Any]:
@ -967,6 +1216,188 @@ def prepull_host_images(host_entry: dict[str, Any]) -> dict[str, Any]:
    }


+def prepare_frontend_remote_tmp(host: dict[str, Any]) -> str:
+    # 先在远端站点目录旁边建一份临时上传目录，避免直接覆盖线上内容。
+    site_dir = str(ADMIN_FRONTEND_SITE_DIR or "").rstrip("/")
+    parent_dir = str(Path(site_dir).parent)
+    script = f"""set -eu
+PARENT_DIR={shlex.quote(parent_dir)}
+mkdir -p "$PARENT_DIR"
+mktemp -d "$PARENT_DIR/.hy-app-monitor-site.XXXXXX"
+"""
+    result = run_host_script_checked(
+        host,
+        script,
+        command_name=f"frontend-tmp-{host['host']}",
+        timeout_seconds=30,
+    )
+    tmp_dir = str(result.get("output") or "").strip().splitlines()[-1].strip()
+    if not tmp_dir:
+        raise RuntimeError(f"failed to create frontend temp dir on {target_display(host)}")
+    return tmp_dir
+
+
+def activate_frontend_site(
+    host: dict[str, Any],
+    *,
+    tmp_dir: str,
+    backup_stamp: str,
+) -> dict[str, str]:
+    # 站点目录保留原 inode，只在目录内替换文件，避免 bind mount 指向失效。
+    site_dir = str(ADMIN_FRONTEND_SITE_DIR or "").rstrip("/")
+    remote_dir = str(ADMIN_FRONTEND_REMOTE_DIR or "").rstrip("/")
+    compose_file = str(ADMIN_FRONTEND_COMPOSE_FILE or "").rstrip("/")
+    compose_service = str(ADMIN_FRONTEND_COMPOSE_SERVICE or "").strip() or "caddy"
+    backup_root = str(Path(site_dir).parent / ".hy-app-monitor-site-backups")
+    keep_count = max(1, int(FRONTEND_SITE_BACKUP_KEEP))
+    script = f"""set -eu
+SITE_DIR={shlex.quote(site_dir)}
+TMP_DIR={shlex.quote(tmp_dir)}
+BACKUP_ROOT={shlex.quote(backup_root)}
+BACKUP_DIR="$BACKUP_ROOT/{shlex.quote(backup_stamp)}"
+REMOTE_DIR={shlex.quote(remote_dir)}
+COMPOSE_FILE={shlex.quote(compose_file)}
+COMPOSE_SERVICE={shlex.quote(compose_service)}
+KEEP_COUNT={keep_count}
+
+mkdir -p "$SITE_DIR" "$BACKUP_ROOT"
+if [ -d "$SITE_DIR" ]; then
+  mkdir -p "$BACKUP_DIR"
+  cp -a "$SITE_DIR"/. "$BACKUP_DIR"/ 2>/dev/null || true
+fi
+find "$SITE_DIR" -mindepth 1 -maxdepth 1 -exec rm -rf {{}} +
+cp -a "$TMP_DIR"/. "$SITE_DIR"/
+rm -rf "$TMP_DIR"
+
+if [ -d "$BACKUP_ROOT" ]; then
+  ls -1dt "$BACKUP_ROOT"/* 2>/dev/null | awk "NR>$KEEP_COUNT" | xargs -r rm -rf
+fi
+
+cd "$REMOTE_DIR"
+docker compose -f "$COMPOSE_FILE" up -d "$COMPOSE_SERVICE" >/dev/null
+docker compose -f "$COMPOSE_FILE" ps "$COMPOSE_SERVICE" || true
+"""
+    result = run_host_script_checked(
+        host,
+        script,
+        command_name=f"frontend-activate-{host['host']}",
+        timeout_seconds=max(SERVICE_BUILD_TIMEOUT_SECONDS, 180),
+    )
+    return {
+        "backupPath": f"{backup_root}/{backup_stamp}",
+        "output": str(result.get("output") or "").strip()[-1200:],
+    }
+
+
+def deploy_frontend_site(
+    service_name: str,
+    *,
+    git_ref: str,
+    progress: ProgressLogger | None = None,
+) -> dict[str, Any]:
+    # 前端发布链路：构建 dist -> SCP 到目标机 -> 原地切换站点内容 -> 写版本元数据。
+    host = frontend_host_record(service_name)
+    build_result = build_frontend_bundle(service_name, git_ref, progress=progress)
+    artifact_path = Path(str(build_result.get("artifactPath") or "")).expanduser().resolve()
+    version = str(build_result.get("imageTag") or "").strip()
+    backup_stamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
+    step_stage = f"frontend.{host['host']}.{service_name}"
+    started_at = time.time()
+    step: dict[str, Any] = {
+        "service": service_name,
+        "host": host["host"],
+        "ip": host["ip"],
+        "kind": "frontend",
+        "image": version,
+        "status": "running",
+    }
+    tmp_dir = ""
+
+    try:
+        tmp_dir = prepare_frontend_remote_tmp(host)
+        emit_progress(progress, step_stage, f"上传静态产物：{artifact_path} -> {target_display(host)}:{tmp_dir}")
+        copy_local_path_to_host(
+            host,
+            f"{artifact_path}/.",
+            f"{tmp_dir}/",
+            recursive=True,
+            timeout_seconds=max(SERVICE_BUILD_TIMEOUT_SECONDS, 180),
+        )
+        emit_progress(progress, step_stage, "上传完成，开始切换站点目录")
+        activate_result = activate_frontend_site(host, tmp_dir=tmp_dir, backup_stamp=backup_stamp)
+    except Exception as exc:  # noqa: BLE001
+        # 上传或切换失败时尽量收尾远端临时目录，避免后续发布被历史垃圾影响。
+        cleanup_script = f"""set -eu
+TMP_DIR={shlex.quote(tmp_dir)}
+if [ -d "$TMP_DIR" ]; then
+  rm -rf "$TMP_DIR"
+fi
+"""
+        if tmp_dir:
+            try:
+                run_host_script_checked(
+                    host,
+                    cleanup_script,
+                    command_name=f"frontend-cleanup-{host['host']}",
+                    timeout_seconds=30,
+                )
+            except Exception:  # noqa: BLE001
+                pass
+        step["status"] = "failed"
+        step["error"] = str(exc)
+        step["durationSeconds"] = round(time.time() - started_at, 2)
+        emit_progress(progress, step_stage, f"前端发布失败：{step['error']}", level="error")
+        return {
+            "ok": False,
+            "updatedAt": utc_now(),
+            "services": [service_name],
+            "build": [build_result],
+            "templates": [],
+            "preparedHosts": [],
+            "steps": [step],
+            "error": step["error"],
+        }
+
+    deployed_at = utc_now()
+    metadata = {
+        "service": service_name,
+        "kind": "frontend",
+        "repoUrl": str(ADMIN_FRONTEND_REPO_URL or "").strip(),
+        "repoRoot": str(ADMIN_FRONTEND_REPO_ROOT),
+        "gitRef": str(build_result.get("gitRef") or "").strip(),
+        "commit": str(build_result.get("commit") or "").strip(),
+        "version": version,
+        "publicUrl": ADMIN_FRONTEND_PUBLIC_URL,
+        "siteDir": ADMIN_FRONTEND_SITE_DIR,
+        "composeFile": ADMIN_FRONTEND_COMPOSE_FILE,
+        "deployedAt": deployed_at,
+        "backupPath": activate_result["backupPath"],
+        "deployedBy": "hy-app-monitor",
+    }
+    write_remote_text(
+        host,
+        ADMIN_FRONTEND_RELEASE_META_PATH,
+        json.dumps(metadata, ensure_ascii=False, indent=2) + "\n",
+        timeout_seconds=30,
+    )
+    emit_progress(progress, step_stage, f"版本元数据已写入：{ADMIN_FRONTEND_RELEASE_META_PATH}")
+    step["status"] = "success"
+    step["remoteStatus"] = "SUCCESS"
+    step["remoteOutput"] = activate_result["output"]
+    step["durationSeconds"] = round(time.time() - started_at, 2)
+
+    return {
+        "ok": True,
+        "updatedAt": deployed_at,
+        "services": [service_name],
+        "build": [build_result],
+        "templates": [],
+        "preparedHosts": [],
+        "steps": [step],
+        "message": "service update complete",
+    }
+
+
 def build_update_targets_payload() -> dict[str, Any]:
    # 页面需要展示当前可更新服务、宿主机，以及本地构建所依赖的仓路径。
    records_map = service_records_by_name()
@ -982,12 +1413,18 @@ def build_update_targets_payload() -> dict[str, Any]:
            image_ref = str(((model.get("services") or {}).get(service_name) or {}).get("image") or "").strip()
            if image_ref and image_ref not in current_images:
                current_images.append(image_ref)
+        display = image_release_display(current_images)

        items.append(
            {
                "service": service_name,
                "kind": service_kind(service_name),
                "repository": service_repository_ref(service_name),
+                "currentLabel": display["currentLabel"],
+                "currentTitle": display["currentTitle"],
+                "detailLabel": display["detailLabel"],
+                "detailTitle": display["detailTitle"],
+                "copyValue": display["copyValue"],
                "currentImages": current_images,
                "hosts": [
                    {
@ -1002,12 +1439,39 @@ def build_update_targets_payload() -> dict[str, Any]:
            }
        )

+    if frontend_release_enabled():
+        frontend_service = ADMIN_FRONTEND_SERVICE_NAME
+        display = frontend_release_display(frontend_service)
+        items.append(
+            {
+                "service": frontend_service,
+                "kind": "frontend",
+                "repository": str(ADMIN_FRONTEND_REPO_URL or ADMIN_FRONTEND_REPO_ROOT),
+                "currentLabel": display["currentLabel"],
+                "currentTitle": display["currentTitle"],
+                "detailLabel": display["detailLabel"],
+                "detailTitle": display["detailTitle"],
+                "copyValue": display["copyValue"],
+                "currentImages": [],
+                "hosts": [
+                    {
+                        "host": ADMIN_FRONTEND_HOST,
+                        "ip": ADMIN_FRONTEND_IP,
+                        "instanceId": ADMIN_FRONTEND_INSTANCE_ID,
+                        "port": 0,
+                        "path": ADMIN_FRONTEND_SITE_DIR,
+                    }
+                ],
+            }
+        )
+
    return {
        "ok": True,
        "updatedAt": utc_now(),
        "defaults": {
            "javaGitRef": UPDATE_DEFAULT_JAVA_GIT_REF,
            "goGitRef": UPDATE_DEFAULT_GO_GIT_REF,
+            "adminGitRef": UPDATE_DEFAULT_ADMIN_GIT_REF,
            "dockerPlatform": BUILD_DOCKER_PLATFORM,
        },
        "builder": {
@ -1015,12 +1479,15 @@ def build_update_targets_payload() -> dict[str, Any]:
            "harborProject": HARBOR_PROJECT,
            "javaRepoRoot": str(Path(JAVA_REPO_ROOT).expanduser()),
            "golangRepoRoot": str(Path(GOLANG_REPO_ROOT).expanduser()),
+            "frontendRepoRoot": str(Path(ADMIN_FRONTEND_REPO_ROOT).expanduser()),
+            "frontendRepoUrl": str(ADMIN_FRONTEND_REPO_URL or "").strip(),
+            "frontendPublicUrl": ADMIN_FRONTEND_PUBLIC_URL,
        },
        "items": items,
    }


-def deploy_updated_services_payload(
+def deploy_runtime_services_payload(
    service_names: list[str],
    *,
    java_git_ref: str,
@ -1029,7 +1496,7 @@ def deploy_updated_services_payload(
    skip_maven_build: bool,
    progress: ProgressLogger | None = None,
 ) -> dict[str, Any]:
-    # 这是“服务更新”主链路：本地 build -> Harbor digest -> 完整 compose -> 预拉 -> 滚动重建。
+    # Java / Go 服务沿用镜像构建、完整 compose、预拉、滚动重建这条链路。
    normalized = normalize_service_names(service_names)
    if not normalized:
        raise RuntimeError("services are required")
@ -1115,12 +1582,12 @@ def deploy_updated_services_payload(
                try:
                    emit_progress(progress, step_stage, f"开始滚动更新：{service['host']} / {service_name}")
                    restart_result = restart_service_instance(service)
-                    step["tatStatus"] = restart_result.get("status")
-                    step["tatOutput"] = str(restart_result.get("output") or "").strip()[:500]
+                    step["remoteStatus"] = restart_result.get("status")
+                    step["remoteOutput"] = str(restart_result.get("output") or "").strip()[:500]
                    emit_progress(
                        progress,
                        step_stage,
-                        f"TAT 重建完成：{restart_result.get('status')} {short_text(step['tatOutput'])}",
+                        f"远端重建完成：{restart_result.get('status')} {short_text(step['remoteOutput'])}",
                    )

                    http_result = wait_http_ready(service, timeout_seconds=ROLLING_RESTART_TIMEOUT_SECONDS)
@ -1184,7 +1651,6 @@ def deploy_updated_services_payload(
        clear_operation_caches()
        emit_progress(progress, "finish", "已清理监控缓存")

-    emit_progress(progress, "finish", "全部服务更新完成")
    return {
        "ok": True,
        "updatedAt": utc_now(),
@ -1195,3 +1661,87 @@ def deploy_updated_services_payload(
        "steps": steps,
        "message": "service update complete",
    }
+
+
+def merge_update_results(service_names: list[str], results: list[dict[str, Any]]) -> dict[str, Any]:
+    # 一个任务里允许同时带镜像服务和前端服务，这里统一把结果拼回页面结构。
+    merged = {
+        "ok": True,
+        "updatedAt": utc_now(),
+        "services": list(service_names),
+        "build": [],
+        "templates": [],
+        "preparedHosts": [],
+        "steps": [],
+        "message": "service update complete",
+    }
+
+    for result in results:
+        if not result:
+            continue
+        merged["ok"] = merged["ok"] and bool(result.get("ok"))
+        merged["build"].extend(list(result.get("build") or result.get("builds") or []))
+        merged["templates"].extend(list(result.get("templates") or []))
+        merged["preparedHosts"].extend(list(result.get("preparedHosts") or []))
+        merged["steps"].extend(list(result.get("steps") or []))
+        if not result.get("ok"):
+            merged["error"] = str(result.get("error") or "service update failed").strip()
+            merged["message"] = str(result.get("message") or merged["error"] or "service update failed").strip()
+            return merged
+        if str(result.get("message") or "").strip():
+            merged["message"] = str(result.get("message") or "").strip()
+
+    return merged
+
+
+def deploy_updated_services_payload(
+    service_names: list[str],
+    *,
+    java_git_ref: str,
+    go_git_ref: str,
+    admin_git_ref: str,
+    image_tag: str,
+    skip_maven_build: bool,
+    progress: ProgressLogger | None = None,
+) -> dict[str, Any]:
+    # 顶层按服务类型拆分：容器服务走 Harbor/compose，Admin 前端走 pnpm build + SSH 发布。
+    normalized = normalize_service_names(service_names)
+    if not normalized:
+        raise RuntimeError("services are required")
+
+    unsupported = [service_name for service_name in normalized if service_name not in buildable_service_names()]
+    if unsupported:
+        raise RuntimeError(f"unsupported update services: {', '.join(unsupported)}")
+
+    runtime_services = [service_name for service_name in normalized if service_kind(service_name) in {"java", "golang"}]
+    frontend_services = [service_name for service_name in normalized if service_kind(service_name) == "frontend"]
+    records_map = service_records_by_name()
+
+    for service_name in runtime_services:
+        if service_name not in records_map:
+            raise RuntimeError(f"service not found in runtime topology: {service_name}")
+
+    emit_progress(progress, "prepare", f"准备更新服务：{', '.join(normalized)}")
+    results: list[dict[str, Any]] = []
+
+    if runtime_services:
+        runtime_result = deploy_runtime_services_payload(
+            runtime_services,
+            java_git_ref=java_git_ref,
+            go_git_ref=go_git_ref,
+            image_tag=image_tag,
+            skip_maven_build=skip_maven_build,
+            progress=progress,
+        )
+        results.append(runtime_result)
+        if not runtime_result.get("ok"):
+            return merge_update_results(normalized, results)
+
+    for service_name in frontend_services:
+        frontend_result = deploy_frontend_site(service_name, git_ref=admin_git_ref, progress=progress)
+        results.append(frontend_result)
+        if not frontend_result.get("ok"):
+            return merge_update_results(normalized, results)
+
+    emit_progress(progress, "finish", "全部服务更新完成")
+    return merge_update_results(normalized, results)
--- a/monitor/service_update_jobs.py
+++ b/monitor/service_update_jobs.py
@ -156,6 +156,7 @@ def create_job_payload(
    *,
    java_git_ref: str,
    go_git_ref: str,
+    admin_git_ref: str,
    image_tag: str,
    skip_maven_build: bool,
 ) -> dict[str, Any]:
@ -180,6 +181,7 @@ def create_job_payload(
            "services": normalized,
            "javaGitRef": str(java_git_ref or "").strip(),
            "goGitRef": str(go_git_ref or "").strip(),
+            "adminGitRef": str(admin_git_ref or "").strip(),
            "imageTag": str(image_tag or "").strip(),
            "skipMavenBuild": bool(skip_maven_build),
        },
@ -284,6 +286,7 @@ def run_update_job(
    *,
    java_git_ref: str,
    go_git_ref: str,
+    admin_git_ref: str,
    image_tag: str,
    skip_maven_build: bool,
 ) -> None:
@ -296,6 +299,7 @@ def run_update_job(
            service_names,
            java_git_ref=java_git_ref,
            go_git_ref=go_git_ref,
+            admin_git_ref=admin_git_ref,
            image_tag=image_tag,
            skip_maven_build=skip_maven_build,
            progress=job_progress_logger(operation_id),
@ -324,6 +328,7 @@ def start_update_job(
    *,
    java_git_ref: str,
    go_git_ref: str,
+    admin_git_ref: str,
    image_tag: str,
    skip_maven_build: bool,
 ) -> dict[str, Any]:
@ -332,6 +337,7 @@ def start_update_job(
        service_names,
        java_git_ref=java_git_ref,
        go_git_ref=go_git_ref,
+        admin_git_ref=admin_git_ref,
        image_tag=image_tag,
        skip_maven_build=skip_maven_build,
    )
@ -344,6 +350,7 @@ def start_update_job(
            "service_names": list(service_names),
            "java_git_ref": java_git_ref,
            "go_git_ref": go_git_ref,
+            "admin_git_ref": admin_git_ref,
            "image_tag": image_tag,
            "skip_maven_build": skip_maven_build,
        },
--- a/ops/bootstrap_ssh_via_tat.py
+++ b/ops/bootstrap_ssh_via_tat.py
@ -0,0 +1,543 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import os
+import shlex
+import site
+import sys
+import time
+from pathlib import Path
+from typing import Callable, TypeVar
+
+
+# 项目根目录，后面需要从这里读取 .env 和 targets.json。
+ROOT = Path(__file__).resolve().parents[1]
+# 默认 .env 路径。
+ENV_FILE = Path(os.environ.get("APP_ENV_FILE", ROOT / ".env")).expanduser()
+# 泛型只用于 invoke 的返回值标注。
+T = TypeVar("T")
+
+
+def load_project_venv() -> None:
+    # 如果当前本来就在项目虚拟环境里运行，就不需要重复处理。
+    if str(ROOT / ".venv") in sys.prefix:
+        return
+
+    # 遍历项目 .venv 下可能存在的 site-packages 目录。
+    for candidate in (ROOT / ".venv" / "lib").glob("python*/site-packages"):
+        # 把依赖目录加入解释器搜索路径。
+        site.addsitedir(str(candidate))
+
+
+# 先注入项目虚拟环境，再导入腾讯云 SDK。
+load_project_venv()
+
+# 腾讯云凭据对象。
+from tencentcloud.common import credential
+# 腾讯云 SDK 统一异常类型。
+from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
+# 客户端公共配置。
+from tencentcloud.common.profile.client_profile import ClientProfile
+# HTTP endpoint 配置。
+from tencentcloud.common.profile.http_profile import HttpProfile
+# CVM 用来解析实例私网 IP 和安全组。
+from tencentcloud.cvm.v20170312 import cvm_client, models as cvm_models
+# TAT 用来做一次性带外引导。
+from tencentcloud.tat.v20201028 import models as tat_models, tat_client
+# VPC 用来写安全组规则。
+from tencentcloud.vpc.v20170312 import models as vpc_models, vpc_client
+
+
+def load_env_file(path: Path) -> None:
+    # .env 文件不存在时允许继续使用外部环境变量。
+    if not path.exists():
+        return
+
+    # 逐行解析 KEY=VALUE。
+    for raw_line in path.read_text(encoding="utf-8").splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, value = line.split("=", 1)
+        key = key.strip()
+        value = value.strip()
+        if value.startswith('"') and value.endswith('"'):
+            value = value[1:-1]
+        if value.startswith("'") and value.endswith("'"):
+            value = value[1:-1]
+        os.environ.setdefault(key, value)
+
+
+def env_truthy(name: str, default: bool = False) -> bool:
+    # 兼容 .env 里常见的布尔字符串写法。
+    raw = os.environ.get(name)
+    if raw is None:
+        return default
+    return raw.strip().lower() in {"1", "true", "yes", "on"}
+
+
+# 先加载项目根目录的 .env。
+load_env_file(ENV_FILE)
+
+# 默认 deploy 实例 ID。
+DEFAULT_DEPLOY_INSTANCE_ID = os.environ.get("DEPLOY_INSTANCE_ID", "").strip()
+# 默认区域。
+DEFAULT_REGION = os.environ.get("DEPLOY_REGION", "").strip()
+# 默认监控目标文件。
+TARGETS_FILE = Path(os.environ.get("TARGETS_FILE", ROOT / "config" / "targets.json")).expanduser()
+# deploy 机上生成的专用 SSH 私钥路径。
+DEFAULT_REMOTE_SSH_KEY_PATH = os.environ.get("SSH_PRIVATE_KEY_PATH", "/opt/hy-app-monitor/run/ssh/id_ed25519").strip()
+# 默认远端 SSH 端口。
+DEFAULT_SSH_PORT = int(os.environ.get("SSH_PORT", "22"))
+# 默认远端 SSH 用户。
+DEFAULT_SSH_USER = os.environ.get("SSH_USER", "root").strip() or "root"
+# 独立 admin 前端目标机配置；配置齐备时会一并加入 SSH 引导。
+ADMIN_FRONTEND_ENABLED = env_truthy("ADMIN_FRONTEND_ENABLED", False)
+ADMIN_FRONTEND_HOST = os.environ.get("ADMIN_FRONTEND_HOST", "").strip()
+ADMIN_FRONTEND_IP = os.environ.get("ADMIN_FRONTEND_IP", "").strip()
+ADMIN_FRONTEND_INSTANCE_ID = os.environ.get("ADMIN_FRONTEND_INSTANCE_ID", "").strip()
+
+
+def parse_args() -> argparse.Namespace:
+    # 默认对 targets.json 里的所有实例做 SSH 引导。
+    parser = argparse.ArgumentParser(description="bootstrap deploy->target ssh connectivity via TAT")
+    parser.add_argument("--hosts", nargs="*", default=[], help="target host names, default all hosts from targets.json")
+    parser.add_argument("--deploy-instance-id", default=DEFAULT_DEPLOY_INSTANCE_ID, help="deploy machine instance id")
+    parser.add_argument("--region", default=DEFAULT_REGION, help="tencent cloud region")
+    parser.add_argument("--ssh-user", default=DEFAULT_SSH_USER, help="remote ssh login user, default root")
+    parser.add_argument("--ssh-port", type=int, default=DEFAULT_SSH_PORT, help="remote ssh port")
+    parser.add_argument(
+        "--remote-key-path",
+        default=DEFAULT_REMOTE_SSH_KEY_PATH,
+        help="private key path generated on deploy machine",
+    )
+    parser.add_argument(
+        "--deploy-ip",
+        default="",
+        help="override deploy private ip used in security group and firewall rules",
+    )
+    return parser.parse_args()
+
+
+def env_required(name: str) -> str:
+    # 读取环境变量并去掉两端空白。
+    value = os.environ.get(name, "").strip()
+    if not value:
+        raise SystemExit(f"missing env: {name}")
+    return value
+
+
+def build_cvm(secret_id: str, secret_key: str, region: str) -> cvm_client.CvmClient:
+    # 构造腾讯云凭据对象。
+    cred = credential.Credential(secret_id, secret_key)
+    return cvm_client.CvmClient(
+        cred,
+        region,
+        ClientProfile(httpProfile=HttpProfile(endpoint="cvm.tencentcloudapi.com")),
+    )
+
+
+def build_tat(secret_id: str, secret_key: str, region: str) -> tat_client.TatClient:
+    # 构造腾讯云凭据对象。
+    cred = credential.Credential(secret_id, secret_key)
+    return tat_client.TatClient(
+        cred,
+        region,
+        ClientProfile(httpProfile=HttpProfile(endpoint="tat.tencentcloudapi.com")),
+    )
+
+
+def build_vpc(secret_id: str, secret_key: str, region: str) -> vpc_client.VpcClient:
+    # 构造腾讯云凭据对象。
+    cred = credential.Credential(secret_id, secret_key)
+    return vpc_client.VpcClient(
+        cred,
+        region,
+        ClientProfile(httpProfile=HttpProfile(endpoint="vpc.tencentcloudapi.com")),
+    )
+
+
+def invoke(action: Callable[[], T], label: str, retries: int = 4, delay_seconds: int = 2) -> T:
+    # 保存最后一次异常，方便重试结束后抛出。
+    last_error: Exception | None = None
+    for attempt in range(1, retries + 1):
+        try:
+            return action()
+        except TencentCloudSDKException as exc:
+            last_error = exc
+            if attempt == retries:
+                break
+            time.sleep(delay_seconds * attempt)
+
+    if last_error is not None:
+        raise last_error
+    raise RuntimeError(f"{label} failed without error")
+
+
+def describe_instances(client: cvm_client.CvmClient, instance_ids: list[str]) -> dict[str, dict[str, object]]:
+    # 一次性把 deploy 和目标机实例详情取回来，减少 API 往返。
+    req = cvm_models.DescribeInstancesRequest()
+    req.from_json_string(json.dumps({"InstanceIds": instance_ids}))
+    resp = json.loads(invoke(lambda: client.DescribeInstances(req), "DescribeInstances").to_json_string())
+    items = resp.get("InstanceSet") or []
+    return {
+        str(item.get("InstanceId") or ""): {
+            "privateIp": str((item.get("PrivateIpAddresses") or [""])[0] or "").strip(),
+            "securityGroupIds": [str(group_id) for group_id in (item.get("SecurityGroupIds") or []) if str(group_id)],
+            "instanceName": str(item.get("InstanceName") or "").strip(),
+        }
+        for item in items
+    }
+
+
+def load_target_hosts(host_names: list[str]) -> list[dict[str, str]]:
+    # 读取 targets.json，按 host 名过滤后返回唯一实例列表。
+    if not TARGETS_FILE.exists():
+        raise SystemExit(f"targets file not found: {TARGETS_FILE}")
+    payload = json.loads(TARGETS_FILE.read_text(encoding="utf-8"))
+    configured = payload.get("hosts") or []
+    selected = {str(name).strip() for name in host_names if str(name).strip()}
+    items: list[dict[str, str]] = []
+    seen_instance_ids: set[str] = set()
+
+    for host in configured:
+        host_name = str(host.get("name") or "").strip()
+        instance_id = str(host.get("instanceId") or "").strip()
+        if not host_name or not instance_id:
+            continue
+        if selected and host_name not in selected:
+            continue
+        if instance_id in seen_instance_ids:
+            continue
+        seen_instance_ids.add(instance_id)
+        items.append(
+            {
+                "host": host_name,
+                "ip": str(host.get("ip") or "").strip(),
+                "instanceId": instance_id,
+            }
+        )
+
+    # Admin 前端机器不在 targets.json 时，也允许通过 .env 把它并进同一轮 SSH 引导。
+    if ADMIN_FRONTEND_ENABLED and ADMIN_FRONTEND_HOST and ADMIN_FRONTEND_INSTANCE_ID:
+        if (not selected) or ADMIN_FRONTEND_HOST in selected:
+            if ADMIN_FRONTEND_INSTANCE_ID not in seen_instance_ids:
+                seen_instance_ids.add(ADMIN_FRONTEND_INSTANCE_ID)
+                items.append(
+                    {
+                        "host": ADMIN_FRONTEND_HOST,
+                        "ip": ADMIN_FRONTEND_IP,
+                        "instanceId": ADMIN_FRONTEND_INSTANCE_ID,
+                    }
+                )
+    return items
+
+
+def render_keygen_script(remote_key_path: str) -> str:
+    # deploy 机上只生成一次专用 SSH key，后续运行链路直接复用。
+    return f"""set -euo pipefail
+KEY_PATH={json.dumps(remote_key_path)}
+KEY_DIR="$(dirname "$KEY_PATH")"
+mkdir -p "$KEY_DIR"
+chmod 700 "$KEY_DIR"
+if [ ! -f "$KEY_PATH" ]; then
+  ssh-keygen -t ed25519 -N '' -C 'hy-app-monitor-deploy' -f "$KEY_PATH" >/dev/null
+fi
+touch "$KEY_DIR/known_hosts"
+chmod 600 "$KEY_PATH" "$KEY_PATH.pub" "$KEY_DIR/known_hosts"
+cat "$KEY_PATH.pub"
+"""
+
+
+def render_target_bootstrap_script(public_key: str, deploy_ip: str, ssh_user: str, ssh_port: int) -> str:
+    # 目标机侧只做 SSH 基础设施准备，不夹带业务运维逻辑。
+    quoted_key = shlex.quote(public_key)
+    quoted_ip = shlex.quote(deploy_ip)
+    quoted_user = shlex.quote(ssh_user)
+    root_login = "yes" if ssh_user == "root" else "no"
+
+    if ssh_user == "root":
+        principal_block = """
+HOME_DIR=/root
+install -d -m 700 "$HOME_DIR/.ssh"
+touch "$HOME_DIR/.ssh/authorized_keys"
+chmod 600 "$HOME_DIR/.ssh/authorized_keys"
+grep -qxF "$PUBKEY" "$HOME_DIR/.ssh/authorized_keys" || printf '%s\\n' "$PUBKEY" >> "$HOME_DIR/.ssh/authorized_keys"
+"""
+    else:
+        principal_block = """
+if ! id -u "$SSH_USER" >/dev/null 2>&1; then
+  useradd -m -s /bin/bash "$SSH_USER"
+fi
+if getent group docker >/dev/null 2>&1; then
+  usermod -aG docker "$SSH_USER" >/dev/null 2>&1 || true
+fi
+HOME_DIR="$(getent passwd "$SSH_USER" | cut -d: -f6)"
+install -d -o "$SSH_USER" -g "$SSH_USER" -m 700 "$HOME_DIR/.ssh"
+touch "$HOME_DIR/.ssh/authorized_keys"
+chown "$SSH_USER":"$SSH_USER" "$HOME_DIR/.ssh/authorized_keys"
+chmod 600 "$HOME_DIR/.ssh/authorized_keys"
+grep -qxF "$PUBKEY" "$HOME_DIR/.ssh/authorized_keys" || printf '%s\\n' "$PUBKEY" >> "$HOME_DIR/.ssh/authorized_keys"
+chown "$SSH_USER":"$SSH_USER" "$HOME_DIR/.ssh/authorized_keys"
+printf '%s\\n' "$SSH_USER ALL=(ALL) NOPASSWD:ALL" > /etc/sudoers.d/hy-app-monitor-ssh-user
+chmod 440 /etc/sudoers.d/hy-app-monitor-ssh-user
+"""
+
+    return f"""set -euo pipefail
+PUBKEY={quoted_key}
+DEPLOY_IP={quoted_ip}
+SSH_USER={quoted_user}
+SSH_PORT={int(ssh_port)}
+DEPLOY_CIDR="${{DEPLOY_IP}}/32"
+
+if ! command -v sshd >/dev/null 2>&1; then
+  if command -v apt-get >/dev/null 2>&1; then
+    export DEBIAN_FRONTEND=noninteractive
+    apt-get update -y >/dev/null 2>&1 || true
+    apt-get install -y openssh-server >/dev/null
+  elif command -v dnf >/dev/null 2>&1; then
+    dnf install -y openssh-server >/dev/null
+  elif command -v yum >/dev/null 2>&1; then
+    yum install -y openssh-server >/dev/null
+  else
+    echo "openssh-server unavailable" >&2
+    exit 1
+  fi
+fi
+
+{principal_block}
+
+mkdir -p /etc/ssh/sshd_config.d
+if ! grep -Eq '^[[:space:]]*Include[[:space:]]+/etc/ssh/sshd_config\\.d/\\*\\.conf' /etc/ssh/sshd_config; then
+  printf '\\nInclude /etc/ssh/sshd_config.d/*.conf\\n' >> /etc/ssh/sshd_config
+fi
+
+cat > /etc/ssh/sshd_config.d/hy-app-monitor.conf <<EOF
+PubkeyAuthentication yes
+AuthorizedKeysFile .ssh/authorized_keys
+PermitRootLogin {root_login}
+Port {int(ssh_port)}
+EOF
+
+if command -v firewall-cmd >/dev/null 2>&1; then
+  firewall-cmd --permanent --add-rich-rule="rule family=ipv4 source address=${{DEPLOY_CIDR}} port protocol=tcp port=${{SSH_PORT}} accept" >/dev/null 2>&1 || true
+  firewall-cmd --reload >/dev/null 2>&1 || true
+fi
+
+if command -v ufw >/dev/null 2>&1; then
+  ufw allow from "${{DEPLOY_IP}}" to any port "${{SSH_PORT}}" proto tcp >/dev/null 2>&1 || true
+fi
+
+if command -v iptables >/dev/null 2>&1; then
+  iptables -C INPUT -s "${{DEPLOY_CIDR}}" -p tcp --dport "${{SSH_PORT}}" -j ACCEPT >/dev/null 2>&1 || iptables -I INPUT -s "${{DEPLOY_CIDR}}" -p tcp --dport "${{SSH_PORT}}" -j ACCEPT || true
+fi
+
+if command -v sshd >/dev/null 2>&1; then
+  sshd -t
+fi
+
+if systemctl list-unit-files sshd.service >/dev/null 2>&1; then
+  systemctl enable sshd >/dev/null 2>&1 || true
+  systemctl restart sshd
+elif systemctl list-unit-files ssh.service >/dev/null 2>&1; then
+  systemctl enable ssh >/dev/null 2>&1 || true
+  systemctl restart ssh
+else
+  echo "ssh systemd service not found" >&2
+  exit 1
+fi
+
+echo "ssh_bootstrap_ok user=${{SSH_USER}} port=${{SSH_PORT}}"
+"""
+
+
+def run_tat_script(client: tat_client.TatClient, instance_id: str, script: str, timeout_seconds: int = 600) -> str:
+    # 创建执行命令请求。
+    req = tat_models.RunCommandRequest()
+    req.from_json_string(
+        json.dumps(
+            {
+                "CommandName": "hy-app-monitor-bootstrap-ssh",
+                "CommandType": "SHELL",
+                "Content": base64.b64encode(script.encode("utf-8")).decode("ascii"),
+                "InstanceIds": [instance_id],
+                "Timeout": timeout_seconds,
+                "WorkingDirectory": "/root",
+            }
+        )
+    )
+
+    # 发起命令执行。
+    resp = json.loads(invoke(lambda: client.RunCommand(req), "RunCommand").to_json_string())
+    invocation_id = resp["InvocationId"]
+    deadline = time.time() + timeout_seconds + 120
+
+    # 轮询远端执行结果。
+    while time.time() < deadline:
+        time.sleep(3)
+        query = tat_models.DescribeInvocationTasksRequest()
+        query.from_json_string(
+            json.dumps(
+                {
+                    "HideOutput": False,
+                    "Limit": 20,
+                    "Filters": [{"Name": "invocation-id", "Values": [invocation_id]}],
+                }
+            )
+        )
+        data = json.loads(invoke(lambda: client.DescribeInvocationTasks(query), "DescribeInvocationTasks").to_json_string())
+        tasks = data.get("InvocationTaskSet") or []
+        if not tasks:
+            continue
+        task = tasks[0]
+        status = task.get("TaskStatus") or ""
+        if status in {"SUCCESS", "FAILED", "TIMEOUT", "CANCELLED", "START_FAILED"}:
+            output = base64.b64decode(task.get("TaskResult", {}).get("Output", "")).decode("utf-8", errors="replace")
+            if status != "SUCCESS":
+                raise SystemExit(output or f"TAT task failed: {status}")
+            return output
+
+    raise SystemExit("TAT task timed out")
+
+
+def ensure_security_group_rule(
+    client: vpc_client.VpcClient,
+    security_group_id: str,
+    *,
+    deploy_ip: str,
+    ssh_port: int,
+) -> str:
+    # 安全组只允许 deploy 私网 IP 访问 22，避免直接对整段网开放 SSH。
+    req = vpc_models.CreateSecurityGroupPoliciesRequest()
+    req.from_json_string(
+        json.dumps(
+            {
+                "SecurityGroupId": security_group_id,
+                "SecurityGroupPolicySet": {
+                    "Ingress": [
+                        {
+                            "Protocol": "TCP",
+                            "Port": str(ssh_port),
+                            "CidrBlock": f"{deploy_ip}/32",
+                            "Action": "ACCEPT",
+                            "PolicyDescription": f"hy-app-monitor ssh {deploy_ip}",
+                        }
+                    ]
+                },
+            }
+        )
+    )
+
+    try:
+        invoke(lambda: client.CreateSecurityGroupPolicies(req), "CreateSecurityGroupPolicies")
+        return "created"
+    except TencentCloudSDKException as exc:
+        if "Duplicate" in str(exc) or "already exists" in str(exc):
+            return "exists"
+        raise
+
+
+def prepare_deploy_public_key(
+    client: tat_client.TatClient,
+    deploy_instance_id: str,
+    remote_key_path: str,
+) -> str:
+    # deploy 机上生成完专用 key 后，把公钥文本拿回来继续给目标机铺设 authorized_keys。
+    output = run_tat_script(client, deploy_instance_id, render_keygen_script(remote_key_path), timeout_seconds=300)
+    public_key = output.strip().splitlines()[-1].strip()
+    if not public_key.startswith("ssh-"):
+        raise SystemExit(f"invalid public key output: {public_key or output.strip()}")
+    return public_key
+
+
+def main() -> None:
+    args = parse_args()
+    if not args.deploy_instance_id:
+        raise SystemExit("deploy instance id is required; set DEPLOY_INSTANCE_ID or pass --deploy-instance-id")
+    if not args.region:
+        raise SystemExit("deploy region is required; set DEPLOY_REGION or pass --region")
+
+    secret_id = env_required("TENCENT_SECRET_ID")
+    secret_key = env_required("TENCENT_SECRET_KEY")
+    targets = load_target_hosts(args.hosts)
+    if not targets:
+        raise SystemExit("no target hosts found in config/targets.json")
+
+    cvm = build_cvm(secret_id, secret_key, args.region)
+    tat = build_tat(secret_id, secret_key, args.region)
+    vpc = build_vpc(secret_id, secret_key, args.region)
+
+    instance_ids = [args.deploy_instance_id, *[target["instanceId"] for target in targets]]
+    instance_map = describe_instances(cvm, instance_ids)
+    deploy_detail = instance_map.get(args.deploy_instance_id)
+    if not deploy_detail:
+        raise SystemExit(f"deploy instance not found: {args.deploy_instance_id}")
+
+    deploy_ip = str(args.deploy_ip or deploy_detail.get("privateIp") or "").strip()
+    if not deploy_ip:
+        raise SystemExit(f"missing deploy private ip: {args.deploy_instance_id}")
+
+    public_key = prepare_deploy_public_key(tat, args.deploy_instance_id, args.remote_key_path)
+    results: list[dict[str, object]] = []
+
+    for target in targets:
+        detail = instance_map.get(target["instanceId"])
+        if not detail:
+            raise SystemExit(f"instance not found: {target['instanceId']}")
+
+        output = run_tat_script(
+            tat,
+            target["instanceId"],
+            render_target_bootstrap_script(public_key, deploy_ip, args.ssh_user, args.ssh_port),
+            timeout_seconds=600,
+        ).strip()
+
+        security_groups = list(detail.get("securityGroupIds") or [])
+        sg_results = [
+            {
+                "securityGroupId": security_group_id,
+                "status": ensure_security_group_rule(
+                    vpc,
+                    security_group_id,
+                    deploy_ip=deploy_ip,
+                    ssh_port=args.ssh_port,
+                ),
+            }
+            for security_group_id in security_groups
+        ]
+
+        results.append(
+            {
+                "host": target["host"],
+                "instanceId": target["instanceId"],
+                "ip": target["ip"] or detail.get("privateIp") or "",
+                "sshUser": args.ssh_user,
+                "sshPort": args.ssh_port,
+                "bootstrap": output,
+                "securityGroups": sg_results,
+            }
+        )
+
+    print(
+        json.dumps(
+            {
+                "ok": True,
+                "deployInstanceId": args.deploy_instance_id,
+                "deployIp": deploy_ip,
+                "remoteKeyPath": args.remote_key_path,
+                "sshUser": args.ssh_user,
+                "sshPort": args.ssh_port,
+                "targets": results,
+            },
+            ensure_ascii=False,
+            indent=2,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()
--- a/ops/deploy_via_ssh.py
+++ b/ops/deploy_via_ssh.py
@ -0,0 +1,244 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import argparse
+import os
+from pathlib import Path
+import shlex
+import subprocess
+
+
+# 项目根目录，后面会从这里读取 .env 并打包代码。
+ROOT = Path(__file__).resolve().parents[1]
+# 默认 .env 路径。
+ENV_FILE = Path(os.environ.get("APP_ENV_FILE", ROOT / ".env")).expanduser()
+
+
+def load_env_file(path: Path) -> None:
+    # .env 文件不存在时允许继续使用外部环境变量。
+    if not path.exists():
+        return
+
+    # 逐行读取 .env 内容。
+    for raw_line in path.read_text(encoding="utf-8").splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, value = line.split("=", 1)
+        key = key.strip()
+        value = value.strip()
+        if value.startswith('"') and value.endswith('"'):
+            value = value[1:-1]
+        if value.startswith("'") and value.endswith("'"):
+            value = value[1:-1]
+        os.environ.setdefault(key, value)
+
+
+# 先加载项目根目录 .env，保证默认 SSH 目标也能从这里读取。
+load_env_file(ENV_FILE)
+# 默认远端部署目录。
+DEFAULT_REMOTE_DIR = os.environ.get("MONITOR_REMOTE_DIR", "/opt/hy-app-monitor")
+# 默认远端监听端口。
+DEFAULT_APP_PORT = os.environ.get("APP_PORT", "2026")
+# 默认 SSH 主机、用户和端口。
+DEFAULT_DEPLOY_SSH_HOST = os.environ.get("DEPLOY_SSH_HOST", "").strip()
+DEFAULT_DEPLOY_SSH_USER = os.environ.get("DEPLOY_SSH_USER", "root").strip() or "root"
+DEFAULT_DEPLOY_SSH_PORT = int(os.environ.get("DEPLOY_SSH_PORT", "22"))
+DEFAULT_DEPLOY_SSH_IDENTITY_FILE = os.environ.get("DEPLOY_SSH_IDENTITY_FILE", "").strip()
+
+
+def parse_args() -> argparse.Namespace:
+    # 这里默认按当前项目目录直接同步到远端，不再依赖远端 git clone。
+    parser = argparse.ArgumentParser(description="deploy hy-app-monitor to deploy host via ssh/scp")
+    parser.add_argument("--host", default=DEFAULT_DEPLOY_SSH_HOST, help="deploy ssh host")
+    parser.add_argument("--user", default=DEFAULT_DEPLOY_SSH_USER, help="deploy ssh user")
+    parser.add_argument("--port", type=int, default=DEFAULT_DEPLOY_SSH_PORT, help="deploy ssh port")
+    parser.add_argument("--identity-file", default=DEFAULT_DEPLOY_SSH_IDENTITY_FILE, help="optional ssh identity file")
+    parser.add_argument("--remote-dir", default=DEFAULT_REMOTE_DIR, help="remote project directory")
+    parser.add_argument("--app-port", default=DEFAULT_APP_PORT, help="remote app port")
+    return parser.parse_args()
+
+
+def require_env_file() -> Path:
+    # 当前项目约定 .env 是唯一运行时配置源，部署时必须显式同步。
+    if not ENV_FILE.exists():
+        raise SystemExit(f"missing local env file: {ENV_FILE}")
+    return ENV_FILE
+
+
+def ssh_base_args(host: str, *, user: str, port: int, identity_file: str) -> list[str]:
+    # 所有 SSH / SCP 命令统一复用同一组参数，避免行为不一致。
+    args = [
+        "-p",
+        str(port),
+        "-o",
+        "BatchMode=yes",
+        "-o",
+        "ConnectTimeout=10",
+        "-o",
+        "ServerAliveInterval=15",
+        "-o",
+        "ServerAliveCountMax=3",
+        "-o",
+        "StrictHostKeyChecking=accept-new",
+    ]
+    if identity_file:
+        args.extend(["-i", identity_file])
+    args.append(f"{user}@{host}")
+    return args
+
+
+def scp_base_args(*, port: int, identity_file: str) -> list[str]:
+    # SCP 的端口参数和 SSH 不同，这里单独组装一份。
+    args = [
+        "-P",
+        str(port),
+        "-o",
+        "BatchMode=yes",
+        "-o",
+        "ConnectTimeout=10",
+        "-o",
+        "StrictHostKeyChecking=accept-new",
+    ]
+    if identity_file:
+        args.extend(["-i", identity_file])
+    return args
+
+
+def run_command(command: list[str], *, input_bytes: bytes | None = None, label: str) -> bytes:
+    # 小工具统一把 stdout/stderr 合并，失败时直接抛完整错误文本。
+    try:
+        result = subprocess.run(
+            command,
+            input=input_bytes,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            check=False,
+        )
+    except FileNotFoundError as exc:
+        raise SystemExit(f"missing command for {label}: {command[0]}") from exc
+
+    if result.returncode != 0:
+        raise SystemExit((result.stdout or b"").decode("utf-8", errors="replace").strip() or f"{label} failed")
+    return result.stdout or b""
+
+
+def sync_project_tree(*, remote_dir: str, ssh_args: list[str]) -> None:
+    # 远端保留 .env / .venv / run，其他项目内容全部按当前工作区覆盖。
+    remote_prepare_script = f"""set -euo pipefail
+REMOTE_DIR={shlex.quote(remote_dir)}
+mkdir -p "$REMOTE_DIR"
+find "$REMOTE_DIR" -mindepth 1 -maxdepth 1 \\
+  ! -name '.env' \\
+  ! -name '.venv' \\
+  ! -name 'run' \\
+  -exec rm -rf {{}} +
+"""
+    run_command(["ssh", *ssh_args, "bash", "-se"], input_bytes=remote_prepare_script.encode("utf-8"), label="remote prepare")
+
+    # 本地 tar 流直接送到远端解压，避免要求远端额外安装 rsync。
+    tar_command = [
+        "tar",
+        "-czf",
+        "-",
+        "--exclude=.git",
+        "--exclude=.env",
+        "--exclude=.venv",
+        "--exclude=run",
+        "--exclude=__pycache__",
+        "--exclude=.pytest_cache",
+        "--exclude=.mypy_cache",
+        "--exclude=.DS_Store",
+        ".",
+    ]
+    ssh_command = ["ssh", *ssh_args, "tar", "-xzf", "-", "-C", remote_dir]
+
+    try:
+        tar_process = subprocess.Popen(tar_command, cwd=str(ROOT), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    except FileNotFoundError as exc:
+        raise SystemExit("missing command for tar sync: tar") from exc
+
+    try:
+        ssh_process = subprocess.run(
+            ssh_command,
+            stdin=tar_process.stdout,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            check=False,
+        )
+        if tar_process.stdout is not None:
+            tar_process.stdout.close()
+        tar_stderr = (tar_process.stderr.read() if tar_process.stderr is not None else b"").decode("utf-8", errors="replace")
+        tar_return = tar_process.wait()
+    finally:
+        if tar_process.poll() is None:
+            tar_process.kill()
+            tar_process.wait()
+
+    if tar_return != 0:
+        raise SystemExit(tar_stderr.strip() or "local tar packaging failed")
+    if ssh_process.returncode != 0:
+        raise SystemExit((ssh_process.stdout or b"").decode("utf-8", errors="replace").strip() or "remote extract failed")
+
+
+def copy_env_file(
+    target: str,
+    *,
+    remote_dir: str,
+    env_path: Path,
+    port: int,
+    identity_file: str,
+) -> None:
+    # 运行配置单独走 scp，和代码同步解耦，避免误把远端 .env 覆盖成空文件。
+    run_command(
+        ["scp", *scp_base_args(port=port, identity_file=identity_file), str(env_path), f"{target}:{remote_dir}/.env"],
+        label="copy env",
+    )
+
+
+def run_remote_install(*, remote_dir: str, app_port: str, ssh_args: list[str]) -> str:
+    # 远端只负责安装依赖、重启 systemd 并校验监听。
+    script = f"""set -euo pipefail
+REMOTE_DIR={shlex.quote(remote_dir)}
+APP_PORT={shlex.quote(app_port)}
+cd "$REMOTE_DIR"
+chmod +x scripts/*.sh
+python3 -m venv .venv
+.venv/bin/pip install -r requirements-ops.txt
+bash scripts/install_systemd.sh
+systemctl restart hy-app-monitor.service
+systemctl is-active hy-app-monitor.service
+ss -lntp | grep ":$APP_PORT" || true
+"""
+    output = run_command(
+        ["ssh", *ssh_args, "bash", "-se"],
+        input_bytes=script.encode("utf-8"),
+        label="remote install",
+    )
+    return output.decode("utf-8", errors="replace").strip()
+
+
+def main() -> None:
+    args = parse_args()
+    if not args.host:
+        raise SystemExit("deploy ssh host is required; set DEPLOY_SSH_HOST or pass --host")
+
+    env_path = require_env_file()
+    target = f"{args.user}@{args.host}"
+    ssh_args = ssh_base_args(args.host, user=args.user, port=args.port, identity_file=args.identity_file)
+
+    sync_project_tree(remote_dir=args.remote_dir, ssh_args=ssh_args)
+    copy_env_file(
+        target,
+        remote_dir=args.remote_dir,
+        env_path=env_path,
+        port=args.port,
+        identity_file=args.identity_file,
+    )
+    output = run_remote_install(remote_dir=args.remote_dir, app_port=args.app_port, ssh_args=ssh_args)
+    print(output)
+
+
+if __name__ == "__main__":
+    main()
--- a/scripts/common.sh
+++ b/scripts/common.sh
@ -31,17 +31,31 @@ mkdir -p "$RUN_DIR"

 read_pid() {
  if [ -f "$PID_FILE" ]; then
-    cat "$PID_FILE"
+    local pid
+    pid="$(cat "$PID_FILE")"
+    if pid_matches_project "$pid"; then
+      echo "$pid"
+      return 0
+    fi
  fi
+
+  discover_project_pid
 }

 is_running() {
  local pid
  pid="$(read_pid || true)"
  if [ -z "${pid:-}" ]; then
+    rm -f "$PID_FILE"
    return 1
  fi
-  kill -0 "$pid" >/dev/null 2>&1
+  if ! kill -0 "$pid" >/dev/null 2>&1; then
+    rm -f "$PID_FILE"
+    return 1
+  fi
+
+  echo "$pid" >"$PID_FILE"
+  return 0
 }

 has_systemd_unit() {
@ -68,3 +82,42 @@ stop_legacy_process() {

  rm -f "$PID_FILE"
 }
+
+discover_project_pid() {
+  if ! command -v lsof >/dev/null 2>&1; then
+    return 1
+  fi
+
+  local pid
+  while IFS= read -r pid; do
+    if pid_matches_project "$pid"; then
+      echo "$pid"
+      return 0
+    fi
+  done < <(lsof -tiTCP:"$APP_PORT" -sTCP:LISTEN 2>/dev/null || true)
+
+  return 1
+}
+
+pid_matches_project() {
+  local pid="$1"
+  if [ -z "${pid:-}" ] || ! kill -0 "$pid" >/dev/null 2>&1; then
+    return 1
+  fi
+
+  local command
+  command="$(ps -p "$pid" -o command= 2>/dev/null || true)"
+  if [[ "$command" != *"$ROOT_DIR/server.py"* ]] && [[ "$command" != *" server.py"* ]]; then
+    return 1
+  fi
+
+  if command -v lsof >/dev/null 2>&1; then
+    local cwd
+    cwd="$(lsof -a -p "$pid" -d cwd -Fn 2>/dev/null | sed -n 's/^n//p' | head -n 1)"
+    if [ -n "$cwd" ] && [ "$cwd" != "$ROOT_DIR" ]; then
+      return 1
+    fi
+  fi
+
+  return 0
+}
--- a/static/app.css
+++ b/static/app.css
--- a/static/app.js
+++ b/static/app.js
--- a/static/index.html
+++ b/static/index.html