---
phase: 03-cluster
plan: 02
type: execute
wave: 2
depends_on:
  - 03-01
files_modified:
  - Cargo.toml
  - src/cluster/ssh.rs
  - src/cluster/preflight.rs
  - src/cluster/health.rs
autonomous: false
requirements:
  - CLUS-01
  - CLUS-02
  - QUAL-01
must_haves:
  truths:
    - "Cargo.toml 引入 russh 0.61.2 + russh-sftp 2.3.0，编译通过且不引入 OpenSSL C 依赖"
    - "SshError 枚举覆盖连接失败、命令执行非零退出、SFTP 上传失败三种场景"
    - "preflight 模块定义 `CommandRunner` trait，使三项检查（sudo/端口/磁盘）可通过 mock 实现单元测试；`preflight_all_nodes` 签名统一使用 `Arc<dyn CommandRunner>` 以便 Plan 03 多步复用 runner"
    - "preflight::check_node 在所有三项通过时返回 Ok，任一失败时返回 Err 含节点 host 和失败检查项名"
    - "wait_tcp_ready 在指定超时内 TCP 不可连时返回 Err，错误消息含 host:port 和超时秒数"
  artifacts:
    - path: "Cargo.toml"
      provides: "russh 和 russh-sftp 依赖声明"
      contains: "russh"
    - path: "src/cluster/ssh.rs"
      provides: "SshError 类型 + SshSession 抽象 + TofuHandler + CommandRunner trait"
      contains: "pub enum SshError"
    - path: "src/cluster/preflight.rs"
      provides: "check_sudo_nopass / check_port_available / check_disk_space / check_node 函数"
      contains: "pub async fn check_node"
    - path: "src/cluster/health.rs"
      provides: "wait_tcp_ready 异步函数"
      contains: "pub async fn wait_tcp_ready"
  key_links:
    - from: "src/cluster/preflight.rs"
      to: "src/cluster/ssh.rs"
      via: "CommandRunner trait 注入"
      pattern: "impl CommandRunner|dyn CommandRunner"
    - from: "src/cluster/ssh.rs"
      to: "russh::client + russh_sftp::client"
      via: "use russh::client"
      pattern: "use russh"
---

<objective>
建立 Phase 3 SSH 远程操作基础设施：依赖引入、SSH 会话抽象、SSH 预检查、TCP 健康轮询。本计划是垂直切片的"传输层"，与 Plan 01（数据层）并行执行——两个计划无文件交集。

Purpose: CLUS-01 的 SSH 远程操作能力、QUAL-01 的预检查、CLUS-02 的健康判据都需要可单元测试的 SSH/TCP 原语。引入 CommandRunner trait 让 preflight 用 mock 实现单元测试，避免本计划阶段就要起真实 SSH 服务器。

Output:
- `Cargo.toml` 新增 russh + russh-sftp 依赖（含包合法性 checkpoint）
- `cluster/ssh.rs` SshError 类型 + SshSession 真实实现 + CommandRunner trait + TofuHandler
- `cluster/preflight.rs` 三项预检查 + 节点级 check_node 编排（mock 测试覆盖）
- `cluster/health.rs` wait_tcp_ready 异步函数 + 超时测试
</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

# Phase 2 现有代码参照
@Cargo.toml
@src/install/checksum.rs

<interfaces>
<!-- russh 0.61.2 + russh-sftp 2.3.0 关键 API（来自 RESEARCH §Pattern 1/2 + docs.rs） -->

From russh::client:
- `pub trait Handler { type Error; async fn check_server_key(&mut self, server_public_key: &ssh_key::PublicKey) -> Result<bool, Self::Error>; }`
- `pub struct Handle<H: Handler>` — SSH 会话句柄，方法：`channel_open_session().await -> Channel`
- `pub enum ChannelMsg { Data { data: CryptoVec }, ExitStatus { exit_status: u32 }, Eof, ... }`
- `Channel::exec(want_reply: bool, command: &str)`
- `Channel::wait() -> Option<ChannelMsg>`

