feat(failover): 补全主备切换闭环 — 六步状态机 + 逆向回滚 + 健康判定器

依据现有实现补全「自动识别故障 → 人工确认切换 → 切回」运维核心闭环。
触发模式 1A+2B+3A：VIP 浮动 + 半自动 + 切前隔离原主（防脑裂）。

【表结构】
  clusters 加 vip/failover_mode；node_commands 加 command_type
  新增 switchover_steps（切换步骤级审计）

【Agent 六动作（脚本化、幂等、结构化返回）】
  fence(停MySQL+摘VIP)/unfence(恢复)/promote_master(STOP+RESET SLAVE)
  revert_promote(降回从库)/flip_sync(反转同步)/verify(可写+VIP校验)

【切换编排器 foms/failover.py】
  六步状态机：precheck→fence→promote→flip_sync→verify→converge
  失败 → 已完成步骤逆向回滚 → switchover_logs=failed
  回滚也失败 → rollback_failed + 最高级告警冒泡人工
  防脑裂不变量：fence 先于 promote，fence 失败硬中止绝不 promote

【健康判定器】
  后台循环（默认5s）扫 active 集群，心跳超时/复制异常判定
  防抖：连续 failover_threshold 次命中才判故障（防瞬时抖动误判）
  半自动：只告警（写 switchover_logs + hostwatch_events）不切，等人确认

【切回 + 回滚语义】
  trigger_switchback 反向六步；回滚用 unfence/revert_promote/反转/换回

【部署文档】
  docs/deploy-operations.md：架构/三机部署/keepalived/Agent 环境变量/
  日常运维/失败语义/防脑裂/已知限制

验证：语法全过；编排器↔Agent 六命令类型对齐；六步顺序正确；
防脑裂不变量成立（fence idx=1 < promote idx=2）；注册链完整。

docs/deploy-operations.md

@@ -0,0 +1,173 @@
+# FOMS 部署与运维手册
+
+> 版本：2026-08-26（补全主备切换闭环后）
+> 适用：MySQL 主从 + rsync 文件同步的双机主备，VIP 浮动（1A+2B+3A）
+
+## 一、架构
+
+```
+                管理中心 C（foms 应用 + 管理库 foms）
+                     ▲  HTTP（Bearer token）
+        ┌────────────┴────────────┐
+   主机 A（master）          主机 B（standby）
+   · MySQL 主库              · MySQL 从库（binlog 复制）
+   · 业务 + keepalived       · 业务 + keepalived
+   · foms-agent              · foms-agent
+        └──── VIP（浮动，业务连 VIP）────┘
+```
+
+三个角色：
+- **管理中心 C**：跑 foms（核心 + dbbackup/filesync/hostwatch 模块），存配置与切换记录。
+- **主机 A / 备机 B**：跑业务 + MySQL + keepalived + foms-agent（守护进程）。
+- **VIP**：业务流量入口，平时在 A，故障时漂到 B（keepalived 兜底 + FOMS 主动操作）。
+
+## 二、管理中心 C 部署
+
+```bash
+cd foms
+./build.sh                    # 建 venv、装依赖（含三模块）、生成 DDL/CRUD
+```
+
+**建库**：
+```bash
+mysql -uroot -p < models/mysql.ddl.sql              # 核心表
+mysql -uroot -p < pkgs/dbbackup/models/mysql.ddl.sql    # dbbackup 表
+mysql -uroot -p < pkgs/filesync/models/mysql.ddl.sql    # filesync 表
+mysql -uroot -p < pkgs/hostwatch/models/mysql.ddl.sql   # hostwatch 表
+```
+
+**配置数据库连接**：编辑 `conf/config.json` 的 `databases.foms`（host/user/password/db）。
+
+**启动**：
+```bash
+py3/bin/python app/foms.py    # 监听 conf 里 website.port（默认 9080）
+```
+启动时 `load_foms()` 会自动注册切换编排器并拉起健康判定器后台任务。
+
+## 三、管理中心登记拓扑（部署后必做）
+
+在 FOMS 界面（或 API）依次创建：
+1. **节点**（nodes）：A、B 各一条，填 host、ip、ssh_port、ssh_user、role（master/standby）。
+2. **集群**（clusters）：master_node_id=A、standby_node_id=B、`vip`（如 `192.168.1.100/24`）、
+   `check_interval`（检测间隔秒）、`failover_threshold`（连续故障次数）、`failover_mode=semi`。
+3. **复制配置**（dbbackup_replications）：主库/从库地址、复制账号。
+4. **文件同步**（filesync_directories）：要 rsync 的目录（source→dest）。
+
+## 四、被管主机 A / B 部署
+
+### 4.1 安装依赖
+```bash
+apt install -y mysql-server rsync keepalived ipvsadm python3
+pip3 install psutil          # Agent 采集系统指标需要
+```
+
+### 4.2 配置 keepalived（VIP 浮动）
+A（`/etc/keepalived/keepalived.conf`，MASTER）：
+```
+vrrp_instance VI_1 {
+    state MASTER
+    interface eth0
+    virtual_router_id 51
+    priority 100                 # A 高优先级
+    advert_int 1
+    virtual_ipaddress { 192.168.1.100/24 }
+}
+```
+B 相同，但 `state BACKUP`、`priority 90`。
+keepalived 是**兜底**：机器全宕机时 VIP 自动漂移；"机器活、服务挂"的场景由 FOMS 主动 fence/promote 处理。
+
+### 4.3 配置并启动 Agent
+环境变量（A、B 各自的值不同）：
+```bash
+export FOMS_MANAGEMENT_URL=http://<C的IP>:9080   # 管理中心地址
+export FOMS_NODE_ID=<本节点在nodes表的id>        # A/B 各自
+export FOMS_AGENT_TOKEN=<鉴权token>              # 与管理中心约定
+export FOMS_HEARTBEAT_INTERVAL=10
+# 切换动作专用
+export FOMS_MYSQL_SERVICE=mysql                  # MySQL systemd 服务名
+export FOMS_VIP=192.168.1.100/24                 # VIP（CIDR）
+export FOMS_VIP_IFACE=eth0                       # VIP 所在网卡
+export MYSQL_ROOT_PASSWORD=<root密码>            # promote/verify 用
+# 回滚降从库用（指向"对端"，即本机的复制上游）
+export FOMS_MASTER_HOST=<对端IP>                 # B 上填 A 的 IP；A 上填 B 的 IP
+export FOMS_MASTER_PORT=3306
+export FOMS_REPL_USER=repl
+export FOMS_REPL_PASSWORD=<复制密码>
+```
+启动：
+```bash
+python3 foms-agent.py        # 建议用 systemd 托管（见下）
+```
+
+**systemd 单元**（`/etc/systemd/system/foms-agent.service`）：
+```
+[Unit]
+Description=FOMS Agent
+After=network.target
+[Service]
+EnvironmentFile=/etc/foms/agent.env      # 上面的环境变量写这里
+ExecStart=/usr/bin/python3 /opt/foms/foms-agent.py
+Restart=always
+[Install]
+WantedBy=multi-user.target
+```
+```bash
+systemctl enable --now foms-agent
+```
+
+### 4.4 Agent 需要的权限
+Agent 以能执行 `systemctl`/`ip`/`mysql` 的用户运行（通常 root）。**生产环境应收窄**：
+用 sudoers 白名单只允许这几条命令，而非完整 root。
+
+## 五、日常运维
+
+### 5.1 故障识别（自动）
+健康判定器后台循环（默认 5s 扫一次）：
+- 判定条件：master 心跳超时（> `failover_threshold × check_interval` 秒）**或** 复制状态 error/failed。
+- **防抖**：连续 `failover_threshold` 次命中才判故障（避免瞬时抖动误判）。
+- 判故障 → 写 `switchover_logs(status=detected, trigger=auto_detect)` + 发 `hostwatch_events` 告警。
+- **半自动**：只告警不切，等人确认。
+
+### 5.2 故障切换（人工确认）
+看到告警后，人工确认 → 触发切换：
+- 界面：集群列表 → 切换按钮（`trigger_switchover`）
+- API：`POST /api/trigger_switchover.dspy  {cluster_id}`
+
+编排器异步执行六步（`switchover_steps` 表可查每步状态）：
+```
+1. precheck   目标节点在线+心跳新鲜+复制延迟≤5s（防丢数据）
+2. fence      隔离原主：停 MySQL + 摘 VIP        ← 防脑裂，失败必中止
+3. promote    提升新主：STOP SLAVE + RESET SLAVE ALL（原子）
+4. flip_sync  反转文件同步方向
+5. verify     新主可写测试 + VIP 归属检查（业务层请人工确认）
+6. converge   集群主备互换 + 节点角色互换 + 状态收敛
+```
+
+### 5.3 故障维修与切回
+1. 修好原主 A（硬件/系统/MySQL）。
+2. 把 A 重新挂为 B 的从库（`CHANGE MASTER TO` 指向 B，`START SLAVE`），等复制追平。
+3. 确认 A 健康后，触发切回：`trigger_switchback`（反向六步）。
+
+### 5.4 失败与回滚语义
+- 任何一步失败 → 已完成步骤**逆向回滚**，`switchover_logs=failed`。
+- 回滚也失败 → `switchover_logs=rollback_failed` + 最高级告警，**人工介入**。
+- 回滚动作：unfence（恢复原主）、revert_promote（降回从库）、flip_sync 反转、converge 换回。
+
+## 六、防脑裂三重保险
+1. **fence 先于 promote**：先停旧主 MySQL + 摘 VIP，再立新主。
+2. **fence 失败硬中止**：绝不 promote（否则双主）。
+3. **keepalived VRRP 兜底**：主机器全宕机时 VIP 自动漂移。
+
+## 七、已知限制（诚实声明）
+- **切换时 Agent 延迟**：Agent 轮询命令，切换六步串行走完约需数十秒（运维切换通常可接受）。
+- **verify 只到数据层**：验证"MySQL 可写 + VIP 可达"，业务层是否正常需人工确认。
+- **数据丢失窗口**：precheck 要求复制延迟 ≤5s；fence 到 promote 之间若有未同步 binlog 会丢失（已尽量收窄）。
+- **远程命令安全**：命令下发目前无白名单鉴权（P1 待办），生产应收窄。
+
+## 八、模块清单
+| 模块 | 仓库 | 职责 |
+|---|---|---|
+| foms（核心） | yumoqing/foms | 集群/节点/切换编排/心跳/远程命令 |
+| dbbackup | yumoqing/dbbackup | 数据库备份 + 日志同步（binlog 复制） |
+| filesync | yumoqing/filesync | 运行文件同步（rsync） |
+| hostwatch | yumoqing/hostwatch | 日志监控 + 日志分析 + 主机指标 |

