---
phase: 03-cluster
plan: 03
type: execute
wave: 3
depends_on:
  - 03-01
  - 03-02
files_modified:
  - src/cli.rs
  - src/main.rs
  - src/cluster/mod.rs
  - src/cluster/deploy.rs
autonomous: false
requirements:
  - CLUS-01
  - CLUS-02
  - QUAL-01
user_setup:
  - service: "目标节点 SSH 环境"
    why: "端到端 manual 验证需要两台可 SSH 登录、装有 sudo / ss / df 的 Linux 节点 + DM 安装包"
    env_vars: []
    dashboard_config:
      - task: "准备两台 Linux 测试节点（VM 或 Docker sshd 容器），配置 sudo 免密、SSH 密钥登录、达梦安装包路径"
        location: "本地或测试环境"
must_haves:
  truths:
    - "用户运行 `dm-installer cluster deploy --config <toml>` CLI 子命令可被解析"
    - "未提供 --config 时 CLI 解析失败并报错（per D-05）"
    - "cluster::run 编排完整序列：load → preflight → 并发 SFTP+install → dminit → 推 4 个 INI → 启主 → wait_tcp_ready(primary) → 主 disql SQL → 启备 → wait_tcp_ready(standby) → 备 disql SQL → 启 dmwatcher"
    - "有序启动可从 tracing 日志中观察到：`[node:primary]` 健康确认日志在 `[node:standby]` 启动日志之前（per CLUS-02 SC3）"
    - "预检查失败时 cluster::run 在执行任何安装操作前 bail（per D-08）"
  artifacts:
    - path: "src/cli.rs"
      provides: "Commands::Cluster(ClusterArgs) variant + ClusterArgs + ClusterSubcommand::Deploy + ClusterDeployArgs"
      contains: "Cluster(ClusterArgs)"
    - path: "src/main.rs"
      provides: "Commands::Cluster 分发分支调用 cluster::run"
      contains: "Commands::Cluster"
    - path: "src/cluster/mod.rs"
      provides: "cluster::run 真实实现（替换 Plan 01 的 unimplemented stub），签名 pub async fn run(args: &cli::ClusterDeployArgs) -> Result<()>"
      contains: "pub async fn run"
    - path: "src/cluster/deploy.rs"
      provides: "build_dminit_args / distribute_configs / configure_database_role / start_dmserver_mount / start_dmwatcher / deploy_node 等编排函数"
      contains: "pub async fn deploy_node"
  key_links:
    - from: "src/main.rs"
      to: "src/cluster/mod.rs"
      via: "match Commands::Cluster(args) => cluster::run(deploy_args).await"
      pattern: "Commands::Cluster"
    - from: "src/cluster/mod.rs"
      to: "src/cluster/deploy.rs"
      via: "deploy::deploy_node(...) 调用链"
      pattern: "deploy::"
    - from: "src/cluster/mod.rs"
      to: "src/cluster/health.rs"
      via: "health::wait_tcp_ready(&primary.host, primary.port, 60).await"
      pattern: "wait_tcp_ready"
---

<objective>
将 Plan 01（配置层 + 模板）和 Plan 02（SSH + preflight + health）整合为完整的 `dm-installer cluster deploy --config X.toml` 端到端流程，使 Phase 3 三项需求（CLUS-01 / CLUS-02 / QUAL-01）真实可用。

Purpose: 这是 Phase 3 的垂直切片闭合点。本计划完成后，一个用户输入（cluster.toml）+ 一个 CLI 命令 → 双节点完整部署，期间走完预检查、SFTP 推送、远端 dminit、配置分发、有序启动（主节点 TCP 健康后才启动备节点）、dmwatcher 启动。

Output:
- `cli.rs` 新增 `Cluster(ClusterArgs)` variant + 嵌套 `cluster deploy --config` 子命令 + CLI 解析测试
- `main.rs` 新增 dispatch 分支
- `cluster/deploy.rs` 节点级安装编排（SFTP 包 → 远端 DMInstall.bin → dminit → 推 INI → 启动 → disql SQL → dmwatcher）
- `cluster/mod.rs` `run` 顶层编排器（load_config → preflight → 主备并发安装 → 推配置 → 有序启动 → 启 dmwatcher）
- manual checkpoint 验证端到端日志中主节点健康确认日志在备节点启动日志之前（CLUS-02 SC3）
</objective>

<execution_context>
@$HOME/.claude/get-shit-done/workflows/execute-plan.md
@$HOME/.claude/get-shit-done/templates/summary.md
</execution_context>

<context>
@.planning/PROJECT.md
@.planning/ROADMAP.md
@.planning/STATE.md
@.planning/phases/03-cluster/03-CONTEXT.md
@.planning/phases/03-cluster/03-RESEARCH.md
@.planning/phases/03-cluster/03-PATTERNS.md
@.planning/phases/03-cluster/03-01-SUMMARY.md
@.planning/phases/03-cluster/03-02-SUMMARY.md
@src/cli.rs
@src/main.rs
@src/install/mod.rs
@src/install/init.rs

<interfaces>
<!-- Plan 01 输出的关键 API（来自 03-01-SUMMARY.md） -->
From src/config/cluster.rs:
- pub fn load_cluster_config(path: &Path) -> anyhow::Result&lt;ClusterConfig&gt;
- pub struct ClusterConfig { pub cluster: ClusterSection }
- pub struct ClusterSection { pub installer_package: PathBuf, pub oguid: u32, pub nodes: Vec&lt;NodeConfig&gt; }
- pub struct NodeConfig (字段: role, host, port, instance_name, install_path, data_path, mal_port, dw_port, inst_dw_port, page_size, charset, case_sensitive, extent_size, ssh)
- pub enum NodeRole { Primary, Standby }