From russh_sftp::client:
- `pub struct SftpSession`
- `SftpSession::new(stream).await -> Result<SftpSession, russh_sftp::Error>`
- `SftpSession::write(path: &str, data: &[u8]) -> Result<(), russh_sftp::Error>`

From Phase 2 (复用):
- `use anyhow::{Context, Result, bail};`
- `tokio::net::TcpStream::connect(addr).await -> io::Result<TcpStream>`
- `tokio::time::timeout(dur, fut).await -> Result<T, Elapsed>`
- `tokio::time::sleep(dur).await`
</interfaces>
</context>

<tasks>

<task type="checkpoint:human-verify" gate="blocking-human">
  <name>Task 1: russh / russh-sftp 包合法性人工校验（slopcheck 不可用）</name>
  <what-built>
    Plan 02 准备引入 russh 0.61.2 + russh-sftp 2.3.0；RESEARCH §Package Legitimacy Audit 标记两者为 [ASSUMED]（slopcheck 不可用），按 scope_reduction_prohibition 规则需用户在 cargo add 前人工核验。
  </what-built>
  <how-to-verify>
    1. 在浏览器或 `curl` 中分别打开:
       - https://crates.io/crates/russh — 确认版本 0.61.2 存在、最近一年内有发布、维护者为 `eugeny` 或 `warp-tech` 组织
       - https://crates.io/crates/russh-sftp — 确认版本 2.3.0 存在、最近一年内有发布、与 russh 兼容
    2. 访问对应 GitHub 仓库链接：
       - https://github.com/Eugeny/russh — 仓库存在，star ≥ 500，最近 commit 在 6 个月内
       - https://github.com/AspectUnk/russh-sftp — 仓库存在，star ≥ 50
    3. 确认两者均无最近 90 天内的 RustSec / GHSA 高危公告（可查 https://rustsec.org/advisories/）
    4. 确认 D-07 TOFU 策略与 russh 0.61.2 的 Handler::check_server_key API 一致（docs.rs/russh/0.61.2/russh/client/trait.Handler.html）
  </how-to-verify>
  <resume-signal>
    回复 "approved" 表示两个包合法可用，executor 继续执行 Task 2 的 cargo add；或回复 "rejected: <reason>" 终止本计划。
  </resume-signal>
</task>

<task type="auto">
  <name>Task 2: 引入 russh / russh-sftp 依赖</name>
  <files>Cargo.toml</files>
  <read_first>
    Cargo.toml,
    .planning/phases/03-cluster/03-RESEARCH.md
  </read_first>
  <action>
1. 在 `[dependencies]` 段（现有 `clap_complete = "4.6.5"` 之后）新增两行：
   - `russh = { version = "0.61.2", default-features = false, features = ["async-trait", "client"] }` — 显式关闭 default-features 避免引入 server 端代码（per RESEARCH §Architectural Responsibility Map：本工具只做客户端）
   - `russh-sftp = "2.3.0"`