foms/failover.py

@@ -0,0 +1,504 @@
+"""FOMS 切换编排器 — 六步状态机 + 逆向回滚 + 防脑裂不变量。
+
+设计（2026-08-26 定调，1A+2B+3A）：
+  · 流量切换：VIP 浮动（keepalived 兜底）
+  · 触发模式：半自动（健康判定器告警 + 人工确认后切）
+  · 防脑裂：fence（隔离原主）先于 promote（立新主）；fence 失败硬中止
+
+切换六步（failover A→B）：
+  step1 precheck     编排器本地：B 在线 + 心跳新鲜 + 复制延迟 < 阈值（防丢数据）
+  step2 fence        下发 A：停 MySQL + 摘 VIP          ← 防脑裂关键，失败必中止
+  step3 promote      下发 B：STOP SLAVE + RESET SLAVE ALL（脚本化原子执行）
+  step4 flip_sync    DB 层反转 filesync 方向 + 下发 B 确认
+  step5 verify       下发 B：连库读写 + VIP 归属检查（业务层留人工确认）
+  step6 converge     收敛：集群主备互换 + 节点角色互换 + 日志终态
+
+任何一步 failed → 已完成步骤逆向回滚 → switchover_logs=failed；
+回滚也失败 → rollback_failed + 最高级告警冒泡人工。
+
+松耦合约束：
+  · precheck/flip_sync 读 dbbackup_replications / filesync_directories（同库只读）
+  · 告警写 hostwatch_events（与 hostwatch 模块约定的事件结构）
+"""
+import asyncio
+import json
+from datetime import datetime
+from appPublic.uniqueID import getID
+from appPublic.timeUtils import curDateString
+from sqlor.dbpools import get_sor_context
+from ahserver.serverenv import ServerEnv
+
+DBNAME = 'foms'
+
+# 幂等标志：防热重载重复启动健康判定器后台任务
+_failover_registered = False
+
+# ═══ 常量（防丢数据 / 超时 / 轮询）═══
+MAX_LAG_SECONDS = 5          # precheck：复制延迟必须 ≤ 该值才允许切换（R4）
+HEARTBEAT_FRESH_SEC = 60     # precheck：目标节点心跳须在 60s 内
+CMD_TIMEOUT = 90             # 等待 Agent 命令完成的最长秒数
+CMD_POLL_INTERVAL = 2        # 轮询命令结果间隔（切换期加密，缓解 R1 延迟）
+JUDGE_INTERVAL = 5           # 健康判定器扫描间隔（秒）
+
+
+def _now():
+    return curDateString()
+
+
+def _parse_dt(v):
+    """把库里取出的时间值解析成 datetime（兼容 str/datetime）。"""
+    if v is None:
+        return None
+    if isinstance(v, datetime):
+        return v
+    for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%dT%H:%M:%S'):
+        try:
+            return datetime.strptime(str(v)[:19], fmt)
+        except (ValueError, TypeError):
+            continue
+    return None
+
+
+# ═══════════════════════════════════════════
+# 步骤记录（switchover_steps 审计）
+# ═══════════════════════════════════════════
+
+async def _record_step(sor, switchover_id, seq, step_name, status, output='', error=''):
+    """写一条步骤审计记录（先查后建，保证幂等）。不写 None 值（sqlor 不接受）。"""
+    existing = await sor.sqlExe(
+        'SELECT id FROM switchover_steps WHERE switchover_id=${sw}$ AND seq=${seq}$',
+        {'sw': switchover_id, 'seq': seq})
+    if existing:
+        upd = {
+            'id': getattr(existing[0], 'id', ''),
+            'status': status,
+            'output': (output or '')[:2000],
+            'error': (error or '')[:2000],
+        }
+        if status in ('success', 'failed', 'skipped'):
+            upd['completed_at'] = _now()
+        await sor.U('switchover_steps', upd)
+    else:
+        rec = {
+            'id': getID(),
+            'switchover_id': switchover_id,
+            'seq': seq,
+            'step_name': step_name,
+            'status': status,
+            'output': (output or '')[:2000],
+            'error': (error or '')[:2000],
+        }
+        if status == 'running':
+            rec['started_at'] = _now()
+        await sor.C('switchover_steps', rec)
+
+
+# ═══════════════════════════════════════════
+# Agent 命令下发 + 等待（结构化类型）
+# ═══════════════════════════════════════════
+
+async def _send_command(sor, node_id, command_type, command):
+    """向节点下发结构化命令，返回 cmd_id。"""
+    cmd_id = getID()
+    await sor.C('node_commands', {
+        'id': cmd_id,
+        'node_id': node_id,
+        'command_type': command_type,
+        'command': command or command_type,
+        'status': 'pending',
+        'created_at': _now(),
+    })
+    return cmd_id
+
+
+async def _wait_command(sor, cmd_id, timeout=CMD_TIMEOUT):
+    """轮询 node_commands 直到终态，返回 (ok, exit_code, output)。"""
+    deadline = asyncio.get_event_loop().time() + timeout
+    while asyncio.get_event_loop().time() < deadline:
+        rows = await sor.sqlExe(
+            'SELECT status, exit_code, output FROM node_commands WHERE id=${cid}$',
+            {'cid': cmd_id})
+        if rows:
+            r = rows[0]
+            status = getattr(r, 'status', '')
+            if status in ('success', 'failed'):
+                return (status == 'success'), getattr(r, 'exit_code', -1), getattr(r, 'output', '')
+        await asyncio.sleep(CMD_POLL_INTERVAL)
+    return False, -1, f'command timeout after {timeout}s'
+
+
+# ═══════════════════════════════════════════
+# 各步骤实现
+# ═══════════════════════════════════════════
+
+async def _step_precheck(sor, ctx):
+    """step1：本地预检 —— 目标节点在线、心跳新鲜、复制延迟达标（防丢数据）。"""
+    master_id, standby_id = ctx['from_node_id'], ctx['to_node_id']
+    target = await sor.R('nodes', {'id': standby_id})
+    if not target:
+        return False, '目标节点不存在'
+    t = target[0]
+    if getattr(t, 'status', '') != 'online':
+        return False, f"目标节点不在线（status={getattr(t, 'status', '')}）"
+    hb = _parse_dt(getattr(t, 'last_heartbeat', None))
+    if hb is None or (datetime.now() - hb).total_seconds() > HEARTBEAT_FRESH_SEC:
+        return False, '目标节点心跳不新鲜（超过 %ds）' % HEARTBEAT_FRESH_SEC
+    # 复制延迟检查（松耦合读 dbbackup 表）
+    repls = await sor.sqlExe(
+        "SELECT db_name, seconds_behind, sync_status FROM dbbackup_replications "
+        "WHERE cluster_id=${cid}$ AND enabled='1'", {'cid': ctx['cluster_id']})
+    for r in (repls or []):
+        lag = getattr(r, 'seconds_behind', None)
+        if lag is None:
+            lag = 999
+        if int(lag) > MAX_LAG_SECONDS:
+            return False, f"复制延迟过大：{getattr(r, 'db_name', '')} behind {lag}s（阈值 {MAX_LAG_SECONDS}s，防丢数据）"
+    return True, 'precheck passed'
+
+
+async def _step_fence(sor, ctx):
+    """step2：隔离原主（停 MySQL + 摘 VIP）。防脑裂关键，失败必中止。"""
+    cmd_id = await _send_command(sor, ctx['from_node_id'], 'fence', 'fence')
+    ok, code, out = await _wait_command(sor, cmd_id)
+    return ok, out
+
+
+async def _step_promote(sor, ctx):
+    """step3：提升新主（STOP SLAVE + RESET SLAVE ALL，脚本化原子）。"""
+    cmd_id = await _send_command(sor, ctx['to_node_id'], 'promote_master', 'promote_master')
+    ok, code, out = await _wait_command(sor, cmd_id)
+    return ok, out
+
+
+async def _step_flip_sync(sor, ctx):
+    """step4：DB 层反转 filesync 方向 + 下发目标节点确认。"""
+    from_id, to_id = ctx['from_node_id'], ctx['to_node_id']
+    # 取原主/新主的 host，用于方向反转
+    nodes = {}
+    for nid in (from_id, to_id):
+        recs = await sor.R('nodes', {'id': nid})
+        if recs:
+            nodes[nid] = getattr(recs[0], 'host', '')
+    from_host, to_host = nodes.get(from_id, ''), nodes.get(to_id, '')
+    # 反转：source_host 从原主指向新主（A→B 变 B→A）
+    await sor.sqlExe(
+        'UPDATE filesync_directories SET source_host=${th}$, dest_host=${fh}$, '
+        'updated_at=${now}$ WHERE cluster_id=${cid}$ AND source_host=${fh}$',
+        {'th': to_host, 'fh': from_host, 'now': _now(), 'cid': ctx['cluster_id']})
+    cmd_id = await _send_command(sor, to_id, 'flip_sync', 'flip_sync')
+    ok, code, out = await _wait_command(sor, cmd_id)
+    return ok, f'flip db done; agent: {out}'
+
+
+async def _step_verify(sor, ctx):
+    """step5：校验新主可写 + VIP 归属（业务层留人工确认）。"""
+    cmd_id = await _send_command(sor, ctx['to_node_id'], 'verify', 'verify')
+    ok, code, out = await _wait_command(sor, cmd_id)
+    if not ok:
+        return False, out
+    return True, out + '（业务层请人工确认）'
+
+
+async def _step_converge(sor, ctx):
+    """step6：收敛 —— 集群主备互换 + 节点角色互换 + 集群恢复 running。"""
+    from_id, to_id = ctx['from_node_id'], ctx['to_node_id']
+    await sor.U('clusters', {
+        'id': ctx['cluster_id'],
+        'master_node_id': to_id,
+        'standby_node_id': from_id,
+        'status': 'running',
+        'updated_at': _now(),
+    })
+    await sor.U('nodes', {'id': to_id, 'role': 'master', 'updated_at': _now()})
+    await sor.U('nodes', {'id': from_id, 'role': 'standby', 'updated_at': _now()})
+    return True, 'converged: master/standby swapped'
+
+
+# 步骤表（正序）
+STEPS = [
+    ('precheck', _step_precheck),
+    ('fence', _step_fence),
+    ('promote', _step_promote),
+    ('flip_sync', _step_flip_sync),
+    ('verify', _step_verify),
+    ('converge', _step_converge),
+]
+
+
+# ═══════════════════════════════════════════
+# 逆向回滚
+# ═══════════════════════════════════════════
+
+async def _rollback(sor, ctx, completed_steps):
+    """逆向回滚已完成步骤。返回 (ok, detail)。
+
+    回滚语义：
+      fence      → 恢复原主（start MySQL + 还 VIP）
+      promote    → 把新主降回从库（CHANGE MASTER 指向原主）——需原主已恢复
+      flip_sync  → 反转回来
+      verify     → 无需回滚
+      converge   → 主备换回
+    """
+    detail = []
+    ok_all = True
+    for step_name in reversed(completed_steps):
+        if step_name == 'fence':
+            # 恢复原主：start MySQL + 还 VIP（用 unfence 类型，Agent 幂等执行）
+            cmd_id = await _send_command(sor, ctx['from_node_id'], 'unfence', 'unfence')
+            ok, _, out = await _wait_command(sor, cmd_id)
+            detail.append(f'unfence {ctx["from_node_id"]}: {"OK" if ok else "FAILED"} {out[:200]}')
+            ok_all = ok_all and ok
+        elif step_name == 'promote':
+            # 降回从库（revert_promote：重新 CHANGE MASTER 指向原主）
+            cmd_id = await _send_command(sor, ctx['to_node_id'], 'revert_promote', 'revert_promote')
+            ok, _, out = await _wait_command(sor, cmd_id)
+            detail.append(f'revert_promote {ctx["to_node_id"]}: {"OK" if ok else "FAILED"} {out[:200]}')
+            ok_all = ok_all and ok
+        elif step_name == 'flip_sync':
+            # 再反转一次即恢复
+            from_id, to_id = ctx['from_node_id'], ctx['to_node_id']
+            nodes = {}
+            for nid in (from_id, to_id):
+                recs = await sor.R('nodes', {'id': nid})
+                if recs:
+                    nodes[nid] = getattr(recs[0], 'host', '')
+            from_host, to_host = nodes.get(from_id, ''), nodes.get(to_id, '')
+            await sor.sqlExe(
+                'UPDATE filesync_directories SET source_host=${fh}$, dest_host=${th}$, '
+                'updated_at=${now}$ WHERE cluster_id=${cid}$ AND source_host=${th}$',
+                {'fh': from_host, 'th': to_host, 'now': _now(), 'cid': ctx['cluster_id']})
+            detail.append('flip_sync reverted')
+        elif step_name == 'converge':
+            # 主备换回
+            await sor.U('clusters', {
+                'id': ctx['cluster_id'],
+                'master_node_id': ctx['from_node_id'],
+                'standby_node_id': ctx['to_node_id'],
+                'status': 'running',
+                'updated_at': _now(),
+            })
+            await sor.U('nodes', {'id': ctx['from_node_id'], 'role': 'master', 'updated_at': _now()})
+            await sor.U('nodes', {'id': ctx['to_node_id'], 'role': 'standby', 'updated_at': _now()})
+            detail.append('converge reverted')
+        # verify / precheck 无需回滚
+    return ok_all, '; '.join(detail)
+
+
+# ═══════════════════════════════════════════
+# 编排器主流程（六步状态机）
+# ═══════════════════════════════════════════
+
+async def run_switchover(switchover_id):
+    """执行一次切换（在独立 asyncio task 中运行）。"""
+    env = ServerEnv()
+    async with get_sor_context(env, DBNAME) as sor:
+        logs = await sor.R('switchover_logs', {'id': switchover_id})
+        if not logs:
+            return
+        logrec = logs[0]
+        ctx = {
+            'switchover_id': switchover_id,
+            'cluster_id': getattr(logrec, 'cluster_id', ''),
+            'from_node_id': getattr(logrec, 'from_node_id', ''),
+            'to_node_id': getattr(logrec, 'to_node_id', ''),
+        }
+        completed = []
+        failed_at = None
+        fail_detail = ''
+        for seq, (step_name, fn) in enumerate(STEPS, start=1):
+            await _record_step(sor, switchover_id, seq, step_name, 'running')
+            try:
+                ok, detail = await fn(sor, ctx)
+            except Exception as e:
+                ok, detail = False, f'exception: {e}'
+            if ok:
+                await _record_step(sor, switchover_id, seq, step_name, 'success', output=detail)
+                completed.append(step_name)
+            else:
+                await _record_step(sor, switchover_id, seq, step_name, 'failed', error=detail)
+                failed_at = step_name
+                fail_detail = detail
+                break
+
+        if failed_at is None:
+            # 全部成功
+            await sor.U('switchover_logs', {
+                'id': switchover_id, 'status': 'completed',
+                'details': 'all steps success', 'completed_at': _now(),
+            })
+            return
+
+        # 有失败 → 逆向回滚
+        rb_ok, rb_detail = await _rollback(sor, ctx, completed)
+        final_status = 'completed' if rb_ok else 'rollback_failed'
+        # 回滚成功也意味着切换未达成 → 记 failed（原切换失败），但状态收敛
+        await sor.U('switchover_logs', {
+            'id': switchover_id,
+            'status': 'failed' if rb_ok else 'rollback_failed',
+            'details': f'failed at {failed_at}: {fail_detail} | rollback: {rb_detail}',
+            'completed_at': _now(),
+        })
+        # 集群恢复（回滚后主备未变）
+        await sor.U('clusters', {
+            'id': ctx['cluster_id'], 'status': 'running', 'updated_at': _now(),
+        })
+        # 回滚失败 → 最高级告警冒泡人工
+        if not rb_ok:
+            await _emit_alert(sor, ctx['cluster_id'], 'critical',
+                              f'切换回滚失败，需人工介入：{rb_detail}')
+
+
+async def _emit_alert(sor, cluster_id, severity, message):
+    """写一条最高级告警到 hostwatch_events（松耦合，约定结构）。"""
+    try:
+        await sor.C('hostwatch_events', {
+            'id': getID(),
+            'rule_id': '',
+            'node_id': '',
+            'severity': severity,
+            'match_count': 1,
+            'sample_message': f'[FOMS cluster={cluster_id}] {message}'[:2000],
+            'status': 'open',
+            'notified': '0',
+            'first_seen_at': _now(),
+            'last_seen_at': _now(),
+        })
+    except Exception:
+        pass  # 告警失败不阻断主流程
+
+
+# ═══════════════════════════════════════════
+# 对外入口：确认切换 / 健康判定器
+# ═══════════════════════════════════════════
+
+async def confirm_switchover(request, params_kw):
+    """人工确认后启动切换（半自动模式的执行入口）。
+
+    入参：cluster_id（必填），direction（failover/failback，默认 failover）
+    行为：创建 switchover_logs(status=in_progress, trigger=manual_confirm)
+          + 置集群 switching + 起异步任务执行六步。
+    """
+    env = ServerEnv()
+    cluster_id = params_kw.get('cluster_id', '') if hasattr(params_kw, 'get') else ''
+    direction = params_kw.get('direction', 'failover') if hasattr(params_kw, 'get') else 'failover'
+    if not cluster_id:
+        return {'status': 'error', 'message': 'cluster_id required'}
+    log_id = getID()
+    now = _now()
+    async with get_sor_context(env, DBNAME) as sor:
+        clusters = await sor.R('clusters', {'id': cluster_id})
+        if not clusters:
+            return {'status': 'error', 'message': '集群不存在'}
+        cluster = clusters[0]
+        master_id = getattr(cluster, 'master_node_id', '')
+        standby_id = getattr(cluster, 'standby_node_id', '')
+        if direction == 'failback':
+            from_id, to_id = standby_id, master_id
+        else:
+            from_id, to_id = master_id, standby_id
+        await sor.C('switchover_logs', {
+            'id': log_id,
+            'cluster_id': cluster_id,
+            'event_type': direction,
+            'from_node_id': from_id,
+            'to_node_id': to_id,
+            'trigger': 'manual_confirm',
+            'reason': params_kw.get('reason', '人工确认切换') if hasattr(params_kw, 'get') else '人工确认切换',
+            'status': 'in_progress',
+            'started_at': now,
+            'created_at': now,
+        })
+        await sor.U('clusters', {'id': cluster_id, 'status': 'switching', 'updated_at': now})
+    # 起异步任务执行（不阻塞 HTTP 响应）
+    asyncio.get_event_loop().create_task(run_switchover(log_id))
+    return {'status': 'ok', 'message': '切换已启动（异步执行）', 'log_id': log_id}
+
+
+# ═══════════════════════════════════════════
+# 健康判定器（后台循环 + 防抖 + 半自动告警）
+# ═══════════════════════════════════════════
+
+# 内存防抖计数：{cluster_id: 连续故障次数}
+_fault_counters = {}
+
+
+async def health_judge_loop():
+    """后台健康判定：周期扫 active 集群，连续 N 次命中才判故障（防抖）。
+
+    判定条件（任一）：
+      · master 心跳超时（> failover_threshold * check_interval 秒）
+      · 复制状态 error/failed
+    判故障 → 写 switchover_logs(status=detected, trigger=auto_detect) + 告警。
+    半自动：只告警不切，等人 confirm_switchover。
+    """
+    env = ServerEnv()
+    while True:
+        try:
+            async with get_sor_context(env, DBNAME) as sor:
+                clusters = await sor.sqlExe(
+                    "SELECT id, master_node_id, check_interval, failover_threshold, status "
+                    "FROM clusters WHERE status IN ('running','switching')", {})
+                for c in (clusters or []):
+                    cid = getattr(c, 'id', '')
+                    master_id = getattr(c, 'master_node_id', '')
+                    interval = int(getattr(c, 'check_interval', 10) or 10)
+                    threshold = int(getattr(c, 'failover_threshold', 3) or 3)
+                    if not master_id:
+                        continue
+                    # 心跳超时判定
+                    nodes = await sor.R('nodes', {'id': master_id})
+                    heartbeat_lost = False
+                    if nodes:
+                        hb = _parse_dt(getattr(nodes[0], 'last_heartbeat', None))
+                        if hb is None or (datetime.now() - hb).total_seconds() > threshold * interval:
+                            heartbeat_lost = True
+                    # 复制异常判定（松耦合读 dbbackup 表）
+                    repl_bad = False
+                    repls = await sor.sqlExe(
+                        "SELECT id FROM dbbackup_replications WHERE cluster_id=${cid}$ "
+                        "AND enabled='1' AND sync_status IN ('error','failed') LIMIT 1",
+                        {'cid': cid})
+                    if repls:
+                        repl_bad = True
+                    faulty = heartbeat_lost or repl_bad
+                    if faulty:
+                        _fault_counters[cid] = _fault_counters.get(cid, 0) + 1
+                        if _fault_counters[cid] >= threshold:
+                            # 达到阈值 → 判故障（防抖通过）
+                            reason = ('心跳超时' if heartbeat_lost else '') + \
+                                     ('+' if heartbeat_lost and repl_bad else '') + \
+                                     ('复制异常' if repl_bad else '')
+                            await sor.C('switchover_logs', {
+                                'id': getID(),
+                                'cluster_id': cid,
+                                'event_type': 'fault_detected',
+                                'from_node_id': master_id,
+                                'to_node_id': '',
+                                'trigger': 'auto_detect',
+                                'reason': reason,
+                                'status': 'detected',
+                                'created_at': _now(),
+                            })
+                            await _emit_alert(sor, cid, 'critical',
+                                              f'主节点故障判定：{reason}（已连续{threshold}次），请人工确认切换')
+                            # 重置计数，避免重复告警刷屏（下次再连续 threshold 次才再报）
+                            _fault_counters[cid] = 0
+                    else:
+                        _fault_counters[cid] = 0
+        except Exception:
+            pass  # 判定器自身异常不退出循环
+        await asyncio.sleep(JUDGE_INTERVAL)
+
+
+def register_failover():
+    """注册切换编排器到 ServerEnv + 启动健康判定器后台任务。幂等（防热重载重复起任务）。"""
+    global _failover_registered
+    env = ServerEnv()
+    env.confirm_switchover = confirm_switchover
+    if _failover_registered:
+        return True
+    _failover_registered = True
+    # 健康判定器后台任务（由 webapp 的 startup 机制拉起）
+    from ahserver.configuredServer import add_startup
+    add_startup(health_judge_loop())
+    return True