From src/cluster/templates/:
- pub fn generate_dm_ini_cluster_suffix(node: &NodeConfig) -> String
- pub fn generate_dmmal_ini(nodes: &[NodeConfig]) -> String
- pub fn generate_dmarch_ini(node: &NodeConfig, peer_instance: &str) -> String
- pub fn generate_dmwatcher_ini(node: &NodeConfig, oguid: u32) -> String

<!-- Plan 02 输出的关键 API（来自 03-02-SUMMARY.md） -->
From src/cluster/ssh.rs:
- pub trait CommandRunner: Send + Sync — exec(cmd) -> Result&lt;(Vec&lt;u8&gt;, u32), SshError&gt;; sftp_write(path, bytes) -> Result&lt;(), SshError&gt;
- pub struct SshSession 实现 CommandRunner，构造：pub async fn connect(host, port, user, creds) -> Result&lt;Self, SshError&gt;
- pub struct MockRunner 实现 CommandRunner（Plan 03 测试需要 sftp_log 和 exec_log 访问，若 Plan 02 未暴露则 Task 2 扩展）

From src/cluster/preflight.rs:
- pub async fn check_node(node: &NodeConfig, runner: &dyn CommandRunner) -> Result&lt;()&gt;
- pub async fn preflight_all_nodes(nodes_and_runners: Vec&lt;(&NodeConfig, Arc&lt;dyn CommandRunner&gt;)&gt;) -> Result&lt;()&gt;

From src/cluster/health.rs:
- pub async fn wait_tcp_ready(host: &str, port: u16, max_secs: u64) -> Result&lt;()&gt;

<!-- Phase 2 复用 (CLI 模式) -->
From src/cli.rs:
- #[derive(Subcommand)] pub enum Commands { Install(InstallArgs), Validate(ValidateArgs), Completions { shell } }
- #[derive(clap::Args)] pub struct ValidateArgs { #[arg(long)] pub config: PathBuf } — 必填 PathBuf 无 Option（D-05 同模式）

From src/main.rs:
- match &cli_args.command { Commands::Install(args) => install::run(args).await, ... }
</interfaces>
</context>

<tasks>

<task type="auto" tdd="true">
  <name>Task 1: CLI 新增 cluster deploy 子命令 + main.rs dispatch</name>
  <files>
    src/cli.rs,
    src/main.rs,
    Cargo.toml
  </files>
  <read_first>
    src/cli.rs,
    src/main.rs,
    .planning/phases/03-cluster/03-01-SUMMARY.md,
    .planning/phases/03-cluster/03-02-SUMMARY.md
  </read_first>
  <behavior>
    - Test 1 (cluster deploy 带 --config 解析成功): Cli::try_parse_from(["dm-installer", "cluster", "deploy", "--config", "/etc/cluster.toml"]) 返回 Ok；模式匹配 Commands::Cluster(args) 后 args.command 为 ClusterSubcommand::Deploy(deploy_args)，deploy_args.config == PathBuf::from("/etc/cluster.toml")
    - Test 2 (cluster deploy 不带 --config 失败): Cli::try_parse_from(["dm-installer", "cluster", "deploy"]) 返回 Err（per D-05）
    - Test 3 (cluster 不带子命令失败): Cli::try_parse_from(["dm-installer", "cluster"]) 返回 Err
    - Test 4 (不影响 Phase 2 测试): 现有 test_install_args_defaults、test_validate_args_config 等仍通过
  </behavior>
  <action>
1. 修改 src/cli.rs：在 Commands 枚举中（Validate(ValidateArgs) 之后）新增 doc 注释 `/// 集群部署子命令` 和一行 variant `Cluster(ClusterArgs),`。在 ValidateArgs 结构体之后追加：(a) `#[derive(clap::Args)] pub struct ClusterArgs { #[command(subcommand)] pub command: ClusterSubcommand }`；(b) `#[derive(Subcommand)] pub enum ClusterSubcommand { /// 部署主备集群 Deploy(ClusterDeployArgs) }`；(c) `#[derive(clap::Args)] pub struct ClusterDeployArgs { /// 集群 TOML 配置文件路径（必填） #[arg(long)] pub config: PathBuf }`。

2. 在 cli.rs 的 mod tests 中追加 3 个新测试，参照 test_validate_args_config / test_validate_requires_config 模式（cli.rs lines 96-114）。测试函数名严格用：test_cluster_deploy_args_config、test_cluster_deploy_requires_config、test_cluster_requires_subcommand。

3. 修改 src/main.rs：在现有 match &cli_args.command 块中 cli::Commands::Validate(args) 之后新增分支 `cli::Commands::Cluster(args) => match &args.command { cli::ClusterSubcommand::Deploy(deploy_args) => cluster::run(deploy_args).await }`。Plan 01 已让 main.rs 含 mod cluster；本 Plan 不再重复 mod 声明。

4. 修改 Cargo.toml [dev-dependencies]：新增 `tracing-test = "0.2"`（Task 3 cluster::run 集成测试需要 `tracing_test::traced_test` 宏捕获 tracing event 顺序以验证 CLUS-02 SC3 主节点健康确认在备节点启动之前）。若已存在则跳过。如 [dev-dependencies] 段不存在则新建。

5. 不修改任何 Phase 2 已有 args 字段、已有测试、已有 dispatch 分支（仅新增）。