2. 同时新增 `futures = "0.3"` 用于 preflight 的 `futures::future::join_all`（per RESEARCH §Pattern 3 与 §Don't Hand-Roll；如已有则跳过）
3. 不修改 Phase 2 已有任何依赖版本
4. 执行 `cargo build` 验证编译通过；若 russh 的 rustls backend 缺失需要额外 feature，executor 应基于 cargo 错误提示修复（如可能需要追加 `"rustls"` feature 名；RESEARCH §Assumption A1 已标记此为 [ASSUMED]）
  </action>
  <verify>
    <automated>cargo build --lib 2>&amp;1 | tee /tmp/p3-02-t2.log; cargo tree | grep -E "^(russh|russh-sftp|openssl|native-tls)" | tee /tmp/p3-02-t2-tree.log; grep -E "^(openssl|native-tls) " /tmp/p3-02-t2-tree.log; test $? -ne 0  # 期望 openssl / native-tls 不出现</automated>
  </verify>
  <acceptance_criteria>
    - `cargo build --lib` exits 0
    - `grep -q '^russh = ' Cargo.toml` 成功（带版本声明）
    - `grep -q '^russh-sftp = "2.3.0"$' Cargo.toml` 成功
    - `cargo tree | grep -E '^(openssl|native-tls) v'` 返回非零（即依赖树中无 openssl 或 native-tls，per CLAUDE.md §What NOT to Use）
    - `cargo tree -p russh` 输出含 `russh v0.61.2`
  </acceptance_criteria>
  <done>
    russh 与 russh-sftp 依赖加入，cargo build 通过，依赖树不含 openssl / native-tls。
  </done>
</task>

<task type="auto" tdd="true">
  <name>Task 3: cluster/ssh.rs — SshError 类型 + CommandRunner trait + 真实 SshSession 实现</name>
  <files>src/cluster/ssh.rs</files>
  <read_first>
    src/cluster/mod.rs,
    src/config/cluster.rs,
    .planning/phases/03-cluster/03-RESEARCH.md
  </read_first>
  <behavior>
    - Test 1 (SshError Display): `SshError::ExecFailed { command: "sudo -n true".into(), exit_code: 1 }.to_string()` 含子串 `"SSH 命令执行失败"` 和 `"exit 1"` 和 `"sudo -n true"`
    - Test 2 (SshError 类型完整): SshError 枚举含至少 3 个 variant：`Connect { host, source }`、`ExecFailed { command, exit_code }`、`SftpUpload { remote_path, source }`
    - Test 3 (CommandRunner trait 默认 mock 实现): 定义 `MockRunner` 结构体实现 `CommandRunner`，可预设 `(command_pattern, exit_code, stdout)` 元组列表；Test 调用 `MockRunner::new(vec![("sudo -n true".into(), 0, vec![])]).exec("sudo -n true").await` 返回 `Ok((vec![], 0))`
  </behavior>
  <action>
1. 新建 `src/cluster/ssh.rs`：
   - 顶部 imports：`use anyhow::Result;`、`use async_trait::async_trait;`（NOTE：若 russh 已重导出 async_trait 则用 `use russh::async_trait;`；否则在 Cargo.toml 加 `async-trait = "0.1"` 依赖，executor 视编译错误决定）、`use std::sync::Arc;`、`use thiserror::Error;`、`use russh::{client, ChannelMsg};`、`use russh_sftp::client::SftpSession;`
   - 定义 `pub enum SshError`，加 `#[derive(Debug, Error)]`，三个 variant：
     - `#[error("SSH 连接失败 {host}: {source}")]` `Connect { host: String, #[source] source: russh::Error }`
     - `#[error("SSH 命令执行失败 (exit {exit_code}): {command}")]` `ExecFailed { command: String, exit_code: u32 }`
     - `#[error("SFTP 上传失败 {remote_path}: {source}")]` `SftpUpload { remote_path: String, #[source] source: russh_sftp::client::error::Error }`（具体 error 类型 path 视 russh-sftp 0.61 重导出而定，executor 按 docs.rs 调整）
   - 定义 `pub trait CommandRunner: Send + Sync`，加 `#[async_trait]` 属性，方法：`async fn exec(&self, command: &str) -> Result<(Vec<u8>, u32), SshError>;`，方法：`async fn sftp_write(&self, remote_path: &str, bytes: &[u8]) -> Result<(), SshError>;`
   - 定义 `pub struct TofuHandler { pub accepted_keys: std::sync::Mutex<Vec<russh::keys::PublicKey>> }`（注意 0.61 中 PublicKey 路径），实现 `impl client::Handler for TofuHandler`：
     - `type Error = russh::Error;`
     - `async fn check_server_key(&mut self, key: &russh::keys::PublicKey) -> Result<bool, russh::Error> { self.accepted_keys.lock().unwrap().push(key.clone()); Ok(true) }`
   - 定义 `pub struct SshSession { handle: client::Handle<TofuHandler> }`，方法：
     - `pub async fn connect(host: &str, port: u16, user: &str, creds: &crate::config::cluster::SshCredentials) -> Result<Self, SshError>` — 使用 `client::Config::default()`，先尝试 `creds.identity_file`（用 `russh::keys::load_secret_key(path, None)` 加载），失败再尝试 `creds.password`；都失败则 bail。完整 connect 逻辑参考 RESEARCH §Pattern 1
     - 实现 `impl CommandRunner for SshSession`：`exec` 内部走 `handle.channel_open_session() → channel.exec(true, command) → loop channel.wait() 匹配 Data/ExitStatus/Eof|None` （per RESEARCH §Pattern 1 完整代码，注意 Pitfall 6：同时处理 Eof 和 None 才能 break）
     - `sftp_write`：`channel.request_subsystem(true, "sftp") → SftpSession::new(channel.into_stream()) → sftp.write(path, bytes)`（per RESEARCH §Pattern 2）

