feat(#60): 一键部署编排与回滚引擎

纯标准库实现，对齐 PRD 5.6「⑥ 部署底座」（不真执行 k8s/helm，模拟编排）：

- deploy_plan.py：DeployStep（4 类：pre_check/deploy/health_check/post_check）
  + DeployPlan（顺序约束自动校验，default_helm_release 默认 4 步计划）+
  DeployOrchestrator（按序执行，必需步骤失败即中止并标记 needs_rollback，
  deploy/health_check 为回滚检查点，pre_check 失败无需回滚）+
  HealthCheckContract（对齐 deploy/k8s/healthz 99.8% 目标）。
- rollback.py：RollbackPoint（部署前快照，values_hash SHA1 检测漂移）+
  RollbackManager（快照栈 + rollback_to_latest/rollback(n)，失败压回栈顶
  保持一致性，FIFO 淘汰，支持回滚后健康检查）+ RollbackResult（4 状态）。
- tests：30 用例覆盖计划构造/顺序约束、全成功/各类失败、回滚检查点判定、
  action 异常、dry_run、空计划、快照压栈/FIFO、回滚成功/失败/多版本/空栈/
  异常、健康检查契约、values 漂移、报告序列化。
- _sanity_check.py：冒烟验证快照→部署成功→部署失败→回滚全流程。

deploy/orchestration/README.md

@@ -0,0 +1,66 @@
+# 一键部署编排 + 回滚引擎（Issue #60 / PRD 5.6）
+
+> 父 Issue「⑥ 一键部署脚本与回滚机制 · 0.5d」
+
+把一键部署与回滚落为**可测试的纯标准库编排引擎**——不真执行 k8s/helm，但模拟
+完整编排序列（pre_check → deploy → health_check → post_check）+ 部署前快照 +
+失败回滚。对齐 `deploy/k8s/helm/iaop`（Chart）与 `deploy/k8s/healthz`（探针）。
+
+## 1. 部署编排（`deploy_plan.py`）
+
+- `DeployStep` —— 声明式步骤：`kind`（pre_check/deploy/health_check/post_check）
+  + `name` + `action` 回调（返回 `(ok, detail)`，默认模拟成功）。
+- `DeployPlan` —— 步骤序列 + 发布元信息（release/namespace/chart/version）；
+  `default_helm_release()` 生成默认 4 步 Helm 发布计划；步骤顺序约束
+  （pre_check < deploy < health_check < post_check）自动校验。
+- `DeployOrchestrator` —— 按序执行，任何必需步骤失败即中止并标记回滚
+  （`DeployOutcome.needs_rollback`），产出 `DeployReport`（逐步结果 + 日志 +
+  可解释 reason）。deploy/health_check 是回滚检查点；pre_check 失败无需回滚。
+- `HealthCheckContract` —— 健康检查契约（端点/超时/重试/可用率目标，对齐
+  `deploy/k8s/healthz` 的 99.8%）。
+
+```python
+from deploy_plan import DeployPlan, DeployOrchestrator
+
+plan = DeployPlan.default_helm_release(release="iaop", backend="gpu")
+report = DeployOrchestrator(plan).execute()
+if not report.succeeded and report.outcome.needs_rollback:
+    print("需回滚：", report.outcome.reason)
+```
+
+## 2. 回滚引擎（`rollback.py`）
+
+- `RollbackPoint` —— 部署前快照（release/namespace/version/chart/values_hash/
+  created_at，不可变）。`values_hash`（SHA1）检测配置漂移。
+- `RollbackManager` —— 快照栈 + 回滚执行：`snapshot(plan)` 部署前压栈；
+  `rollback_to_latest()` / `rollback(n)` 失败时回滚到稳定版本。回滚失败把目标压回
+  栈顶（保持栈一致性）；支持回滚后健康检查。FIFO 淘汰防内存膨胀。
+- `RollbackResult` —— 回滚结果（SUCCESS/FAILED/NO_TARGET/SKIPPED + 可解释 reason）。
+
+```python
+from deploy_plan import DeployPlan, DeployOrchestrator
+from rollback import RollbackManager
+
+mgr = RollbackManager()
+mgr.snapshot(plan)              # 部署前快照
+report = DeployOrchestrator(plan).execute()
+if not report.succeeded:
+    result = mgr.rollback_to_latest(reason=report.outcome.reason)
+    print(result.status.value, result.detail)
+```
+
+## 与现有部署资产的关系
+
+- 真实环境把 `action` / `restore_action` 注入为 `helm upgrade` / `helm rollback`
+  等真回调即可；本引擎只做编排与状态管理，不耦合 k8s 客户端。
+- 健康检查契约对齐 `deploy/k8s/healthz/probe_availability.py`（99.8% 可用率目标）。
+
+## 测试
+
+```bash
+python -m unittest discover -s deploy/orchestration/tests -p "test_*.py" -v
+```
+
+覆盖正常 + 边界 + 错误（30 用例）：4 步计划构造、顺序约束、全成功/各类失败、
+回滚检查点判定、action 异常、dry_run、空计划、快照压栈/FIFO、回滚成功/失败/
+多版本/空栈/异常、健康检查契约校验、values 漂移、报告序列化。

deploy/orchestration/__init__.py

@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+"""一键部署编排 + 回滚引擎包（Issue #60 / PRD 5.6「⑥ 部署底座」）。
+
+把部署与回滚落为**可测试的纯标准库编排引擎**——不真执行 k8s/helm，但模拟
+完整编排序列（pre_check → deploy → health_check → post_check）+ 部署前快照
++ 失败回滚。零运行时依赖，与 iAOP 部署底座一致。
+"""
+from .deploy_plan import (
+    DeployOrchestrator,
+    DeployOutcome,
+    DeployPlan,
+    DeployReport,
+    DeployStep,
+    DeployStepKind,
+    DeployStepResult,
+    DeployStatus,
+    HealthCheckContract,
+)
+from .rollback import (
+    RollbackManager,
+    RollbackPoint,
+    RollbackResult,
+    RollbackStatus,
+)
+
+__all__ = [
+    # deploy_plan
+    "DeployPlan",
+    "DeployStep",
+    "DeployStepKind",
+    "DeployStepResult",
+    "DeployStatus",
+    "DeployOutcome",
+    "DeployReport",
+    "DeployOrchestrator",
+    "HealthCheckContract",
+    # rollback
+    "RollbackPoint",
+    "RollbackManager",
+    "RollbackResult",
+    "RollbackStatus",
+]

deploy/orchestration/_sanity_check.py

@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+"""部署编排 + 回滚冒烟脚本（Issue #60）。
+
+直接运行 ``python _sanity_check.py`` 验证：默认 Helm 计划全成功部署、
+快照压栈、模拟部署失败触发回滚到上一稳定版本。零第三方依赖。
+"""
+import os
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+
+from deploy_plan import (  # noqa: E402
+    DeployOrchestrator, DeployPlan, DeployStepKind)
+from rollback import RollbackManager  # noqa: E402
+
+
+def main() -> int:
+    # 1) 部署前快照（当前稳定版本 v1.0.0）
+    mgr = RollbackManager()
+    stable = DeployPlan.default_helm_release(version="v1.0.0")
+    mgr.snapshot(stable, reason="部署 v1.1.0 前的稳定版本")
+    print(f"[OK] 快照压栈：稳定版本 {mgr.latest().version}")
+
+    # 2) 部署 v1.1.0 全成功
+    new_plan = DeployPlan.default_helm_release(version="v1.1.0")
+    report = DeployOrchestrator(new_plan).execute()
+    assert report.succeeded, report.outcome.reason
+    mgr.snapshot(new_plan, reason="v1.1.0 部署成功")
+    print(f"[OK] 部署 v1.1.0 成功：{len(report.results)} 步，无需回滚")
+
+    # 3) 模拟 v1.2.0 部署失败（health_check 失败）→ 触发回滚
+    def _bad_health(ctx):
+        return False, "/health 503（服务未就绪）"
+    bad_plan = DeployPlan.default_helm_release(
+        version="v1.2.0",
+        step_actions={DeployStepKind.HEALTH_CHECK: _bad_health})
+    bad_report = DeployOrchestrator(bad_plan).execute()
+    assert not bad_report.succeeded
+    assert bad_report.outcome.needs_rollback
+    print(f"[OK] v1.2.0 部署失败：{bad_report.outcome.reason}")
+
+    result = mgr.rollback_to_latest(reason="v1.2.0 健康检查失败")
+    assert result.succeeded, result.detail
+    print(f"[OK] 回滚到 {result.target.version}：{result.detail}")
+    print("部署编排 + 回滚冒烟通过 ✅")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

deploy/orchestration/deploy_plan.py

@@ -0,0 +1,428 @@
+# -*- coding: utf-8 -*-
+"""一键部署编排引擎（Issue #60 / PRD 5.6「⑥ 部署底座」）。
+
+PRD 5.6 验收：一套 Helm Chart 部署内核 + 模板，GPU/NPU 推理后端可插拔。
+本模块把部署落为**可测试的编排引擎**——把部署拆为有序步骤序列
+（pre_check → deploy → health_check → post_check），编排器按序执行（模拟，
+不真调 k8s/helm），逐步产出步骤日志与状态，失败即触发回滚。
+
+设计要点
+--------
+1. **部署即编排序列**（``DeployStep``）：每步声明 ``kind``（四类）+ ``name``
+   + 可执行动作（``action`` 回调，返回 ok/detail）；步骤序列即 :class:`DeployPlan`。
+   - ``pre_check``   部署前检查（集群可达 / 镜像存在 / values 合法 / 资源配额）；
+   - ``deploy``      部署动作（helm install/upgrade，模拟）；
+   - ``health_check``健康检查（探活 /health，对齐 ``deploy/k8s/healthz``）；
+   - ``post_check``  部署后验证（业务接口回归 / 监控接入）。
+2. **编排器**（``DeployOrchestrator``）：按序执行步骤，任何一步失败即中止并
+   标记需回滚（``DeployOutcome.needs_rollback``），产出 :class:`DeployReport`
+   （逐步结果 + 总状态 + 可解释 reason）。
+3. **健康检查契约**（``HealthCheckContract``）：定义健康检查的判定口径
+   （超时/重试/目标可用率，对齐 ``deploy/k8s/healthz`` 的 99.8% 目标）。
+4. **纯标准库**：action 为回调，默认实现模拟成功；真实环境注入真 action 即可。
+5. **不真执行 k8s/helm**：编排只产日志与状态，便于离线测试与 CI 集成。
+
+用法::
+
+    plan = DeployPlan.default_helm_release(release="iaop", namespace="iaop")
+    report = DeployOrchestrator(plan).execute()
+    if not report.outcome.succeeded:
+        print("需回滚：", report.outcome.reason)
+"""
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Callable, Dict, List, Optional, Sequence, Tuple
+
+#: 健康检查默认超时（秒，对齐 deploy/k8s/healthz 探针口径）。
+DEFAULT_HEALTH_TIMEOUT_S = 30.0
+#: 健康检查默认重试次数。
+DEFAULT_HEALTH_RETRIES = 3
+#: 可用率目标（对齐 deploy/k8s/healthz probe_availability 的 99.8%）。
+DEFAULT_AVAILABILITY_TARGET = 0.998
+
+
+class DeployError(ValueError):
+    """部署计划声明/编排错误（步骤序列非法、kind 未知、回调异常等）。"""
+
+
+class DeployStepKind(str, Enum):
+    """部署步骤类型（决定执行顺序与失败后果）。"""
+
+    PRE_CHECK = "pre_check"        # 部署前检查
+    DEPLOY = "deploy"              # 部署动作
+    HEALTH_CHECK = "health_check"  # 健康检查
+    POST_CHECK = "post_check"      # 部署后验证
+
+    @property
+    def label(self) -> str:
+        return {
+            DeployStepKind.PRE_CHECK: "部署前检查",
+            DeployStepKind.DEPLOY: "部署动作",
+            DeployStepKind.HEALTH_CHECK: "健康检查",
+            DeployStepKind.POST_CHECK: "部署后验证",
+        }[self]
+
+    @property
+    def is_checkpoint(self) -> bool:
+        """该步失败是否触发回滚（deploy/health_check 失败必须回滚）。"""
+        return self in (DeployStepKind.DEPLOY, DeployStepKind.HEALTH_CHECK)
+
+
+#: 步骤动作回调：返回 (ok, detail)。
+StepAction = Callable[[Dict[str, object]], Tuple[bool, str]]
+
+
+def _default_action(ctx: Dict[str, object]) -> Tuple[bool, str]:
+    """默认动作：模拟成功（真实环境注入真回调）。"""
+    return True, "模拟成功（无 k8s/helm 环境）"
+
+
+@dataclass
+class DeployStep:
+    """单个部署步骤（声明式）。
+
+    Attributes:
+        kind:     步骤类型。
+        name:     步骤名（可读，如 "helm upgrade"）。
+        action:   可执行动作（ctx → (ok, detail)）；默认模拟成功。
+        required: 是否必需（必需步骤失败即中止并标记回滚；非必需失败仅告警）。
+        detail:   步骤说明（引 PRD/SOP，可解释）。
+    """
+
+    kind: DeployStepKind
+    name: str
+    action: StepAction = field(default=_default_action, repr=False)
+    required: bool = True
+    detail: str = ""
+
+    def __post_init__(self) -> None:
+        if not self.name:
+            raise DeployError("DeployStep.name 不能为空")
+        if not isinstance(self.kind, DeployStepKind):
+            raise DeployError(f"kind 必须是 DeployStepKind，实际 {type(self.kind)}")
+
+
+@dataclass
+class HealthCheckContract:
+    """健康检查契约（对齐 deploy/k8s/healthz 的探针口径）。
+
+    Attributes:
+        endpoint:             健康端点（如 http://iaop:8000/health）。
+        timeout_s:            单次探测超时。
+        retries:              重试次数（窗口内达到目标可用率即通过）。
+        availability_target:  可用率目标（默认 0.998 = 99.8%）。
+    """
+
+    endpoint: str = "http://iaop:8000/health"
+    timeout_s: float = DEFAULT_HEALTH_TIMEOUT_S
+    retries: int = DEFAULT_HEALTH_RETRIES
+    availability_target: float = DEFAULT_AVAILABILITY_TARGET
+
+    def __post_init__(self) -> None:
+        if self.timeout_s <= 0:
+            raise DeployError(f"timeout_s 必须 > 0，实际 {self.timeout_s}")
+        if self.retries < 1:
+            raise DeployError(f"retries 必须 ≥ 1，实际 {self.retries}")
+        if not (0.0 < self.availability_target <= 1.0):
+            raise DeployError(
+                f"availability_target 须在 (0,1]，实际 {self.availability_target}")
+
+
+@dataclass
+class DeployPlan:
+    """部署计划：步骤序列 + 发布元信息。
+
+    步骤顺序约束：pre_check 必须在 deploy 前；health_check 必须在 deploy 后、
+    post_check 前。``__post_init__`` 校验该顺序，违例即拒绝（防编排漂移）。
+
+    Attributes:
+        name:      计划名（如 "iaop-helm-upgrade"）。
+        release:   Helm release 名（如 "iaop"）。
+        namespace: 命名空间。
+        chart:     Chart 引用（如 "deploy/k8s/helm/iaop"）。
+        version:   目标版本（appVersion，如 "v1.0.0"）。
+        steps:     步骤序列（按执行顺序）。
+        health:    健康检查契约（health_check 步骤引用）。
+    """
+
+    name: str
+    release: str = "iaop"
+    namespace: str = "iaop"
+    chart: str = "deploy/k8s/helm/iaop"
+    version: str = "v1.0.0"
+    steps: List[DeployStep] = field(default_factory=list)
+    health: HealthCheckContract = field(default_factory=HealthCheckContract)
+
+    def __post_init__(self) -> None:
+        if not self.name:
+            raise DeployError("DeployPlan.name 不能为空")
+        self._validate_order()
+
+    # ------------------------------------------------------------------
+    def _validate_order(self) -> None:
+        """校验步骤顺序：pre_check < deploy < health_check < post_check。"""
+        if not self.steps:
+            return  # 空计划在 execute() 时报错
+        order_rank = {
+            DeployStepKind.PRE_CHECK: 0,
+            DeployStepKind.DEPLOY: 1,
+            DeployStepKind.HEALTH_CHECK: 2,
+            DeployStepKind.POST_CHECK: 3,
+        }
+        prev_rank = -1
+        for s in self.steps:
+            r = order_rank[s.kind]
+            if r < prev_rank:
+                raise DeployError(
+                    f"步骤顺序违例：{s.kind.value}({s.name}) 不能排在"
+                    f" rank {prev_rank} 之后（pre_check < deploy < "
+                    f"health_check < post_check）")
+            prev_rank = r
+
+    # ------------------------------------------------------------------
+    @classmethod
+    def default_helm_release(
+        cls,
+        release: str = "iaop",
+        namespace: str = "iaop",
+        version: str = "v1.0.0",
+        backend: str = "gpu",
+        health: Optional[HealthCheckContract] = None,
+        step_actions: Optional[Dict[DeployStepKind, StepAction]] = None,
+    ) -> "DeployPlan":
+        """默认 Helm 发布计划（pre_check → deploy → health_check → post_check）。
+
+        Args:
+            release:      release 名。
+            namespace:    命名空间。
+            version:      目标 appVersion。
+            backend:      推理后端（gpu|npu，对齐 PRD 5.6 配置点）。
+            health:       健康检查契约（默认 :class:`HealthCheckContract`）。
+            step_actions: 各类步骤的动作回调覆盖（默认模拟成功）。
+        """
+        actions = step_actions or {}
+        plan = cls(
+            name=f"{release}-helm-upgrade",
+            release=release,
+            namespace=namespace,
+            version=version,
+            health=health or HealthCheckContract(),
+            steps=[
+                DeployStep(
+                    kind=DeployStepKind.PRE_CHECK,
+                    name="pre-check-cluster-values",
+                    action=actions.get(DeployStepKind.PRE_CHECK, _default_action),
+                    detail="部署前检查：集群可达 / 镜像存在 / values 合法 / 资源配额"),
+                DeployStep(
+                    kind=DeployStepKind.DEPLOY,
+                    name=f"helm-upgrade-{backend}",
+                    action=actions.get(DeployStepKind.DEPLOY, _default_action),
+                    detail=f"helm upgrade --install {release}（backend={backend}，PRD 5.6）"),
+                DeployStep(
+                    kind=DeployStepKind.HEALTH_CHECK,
+                    name="health-check-probe",
+                    action=actions.get(DeployStepKind.HEALTH_CHECK, _default_action),
+                    detail="健康检查：探活 /health（目标可用率 99.8%，对齐 healthz）"),
+                DeployStep(
+                    kind=DeployStepKind.POST_CHECK,
+                    name="post-check-smoke",
+                    action=actions.get(DeployStepKind.POST_CHECK, _default_action),
+                    required=False,  # 烟测非阻断
+                    detail="部署后验证：业务接口烟测 / 监控接入（非阻断）"),
+            ],
+        )
+        return plan
+
+
+# ---------------------------------------------------------------------------
+# 执行结果
+# ---------------------------------------------------------------------------
+
+
+class DeployStatus(str, Enum):
+    """单步执行状态。"""
+
+    PENDING = "pending"
+    SUCCESS = "success"
+    FAILED = "failed"
+    SKIPPED = "skipped"
+
+
+@dataclass
+class DeployStepResult:
+    """单步执行结果。"""
+
+    step: DeployStep
+    status: DeployStatus = DeployStatus.PENDING
+    detail: str = ""
+    duration_ms: float = 0.0
+    index: int = 0
+
+
+@dataclass
+class DeployOutcome:
+    """部署总结果（可解释：是否成功 + 是否需回滚 + reason）。"""
+
+    succeeded: bool
+    needs_rollback: bool
+    failed_step: Optional[str] = None
+    reason: str = ""
+
+
+@dataclass
+class DeployReport:
+    """部署编排报告（逐步结果 + 总状态 + 日志行）。"""
+
+    plan_name: str
+    results: List[DeployStepResult] = field(default_factory=list)
+    outcome: Optional[DeployOutcome] = None
+    log_lines: List[str] = field(default_factory=list)
+    context: Dict[str, object] = field(default_factory=dict)
+
+    @property
+    def succeeded(self) -> bool:
+        return self.outcome.succeeded if self.outcome else False
+
+    def to_dict(self) -> dict:
+        return {
+            "plan_name": self.plan_name,
+            "succeeded": self.succeeded,
+            "needs_rollback": self.outcome.needs_rollback if self.outcome else False,
+            "reason": self.outcome.reason if self.outcome else "",
+            "steps": [
+                {"index": r.index, "kind": r.step.kind.value, "name": r.step.name,
+                 "status": r.status.value, "detail": r.detail,
+                 "duration_ms": round(r.duration_ms, 2)}
+                for r in self.results
+            ],
+            "log": list(self.log_lines),
+        }
+
+
+# ---------------------------------------------------------------------------
+# 编排器
+# ---------------------------------------------------------------------------
+
+
+class DeployOrchestrator:
+    """部署编排器：按序执行 DeployPlan 步骤，失败即中止并标记回滚。
+
+    Args:
+        plan:     部署计划。
+        dry_run:  空跑模式（不执行 action，全部标记 SKIPPED；用于计划校验）。
+        logger:   日志回调（默认 print 到内部 log_lines 缓冲）。
+    """
+
+    def __init__(
+        self,
+        plan: DeployPlan,
+        dry_run: bool = False,
+        logger: Optional[Callable[[str], None]] = None,
+    ) -> None:
+        self.plan = plan
+        self.dry_run = bool(dry_run)
+        self._log: List[str] = []
+        self._logger = logger
+
+    # ------------------------------------------------------------------
+    def _log_line(self, line: str) -> None:
+        self._log.append(line)
+        if self._logger is not None:
+            self._logger(line)
+
+    # ------------------------------------------------------------------
+    def execute(self, context: Optional[Dict[str, object]] = None) -> DeployReport:
+        """执行部署计划，返回报告。"""
+        report = DeployReport(plan_name=self.plan.name)
+        ctx: Dict[str, object] = {
+            "release": self.plan.release,
+            "namespace": self.plan.namespace,
+            "chart": self.plan.chart,
+            "version": self.plan.version,
+            "health": {
+                "endpoint": self.plan.health.endpoint,
+                "timeout_s": self.plan.health.timeout_s,
+                "retries": self.plan.health.retries,
+                "availability_target": self.plan.health.availability_target,
+            },
+        }
+        if context:
+            ctx.update(context)
+        report.context = dict(ctx)
+
+        if not self.plan.steps:
+            report.outcome = DeployOutcome(
+                succeeded=False, needs_rollback=False,
+                reason="部署计划无步骤（DeployPlan.steps 为空）")
+            self._log_line("[FAIL] 部署计划无步骤，未执行")
+            report.log_lines = list(self._log)
+            return report
+
+        self._log_line(
+            f"[START] 部署 {self.plan.name}：release={self.plan.release} "
+            f"namespace={self.plan.namespace} version={self.plan.version} "
+            f"steps={len(self.plan.steps)}")
+
+        failed_result: Optional[DeployStepResult] = None
+        for idx, step in enumerate(self.plan.steps):
+            result = DeployStepResult(step=step, index=idx)
+
+            if self.dry_run:
+                result.status = DeployStatus.SKIPPED
+                result.detail = "dry-run 跳过"
+                report.results.append(result)
+                self._log_line(
+                    f"[SKIP] #{idx} {step.kind.label}/{step.name}（dry-run）")
+                continue
+
+            self._log_line(f"[RUN] #{idx} {step.kind.label}/{step.name}")
+            started = time.monotonic()
+            try:
+                ok, detail = step.action(ctx)
+            except Exception as exc:  # noqa: BLE001 - 任意 action 异常视为失败
+                ok, detail = False, f"action 异常：{exc}"
+            result.duration_ms = (time.monotonic() - started) * 1000.0
+            result.detail = detail
+            result.status = DeployStatus.SUCCESS if ok else DeployStatus.FAILED
+            report.results.append(result)
+
+            tag = "OK" if ok else "FAIL"
+            self._log_line(
+                f"[{tag}] #{idx} {step.name}：{detail} "
+                f"({result.duration_ms:.0f}ms)")
+
+            if not ok:
+                if step.required:
+                    failed_result = result
+                    break
+                # 非必需步骤失败：告警但继续
+                self._log_line(
+                    f"[WARN] #{idx} {step.name} 非必需步骤失败，继续编排")
+
+        report.log_lines = list(self._log)
+
+        if self.dry_run:
+            report.outcome = DeployOutcome(
+                succeeded=True, needs_rollback=False,
+                reason=f"dry-run：{len(self.plan.steps)} 步全部跳过（计划校验通过）")
+            return report
+
+        if failed_result is not None:
+            needs_rollback = failed_result.step.kind.is_checkpoint
+            report.outcome = DeployOutcome(
+                succeeded=False,
+                needs_rollback=needs_rollback,
+                failed_step=failed_result.step.name,
+                reason=(
+                    f"步骤 #{failed_result.index} {failed_result.step.kind.label}/"
+                    f"{failed_result.step.name} 失败：{failed_result.detail}"
+                    + ("，需回滚到上一稳定版本" if needs_rollback
+                       else "（非回滚检查点，无需回滚）")))
+        else:
+            report.outcome = DeployOutcome(
+                succeeded=True, needs_rollback=False,
+                reason=f"全部 {len(report.results)} 步执行成功，部署完成")
+        return report

deploy/orchestration/rollback.py

@@ -0,0 +1,265 @@
+# -*- coding: utf-8 -*-
+"""部署回滚引擎（Issue #60 / PRD 5.6「⑥ 部署底座」）。
+
+PRD 5.6 一键部署需配套**回滚机制**——部署失败时恢复到部署前的稳定配置版本，
+保证可用性。本模块把回滚落为**可测试的纯标准库引擎**：部署前对当前配置版本
+打快照（:class:`RollbackPoint`），失败时按快照恢复（:class:`RollbackManager`），
+产出可解释的 :class:`RollbackResult`。
+
+设计要点
+--------
+1. **快照即配置版本**（``RollbackPoint``）：部署前记录当前稳定版本的元信息
+   （release/namespace/version/chart/values_hash/创建时间），存入快照栈。
+   快照不可变（只读），保证回滚目标确定。
+2. **快照栈**（``RollbackManager``）：每次成功部署压栈；失败时弹出最近快照执行
+   回滚。支持多版本回溯（rollback(n) 回滚 n 个版本）。
+3. **回滚动作**（``restore_action`` 回调）：默认模拟成功；真实环境注入
+   ``helm rollback`` 等真回调。回滚后重新健康检查（契约对齐 deploy_plan）。
+4. **纯标准库**：无 k8s/helm 依赖，便于离线测试与 CI 集成。
+
+用法::
+
+    mgr = RollbackManager()
+    mgr.snapshot(plan)              # 部署前快照
+    report = orchestrator.execute() # 部署
+    if not report.succeeded:
+        result = mgr.rollback_to_latest(reason=report.outcome.reason)
+        print(result.status, result.detail)
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Callable, Dict, List, Optional, Tuple
+
+from deploy_plan import (
+    DEFAULT_AVAILABILITY_TARGET,
+    DEFAULT_HEALTH_RETRIES,
+    DEFAULT_HEALTH_TIMEOUT_S,
+    DeployError,
+    DeployPlan,
+    HealthCheckContract,
+)
+
+#: 回滚动作回调：(snapshot, ctx) → (ok, detail)。
+RestoreAction = Callable[["RollbackPoint", Dict[str, object]], Tuple[bool, str]]
+
+
+def _default_restore(snapshot: "RollbackPoint",
+                     ctx: Dict[str, object]) -> Tuple[bool, str]:
+    """默认回滚动作：模拟成功（真实环境注入 helm rollback 回调）。"""
+    return True, f"模拟回滚到 {snapshot.version}（无 k8s/helm 环境）"
+
+
+class RollbackStatus(str, Enum):
+    """回滚执行状态。"""
+
+    SUCCESS = "success"     # 回滚成功（含回滚后健康检查通过）
+    FAILED = "failed"       # 回滚失败（restore 动作失败或健康检查不过）
+    NO_TARGET = "no_target" # 无可回滚快照（栈空）
+    SKIPPED = "skipped"     # 跳过（手动）
+
+
+@dataclass
+class RollbackPoint:
+    """部署前快照（不可变配置版本）。
+
+    Attributes:
+        release:     Helm release 名。
+        namespace:   命名空间。
+        version:     快照时的版本（appVersion，回滚目标）。
+        chart:       Chart 引用。
+        values_hash: values 配置的哈希（检测配置漂移）。
+        created_at:  快照创建时间戳（epoch 秒）。
+        reason:      快照说明（如 "部署 v1.1.0 前的稳定版本 v1.0.0"）。
+    """
+
+    release: str
+    namespace: str
+    version: str
+    chart: str
+    values_hash: str
+    created_at: float = field(default_factory=time.time)
+    reason: str = ""
+
+    def to_dict(self) -> dict:
+        return {
+            "release": self.release,
+            "namespace": self.namespace,
+            "version": self.version,
+            "chart": self.chart,
+            "values_hash": self.values_hash,
+            "created_at": self.created_at,
+            "reason": self.reason,
+        }
+
+
+@dataclass
+class RollbackResult:
+    """回滚执行结果（可解释：状态 + 目标版本 + reason）。"""
+
+    status: RollbackStatus
+    target: Optional[RollbackPoint] = None
+    detail: str = ""
+    reason: str = ""
+
+    @property
+    def succeeded(self) -> bool:
+        return self.status is RollbackStatus.SUCCESS
+
+    def to_dict(self) -> dict:
+        return {
+            "status": self.status.value,
+            "succeeded": self.succeeded,
+            "target": self.target.to_dict() if self.target else None,
+            "detail": self.detail,
+            "reason": self.reason,
+        }
+
+
+class RollbackManager:
+    """部署回滚管理器：快照栈 + 回滚执行。
+
+    Args:
+        restore_action:    回滚动作回调（默认模拟成功）。
+        post_rollback_health: 回滚后是否健康检查（默认 False，由编排器单独跑）。
+        max_snapshots:     快照栈上限（FIFO 淘汰，防内存膨胀；默认 10）。
+    """
+
+    def __init__(
+        self,
+        restore_action: RestoreAction = _default_restore,
+        post_rollback_health: bool = False,
+        max_snapshots: int = 10,
+    ) -> None:
+        if max_snapshots < 1:
+            raise DeployError(f"max_snapshots 必须 ≥ 1，实际 {max_snapshots}")
+        self._stack: List[RollbackPoint] = []
+        self.restore_action = restore_action
+        self.post_rollback_health = bool(post_rollback_health)
+        self.max_snapshots = int(max_snapshots)
+
+    # ------------------------------------------------------------------
+    # 快照栈
+    # ------------------------------------------------------------------
+    @property
+    def snapshots(self) -> List[RollbackPoint]:
+        """当前快照栈（栈顶 = 最近快照 = 列表末尾）。只读视图。"""
+        return list(self._stack)
+
+    def latest(self) -> Optional[RollbackPoint]:
+        """栈顶快照（最近一次成功版本）；栈空返回 None。"""
+        return self._stack[-1] if self._stack else None
+
+    def snapshot(self, plan: DeployPlan, values: Optional[Dict[str, object]] = None,
+                 reason: str = "") -> RollbackPoint:
+        """部署前打快照（压栈）。
+
+        Args:
+            plan:    部署计划（取 release/namespace/version/chart）。
+            values:  当前 values 配置（用于计算 values_hash，检测漂移）。
+            reason:  快照说明。
+        """
+        values_hash = _hash_values(values or {})
+        point = RollbackPoint(
+            release=plan.release,
+            namespace=plan.namespace,
+            version=plan.version,
+            chart=plan.chart,
+            values_hash=values_hash,
+            reason=reason or f"部署 {plan.name} 前快照（version={plan.version}）",
+        )
+        self._stack.append(point)
+        # FIFO 淘汰（保留最近 max_snapshots 个）
+        if len(self._stack) > self.max_snapshots:
+            self._stack = self._stack[-self.max_snapshots:]
+        return point
+
+    # ------------------------------------------------------------------
+    # 回滚
+    # ------------------------------------------------------------------
+    def rollback_to_latest(self, reason: str = "") -> RollbackResult:
+        """回滚到最近快照（弹出栈顶）。"""
+        return self._rollback(n=1, reason=reason)
+
+    def rollback(self, n: int = 1, reason: str = "") -> RollbackResult:
+        """回滚 n 个版本（弹出 n 个快照，回滚到第 n 个）。"""
+        return self._rollback(n=n, reason=reason)
+
+    def _rollback(self, n: int, reason: str) -> RollbackResult:
+        if n < 1:
+            return RollbackResult(
+                status=RollbackStatus.SKIPPED,
+                detail=f"n={n}，跳过回滚",
+                reason="回滚步数 n 必须 ≥ 1")
+        if len(self._stack) < n:
+            return RollbackResult(
+                status=RollbackStatus.NO_TARGET,
+                detail=f"快照栈仅 {len(self._stack)} 个，无法回滚 {n} 个版本",
+                reason="无可回滚的稳定版本快照")
+
+        # 弹出 n 个快照，回滚目标 = 第 n 个（最后弹出的）
+        popped: List[RollbackPoint] = []
+        for _ in range(n):
+            popped.append(self._stack.pop())
+        target = popped[-1]
+
+        ctx: Dict[str, object] = {
+            "release": target.release,
+            "namespace": target.namespace,
+            "version": target.version,
+            "reason": reason or f"部署失败，回滚到 {target.version}",
+        }
+        try:
+            ok, detail = self.restore_action(target, ctx)
+        except Exception as exc:  # noqa: BLE001
+            ok, detail = False, f"restore action 异常：{exc}"
+
+        if not ok:
+            # 回滚失败：把目标压回栈顶（保持快照栈一致性）
+            self._stack.append(target)
+            return RollbackResult(
+                status=RollbackStatus.FAILED,
+                target=target,
+                detail=detail,
+                reason=f"回滚到 {target.version} 失败：{detail}")
+
+        full_detail = detail
+        # 回滚后健康检查（可选）
+        if self.post_rollback_health:
+            health_ok, health_detail = self._health_check(target)
+            if not health_ok:
+                self._stack.append(target)
+                return RollbackResult(
+                    status=RollbackStatus.FAILED,
+                    target=target,
+                    detail=f"回滚动作成功但健康检查失败：{health_detail}",
+                    reason=f"回滚到 {target.version} 后健康检查未通过")
+            full_detail = f"{detail}；健康检查通过"
+
+        return RollbackResult(
+            status=RollbackStatus.SUCCESS,
+            target=target,
+            detail=full_detail,
+            reason=f"回滚到 {target.version} 成功"
+                   + (f"（{reason}）" if reason else ""))
+
+    # ------------------------------------------------------------------
+    def _health_check(self, target: RollbackPoint) -> Tuple[bool, str]:
+        """回滚后健康检查（模拟，默认通过；真实环境注入 restore_action 内）。"""
+        # 纯标准库模拟：回滚后假设服务恢复（真实健康检查由 deploy_plan 编排）
+        return True, f"回滚后 {target.version} 服务健康（模拟）"
+
+
+# ---------------------------------------------------------------------------
+# 辅助
+# ---------------------------------------------------------------------------
+
+def _hash_values(values: Dict[str, object]) -> str:
+    """计算 values 配置的短哈希（检测配置漂移，SHA1 前 12 位）。"""
+    payload = json.dumps(values, sort_keys=True, ensure_ascii=False,
+                         default=str)
+    return hashlib.sha1(payload.encode("utf-8")).hexdigest()[:12]

deploy/orchestration/tests/_bootstrap.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+"""测试引导：把 ``deploy/orchestration`` 挂到 sys.path，使 deploy_plan/rollback
+可被 ``from deploy_plan import ...`` 导入（orchestration 包内相对导入需要）。
+
+与 core 模块测试引导同款模式。
+"""
+import os
+import sys
+import types
+
+PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if PKG_DIR not in sys.path:
+    sys.path.insert(0, PKG_DIR)

deploy/orchestration/tests/test_deploy_plan.py

@@ -0,0 +1,200 @@
+# -*- coding: utf-8 -*-
+"""部署编排引擎测试（Issue #60）。
+
+覆盖：
+1. 默认 Helm 计划构造（4 步：pre_check/deploy/health_check/post_check）；
+2. 步骤顺序约束校验（违例 → DeployError）；
+3. 全成功执行（succeeded + 无需回滚）；
+4. deploy 步骤失败（needs_rollback + failed_step）；
+5. pre_check 失败（needs_rollback=False，非回滚检查点）；
+6. 非必需 post_check 失败（继续编排，succeeded）；
+7. action 异常（视为失败）；
+8. dry_run（全部 SKIPPED）；
+9. 空计划（succeeded=False）；
+10. HealthCheckContract 校验；
+11. 报告序列化。
+"""
+import unittest
+
+import _bootstrap  # noqa: F401  (sys.path 挂载)
+
+from deploy_plan import (
+    DEFAULT_AVAILABILITY_TARGET,
+    DeployError,
+    DeployOrchestrator,
+    DeployOutcome,
+    DeployPlan,
+    DeployStep,
+    DeployStepKind,
+    DeployStatus,
+    HealthCheckContract,
+)
+
+
+def _action(ok: bool, detail: str = ""):
+    """构造固定返回值的 action。"""
+    def _fn(ctx):
+        return ok, (detail or ("成功" if ok else "失败"))
+    return _fn
+
+
+def _raising_action(exc_type=RuntimeError):
+    def _fn(ctx):
+        raise exc_type("boom")
+    return _fn
+
+
+class TestDeployPlanModel(unittest.TestCase):
+    """部署计划模型与顺序约束。"""
+
+    def test_default_helm_release_has_four_steps(self):
+        plan = DeployPlan.default_helm_release()
+        kinds = [s.kind for s in plan.steps]
+        self.assertEqual(kinds, [
+            DeployStepKind.PRE_CHECK,
+            DeployStepKind.DEPLOY,
+            DeployStepKind.HEALTH_CHECK,
+            DeployStepKind.POST_CHECK,
+        ])
+        # post_check 默认非必需
+        self.assertFalse(plan.steps[-1].required)
+
+    def test_step_order_violation_rejected(self):
+        # health_check 排在 deploy 前 → 违例
+        with self.assertRaises(DeployError):
+            DeployPlan(
+                name="bad",
+                steps=[
+                    DeployStep(DeployStepKind.HEALTH_CHECK, "hc"),
+                    DeployStep(DeployStepKind.DEPLOY, "d"),
+                ])
+
+    def test_empty_step_name_rejected(self):
+        with self.assertRaises(DeployError):
+            DeployStep(DeployStepKind.DEPLOY, "")
+
+    def test_plan_name_required(self):
+        with self.assertRaises(DeployError):
+            DeployPlan(name="")
+
+
+class TestOrchestratorSuccess(unittest.TestCase):
+    """全成功编排。"""
+
+    def test_all_steps_succeed(self):
+        plan = DeployPlan.default_helm_release()
+        report = DeployOrchestrator(plan).execute()
+        self.assertTrue(report.succeeded)
+        self.assertFalse(report.outcome.needs_rollback)
+        self.assertEqual(len(report.results), 4)
+        self.assertEqual(
+            [r.status for r in report.results],
+            [DeployStatus.SUCCESS] * 4)
+        # 日志含 START/RUN/OK
+        self.assertTrue(any("[START]" in ln for ln in report.log_lines))
+        self.assertTrue(any("[OK]" in ln for ln in report.log_lines))
+
+    def test_context_populated(self):
+        plan = DeployPlan.default_helm_release()
+        report = DeployOrchestrator(plan).execute()
+        self.assertEqual(report.context["release"], "iaop")
+        self.assertEqual(report.context["namespace"], "iaop")
+        self.assertIn("health", report.context)
+
+
+class TestOrchestratorFailures(unittest.TestCase):
+    """失败场景 + 回滚标记。"""
+
+    def test_deploy_failure_triggers_rollback(self):
+        plan = DeployPlan.default_helm_release(
+            step_actions={DeployStepKind.DEPLOY: _action(False, "helm 失败")})
+        report = DeployOrchestrator(plan).execute()
+        self.assertFalse(report.succeeded)
+        self.assertTrue(report.outcome.needs_rollback)  # deploy 是回滚检查点
+        self.assertIsNotNone(report.outcome.failed_step)
+        # deploy 之后的 health_check/post_check 未执行
+        executed_kinds = [r.step.kind for r in report.results]
+        self.assertNotIn(DeployStepKind.HEALTH_CHECK, executed_kinds)
+        self.assertNotIn(DeployStepKind.POST_CHECK, executed_kinds)
+
+    def test_pre_check_failure_no_rollback(self):
+        plan = DeployPlan.default_helm_release(
+            step_actions={DeployStepKind.PRE_CHECK: _action(False, "集群不可达")})
+        report = DeployOrchestrator(plan).execute()
+        self.assertFalse(report.succeeded)
+        # pre_check 失败：尚未部署，无需回滚
+        self.assertFalse(report.outcome.needs_rollback)
+        self.assertIn("集群不可达", report.outcome.reason)
+
+    def test_non_required_post_check_failure_continues(self):
+        plan = DeployPlan.default_helm_release(
+            step_actions={DeployStepKind.POST_CHECK: _action(False, "烟测失败")})
+        report = DeployOrchestrator(plan).execute()
+        # post_check 非必需，失败仍判成功（核心步骤都过了）
+        self.assertTrue(report.succeeded)
+        self.assertFalse(report.outcome.needs_rollback)
+        # 但有 WARN 日志
+        self.assertTrue(any("[WARN]" in ln for ln in report.log_lines))
+
+    def test_action_exception_treated_as_failure(self):
+        plan = DeployPlan.default_helm_release(
+            step_actions={DeployStepKind.DEPLOY: _raising_action()})
+        report = DeployOrchestrator(plan).execute()
+        self.assertFalse(report.succeeded)
+        self.assertTrue(report.outcome.needs_rollback)
+        self.assertIn("action 异常", report.outcome.reason)
+
+    def test_health_check_failure_triggers_rollback(self):
+        plan = DeployPlan.default_helm_release(
+            step_actions={DeployStepKind.HEALTH_CHECK: _action(False, "/health 不可用")})
+        report = DeployOrchestrator(plan).execute()
+        self.assertFalse(report.succeeded)
+        self.assertTrue(report.outcome.needs_rollback)
+
+
+class TestDryRunAndEmpty(unittest.TestCase):
+    """dry_run + 空计划。"""
+
+    def test_dry_run_skips_all(self):
+        plan = DeployPlan.default_helm_release()
+        report = DeployOrchestrator(plan, dry_run=True).execute()
+        self.assertTrue(report.succeeded)
+        self.assertEqual(
+            [r.status for r in report.results],
+            [DeployStatus.SKIPPED] * 4)
+
+    def test_empty_plan_fails(self):
+        plan = DeployPlan(name="empty", steps=[])
+        report = DeployOrchestrator(plan).execute()
+        self.assertFalse(report.succeeded)
+        self.assertIn("无步骤", report.outcome.reason)
+
+
+class TestHealthContractAndReport(unittest.TestCase):
+    """HealthCheckContract 校验 + 报告序列化。"""
+
+    def test_health_contract_defaults(self):
+        h = HealthCheckContract()
+        self.assertEqual(h.endpoint, "http://iaop:8000/health")
+        self.assertEqual(h.retries, 3)
+        self.assertAlmostEqual(h.availability_target, DEFAULT_AVAILABILITY_TARGET)
+
+    def test_health_contract_validation(self):
+        with self.assertRaises(DeployError):
+            HealthCheckContract(timeout_s=0)
+        with self.assertRaises(DeployError):
+            HealthCheckContract(retries=0)
+        with self.assertRaises(DeployError):
+            HealthCheckContract(availability_target=1.5)
+
+    def test_report_to_dict(self):
+        plan = DeployPlan.default_helm_release()
+        report = DeployOrchestrator(plan).execute()
+        d = report.to_dict()
+        self.assertTrue(d["succeeded"])
+        self.assertEqual(len(d["steps"]), 4)
+        self.assertEqual(d["steps"][0]["kind"], "pre_check")
+
+
+if __name__ == "__main__":
+    unittest.main()

deploy/orchestration/tests/test_rollback.py

@@ -0,0 +1,185 @@
+# -*- coding: utf-8 -*-
+"""部署回滚引擎测试（Issue #60）。
+
+覆盖：
+1. 快照压栈 + latest 查询；
+2. FIFO 淘汰（超 max_snapshots）；
+3. rollback_to_latest 成功（弹出栈顶 + 回滚目标版本）；
+4. rollback(n) 多版本回溯；
+5. 空栈回滚（NO_TARGET）；
+6. restore 动作失败（FAILED + 压回栈顶）；
+7. restore 异常（FAILED）；
+8. post_rollback_health 失败（FAILED + 压回）；
+9. n<1 跳过；
+10. values_hash 漂移检测；
+11. 报告序列化。
+"""
+import unittest
+
+import _bootstrap  # noqa: F401  (sys.path 挂载)
+
+from deploy_plan import DeployPlan
+from rollback import (
+    DeployError,
+    RollbackManager,
+    RollbackPoint,
+    RollbackStatus,
+)
+
+
+def _restore(ok: bool, detail: str = ""):
+    def _fn(snapshot, ctx):
+        return ok, (detail or ("回滚成功" if ok else "回滚失败"))
+    return _fn
+
+
+def _raising_restore():
+    def _fn(snapshot, ctx):
+        raise RuntimeError("restore boom")
+    return _fn
+
+
+def _make_plan(version="v1.0.0", release="iaop"):
+    return DeployPlan.default_helm_release(release=release, version=version)
+
+
+class TestSnapshotStack(unittest.TestCase):
+    """快照压栈与查询。"""
+
+    def test_snapshot_push_and_latest(self):
+        mgr = RollbackManager()
+        # snapshot 即压栈（部署前快照），latest 即栈顶
+        mgr.snapshot(_make_plan("v1.0.0"), values={"k": 1}, reason="第一次")
+        self.assertEqual(mgr.latest().version, "v1.0.0")
+        self.assertEqual(len(mgr.snapshots), 1)
+        mgr.snapshot(_make_plan("v1.1.0"), values={"k": 2})
+        self.assertEqual(mgr.latest().version, "v1.1.0")
+        self.assertEqual(len(mgr.snapshots), 2)
+        # 空栈时 latest 返回 None
+        self.assertIsNotNone(mgr.latest())
+        mgr2 = RollbackManager()
+        self.assertIsNone(mgr2.latest())
+
+    def test_fifo_eviction(self):
+        mgr = RollbackManager(max_snapshots=2)
+        mgr.snapshot(_make_plan("v1.0.0"))
+        mgr.snapshot(_make_plan("v1.1.0"))
+        mgr.snapshot(_make_plan("v1.2.0"))
+        # 上限 2，淘汰最旧
+        self.assertEqual(len(mgr.snapshots), 2)
+        self.assertEqual([s.version for s in mgr.snapshots], ["v1.1.0", "v1.2.0"])
+
+    def test_max_snapshots_validation(self):
+        with self.assertRaises(DeployError):
+            RollbackManager(max_snapshots=0)
+
+    def test_values_hash_changes_with_config(self):
+        mgr = RollbackManager()
+        p1 = mgr.snapshot(_make_plan(), values={"backend": "gpu"})
+        p2 = mgr.snapshot(_make_plan(), values={"backend": "npu"})
+        self.assertNotEqual(p1.values_hash, p2.values_hash)
+
+
+class TestRollbackSuccess(unittest.TestCase):
+    """回滚成功场景。"""
+
+    def test_rollback_to_latest_success(self):
+        mgr = RollbackManager(restore_action=_restore(True))
+        mgr.snapshot(_make_plan("v1.0.0"))
+        mgr.snapshot(_make_plan("v1.1.0"))
+        result = mgr.rollback_to_latest(reason="部署 v1.2.0 失败")
+        self.assertTrue(result.succeeded)
+        self.assertEqual(result.status, RollbackStatus.SUCCESS)
+        self.assertEqual(result.target.version, "v1.1.0")
+        # 弹出栈顶后栈剩 1 个
+        self.assertEqual(len(mgr.snapshots), 1)
+        self.assertEqual(mgr.latest().version, "v1.0.0")
+        self.assertIn("v1.2.0 失败", result.reason)
+
+    def test_rollback_n_versions(self):
+        mgr = RollbackManager(restore_action=_restore(True))
+        for v in ["v1.0.0", "v1.1.0", "v1.2.0"]:
+            mgr.snapshot(_make_plan(v))
+        # 回滚 2 个版本 → 目标 v1.1.0（弹出 v1.2.0 和 v1.1.0）
+        result = mgr.rollback(n=2)
+        self.assertTrue(result.succeeded)
+        self.assertEqual(result.target.version, "v1.1.0")
+        self.assertEqual(len(mgr.snapshots), 1)
+
+
+class TestRollbackFailures(unittest.TestCase):
+    """回滚失败与边界。"""
+
+    def test_empty_stack_no_target(self):
+        mgr = RollbackManager()
+        result = mgr.rollback_to_latest()
+        self.assertEqual(result.status, RollbackStatus.NO_TARGET)
+        self.assertFalse(result.succeeded)
+
+    def test_insufficient_stack_for_n(self):
+        mgr = RollbackManager(restore_action=_restore(True))
+        mgr.snapshot(_make_plan("v1.0.0"))
+        # 栈仅 1 个，回滚 2 个 → NO_TARGET
+        result = mgr.rollback(n=2)
+        self.assertEqual(result.status, RollbackStatus.NO_TARGET)
+
+    def test_restore_failure_pushes_target_back(self):
+        mgr = RollbackManager(restore_action=_restore(False, "helm rollback 失败"))
+        mgr.snapshot(_make_plan("v1.0.0"))
+        mgr.snapshot(_make_plan("v1.1.0"))
+        result = mgr.rollback_to_latest()
+        self.assertEqual(result.status, RollbackStatus.FAILED)
+        # 失败时目标压回栈顶，栈仍 2 个
+        self.assertEqual(len(mgr.snapshots), 2)
+        self.assertEqual(mgr.latest().version, "v1.1.0")
+
+    def test_restore_exception_failure(self):
+        mgr = RollbackManager(restore_action=_raising_restore())
+        mgr.snapshot(_make_plan("v1.0.0"))
+        result = mgr.rollback_to_latest()
+        self.assertEqual(result.status, RollbackStatus.FAILED)
+        self.assertIn("异常", result.detail)
+
+    def test_post_rollback_health_failure(self):
+        # restore 成功但健康检查失败（注入失败的健康检查）
+        def _restore_then_health_fail(snapshot, ctx):
+            return True, "回滚动作成功"
+        mgr = RollbackManager(
+            restore_action=_restore_then_health_fail, post_rollback_health=True)
+        # 覆盖 _health_check 为失败
+        mgr._health_check = lambda target: (False, "/health 仍不可用")
+        mgr.snapshot(_make_plan("v1.0.0"))
+        result = mgr.rollback_to_latest()
+        self.assertEqual(result.status, RollbackStatus.FAILED)
+        self.assertIn("健康检查", result.detail)
+        # 压回栈顶
+        self.assertEqual(len(mgr.snapshots), 1)
+
+    def test_n_less_than_one_skipped(self):
+        mgr = RollbackManager()
+        result = mgr.rollback(n=0)
+        self.assertEqual(result.status, RollbackStatus.SKIPPED)
+
+
+class TestResultExport(unittest.TestCase):
+    """结果序列化。"""
+
+    def test_result_to_dict(self):
+        mgr = RollbackManager(restore_action=_restore(True))
+        mgr.snapshot(_make_plan("v1.0.0"))
+        result = mgr.rollback_to_latest()
+        d = result.to_dict()
+        self.assertEqual(d["status"], "success")
+        self.assertTrue(d["succeeded"])
+        self.assertEqual(d["target"]["version"], "v1.0.0")
+
+    def test_point_to_dict(self):
+        p = RollbackPoint(release="r", namespace="ns", version="v1",
+                          chart="c", values_hash="abc123", reason="x")
+        d = p.to_dict()
+        self.assertEqual(d["version"], "v1")
+        self.assertEqual(d["values_hash"], "abc123")
+
+
+if __name__ == "__main__":
+    unittest.main()