6. 所有新增的 struct / enum / fn 函数体保持 < 40 行（CLAUDE.md universal rule）。
  </action>
  <verify>
    <automated>cargo test --lib cli --no-fail-fast 2>&amp;1 | tee /tmp/p3-03-t1.log; grep -E "test result:" /tmp/p3-03-t1.log; cargo build --bin dm-installer 2>&amp;1 | tail -5</automated>
  </verify>
  <acceptance_criteria>
    - `cargo test --lib cli` exits 0（Phase 2 现有测试 + 3 个新测试全部通过）
    - `cargo build --bin dm-installer` exits 0
    - `grep -q 'Cluster(ClusterArgs)' src/cli.rs` 成功
    - `grep -q '^pub struct ClusterDeployArgs' src/cli.rs` 成功
    - `grep -q '^pub enum ClusterSubcommand' src/cli.rs` 成功
    - `src/main.rs` 含字符串 `Commands::Cluster`（grep -q "Commands::Cluster" src/main.rs）
    - 至少 3 个新测试函数名包含 cluster（grep -cE 'fn test_cluster' src/cli.rs 应 ≥ 3）
    - `grep -q '^tracing-test' Cargo.toml` 成功（dev-dependency 已声明，per Task 3 集成测试需要捕获 tracing event 顺序验证 CLUS-02 SC3）
  </acceptance_criteria>
  <done>
    CLI 完整支持 `dm-installer cluster deploy --config <path>`；main.rs dispatch 调用 cluster::run；3 个新测试通过；Phase 2 现有测试不破坏。
  </done>
</task>

<task type="auto" tdd="true">
  <name>Task 2: cluster/deploy.rs — 节点级安装编排函数族</name>
  <files>
    src/cluster/deploy.rs,
    src/cluster/mod.rs,
    src/cluster/ssh.rs
  </files>
  <read_first>
    src/cluster/mod.rs,
    src/cluster/ssh.rs,
    src/cluster/templates/mod.rs,
    src/cluster/health.rs,
    src/install/init.rs,
    .planning/phases/03-cluster/03-RESEARCH.md,
    .planning/phases/03-cluster/03-02-SUMMARY.md
  </read_first>
  <behavior>
    - Test 1 (build_dminit_args 格式正确): 给定 NodeConfig { instance_name = "DMSVR01", install_path = "/opt/dmdbms", data_path = "/opt/dmdbms/data/DMSVR01", port = 5236, page_size = 8, charset = 0, case_sensitive = true, extent_size = 16 }, build_dminit_args(&node) 返回 Vec&lt;String&gt;，第一项为 "/opt/dmdbms/bin/dminit"，列表包含字符串 "PATH=/opt/dmdbms/data/DMSVR01"、"INSTANCE_NAME=DMSVR01"、"PORT_NUM=5236"、"PAGE_SIZE=8"（等号两侧无空格——Pitfall 2 防范）
    - Test 2 (distribute_configs 调用 4 次 sftp_write): MockRunner 注入；distribute_configs(&primary_node, &all_nodes, 453331, &runner).await 成功后，从 MockRunner 取出 sftp_log 应有 ≥ 4 项，其中 4 个远端路径分别含子串 "dm.ini"、"dmmal.ini"、"dmarch.ini"、"dmwatcher.ini"
    - Test 3 (configure_database_role primary 含正确 SQL): 用 MockRunner（pre-set exec exit_code = 0），调用 configure_database_role(&primary_node, NodeRole::Primary, 453331, &runner).await，从 MockRunner.exec_log 中至少一条 SSH 命令字符串包含 "sp_set_oguid(453331)" 和 "alter database primary"
    - Test 4 (configure_database_role standby 含正确 SQL): 同上对 Standby 验证含 "alter database standby" 而非 "alter database primary"
    - Test 5 (start_dmserver_mount 使用 mount 模式 + nohup): MockRunner 注入；start_dmserver_mount(&node, &runner).await 后 exec_log 至少一条命令字符串同时含 "dmserver" / "mount" / "nohup" 三个子串（per Pitfall 4 + RESEARCH Open Question Q3 已 RESOLVED 为 nohup 方案）
    - Test 6 (upload_installer_and_install 必须推送 XML — BLOCKER 3): MockRunner 注入；upload_installer_and_install(&node, Path::new("/tmp/fake.iso"), &runner).await 完成后，sftp_log 至少含 2 项：一项远端路径含 `.xml` 子串（XML response file，per CLUS-01），一项含 `.iso` 子串（安装包）。本测试断言 XML 生成 + SFTP push 为强制步骤，不可省略
  </behavior>
  <action>
1. 修改 src/cluster/ssh.rs：在 MockRunner impl 中追加两个公共访问器（如 Plan 02 SUMMARY 未暴露则本 task 补充）：`pub fn exec_log(&self) -> Vec<String>` 返回所有 exec 调用过的命令字符串副本；`pub fn sftp_log(&self) -> Vec<(String, Vec<u8>)>` 返回 (path, bytes) 副本。Mutex 保护的内部 Vec 在被调用时锁 + clone 返回。

2. 新建 src/cluster/deploy.rs：imports 包含 `use anyhow::{Context, Result};`、`use crate::config::cluster::{NodeConfig, NodeRole};`、`use crate::cluster::ssh::CommandRunner;`、`use crate::cluster::templates::{generate_dm_ini_cluster_suffix, generate_dmmal_ini, generate_dmarch_ini, generate_dmwatcher_ini};`、`use std::path::Path;`。