foms/init.py

@@ -126,60 +126,31 @@ async def agent_heartbeat(request, params_kw):
 # ═══════════════════════════════════════════
 # 故障切换编排
 # ═══════════════════════════════════════════
+# 2026-08-26 起：切换不再是"只写日志"，委托给 failover 编排器执行六步状态机
+# （precheck → fence → promote → flip_sync → verify → converge，失败逆向回滚）。
+# 触发模式为半自动（1A+2B+3A）：健康判定器告警 → 人工确认 → 本入口执行。
 
 async def trigger_switchover(request, params_kw):
-    """触发故障切换 A→B"""
-    env = ServerEnv()
-    cluster_id = _get(params_kw, 'cluster_id')
-    reason = _get(params_kw, 'reason', '手动触发')
-    log_id = getID()
-    now = curDateString()
-    async with get_sor_context(env, DBNAME) as sor:
-        clusters = await sor.R('clusters', {'id': cluster_id})
-        if not clusters:
-            return {'status': 'error', 'message': '集群不存在'}
-        cluster = clusters[0]
-        await sor.C('switchover_logs', {
-            'id': log_id,
-            'cluster_id': cluster_id,
-            'event_type': 'failover',
-            'from_node_id': getattr(cluster, 'master_node_id', ''),
-            'to_node_id': getattr(cluster, 'standby_node_id', ''),
-            'trigger': 'manual',
-            'reason': reason,
-            'status': 'in_progress',
-            'started_at': now,
-            'created_at': now,
-        })
-        await sor.U('clusters', {'id': cluster_id, 'status': 'switching', 'updated_at': now})
-    return {'status': 'ok', 'message': '切换流程已触发', 'log_id': log_id}
+    """触发故障切换 A→B（委托编排器执行六步状态机）"""
+    from foms.failover import confirm_switchover
+    ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
+    ns['direction'] = 'failover'
+    result = await confirm_switchover(request, ns)
+    if result.get('status') == 'error':
+        return {'widgettype': 'Message', 'options': {'title': '失败', 'message': result.get('message', ''), 'type': 'error'}}
+    return {'widgettype': 'Message', 'options': {'title': '成功', 'message': f"切换已启动（异步执行），事件ID：{result.get('log_id', '')}", 'type': 'success'}}
 
 
 async def trigger_switchback(request, params_kw):