2. 新建 `pub struct MockRunner { pub responses: std::sync::Mutex<Vec<(String, i32, Vec<u8>)>> }`（cfg(test) 或 pub 都可——选 `pub` 让 Plan 03 也能用于集成测试 mock）：
   - `pub fn new(responses: Vec<(String, u32, Vec<u8>)>) -> Self`
   - 实现 `impl CommandRunner for MockRunner` — `exec` 按 command 字符串前缀匹配 responses 列表第一项（match 后弹出），返回对应 `(stdout, exit_code)`；未匹配返回 `SshError::ExecFailed { command, exit_code: 127 }`
   - `sftp_write` 简单记录到内部 `Vec<(String, Vec<u8>)>`，总是返回 Ok

3. 在 `src/cluster/mod.rs` 中新增 `pub mod ssh;`（与 Plan 01 创建的 `pub mod templates;` 并列）

4. 末尾加 `#[cfg(test)] mod tests`：
   - Test 1 (SshError Display): 构造 `SshError::ExecFailed { command: "sudo -n true".into(), exit_code: 1 }`，`assert!(err.to_string().contains("sudo -n true"))`、`assert!(err.to_string().contains("exit 1"))`、`assert!(err.to_string().contains("SSH 命令执行失败"))`
   - Test 2 (MockRunner 匹配): 构造 `MockRunner::new(vec![("sudo -n true".into(), 0, vec![])])`，调用 `runner.exec("sudo -n true").await`，断言返回 `Ok((vec![], 0))`
   - Test 3 (MockRunner 未匹配): 空 responses 调用 exec 返回 Err 含 exit_code 127

5. 所有函数体 < 40 行；TofuHandler、SshSession::connect、SshSession 的 exec 实现可能较长，必要时拆为 private helper（如 `fn try_key_auth`、`fn try_password_auth`）
  </action>
  <verify>
    <automated>cargo test --lib cluster::ssh --no-fail-fast 2>&amp;1 | tee /tmp/p3-02-t3.log; grep -E "(test result: ok\.|FAILED)" /tmp/p3-02-t3.log</automated>
  </verify>
  <acceptance_criteria>
    - `cargo test --lib cluster::ssh` exits 0
    - `grep -q '^pub enum SshError' src/cluster/ssh.rs`
    - `grep -q '^pub trait CommandRunner' src/cluster/ssh.rs`
    - `grep -q '^pub struct SshSession' src/cluster/ssh.rs`
    - `grep -q '^pub struct TofuHandler' src/cluster/ssh.rs`
    - `grep -q '^pub struct MockRunner' src/cluster/ssh.rs`
    - `src/cluster/mod.rs` 含 `pub mod ssh;`（grep -q "^pub mod ssh;" src/cluster/mod.rs）
    - SshError 三个 variant 名严格为 `Connect`、`ExecFailed`、`SftpUpload`（grep -cE '^\s+(Connect|ExecFailed|SftpUpload) ' src/cluster/ssh.rs ≥ 3）
    - cargo build 通过（无 dead_code warning 除外）
  </acceptance_criteria>
  <done>
    SshError + CommandRunner trait + SshSession（真实 russh 实现）+ TofuHandler + MockRunner 全部就位，3 个单元测试通过，后续 preflight / health 可基于 CommandRunner 注入 mock 测试。
  </done>