3. pub fn build_dminit_args(node: &NodeConfig) -> Vec<String>：参照 Phase 2 src/install/init.rs 的 build_dminit_command 模式（lines 30-46）。返回包含 format!("{}/bin/dminit", node.install_path) 作为第一项，以及 format!("PATH={}", node.data_path)、format!("INSTANCE_NAME={}", node.instance_name)、format!("PORT_NUM={}", node.port)、format!("PAGE_SIZE={}", node.page_size)、format!("CHARSET={}", node.charset)、format!("CASE_SENSITIVE={}", if node.case_sensitive { 1 } else { 0 })、format!("EXTENT_SIZE={}", node.extent_size) — 等号两侧无空格（Pitfall 2）。

4. pub async fn upload_installer_and_install(node: &NodeConfig, package_path: &Path, runner: &dyn CommandRunner) -> Result<()>：**XML 生成与 SFTP 推送为强制步骤**（per CLUS-01 强制要求；DMInstall.bin -q 必须传入有效 XML response file，无 XML 则静默安装失败——不可标记为 TODO 或 executor optional）。
   a. 日志 `tracing::info!("[node:{:?}][1/6] 生成 XML response file", node.role);`
   b. 调用 Phase 2 `crate::install::silent_install::generate_install_xml(...)` 生成 XML 字符串；若函数仅接受 InstallConfig，本函数构造临时 InstallConfig（字段从 NodeConfig 映射：install_path / instance_name / port / page_size / charset / case_sensitive / extent_size）。若 Phase 2 函数为 `pub(crate)` 或更窄，executor 在 silent_install.rs 中改为 `pub`（最小改动，向后兼容；记录于 SUMMARY）。
   c. 远端 XML 路径：`let remote_xml = format!("/tmp/cluster_install_{}.xml", node.instance_name);`
   d. `runner.sftp_write(&remote_xml, xml_str.as_bytes()).await.context("SFTP 上传 XML response file 失败")?;` —— **必须执行，不可省略，不可标 TODO**
   e. 日志 `tracing::info!("[node:{:?}][2/6] 推送安装包", node.role);` 读本地：`let bytes = tokio::fs::read(package_path).await.with_context(|| format!("无法读取安装包 {}", package_path.display()))?;`
   f. 远端 ISO 目标：`let remote_iso = format!("/tmp/dm_installer_{}.iso", node.instance_name);`
   g. `runner.sftp_write(&remote_iso, &bytes).await.context("SFTP 上传安装包失败")?;`
   h. 远端解压并静默安装命令：`format!("cd /tmp && DMInstall.bin -q {}", remote_xml)`；通过 runner.exec 调用；exit_code != 0 时 `anyhow::bail!("DMInstall.bin 失败 (exit {}): {}", exit_code, String::from_utf8_lossy(&stdout));`

5. pub async fn run_dminit_remote(node: &NodeConfig, runner: &dyn CommandRunner) -> Result<()>：日志 [3/6]；`let cmd = build_dminit_args(node).join(" ");` `let (stdout, exit_code) = runner.exec(&cmd).await?;` `anyhow::ensure!(exit_code == 0, "dminit 失败 (exit {}): {}", exit_code, String::from_utf8_lossy(&stdout));`。

6. pub async fn distribute_configs(node: &NodeConfig, all_nodes: &[NodeConfig], oguid: u32, runner: &dyn CommandRunner) -> Result<()>：日志 [4/6]；计算 peer：`let peer = all_nodes.iter().find(|n| n.instance_name != node.instance_name).context("找不到对端节点")?;` 控制机生成 4 个 INI 内容（D-11）：调用 generate_dm_ini_cluster_suffix(node)、generate_dmmal_ini(all_nodes)、generate_dmarch_ini(node, &peer.instance_name)、generate_dmwatcher_ini(node, oguid)。计算远端 base：`let base = format!("{}/{}", node.data_path, node.instance_name);` SFTP 推送 4 次到 `{base}/dm.ini.cluster_suffix`、`{base}/dmmal.ini`、`{base}/dmarch.ini`、`{base}/dmwatcher.ini`，每次都 .as_bytes()。所有路径用 format! + PathBuf 风格拼接（T-03-06 缓解）；不直接拼接 shell 字符串。NOTE：dm.ini 的追加片段需在远端用 `cat {base}/dm.ini.cluster_suffix >> {base}/dm.ini` 合并；本任务可在 SFTP 4 次后追加一条 runner.exec 实现合并。

7. pub async fn start_dmserver_mount(node: &NodeConfig, runner: &dyn CommandRunner) -> Result<()>：日志 [5/6]；命令字符串：`format!("nohup {0}/bin/dmserver {1}/{2}/dm.ini mount > /tmp/dmserver_{2}.log 2>&1 &", node.install_path, node.data_path, node.instance_name)`（per Pitfall 4：必须 mount 模式；per RESEARCH Open Question Q3：nohup + & 防 SIGHUP）；调用 runner.exec(&cmd).await? — 不检查 exit_code 因后台进程，后续由 wait_tcp_ready 验证启动成功。

8. pub async fn configure_database_role(node: &NodeConfig, role: NodeRole, oguid: u32, runner: &dyn CommandRunner) -> Result<()>：根据 role 构造 SQL block（per RESEARCH §部署步骤序列）。SQL 字符串包含 "SP_SET_PARA_VALUE(1,'ALTER_MODE_STATUS',1);" + "sp_set_oguid({oguid});" + "alter database primary;" 或 "alter database standby;" + "SP_SET_PARA_VALUE(1,'ALTER_MODE_STATUS',0);"。disql 调用命令：`format!("echo \"{}\" | {}/bin/disql SYSDBA/SYSDBA@localhost:{}", sql_block, node.install_path, node.port)`（用 stdin pipe 而非 -e，更保守，per RESEARCH Open Q1）。调用 runner.exec 后检查 exit_code == 0。