-    """触发切回 B→A"""
-    env = ServerEnv()
-    cluster_id = _get(params_kw, 'cluster_id')
-    log_id = getID()
-    now = curDateString()
-    async with get_sor_context(env, DBNAME) as sor:
-        clusters = await sor.R('clusters', {'id': cluster_id})
-        if not clusters:
-            return {'status': 'error', 'message': '集群不存在'}
-        cluster = clusters[0]
-        await sor.C('switchover_logs', {
-            'id': log_id,
-            'cluster_id': cluster_id,
-            'event_type': 'failback',
-            'from_node_id': getattr(cluster, 'standby_node_id', ''),
-            'to_node_id': getattr(cluster, 'master_node_id', ''),
-            'trigger': 'manual',
-            'reason': '手动切回',
-            'status': 'in_progress',
-            'started_at': now,
-            'created_at': now,
-        })
-        await sor.U('clusters', {'id': cluster_id, 'status': 'switching', 'updated_at': now})
-    return {'status': 'ok', 'message': '切回流程已触发', 'log_id': log_id}
+    """触发切回 B→A（委托编排器，反向六步）"""
+    from foms.failover import confirm_switchover
+    ns = dict(params_kw) if hasattr(params_kw, 'items') else {}
+    ns['direction'] = 'failback'
+    ns.setdefault('reason', '手动切回')
+    result = await confirm_switchover(request, ns)
+    if result.get('status') == 'error':
+        return {'widgettype': 'Message', 'options': {'title': '失败', 'message': result.get('message', ''), 'type': 'error'}}
+    return {'widgettype': 'Message', 'options': {'title': '成功', 'message': f"切回已启动（异步执行），事件ID：{result.get('log_id', '')}", 'type': 'success'}}
 
 
 # ═══════════════════════════════════════════