</task>

<task type="auto" tdd="true">
  <name>Task 4: cluster/preflight.rs — 三项预检查（QUAL-01）</name>
  <files>src/cluster/preflight.rs</files>
  <read_first>
    src/cluster/ssh.rs,
    src/config/cluster.rs,
    .planning/phases/03-cluster/03-RESEARCH.md
  </read_first>
  <behavior>
    - Test 1 (全通过): MockRunner 预设三条命令均返回 exit_code 0（sudo -n true → 0; ss -tlnp 输出不含 `:5236` → 0 with empty stdout; df -B1 stdout 含 `Available` 列值 ≥ 5 GB）；`check_node` 返回 Ok
    - Test 2 (sudo 失败): MockRunner 第一条 sudo -n true 返回 exit_code 1；check_node 返回 Err，错误消息含 host 字符串和 `"sudo 免密"` 关键字
    - Test 3 (端口被占): MockRunner ss 命令 stdout 含 `:5236 LISTEN`；check_node 返回 Err，消息含 `"端口 5236 已被占用"`
    - Test 4 (磁盘不足): MockRunner df 命令 stdout 表示剩余 ≤ 5 GB（如 "1000000"）；check_node 返回 Err，消息含 `"磁盘空间不足"` 和 `">= 5 GB"` 或 `"5 GB"`
    - Test 5 (多节点并发): 用 `preflight_all_nodes(items)` 接受 `Vec<(NodeConfig, Arc<dyn CommandRunner>)>`，全通过时返回 Ok，一节点失败时返回 Err 含所有失败节点 host。**注：** 统一使用 `Arc<dyn CommandRunner>`（非 `Box<dyn>`），允许 Plan 03 在 preflight / 安装 / 启动多个步骤复用同一 runner 实例（per Plan 03 编排需求）
  </behavior>
  <action>