9. pub async fn start_dmwatcher(node: &NodeConfig, runner: &dyn CommandRunner) -> Result<()>：日志 [6/6]；命令：`format!("nohup {0}/bin/dmwatcher {1}/{2}/dmwatcher.ini > /tmp/dmwatcher_{2}.log 2>&1 &", node.install_path, node.data_path, node.instance_name)`；调用 runner.exec。

10. 修改 src/cluster/mod.rs：新增 `pub mod deploy;`（与 templates / ssh / preflight / health 并列）；run 函数仍保持 unimplemented stub，Task 3 替换为真实实现。

11. 末尾 #[cfg(test)] mod tests 实现 5 个 #[tokio::test] 测试，全部使用 MockRunner 注入（无需真实 SSH）：
    - test_build_dminit_args_format
    - test_distribute_configs_calls_four_sftp_writes
    - test_configure_database_role_primary_sql
    - test_configure_database_role_standby_sql
    - test_start_dmserver_mount_uses_mount_and_nohup

12. 所有 pub fn / pub async fn 函数体保持 < 40 行。distribute_configs 拆 `fn target_path(node: &NodeConfig, filename: &str) -> String { format!("{}/{}/{}", node.data_path, node.instance_name, filename) }` private helper 后单函数行数应满足要求。
  </action>
  <verify>
    <automated>cargo test --lib cluster::deploy --no-fail-fast 2>&amp;1 | tee /tmp/p3-03-t2.log; grep -E "test result:" /tmp/p3-03-t2.log; grep -cE 'pub (async )?fn (build_dminit_args|upload_installer_and_install|run_dminit_remote|distribute_configs|start_dmserver_mount|configure_database_role|start_dmwatcher)' src/cluster/deploy.rs</automated>
  </verify>
  <acceptance_criteria>
    - `cargo test --lib cluster::deploy` exits 0（5 个新测试通过）
    - `cargo build --lib` exits 0
    - `grep -q '^pub fn build_dminit_args' src/cluster/deploy.rs`
    - `grep -q '^pub async fn upload_installer_and_install' src/cluster/deploy.rs`
    - `grep -q '^pub async fn run_dminit_remote' src/cluster/deploy.rs`
    - `grep -q '^pub async fn distribute_configs' src/cluster/deploy.rs`
    - `grep -q '^pub async fn start_dmserver_mount' src/cluster/deploy.rs`
    - `grep -q '^pub async fn configure_database_role' src/cluster/deploy.rs`
    - `grep -q '^pub async fn start_dmwatcher' src/cluster/deploy.rs`
    - `src/cluster/mod.rs` 含 `pub mod deploy;`
    - SshSession.MockRunner 含 `pub fn exec_log` 和 `pub fn sftp_log` 两个访问器（grep -cE 'pub fn (exec_log|sftp_log)' src/cluster/ssh.rs 应 ≥ 2）
    - Test 1 断言 build_dminit_args 输出含 "PORT_NUM=5236" 严格无空格（grep -q 'PORT_NUM=' src/cluster/deploy.rs 测试节）
    - **BLOCKER 3 修复验证**：upload_installer_and_install 必须 SFTP 推送 XML response file。grep 函数体含 `generate_install_xml` 调用和 `.xml` 远端路径（grep -cE 'generate_install_xml|cluster_install_.*\.xml' src/cluster/deploy.rs ≥ 2）。Test 6 存在且通过（grep -E 'fn test_.*xml|fn test_upload_installer' src/cluster/deploy.rs 至少一处匹配，且测试断言 sftp_log 含 `.xml` 路径条目）。
  </acceptance_criteria>
  <done>
    deploy.rs 7 个 pub 函数全部实现并通过 mock 单元测试；MockRunner 增加 exec_log / sftp_log 访问器供集成断言；Pitfall 2（dminit 无空格）、Pitfall 4（mount 模式）、disql 主备 SQL 差异均由测试覆盖。
  </done>
</task>

<task type="auto" tdd="true">
  <name>Task 3: cluster::run 顶层编排器（替换 Plan 01 unimplemented stub）</name>
  <files>src/cluster/mod.rs</files>
  <read_first>
    src/cluster/mod.rs,
    src/cluster/deploy.rs,
    src/cluster/preflight.rs,
    src/cluster/health.rs,
    src/cluster/ssh.rs,
    src/cli.rs,
    src/install/mod.rs,
    .planning/phases/03-cluster/03-RESEARCH.md
  </read_first>
  <behavior>
    - Test 1 (run 拒绝缺 primary 的 toml fixture): #[tokio::test] 调用 `cluster::run(&ClusterDeployArgs { config: PathBuf::from("tests/fixtures/cluster_invalid_no_primary.toml") })` 返回 Err，错误链 ({:#}) 含 "primary"。
    - Test 2 (preflight 失败时 run 在安装前 bail): 用纯函数子层 `pub async fn run_with_runners(config: ClusterConfig, runners: Vec<(NodeConfig, Arc<dyn CommandRunner>)>, health_check_fn: impl Fn(String, u16, u64) -> std::pin::Pin<Box<dyn std::future::Future<Output=anyhow::Result<()>> + Send>>) -> Result<()>`（test-only 注入点），传入 MockRunner 其中一个 sudo 命令返回 exit_code=1；调用后返回 Err 含 "预检查失败"，且 MockRunner 的 exec_log 中不出现 dminit / dmserver / disql 命令（验证 D-08 "任一失败则中止部署"）。
    - Test 3 (顺序：primary 健康确认在 standby 启动前): 注入 `health_check_fn = |_host, _port, _secs| Box::pin(async { Ok(()) })` 使健康检查立即返回 Ok（不连真实 TCP）；主备 runners 均预设 exec 全返回 exit_code=0；调用 `run_with_runners(config, runners, |h,p,s| Box::pin(async move { Ok(()) }))` 完成后断言：primary runner 的 exec_log 中 "alter database primary" 出现位置 < standby runner 的 exec_log 中含 "mount" 命令的位置；用 `#[tracing_test::traced_test]` 捕获日志验证 "主节点就绪" 事件早于 "启动达梦备实例" 事件（per CLUS-02 SC3）。
  </behavior>
  <action>