@@ -245,18 +216,20 @@ async def exec_remote_command(request, params_kw):
 
 
 async def agent_poll_commands(request, params_kw):
-    """Agent 轮询获取待执行命令"""
+    """Agent 轮询获取待执行命令（含 command_type，供 Agent 按类型路由）"""
     env = ServerEnv()
     node_id = _get(params_kw, 'node_id')
     async with get_sor_context(env, DBNAME) as sor:
         rows = await sor.sqlExe(
-            "SELECT id, command FROM node_commands WHERE node_id=${nid}$ AND status='pending' "
+            "SELECT id, command_type, command FROM node_commands WHERE node_id=${nid}$ AND status='pending' "
             "ORDER BY created_at ASC LIMIT 5",
             {'nid': node_id})
         # 取到的命令标记为 running，防止重复下发
         for r in (rows or []):
             await sor.U('node_commands', {'id': getattr(r, 'id', ''), 'status': 'running'})
-    return {'commands': [{'id': getattr(r, 'id', ''), 'command': getattr(r, 'command', '')}
+    return {'commands': [{'id': getattr(r, 'id', ''),
+                          'command_type': getattr(r, 'command_type', 'shell') or 'shell',
+                          'command': getattr(r, 'command', '')}
                          for r in (rows or [])]}
 
 
@@ -306,4 +279,7 @@ def load_foms():
     env.exec_remote_command = exec_remote_command
     env.agent_poll_commands = agent_poll_commands
     env.agent_report_command_result = agent_report_command_result
+    # 切换编排器 + 健康判定器（后台任务）
+    from foms.failover import register_failover
+    register_failover()
     return True

models/clusters.json

@@ -15,6 +15,8 @@
         {"name": "standby_node_id", "title": "备机节点", "type": "str", "length": 32},
         {"name": "check_interval", "title": "检测间隔(秒)", "type": "int", "nullable": "no", "default": "10"},
         {"name": "failover_threshold", "title": "故障判定阈值(次)", "type": "int", "nullable": "no", "default": "3"},
+        {"name": "vip", "title": "VIP地址(CIDR)", "type": "str", "length": 64},
+        {"name": "failover_mode", "title": "触发模式", "type": "str", "length": 16, "nullable": "no", "default": "semi"},
         {"name": "status", "title": "集群状态", "type": "str", "length": 32, "nullable": "no", "default": "stopped"},
         {"name": "created_at", "title": "创建时间", "type": "datetime"},
         {"name": "updated_at", "title": "更新时间", "type": "datetime"}
@@ -24,6 +26,7 @@
     ],
     "codes": [
         {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='cluster_status'"},
+        {"field": "failover_mode", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='failover_mode'"},
         {"field": "master_node_id", "table": "nodes", "valuefield": "id", "textfield": "host"},
         {"field": "standby_node_id", "table": "nodes", "valuefield": "id", "textfield": "host"}
     ]

models/node_commands.json

@@ -10,6 +10,7 @@
     "fields": [
         {"name": "id", "title": "命令ID", "type": "str", "length": 32, "nullable": "no"},
         {"name": "node_id", "title": "节点ID", "type": "str", "length": 32, "nullable": "no"},
+        {"name": "command_type", "title": "命令类型", "type": "str", "length": 32, "nullable": "no", "default": "shell"},
         {"name": "command", "title": "命令内容", "type": "text", "nullable": "no"},
         {"name": "status", "title": "状态", "type": "str", "length": 16, "nullable": "no", "default": "pending"},
         {"name": "exit_code", "title": "退出码", "type": "int"},
@@ -22,6 +23,7 @@
         {"name": "idx_ncmd_status", "idxtype": "index", "idxfields": ["status"]}
     ],
     "codes": [
-        {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='cmd_status'"}
+        {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='cmd_status'"},
+        {"field": "command_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='cmd_type'"}
     ]
 }

models/switchover_steps.json

@@ -0,0 +1,29 @@
+{
+    "summary": [
+        {
+            "name": "switchover_steps",
+            "title": "切换步骤明细",
+            "primary": ["id"],
+            "catelog": "entity"
+        }
+    ],
+    "fields": [
+        {"name": "id", "title": "步骤ID", "type": "str", "length": 32, "nullable": "no"},
+        {"name": "switchover_id", "title": "切换事件", "type": "str", "length": 32, "nullable": "no"},
+        {"name": "seq", "title": "步骤序号", "type": "int", "nullable": "no"},
+        {"name": "step_name", "title": "步骤名称", "type": "str", "length": 32, "nullable": "no"},
+        {"name": "status", "title": "步骤状态", "type": "str", "length": 16, "nullable": "no", "default": "pending"},
+        {"name": "output", "title": "执行输出", "type": "text"},
+        {"name": "error", "title": "错误信息", "type": "text"},
+        {"name": "started_at", "title": "开始时间", "type": "datetime"},
+        {"name": "completed_at", "title": "完成时间", "type": "datetime"}
+    ],
+    "indexes": [
+        {"name": "idx_ss_switchover", "idxtype": "index", "idxfields": ["switchover_id"]},
+        {"name": "idx_ss_seq", "idxtype": "index", "idxfields": ["switchover_id", "seq"]}
+    ],
+    "codes": [
+        {"field": "status", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='step_status'"},
+        {"field": "switchover_id", "table": "switchover_logs", "valuefield": "id", "textfield": "event_type"}
+    ]
+}

scripts/foms-agent.py

@@ -34,6 +34,12 @@ NODE_ID = os.environ.get('FOMS_NODE_ID', '')
 AGENT_TOKEN = os.environ.get('FOMS_AGENT_TOKEN', '')
 HEARTBEAT_INTERVAL = int(os.environ.get('FOMS_HEARTBEAT_INTERVAL', '10'))
 
+# ═══ Failover configuration（切换动作专用）═══
+MYSQL_SERVICE = os.environ.get('FOMS_MYSQL_SERVICE', 'mysql')
+VIP_ADDR = os.environ.get('FOMS_VIP', '')              # 如 192.168.1.100/24
+VIP_IFACE = os.environ.get('FOMS_VIP_IFACE', 'eth0')
+MYSQL_ROOT_PW = os.environ.get('MYSQL_ROOT_PASSWORD', '')
+
 # ═══ Logging ═══
 logging.basicConfig(
     level=logging.INFO,
@@ -321,26 +327,176 @@ class FomsAgent:
         else:
             log.warning(f"Metrics report failed: {result}")
 
+    # ═══ Failover Actions（切换动作，脚本化 + 幂等）═══
+    def _run(self, argv, timeout=30):
+        """执行单条命令，返回 (exit_code, output)。argv 为列表，不走 shell。"""
+        try:
+            proc = subprocess.run(argv, capture_output=True, text=True, timeout=timeout)
+            output = (proc.stdout + proc.stderr).strip()[:2000]
+            return proc.returncode, output
+        except subprocess.TimeoutExpired:
+            return -1, f'TIMEOUT ({timeout}s): {argv[0]}'
+        except Exception as e:
+            return -1, str(e)
+
+    def action_fence(self):
+        """隔离本机（防脑裂）：停 MySQL + 摘 VIP。幂等——已停/已摘时不报错。"""
+        results = []
+        # 1) 停 MySQL 服务（systemd）。已停止时 systemctl stop 幂等返回 0。
+        code, out = self._run(['systemctl', 'stop', MYSQL_SERVICE], timeout=60)
+        results.append({'step': 'stop_mysql', 'exit_code': code, 'output': out})
+        # 2) 摘 VIP（ip addr del）。已不存在时返回非 0，视为已摘除（幂等成功）。
+        if VIP_ADDR:
+            code, out = self._run(['ip', 'addr', 'del', VIP_ADDR, 'dev', VIP_IFACE])
+            if code != 0 and 'Cannot find address' in out:
+                code, out = 0, 'VIP already removed (idempotent)'
+            results.append({'step': 'remove_vip', 'exit_code': code, 'output': out})
+        ok = all(r['exit_code'] == 0 for r in results)
+        log.info(f"fence: {'OK' if ok else 'FAILED'} {results}")
+        return {'ok': ok, 'actions': results}
+
+    def action_promote_master(self):
+        """提升本机为新主：STOP SLAVE + RESET SLAVE ALL。幂等——已是主库视为成功。"""
+        env_sql = ['mysql', '-uroot']
+        if MYSQL_ROOT_PW:
+            env_sql.append(f'-p{MYSQL_ROOT_PW}')
+        # 检查当前是否仍是 slave；已是主库（无 SLAVE STATUS）则直接成功（幂等）
+        code, out = self._run(env_sql + ['-e', 'SHOW SLAVE STATUS\\G'])
+        if code == 0 and 'Slave_IO_State' not in out:
+            log.info("promote_master: already master (no slave status), idempotent OK")
+            return {'ok': True, 'actions': [{'step': 'already_master', 'exit_code': 0, 'output': out[:300]}]}
+        # 原子执行：STOP SLAVE; RESET SLAVE ALL;（单条 -e 保证原子）
+        promote_sql = 'STOP SLAVE; RESET SLAVE ALL;'
+        code, out = self._run(env_sql + ['-e', promote_sql], timeout=60)
+        results = [{'step': 'stop_reset_slave', 'exit_code': code, 'output': out}]
+        # 校验：确认不再有 slave 状态
+        vcode, vout = self._run(env_sql + ['-e', 'SHOW SLAVE STATUS\\G'])
+        promoted = (vcode == 0 and 'Slave_IO_State' not in vout)
+        results.append({'step': 'verify_no_slave', 'exit_code': 0 if promoted else 1,
+                        'output': ('promoted' if promoted else vout[:300])})
+        ok = (code == 0 and promoted)
+        log.info(f"promote_master: {'OK' if ok else 'FAILED'} {results}")
+        return {'ok': ok, 'actions': results}
+
+    def action_flip_sync(self):
+        """反转/暂停文件同步方向。通过通知管理中心翻转 filesync 配置方向实现。
+        本动作只做「本机侧的确认与暂停」：返回当前同步配置状态，由编排器在 DB 层反转。"""
+        results = [{'step': 'flip_sync_ack', 'exit_code': 0,
+                    'output': f'node={self.node_id} ready_for_flip'}]
+        log.info(f"flip_sync: ACK {results}")
+        return {'ok': True, 'actions': results}
+
+    def action_unfence(self):
+        """恢复本机（回滚用）：启动 MySQL + 还 VIP。幂等——已启动/已挂时不报错。"""
+        results = []
+        # 1) 启动 MySQL 服务（systemd）。已运行时 systemctl start 幂等返回 0。
+        code, out = self._run(['systemctl', 'start', MYSQL_SERVICE], timeout=60)
+        results.append({'step': 'start_mysql', 'exit_code': code, 'output': out})
+        # 2) 挂回 VIP（ip addr add）。已存在时返回非 0，视为已挂载（幂等成功）。
+        if VIP_ADDR:
+            code, out = self._run(['ip', 'addr', 'add', VIP_ADDR, 'dev', VIP_IFACE])
+            if code != 0 and 'File exists' in out:
+                code, out = 0, 'VIP already present (idempotent)'
+            results.append({'step': 'add_vip', 'exit_code': code, 'output': out})
+        ok = all(r['exit_code'] == 0 for r in results)
+        log.info(f"unfence: {'OK' if ok else 'FAILED'} {results}")
+        return {'ok': ok, 'actions': results}
+
+    def action_revert_promote(self):
+        """降回从库（回滚用）：重新 CHANGE MASTER 指向原主。
+        需要原主连接信息——从本机复制配置读取（若原主已恢复）。
+        幂等：若本机已是 slave 则直接成功。"""
+        env_sql = ['mysql', '-uroot']
+        if MYSQL_ROOT_PW:
+            env_sql.append(f'-p{MYSQL_ROOT_PW}')
+        # 已是 slave → 幂等成功
+        code, out = self._run(env_sql + ['-e', 'SHOW SLAVE STATUS\\G'])
+        if code == 0 and 'Slave_IO_State' in out:
+            log.info("revert_promote: already slave, idempotent OK")
+            return {'ok': True, 'actions': [{'step': 'already_slave', 'exit_code': 0, 'output': out[:300]}]}
+        # 读取本机复制配置（CHANGE MASTER 需要原主地址）。
+        # 优先从环境变量 FOMS_MASTER_HOST/PORT 获取（部署时配置），否则无法回滚。
+        master_host = os.environ.get('FOMS_MASTER_HOST', '')
+        master_port = os.environ.get('FOMS_MASTER_PORT', '3306')
+        repl_user = os.environ.get('FOMS_REPL_USER', 'repl')
+        repl_pass = os.environ.get('FOMS_REPL_PASSWORD', '')
+        if not master_host:
+            return {'ok': False, 'actions': [{'step': 'revert_promote', 'exit_code': 1,
+                    'output': 'FOMS_MASTER_HOST not set, cannot revert'}]}
+        change_sql = (f"CHANGE MASTER TO MASTER_HOST='{master_host}', MASTER_PORT={master_port}, "
+                      f"MASTER_USER='{repl_user}', MASTER_PASSWORD='{repl_pass}', MASTER_AUTO_POSITION=1; "
+                      f"START SLAVE;")
+        code, out = self._run(env_sql + ['-e', change_sql], timeout=60)
+        results = [{'step': 'change_master_start_slave', 'exit_code': code, 'output': out}]
+        ok = (code == 0)
+        log.info(f"revert_promote: {'OK' if ok else 'FAILED'} {results}")
+        return {'ok': ok, 'actions': results}
+
+    def action_verify(self):
+        """校验本机可写 + VIP 归属（切换后验证）。业务层留人工确认。"""
+        results = []
+        env_sql = ['mysql', '-uroot']
+        if MYSQL_ROOT_PW:
+            env_sql.append(f'-p{MYSQL_ROOT_PW}')
+        # 1) MySQL 可写测试：建临时表写入再删除
+        test_sql = ("CREATE TABLE IF NOT EXISTS _foms_verify (id INT); "
+                    "INSERT INTO _foms_verify VALUES (1); "
+                    "DROP TABLE _foms_verify;")
+        code, out = self._run(env_sql + ['-e', test_sql], timeout=30)
+        results.append({'step': 'mysql_write_test', 'exit_code': code, 'output': out[:300]})
+        # 2) VIP 归属检查（若配置了 VIP）
+        if VIP_ADDR:
+            vip_ip = VIP_ADDR.split('/')[0]
+            code, out = self._run(['ip', 'addr', 'show', VIP_IFACE])
+            vip_here = (code == 0 and vip_ip in out)
+            results.append({'step': 'vip_check', 'exit_code': 0 if vip_here else 1,
+                            'output': f'vip={vip_ip} on {VIP_IFACE}: {"YES" if vip_here else "NO"}'})
+        ok = all(r['exit_code'] == 0 for r in results)
+        log.info(f"verify: {'OK' if ok else 'FAILED'} {results}")
+        return {'ok': ok, 'actions': results}
+
+    # ═══ Command dispatcher（按 command_type 路由）═══
+    def dispatch_command(self, cmd_id, command_type, command):
+        """根据 command_type 分发执行，上报结构化结果。"""
+        if command_type == 'fence':
+            res = self.action_fence()
+        elif command_type == 'unfence':
+            res = self.action_unfence()
+        elif command_type == 'promote_master':
+            res = self.action_promote_master()
+        elif command_type == 'revert_promote':
+            res = self.action_revert_promote()
+        elif command_type == 'flip_sync':
+            res = self.action_flip_sync()
+        elif command_type == 'verify':
+            res = self.action_verify()
+        else:
+            # 普通 shell 命令（保留原有行为）
+            proc = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30)
+            res = {'ok': proc.returncode == 0,
+                   'actions': [{'step': 'shell', 'exit_code': proc.returncode,
+                                'output': (proc.stdout + proc.stderr)[:2000]}]}
+        exit_code = 0 if res.get('ok') else 1
+        output = json.dumps(res, ensure_ascii=False)[:2000]
+        self._api_post('agent_report_command_result.dspy', {
+            'node_id': self.node_id, 'cmd_id': cmd_id,
+            'output': output, 'exit_code': exit_code,
+        })
+        log.info(f"Command [{cmd_id}] type={command_type} done: exit={exit_code}")
+
     # ═══ Remote Command Execution ═══
     def poll_and_execute_commands(self):
-        """Poll management for pending commands, execute, report results"""
+        """Poll management for pending commands, dispatch by type, report results"""
         result = self._api_get('agent_poll_commands.dspy?node_id=' + self.node_id)
         if not result or not result.get('commands'):
             return
         for cmd in result['commands']:
             cmd_id = cmd.get('id', '')
+            command_type = cmd.get('command_type', 'shell')
             command = cmd.get('command', '')
-            log.info(f"Executing command [{cmd_id}]: {command}")
+            log.info(f"Executing command [{cmd_id}] type={command_type}")
             try:
-                proc = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30)
-                output = (proc.stdout + proc.stderr)[:2000]
-                self._api_post('agent_report_command_result.dspy', {
-                    'node_id': self.node_id,
-                    'cmd_id': cmd_id,
-                    'output': output,
-                    'exit_code': proc.returncode,
-                })
-                log.info(f"Command [{cmd_id}] done: exit={proc.returncode}")
+                self.dispatch_command(cmd_id, command_type, command)
             except subprocess.TimeoutExpired:
                 self._api_post('agent_report_command_result.dspy', {
                     'node_id': self.node_id, 'cmd_id': cmd_id,