1. 新建 `src/cluster/preflight.rs`：
   - imports：`use anyhow::{bail, Result, Context};`、`use crate::cluster::ssh::CommandRunner;`、`use crate::config::cluster::NodeConfig;`、`use futures::future::join_all;`、`use std::path::Path;`
   - `pub async fn check_sudo_nopass(runner: &dyn CommandRunner) -> Result<()>`:
     调用 `runner.exec("sudo -n true").await?`，若 exit_code != 0 则 `bail!("[预检查] sudo 免密失败 (exit {})", exit_code)`
   - `pub async fn check_port_available(runner: &dyn CommandRunner, port: u16) -> Result<()>`:
     调用 `runner.exec(&format!("ss -tlnp | grep ':{}'", port)).await?`，若 stdout 字节数 > 0（有匹配行）则 `bail!("[预检查] 端口 {} 已被占用", port)`；exit_code 在 grep 无匹配时为 1 但 stdout 空，不算错误
   - `pub async fn check_disk_space(runner: &dyn CommandRunner, install_path: &str) -> Result<()>`:
     a. 计算父目录：`let parent = std::path::Path::new(install_path).parent().unwrap_or_else(|| Path::new("/"))`；若 install_path 为 `/opt/dmdbms` 则父为 `/opt`，若不存在则用 `/`（Pitfall 7）
     b. 调用 `runner.exec(&format!("df -B1 {}", parent.display())).await?`
     c. 解析 stdout：第二行（跳过 header）按空白分割，取第 4 列（Available bytes）；不可解析则 bail
     d. 若 `available < 5 * 1024 * 1024 * 1024` 则 `bail!("[预检查] 磁盘空间不足: 剩余 {} bytes，需要 >= 5 GB", available)`
   - `pub async fn check_node(node: &NodeConfig, runner: &dyn CommandRunner) -> Result<()>`:
     依次调用三项检查，任一 Err 则用 `.with_context(|| format!("节点 {} ({:?}) 预检查失败", node.host, node.role))?` 包装
   - `pub async fn preflight_all_nodes(items: Vec<(NodeConfig, std::sync::Arc<dyn CommandRunner>)>) -> Result<()>`:
     **签名标准化为 `Arc<dyn CommandRunner>`**（per Plan 03 接口契约）。用 `join_all` 并发执行；每个任务内调用 `check_node(&node, runner.as_ref())`；收集 `Vec<Result<()>>`；过滤失败项；若有失败：
     ```
     let failures: Vec<String> = results.iter().enumerate()
         .filter_map(|(i, r)| r.as_ref().err().map(|e| format!("[{}] {:#}", items[i].0.host, e)))
         .collect();
     bail!("预检查失败 — 中止部署:\n{}", failures.join("\n"));
     ```

2. 在 `src/cluster/mod.rs` 中新增 `pub mod preflight;`

3. 末尾 `#[cfg(test)] mod tests`：实现 5 个 Test，全部用 `MockRunner` 注入预设响应。关键 fixture：用 `NodeConfig::default()` 或手工构造，设 `host = "192.168.1.10".into()`、`port = 5236`、`install_path = "/opt/dmdbms".into()`。
   - Test 5 多节点：构造两个 MockRunner（一个全成功 / 一个 sudo 失败），用 `std::sync::Arc::new(...)` 包装为 `Arc<dyn CommandRunner>`，传入 preflight_all_nodes（Vec 元素类型 `(NodeConfig, Arc<dyn CommandRunner>)`）
  </action>
  <verify>
    <automated>cargo test --lib cluster::preflight --no-fail-fast 2>&amp;1 | tee /tmp/p3-02-t4.log; grep -E "(test result: ok\.|FAILED)" /tmp/p3-02-t4.log</automated>
  </verify>
  <acceptance_criteria>
    - `cargo test --lib cluster::preflight` exits 0
    - `grep -q '^pub async fn check_sudo_nopass' src/cluster/preflight.rs`
    - `grep -q '^pub async fn check_port_available' src/cluster/preflight.rs`
    - `grep -q '^pub async fn check_disk_space' src/cluster/preflight.rs`
    - `grep -q '^pub async fn check_node' src/cluster/preflight.rs`
    - `grep -q '^pub async fn preflight_all_nodes' src/cluster/preflight.rs`
    - `grep -q 'Arc<dyn CommandRunner>' src/cluster/preflight.rs`（验证签名统一为 Arc，per Plan 03 契约）
    - `src/cluster/mod.rs` 含 `pub mod preflight;`
    - `grep -c 'join_all' src/cluster/preflight.rs` ≥ 1（验证 D-08 并发执行）
    - 所有 pub fn 函数体 < 40 行（结合 helper 拆分）
  </acceptance_criteria>
  <done>
    QUAL-01 三项预检查（sudo 免密 / 端口可用 / 磁盘空间）全部实现并通过 mock 单元测试；多节点并发通过 `join_all` + `preflight_all_nodes` 实现；失败时报告所有失败节点。
  </done>
</task>