1. 修改 src/cluster/mod.rs：声明 imports：`use anyhow::{Context, Result};`、`use crate::cli::ClusterDeployArgs;`、`use crate::config::cluster::{load_cluster_config, ClusterConfig, NodeConfig, NodeRole};`、`use crate::cluster::ssh::{CommandRunner, SshSession};`。

2. 替换原 pub async fn run() stub 为：`pub async fn run(args: &ClusterDeployArgs) -> Result<()>`：
   a. `let config = load_cluster_config(&args.config).with_context(|| format!("加载集群配置失败: {}", args.config.display()))?;`
   b. tracing::info!("[cluster][1/6] 建立 SSH 会话");
   c. 对每个 node 调用 `SshSession::connect(&node.host, 22, &node.ssh.user, &node.ssh).await?`，收集到 `Vec<(NodeConfig, Arc<dyn CommandRunner>)>`（NodeConfig 克隆即可）。
   d. 调用内部函数 `run_with_runners(config, runners).await`。

3. 新增内部函数 `pub async fn run_with_runners(config: ClusterConfig, runners: Vec<(NodeConfig, Arc<dyn CommandRunner>)>, health_check_fn: impl Fn(String, u16, u64) -> std::pin::Pin<Box<dyn std::future::Future<Output=anyhow::Result<()>> + Send>>) -> Result<()>`（pub 以便测试 #[cfg(test)] 模块外部仍可调用；如希望仅 test-only 可改 `pub(crate)`）。函数体编排：

   步骤 1（预检查 - QUAL-01）：tracing::info!("[cluster][2/6] SSH 预检查");
   构造 `let preflight_input: Vec<(NodeConfig, std::sync::Arc<dyn CommandRunner>)> = runners.iter().map(|(n, r)| (n.clone(), r.clone())).collect();` —— **Plan 02 修订后已统一为 `Arc<dyn CommandRunner>`**，Arc 克隆零开销，runner 实例在 preflight / 安装 / 启动多步骤间复用，无需重连 SSH。调用 `preflight::preflight_all_nodes(preflight_input).await?;` — 任一失败立即返回 Err。

   步骤 2（并发安装包推送 + DMInstall.bin + dminit）：tracing::info!("[cluster][3/6] 并发推送安装包并 dminit");
   用 `futures::future::try_join_all(nodes.iter().map(|(n, r)| async { upload_installer_and_install(n, &config.cluster.installer_package, r.as_ref()).await?; run_dminit_remote(n, r.as_ref()).await }))` 并发 — 任一失败立即返回。

   步骤 3（配置文件分发）：tracing::info!("[cluster][4/6] 分发配置文件");
   并发 distribute_configs，同上 try_join_all。

   步骤 4（有序启动 - CLUS-02）：tracing::info!("[cluster][5/6] 有序启动主备实例");
   先找 primary 节点：`let (primary_node, primary_runner) = runners.iter().find(|(n, _)| n.role == NodeRole::Primary).context("缺少 primary 节点")?;`
   调用 start_dmserver_mount(primary_node, primary_runner.as_ref()).await?;
   tracing::info!("[node:{:?}] 等待主节点健康 (TCP:{}) ...", primary_node.role, primary_node.port);
   调用 health_check_fn(primary_node.host.clone(), primary_node.port, 60).await?; // per D-09；测试注入 |_,_,_| Box::pin(async { Ok(()) })
   tracing::info!("[node:{:?}] 主节点就绪 ✓", primary_node.role);
   调用 configure_database_role(primary_node, NodeRole::Primary, config.cluster.oguid, primary_runner.as_ref()).await?;
   找 standby 节点：`let (standby_node, standby_runner) = runners.iter().find(|(n, _)| n.role == NodeRole::Standby).context("缺少 standby 节点")?;`
   tracing::info!("[node:{:?}][5/6] 启动达梦备实例", standby_node.role);
   调用 start_dmserver_mount(standby_node, standby_runner.as_ref()).await?;
   health_check_fn(standby_node.host.clone(), standby_node.port, 60).await?;
   configure_database_role(standby_node, NodeRole::Standby, config.cluster.oguid, standby_runner.as_ref()).await?;

   步骤 5（启动 dmwatcher - 并发可选）：tracing::info!("[cluster][6/6] 启动 dmwatcher");
   try_join_all 并发对所有 nodes 调用 deploy::start_dmwatcher。

   tracing::info!("集群部署完成 ✓"); Ok(())。

4. 所有权策略已统一（per Plan 02 修订）：`preflight_all_nodes` 签名为 `Vec<(NodeConfig, Arc<dyn CommandRunner>)>`；cluster::run 与 run_with_runners 同样持有 `Vec<(NodeConfig, Arc<dyn CommandRunner>)>`；deploy.rs 各函数仍接 `&dyn CommandRunner` 参数，通过 `runner.as_ref()` 传入。Arc 克隆零开销，单个 runner 实例在预检查 / 安装包推送 / dminit / 配置分发 / 启动 / disql / dmwatcher 多个步骤间复用。

5. 末尾 #[cfg(test)] mod tests 实现 Test 1 / Test 2 / Test 3：
   - 用 tests/fixtures/cluster_valid.toml 构造 ClusterConfig
   - 构造 MockRunner per node，预设全部 exec 返回 exit_code=0（preflight: sudo/ss/df 均通过；dminit / dmserver / disql 均通过）
   - Test 3 检查日志顺序（per WARNING 4 修复，使用 tracing-test）：用 `#[tracing_test::traced_test]` + `#[tokio::test]` 双注解（dev-dep tracing-test = "0.2" 由 Task 1 步骤 4 引入）；调用 run_with_runners 完成后用 `logs_assert(|lines| { let primary_ready = lines.iter().position(|l| l.contains("主节点就绪")); let standby_start = lines.iter().position(|l| l.contains("启动达梦备实例")); match (primary_ready, standby_start) { (Some(a), Some(b)) if a < b => Ok(()), _ => Err("主节点就绪 must precede 启动达梦备实例".into()) } })` 严格断言 event 顺序（CLUS-02 SC3）。冗余检查：从 primary_runner.exec_log() 验证含 "alter database primary"，从 standby_runner.exec_log() 验证含 "dmserver" + "mount" 
   - Test 2: 让 primary 的 MockRunner sudo 命令返回 exit 1；调用 run_with_runners 后断言 Err；断言 primary_runner.exec_log() 不含 "dminit"、不含 "dmserver"、不含 "disql"（D-08：预检查失败前不执行任何安装步骤）
   - Test 1: 构造 ClusterDeployArgs{ config: PathBuf::from("tests/fixtures/cluster_invalid_no_primary.toml") }，调用 run(&args).await，断言 Err 错误链含 "primary"

6. 所有函数体 < 40 行；run / run_with_runners 必然要拆分多个 private async fn helper：建议 `fn collect_runners(config: &ClusterConfig) -> Result<Vec<...>>`、`fn run_preflight(...)`、`fn run_install_phase(...)`、`fn run_startup_phase(...)`，每个 helper < 40 行。
  </action>
  <verify>
    <automated>cargo test --lib cluster --no-fail-fast 2>&amp;1 | tee /tmp/p3-03-t3.log; grep -E "test result:" /tmp/p3-03-t3.log; cargo build --bin dm-installer 2>&amp;1 | tail -3</automated>
  </verify>
  <acceptance_criteria>
    - `cargo test --lib cluster` exits 0（含 Plan 01 + Plan 02 + Plan 03 全部单元测试）
    - `cargo build --bin dm-installer` exits 0
    - `grep -q '^pub async fn run(args: &ClusterDeployArgs)' src/cluster/mod.rs` 或等价签名
    - `grep -q 'health_check_fn' src/cluster/mod.rs`（可注入健康检查函数；生产路径调用 wait_tcp_ready，测试注入立即返回的闭包）
    - `grep -q 'preflight_all_nodes' src/cluster/mod.rs`（预检查调用）
    - `grep -c 'NodeRole::Primary' src/cluster/mod.rs` ≥ 1（区分主备启动顺序）
    - `grep -c 'NodeRole::Standby' src/cluster/mod.rs` ≥ 1
    - 不再有 `unimplemented!` 字样（grep -L "unimplemented" src/cluster/mod.rs 应失败，即不含）
    - 至少 3 个新测试函数：test_run_rejects_no_primary_fixture、test_run_aborts_on_preflight_failure_before_install、test_run_orders_primary_health_before_standby_start
    - **WARNING 4 修复验证**：Test 3 使用 `tracing_test::traced_test` 或调用 `logs_assert`/`logs_contain`（grep -E 'tracing_test|traced_test|logs_assert|logs_contain' src/cluster/mod.rs 至少一处匹配）
    - **WARNING 5 修复验证**：runners 集合类型统一为 `Arc<dyn CommandRunner>`（grep -q 'Arc<dyn CommandRunner>' src/cluster/mod.rs）
  </acceptance_criteria>
  <done>
    cluster::run 完整编排 6 步序列；预检查失败时安装步骤不执行；主节点 TCP 健康确认 + disql 配置 在备节点启动之前；3 个集成单元测试通过；与 main.rs 的 dispatch 接通。
  </done>
</task>

<task type="checkpoint:human-verify" gate="blocking">
  <name>Task 4: 端到端 manual 验证 — 双节点 cluster deploy 日志顺序（CLUS-02 SC3）</name>
  <what-built>
    Claude 已自动完成所有自动化验证：
    - Plan 01 / 02 / 03 全部单元测试通过
    - cargo build --bin dm-installer 通过
    - cluster::run 编排逻辑覆盖 preflight → install → distribute → ordered startup → dmwatcher
    - tracing 日志包含 [node:primary] 健康确认在 [node:standby] 启动之前的序列证据（单元测试）
  </what-built>
  <how-to-verify>
    1. 准备两台测试节点（任选其一）：
       a) 真实 Linux VM：两台 IP 互通的 Linux x86_64，root 用户 sudo 免密，SSH 密钥登录就绪
       b) Docker sshd 容器：用 docker run -d -p 22:22 takeyamajp/ubuntu-sshd 起两个容器，记录映射端口
    2. 在控制机本地准备一份 cluster.toml（参照 tests/fixtures/cluster_valid.toml，替换 host 字段为实际节点 IP/端口，identity_file 指向你的 SSH 私钥）
    3. 准备 DM 安装包：下载或本地复制一个 dm8 ISO 到本地，并在 cluster.toml 中 installer_package 字段写其路径
    4. 进入工作树目录运行：
       `RUST_LOG=info cargo run --bin dm-installer -- cluster deploy --config /path/to/your/cluster.toml 2>&1 | tee /tmp/cluster_deploy.log`
    5. 验证以下行为：
       a) 命令开始即出现 `[预检查]` 日志条目，分别对两个节点输出 sudo / 端口 / 磁盘三项检查结果
       b) 预检查全通过后才出现 `[node:Primary][1/6]` / `[node:Primary][2/6]` ... 序列
       c) `[node:Primary]` 出现 "主节点就绪 ✓" 字样后，才出现 `[node:Standby][5/6] 启动达梦备实例` 字样（CLUS-02 SC3 关键验证点）
       d) 最终输出 "集群部署完成 ✓"
       e) 在两个节点上手动 `ps -ef | grep dmserver` 和 `ps -ef | grep dmwatcher` 各看到一个进程
    6. 若手头无两个真实节点 / 无 DM 安装包，可只验证 step a-b 的 dry-run 路径：故意提供一个不可达的 host（如 192.0.2.1）观察预检查失败时是否中止且不进入安装步骤。
    7. 报告结果：粘贴 /tmp/cluster_deploy.log 中以下关键行：所有 [预检查] 行 + "主节点就绪 ✓" 行 + 首个 [node:Standby] 启动行 + 最终完成行（或失败行）。
  </how-to-verify>
  <resume-signal>
    回复 "approved" 表示日志顺序正确（主节点健康确认在备节点启动之前）；或回复 "rejected: <具体观察到的错误>" 由 Claude 调查修复。若手头无完整测试环境，回复 "manual-deferred: 无真实节点" 也接受——前提是所有自动化单元测试已绿。
  </resume-signal>