<task type="auto" tdd="true">
  <name>Task 5: cluster/health.rs — TCP 健康轮询（CLUS-02）</name>
  <files>src/cluster/health.rs</files>
  <read_first>
    src/install/checksum.rs,
    .planning/phases/03-cluster/03-RESEARCH.md
  </read_first>
  <behavior>
    - Test 1 (超时返回 Err): 调用 `wait_tcp_ready("127.0.0.1", 1u16, 2)`（无人监听的 1 端口，max_secs=2），返回 Err 在 ~2 秒内，错误消息含 `"127.0.0.1:1"` 和 `"2s"`
    - Test 2 (立即就绪): 使用 `tokio::net::TcpListener::bind("127.0.0.1:0")` 在测试中起一个监听者，传入分配到的实际端口，调用 `wait_tcp_ready("127.0.0.1", port, 5)`，返回 Ok 在 < 200ms（不需要完整等到 3s 轮询间隔，首次连接即成功）
  </behavior>
  <action>
1. 新建 `src/cluster/health.rs`：
   - imports：`use anyhow::{anyhow, Result};`、`use tokio::net::TcpStream;`、`use tokio::time::{sleep, timeout, Duration};`
   - `pub async fn wait_tcp_ready(host: &str, port: u16, max_secs: u64) -> Result<()>`:
     - 内部 const `POLL_INTERVAL: Duration = Duration::from_secs(3)`（per D-09）
     - 构造 `let addr = format!("{}:{}", host, port);`
     - `let result = timeout(Duration::from_secs(max_secs), poll_loop(&addr, POLL_INTERVAL)).await;`
     - `result.map_err(|_| anyhow!("主节点 {} 在 {}s 内未就绪", addr, max_secs))?`
   - private `async fn poll_loop(addr: &str, interval: Duration) -> ()`:
     - `loop { match TcpStream::connect(addr).await { Ok(_) => return, Err(_) => sleep(interval).await } }`
     - 返回 () 给外层 timeout 处理超时

2. 在 `src/cluster/mod.rs` 中新增 `pub mod health;`

3. 末尾 `#[cfg(test)] mod tests`：
   - Test 1：`#[tokio::test] async fn test_wait_tcp_ready_times_out()` — 调用 wait_tcp_ready("127.0.0.1", 1, 2) 并断言 Err；用 `std::time::Instant::now()` 测量耗时，确保 ≥ 1.5s 且 ≤ 3.5s（避免 flaky）
   - Test 2：`#[tokio::test] async fn test_wait_tcp_ready_immediate_success()`：
     ```
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
     let port = listener.local_addr().unwrap().port();
     let start = std::time::Instant::now();
     let result = wait_tcp_ready("127.0.0.1", port, 5).await;
     assert!(result.is_ok());
     assert!(start.elapsed() < Duration::from_millis(500));
     ```

4. wait_tcp_ready 函数体 < 40 行（拆 poll_loop 为 helper 后只剩 timeout 调用 + 错误映射）
  </action>
  <verify>
    <automated>cargo test --lib cluster::health --no-fail-fast 2>&amp;1 | tee /tmp/p3-02-t5.log; grep -E "(test result: ok\.|FAILED)" /tmp/p3-02-t5.log</automated>
  </verify>
  <acceptance_criteria>
    - `cargo test --lib cluster::health` exits 0（两个测试通过）
    - `grep -q '^pub async fn wait_tcp_ready' src/cluster/health.rs`
    - `src/cluster/mod.rs` 含 `pub mod health;`
    - `grep -q 'POLL_INTERVAL' src/cluster/health.rs` 或 `grep -q 'from_secs(3)' src/cluster/health.rs`（验证 D-09 间隔 3 秒）
    - Test 1 测试运行耗时 ≥ 1.5s 且 ≤ 3.5s（用 grep 检查测试断言：`grep -q 'Duration::from_secs(2)' src/cluster/health.rs` 用于 max_secs=2 的 fixture）
    - 完整 cargo build 通过
  </acceptance_criteria>
  <done>
    CLUS-02 的 TCP 健康判据原语完成，超时路径与立即就绪路径均有单元测试验证；为 Plan 03 的有序启动提供可调用的 wait_tcp_ready 函数。
  </done>