</task>

</tasks>

<threat_model>
## Trust Boundaries

| Boundary | Description |
|----------|-------------|
| 用户提供的 cluster.toml 路径（控制机文件系统） → CLI args | 路径来自命令行，可指向任意文件 |
| 控制机进程 → 远端 SSH 22 端口 | 外发连接，TOFU 主机密钥（Plan 02 已建立） |
| 控制机生成的 INI 字符串 → SFTP 远端文件系统 | 字节经由 russh-sftp 协议传输 |
| 远端 SSH stdout（disql 输出）→ 控制机错误判定 | 解析远端文本判断 exit_code |

## STRIDE Threat Register

| Threat ID | Category | Component | Disposition | Mitigation Plan |
|-----------|----------|-----------|-------------|-----------------|
| T-03-08 | Tampering | NodeConfig.host 经 wait_tcp_ready 拼入 socket 地址 | mitigate | host 类型是 String，但 wait_tcp_ready 内部用 format!("{}:{}", host, port) 传给 TcpStream::connect — socket 解析阶段拒绝非法主机名/IP |
| T-03-09 | Tampering | disql SQL 字符串经 echo 管道传至远端 | mitigate | SQL 模板字符串固定，仅 oguid 数值参数化（u32 类型保证只能是数字）；无用户输入直接拼入 SQL 主体 |
| T-03-10 | Information Disclosure | cluster.toml 含 password 明文 → run 日志意外打印 | mitigate | cluster::run 仅 log host / role / 步骤名，不 log SshCredentials 结构体；Plan 01 SshCredentials 已 `#[serde(skip_serializing)]` password |
| T-03-11 | Tampering | configure_database_role 命令通过 SSH exec，disql 客户端非交互 | mitigate | sql_block 经 echo "..." 管道写入，使用双引号转义 | 
| T-03-SC | Tampering | 本计划未引入新依赖 | N/A | 仅复用 Plan 01/02 已安装的 russh / russh-sftp（Plan 02 已通过合法性 checkpoint） |
</threat_model>

<verification>
1. `cargo build --bin dm-installer` exits 0
2. `cargo nextest run --lib` 或 `cargo test --lib` 全部通过（Plan 01 9 个 + Plan 02 10 个 + Plan 03 ≥ 11 个 = ≥ 30 个测试）
3. `cargo run --bin dm-installer -- cluster deploy --help` 输出含 `--config`
4. `cargo run --bin dm-installer -- cluster deploy` (无 --config) 退出码非 0 且 stderr 含 "required"
5. `cargo run --bin dm-installer -- cluster deploy --config tests/fixtures/cluster_invalid_no_primary.toml` 退出码非 0 且 stderr 含 "primary"
6. Manual checkpoint Task 4 完成（approved 或 manual-deferred）
</verification>

<success_criteria>
- CLI `cluster deploy --config <toml>` 子命令可解析并 dispatch 到 cluster::run
- cluster::run 完整 6 步编排：load → preflight → 并发推包+dminit → 分发 INI → 有序启动（主→TCP健康→主disql→备→TCP健康→备disql）→ 启 dmwatcher
- 预检查失败时不执行任何安装步骤（D-08）
- 主节点 TCP 健康确认 + disql 配置完成后才启动备节点（CLUS-02 SC3）
- 全部新增 pub fn 函数体 < 40 行
- 全部 Plan 01 + 02 + 03 单元测试通过
- 端到端 manual 验证 approved 或正当 deferred
</success_criteria>

<output>
Create `.planning/phases/03-cluster/03-03-SUMMARY.md` when done. Record:
- 实际函数签名（特别是 cluster::run 和 run_with_runners 是否需要 Arc 调整）
- MockRunner 暴露的 access methods（exec_log / sftp_log）最终 API
- 端到端 manual checkpoint 的用户响应（approved / manual-deferred）
- 任何 RESEARCH §Open Questions 在实施中得到验证或调整的结论（如 disql 参数格式、nohup 是否真在远端持久化）
- Phase 3 总体收尾：哪些验收准则（ROADMAP §Phase 3 SC1-SC4）已被覆盖
</output>
</content>
</invoke>