</task>

</tasks>

<threat_model>
## Trust Boundaries

| Boundary | Description |
|----------|-------------|
| 控制机 → 远端 SSH server | 出站连接；首次连接需信任未知主机密钥（D-07 TOFU） |
| 远端节点 stdout → 控制机 | 远端 shell 输出进入 preflight 解析路径（df 输出整数解析） |
| TOML 解析后的 SshCredentials.password → russh 鉴权 | 内存中传递，不写日志 |

## STRIDE Threat Register

| Threat ID | Category | Component | Disposition | Mitigation Plan |
|-----------|----------|-----------|-------------|-----------------|
| T-03-04 | Spoofing | 中间人攻击（远端 SSH server 假冒） | accept | D-07 TOFU 策略；首次连接接受任何主机密钥并记入内存，MVP 文档化此假设。生产环境应预填 known_hosts（Phase v2 增强） |
| T-03-05 | Information Disclosure | identity_file 路径或 password 打印到 tracing 日志 | mitigate | ssh.rs 的 connect 函数禁止 `tracing::info!("creds: {:?}", creds)` 模式；只 log host/user，不 log 凭据；SshCredentials 已在 Plan 01 加 `#[serde(skip_serializing)]` |
| T-03-06 | Tampering | 远端命令注入（如 NodeConfig.host 含 `;` 或 shell 元字符进入 SSH exec 字符串） | mitigate | preflight 中 ss/df 命令拼接 `port: u16` 和 `Path` 类型（非任意字符串），类型层防止注入；host 仅用于 SSH connect 的 socket 地址，不进 exec 字符串。Plan 03 SFTP 路径用 `PathBuf::join` 避免 shell 拼接 |
| T-03-07 | Denial of Service | TCP 健康轮询无限循环 | mitigate | `tokio::time::timeout` 强制超时（D-09 默认 60s）；wait_tcp_ready 函数签名要求 max_secs 参数 |
| T-03-SC | Tampering | russh / russh-sftp 供应链 | mitigate | Task 1 blocking-human checkpoint 强制核验 crates.io 包合法性（标记 [ASSUMED]，slopcheck 不可用） |
</threat_model>

<verification>
1. `cargo build --lib` exits 0
2. `cargo tree | grep -E '^(openssl|native-tls) v'` 返回非零（依赖树不含 OpenSSL）
3. `cargo test --lib cluster::ssh` exits 0（3 个测试）
4. `cargo test --lib cluster::preflight` exits 0（5 个测试）
5. `cargo test --lib cluster::health` exits 0（2 个测试）
6. `cargo test --lib` 整库不破坏 Phase 2 测试
</verification>

<success_criteria>
- russh + russh-sftp 依赖加入，依赖树不引入 C FFI
- SshError 含 3 个 variant，CommandRunner trait + 真实 SshSession + MockRunner 三套实现
- TofuHandler 实现 client::Handler，check_server_key 无条件 Ok(true)（D-07）
- preflight 三项检查 + check_node + preflight_all_nodes（并发）全部就位
- wait_tcp_ready 超时路径（D-09 60s 默认 + 3s 间隔）就位
- 10 个单元测试全部通过（3 + 5 + 2 = 10）
- 所有 pub fn 函数体 < 40 行
</success_criteria>

<output>
Create `.planning/phases/03-cluster/03-02-SUMMARY.md` when done. Record:
- Cargo.toml 实际新增依赖行（含 features）
- SshError 三个 variant 的最终错误消息格式
- CommandRunner trait 方法签名（供 Plan 03 调用）
- MockRunner 公共构造和使用示例（供 Plan 03 集成测试参考）
- wait_tcp_ready 实测耗时（Test 1 / Test 2 各自时长）
</output>
</content>
</invoke>