diff --git a/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po
index e00a5449..140208ce 100644
--- a/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po
+++ b/src/iac_code/i18n/locales/de/LC_MESSAGES/messages.po
@@ -4077,6 +4077,15 @@ msgstr ""
 "Eine erfolgreiche Bereitstellung muss warten, bis ros_deploy "
 "CREATE_COMPLETE zurückgibt."
 
+#: src/iac_code/pipeline/engine/complete_step_tool.py
+msgid ""
+"A failed deployment must report the ROS status_reason to the user and "
+"then retry, roll back, or return to candidate selection."
+msgstr ""
+"Eine fehlgeschlagene Bereitstellung muss den ROS-status_reason an den "
+"Benutzer melden und danach erneut versuchen, zurückrollen oder zur "
+"Kandidatenauswahl zurückkehren."
+
 #: src/iac_code/pipeline/engine/complete_step_tool.py
 msgid ""
 "Every explicit user hard constraint must be covered by a satisfied check "
@@ -4603,6 +4612,15 @@ msgstr "Der Template-Dateipfad muss relativ zum Arbeitsverzeichnis sein"
 msgid "Template file path cannot escape the working directory"
 msgstr "Der Template-Dateipfad darf das Arbeitsverzeichnis nicht verlassen"
 
+#: src/iac_code/pipeline/engine/step_executor.py
+msgid "no failure reason reported"
+msgstr "kein Fehlergrund gemeldet"
+
+#: src/iac_code/pipeline/engine/step_executor.py
+#, python-brace-format
+msgid "Step {step_id} reported a failed conclusion: {reason}"
+msgstr "Schritt {step_id} hat ein fehlgeschlagenes Ergebnis gemeldet: {reason}"
+
 #: src/iac_code/pipeline/engine/user_input.py
 msgid "[Image input]"
 msgstr "[Bildeingabe]"
diff --git a/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po
index be131234..9fb08faf 100644
--- a/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po
+++ b/src/iac_code/i18n/locales/es/LC_MESSAGES/messages.po
@@ -4055,6 +4055,14 @@ msgstr ""
 "Un despliegue correcto debe esperar hasta que ros_deploy devuelva "
 "CREATE_COMPLETE."
 
+#: src/iac_code/pipeline/engine/complete_step_tool.py
+msgid ""
+"A failed deployment must report the ROS status_reason to the user and "
+"then retry, roll back, or return to candidate selection."
+msgstr ""
+"Un despliegue fallido debe informar al usuario del status_reason de ROS y"
+" después reintentar, revertir o volver a la selección de candidatos."
+
 #: src/iac_code/pipeline/engine/complete_step_tool.py
 msgid ""
 "Every explicit user hard constraint must be covered by a satisfied check "
@@ -4570,6 +4578,15 @@ msgstr ""
 msgid "Template file path cannot escape the working directory"
 msgstr "La ruta del archivo de plantilla no puede salir del directorio de trabajo"
 
+#: src/iac_code/pipeline/engine/step_executor.py
+msgid "no failure reason reported"
+msgstr "no se informó ningún motivo de fallo"
+
+#: src/iac_code/pipeline/engine/step_executor.py
+#, python-brace-format
+msgid "Step {step_id} reported a failed conclusion: {reason}"
+msgstr "El paso {step_id} informó una conclusión fallida: {reason}"
+
 #: src/iac_code/pipeline/engine/user_input.py
 msgid "[Image input]"
 msgstr "[Entrada de imagen]"
diff --git a/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po
index 67ef1a4c..c9ee32bb 100644
--- a/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po
+++ b/src/iac_code/i18n/locales/fr/LC_MESSAGES/messages.po
@@ -4063,6 +4063,15 @@ msgstr ""
 "Un déploiement réussi doit attendre que ros_deploy retourne "
 "CREATE_COMPLETE."
 
+#: src/iac_code/pipeline/engine/complete_step_tool.py
+msgid ""
+"A failed deployment must report the ROS status_reason to the user and "
+"then retry, roll back, or return to candidate selection."
+msgstr ""
+"Un déploiement en échec doit signaler le status_reason ROS à "
+"l'utilisateur, puis réessayer, revenir en arrière ou retourner à la "
+"sélection des candidats."
+
 #: src/iac_code/pipeline/engine/complete_step_tool.py
 msgid ""
 "Every explicit user hard constraint must be covered by a satisfied check "
@@ -4580,6 +4589,15 @@ msgstr "Le chemin du fichier de modèle doit être relatif au répertoire de tra
 msgid "Template file path cannot escape the working directory"
 msgstr "Le chemin du fichier de modèle ne peut pas sortir du répertoire de travail"
 
+#: src/iac_code/pipeline/engine/step_executor.py
+msgid "no failure reason reported"
+msgstr "aucune raison d'échec signalée"
+
+#: src/iac_code/pipeline/engine/step_executor.py
+#, python-brace-format
+msgid "Step {step_id} reported a failed conclusion: {reason}"
+msgstr "L'étape {step_id} a signalé une conclusion en échec : {reason}"
+
 #: src/iac_code/pipeline/engine/user_input.py
 msgid "[Image input]"
 msgstr "[Entrée d'image]"
diff --git a/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po
index 83869fb5..d5c4b10d 100644
--- a/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po
+++ b/src/iac_code/i18n/locales/ja/LC_MESSAGES/messages.po
@@ -3868,6 +3868,12 @@ msgid ""
 "CREATE_COMPLETE."
 msgstr "デプロイ成功には、ros_deploy が CREATE_COMPLETE を返すまで待つ必要があります。"
 
+#: src/iac_code/pipeline/engine/complete_step_tool.py
+msgid ""
+"A failed deployment must report the ROS status_reason to the user and "
+"then retry, roll back, or return to candidate selection."
+msgstr "デプロイ失敗時は ROS の status_reason をユーザーに報告し、その後リトライ、ロールバック、または候補選択に戻る必要があります。"
+
 #: src/iac_code/pipeline/engine/complete_step_tool.py
 msgid ""
 "Every explicit user hard constraint must be covered by a satisfied check "
@@ -4335,6 +4341,15 @@ msgstr "テンプレートファイルのパスは作業ディレクトリから
 msgid "Template file path cannot escape the working directory"
 msgstr "テンプレートファイルのパスは作業ディレクトリの外に出られません"
 
+#: src/iac_code/pipeline/engine/step_executor.py
+msgid "no failure reason reported"
+msgstr "失敗理由が報告されていません"
+
+#: src/iac_code/pipeline/engine/step_executor.py
+#, python-brace-format
+msgid "Step {step_id} reported a failed conclusion: {reason}"
+msgstr "ステップ {step_id} が失敗の結論を報告しました: {reason}"
+
 #: src/iac_code/pipeline/engine/user_input.py
 msgid "[Image input]"
 msgstr "[画像入力]"
diff --git a/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po
index 953c6a8e..c06cf2a1 100644
--- a/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po
+++ b/src/iac_code/i18n/locales/pt/LC_MESSAGES/messages.po
@@ -4036,6 +4036,14 @@ msgstr ""
 "Uma implantação bem-sucedida deve aguardar até que ros_deploy retorne "
 "CREATE_COMPLETE."
 
+#: src/iac_code/pipeline/engine/complete_step_tool.py
+msgid ""
+"A failed deployment must report the ROS status_reason to the user and "
+"then retry, roll back, or return to candidate selection."
+msgstr ""
+"Uma implantação com falha deve informar o status_reason do ROS ao usuário"
+" e depois tentar novamente, reverter ou voltar à seleção de candidatos."
+
 #: src/iac_code/pipeline/engine/complete_step_tool.py
 msgid ""
 "Every explicit user hard constraint must be covered by a satisfied check "
@@ -4544,6 +4552,15 @@ msgstr ""
 msgid "Template file path cannot escape the working directory"
 msgstr "O caminho do arquivo de template não pode sair do diretório de trabalho"
 
+#: src/iac_code/pipeline/engine/step_executor.py
+msgid "no failure reason reported"
+msgstr "nenhum motivo de falha informado"
+
+#: src/iac_code/pipeline/engine/step_executor.py
+#, python-brace-format
+msgid "Step {step_id} reported a failed conclusion: {reason}"
+msgstr "A etapa {step_id} informou uma conclusão com falha: {reason}"
+
 #: src/iac_code/pipeline/engine/user_input.py
 msgid "[Image input]"
 msgstr "[Entrada de imagem]"
diff --git a/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po b/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po
index a1f9264f..d66bbd5f 100644
--- a/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po
+++ b/src/iac_code/i18n/locales/zh/LC_MESSAGES/messages.po
@@ -3832,6 +3832,12 @@ msgid ""
 "CREATE_COMPLETE."
 msgstr "部署成功必须等待 ros_deploy 返回 CREATE_COMPLETE。"
 
+#: src/iac_code/pipeline/engine/complete_step_tool.py
+msgid ""
+"A failed deployment must report the ROS status_reason to the user and "
+"then retry, roll back, or return to candidate selection."
+msgstr "部署失败必须向用户报告 ROS 返回的 status_reason，然后重试、回滚或返回方案选择。"
+
 #: src/iac_code/pipeline/engine/complete_step_tool.py
 msgid ""
 "Every explicit user hard constraint must be covered by a satisfied check "
@@ -4280,6 +4286,15 @@ msgstr "模板文件路径必须是相对于工作目录的路径"
 msgid "Template file path cannot escape the working directory"
 msgstr "模板文件路径不能跳出工作目录"
 
+#: src/iac_code/pipeline/engine/step_executor.py
+msgid "no failure reason reported"
+msgstr "未报告失败原因"
+
+#: src/iac_code/pipeline/engine/step_executor.py
+#, python-brace-format
+msgid "Step {step_id} reported a failed conclusion: {reason}"
+msgstr "步骤 {step_id} 提交了失败结论：{reason}"
+
 #: src/iac_code/pipeline/engine/user_input.py
 msgid "[Image input]"
 msgstr "[图片输入]"
diff --git a/src/iac_code/pipeline/engine/complete_step_tool.py b/src/iac_code/pipeline/engine/complete_step_tool.py
index 7cabdb76..e1a755d0 100644
--- a/src/iac_code/pipeline/engine/complete_step_tool.py
+++ b/src/iac_code/pipeline/engine/complete_step_tool.py
@@ -56,6 +56,10 @@
         "or confirm that it should not be handled for now."
     ),
     "deploy_wait_create_complete": ("A successful deployment must wait until ros_deploy returns CREATE_COMPLETE."),
+    "deploy_report_failure_reason": (
+        "A failed deployment must report the ROS status_reason to the user and then retry, "
+        "roll back, or return to candidate selection."
+    ),
     "hard_constraint_verification_required": (
         "Every explicit user hard constraint must be covered by a satisfied check with matching parameters "
         "and evidence."
@@ -96,6 +100,10 @@ def _completion_guard_message_i18n_markers() -> tuple[str, ...]:
             "or confirm that it should not be handled for now."
         ),
         _("A successful deployment must wait until ros_deploy returns CREATE_COMPLETE."),
+        _(
+            "A failed deployment must report the ROS status_reason to the user and then retry, "
+            "roll back, or return to candidate selection."
+        ),
         _(
             "Every explicit user hard constraint must be covered by a satisfied check with matching parameters "
             "and evidence."
diff --git a/src/iac_code/pipeline/engine/loader.py b/src/iac_code/pipeline/engine/loader.py
index 88f5e298..d2562362 100644
--- a/src/iac_code/pipeline/engine/loader.py
+++ b/src/iac_code/pipeline/engine/loader.py
@@ -299,6 +299,7 @@ def _parse_steps(raw_steps: list[dict]) -> list[StepSpec]:
                 completion_guards=_parse_completion_guards(raw.get("completion_guards"), step_id),
                 description=raw.get("description", ""),
                 exit_condition=_parse_exit_condition(raw.get("exit_condition"), step_id),
+                failure_condition=_parse_failure_condition(raw.get("failure_condition"), step_id),
                 a2a_artifacts=_parse_a2a_artifacts(raw.get("a2a_artifacts"), step_id),
                 surface_overrides=_parse_surface_overrides(raw.get("surface_overrides"), step_id),
                 config=_parse_mapping(raw.get("config"), "config", step_id),
@@ -410,6 +411,31 @@ def _parse_exit_condition(raw: dict | None, step_id: str) -> dict | None:
     return raw
 
 
+def _parse_failure_condition(raw: dict | None, step_id: str) -> dict | None:
+    if raw is None:
+        return None
+    if not isinstance(raw, dict) or "field" not in raw or "value" not in raw:
+        raise ValueError(
+            f"Step '{step_id}': failure_condition must be a dict with 'field' and 'value' keys, got {raw!r}"
+        )
+    unsupported = set(raw) - {"field", "value", "reason_fields"}
+    if unsupported:
+        unknown = ", ".join(sorted(str(key) for key in unsupported))
+        raise ValueError(
+            f"Step '{step_id}': failure_condition has unsupported keys: {unknown}; "
+            "supported: field, reason_fields, value"
+        )
+    reason_fields = raw.get("reason_fields")
+    if reason_fields is not None and (
+        not isinstance(reason_fields, list) or not all(isinstance(name, str) and name for name in reason_fields)
+    ):
+        raise ValueError(
+            f"Step '{step_id}': failure_condition.reason_fields must be a list of non-empty strings, "
+            f"got {reason_fields!r}"
+        )
+    return raw
+
+
 def _filter_and_relink(steps: list[StepSpec], feature_flags: dict[str, bool]) -> list[StepSpec]:
     """Remove disabled steps and fix forward links."""
     enabled = [s for s in steps if _is_enabled(s, feature_flags)]
diff --git a/src/iac_code/pipeline/engine/step_executor.py b/src/iac_code/pipeline/engine/step_executor.py
index f6a7c31e..da8083f5 100644
--- a/src/iac_code/pipeline/engine/step_executor.py
+++ b/src/iac_code/pipeline/engine/step_executor.py
@@ -15,6 +15,7 @@
 
 from iac_code.agent.message import ContentBlock, Message
 from iac_code.agent.system_prompt import SECTION_BUILDERS, build_base_sections
+from iac_code.i18n import _
 from iac_code.mcp.prompt_dispatch import mcp_prompt_command_stream
 from iac_code.pipeline.engine.complete_step_tool import CompleteStepTool
 from iac_code.pipeline.engine.completion_guard_state import (
@@ -79,6 +80,28 @@ def _completion_guard_tool_result_content(event: ToolResultEvent) -> str:
         return event.result
 
 
+def _failure_error_from_conclusion(step: StepSpec, conclusion: Any) -> str | None:
+    """Return a user-facing error when the conclusion matches the step failure condition."""
+    condition = step.failure_condition
+    if not condition or not isinstance(conclusion, dict):
+        return None
+
+    expected = condition.get("value")
+    actual = conclusion.get(str(condition.get("field", "")))
+    matched = actual is expected if isinstance(expected, bool) else actual == expected
+    if not matched:
+        return None
+
+    reason_fields = condition.get("reason_fields") or []
+    reasons = []
+    for name in reason_fields:
+        value = conclusion.get(name)
+        if isinstance(value, str) and value.strip():
+            reasons.append(f"{name}={value.strip()}")
+    reason = "; ".join(reasons) if reasons else _("no failure reason reported")
+    return _("Step {step_id} reported a failed conclusion: {reason}").format(step_id=step.step_id, reason=reason)
+
+
 @dataclass
 class StepAgentLoopContext:
     """AgentLoop context built by the same path used for step execution."""
@@ -430,15 +453,17 @@ async def consume_complete_step_events(
             conclusion = self._merge_preserved_candidate_selection(preserved_selection, conclusion)
             rollback = complete_step_input.get("rollback_request")
             rollback_tuple = (rollback["target_step"], rollback["reason"]) if rollback else None
+            context.set_conclusion(step.conclusion_field, conclusion)
+            if step.on_exit:
+                step.on_exit(context, conclusion)
+            failure_error = None if rollback_tuple else _failure_error_from_conclusion(step, conclusion)
             step_result = StepResult(
                 step_id=step.step_id,
-                status=StepStatus.COMPLETED,
+                status=StepStatus.FAILED if failure_error else StepStatus.COMPLETED,
                 conclusion=conclusion,
                 rollback_request=rollback_tuple,
+                error=failure_error,
             )
-            context.set_conclusion(step.conclusion_field, conclusion)
-            if step.on_exit:
-                step.on_exit(context, conclusion)
         else:
             step_result = StepResult(
                 step_id=step.step_id,
@@ -610,11 +635,14 @@ def _restore_completed_step_result(
         if isinstance(rollback, dict) and rollback.get("target_step") and rollback.get("reason"):
             rollback_tuple = (str(rollback["target_step"]), str(rollback["reason"]))
 
+        conclusion = conclusion if isinstance(conclusion, dict) else {}
+        failure_error = None if rollback_tuple else _failure_error_from_conclusion(step, conclusion)
         return StepResult(
             step_id=step.step_id,
-            status=StepStatus.COMPLETED,
-            conclusion=conclusion if isinstance(conclusion, dict) else {},
+            status=StepStatus.FAILED if failure_error else StepStatus.COMPLETED,
+            conclusion=conclusion,
             rollback_request=rollback_tuple,
+            error=failure_error,
         )
 
     def _build_full_system_prompt(self, step: StepSpec, context: PipelineContext) -> str:
diff --git a/src/iac_code/pipeline/engine/step_spec.py b/src/iac_code/pipeline/engine/step_spec.py
index b0af6ca3..7d82c382 100644
--- a/src/iac_code/pipeline/engine/step_spec.py
+++ b/src/iac_code/pipeline/engine/step_spec.py
@@ -85,6 +85,7 @@ class StepSpec:
     completion_guards: list[dict] = field(default_factory=list)
     description: str = ""
     exit_condition: dict | None = None
+    failure_condition: dict | None = None
     a2a_artifacts: list[A2AArtifactSpec] = field(default_factory=list)
     surface_overrides: dict[str, StepSurfaceOverride] = field(default_factory=dict)
     config: dict[str, Any] = field(default_factory=dict)
diff --git a/src/iac_code/pipeline/selling/hooks/deploying.py b/src/iac_code/pipeline/selling/hooks/deploying.py
index 87673e3f..411687d0 100644
--- a/src/iac_code/pipeline/selling/hooks/deploying.py
+++ b/src/iac_code/pipeline/selling/hooks/deploying.py
@@ -250,6 +250,19 @@ def on_enter(ctx: PipelineContext) -> None:
     ctx.set_conclusion("selected_plan", normalized)
 
 
+def on_exit(ctx: PipelineContext, conclusion: dict[str, Any]) -> None:
+    """Make sure a failed deployment conclusion always carries a ROS failure reason."""
+    _ = ctx
+    if not isinstance(conclusion, dict) or conclusion.get("status") != "failed":
+        return
+    status_reason = conclusion.get("status_reason")
+    if isinstance(status_reason, str) and status_reason.strip():
+        return
+    error = conclusion.get("error")
+    if isinstance(error, str) and error.strip():
+        conclusion["status_reason"] = error.strip()
+
+
 def on_resource_observed(
     ctx: PipelineContext,
     event: ResourceObservedEvent,
diff --git a/src/iac_code/pipeline/selling/pipeline.yaml b/src/iac_code/pipeline/selling/pipeline.yaml
index af868cb4..26fd983f 100644
--- a/src/iac_code/pipeline/selling/pipeline.yaml
+++ b/src/iac_code/pipeline/selling/pipeline.yaml
@@ -589,6 +589,10 @@ steps:
     interrupt_judge_failure: pause
     context_fields: [intent, selected_plan, evaluated_candidates]
     hooks_file: hooks/deploying.py
+    failure_condition:
+      field: status
+      value: failed
+      reason_fields: [status_reason, error]
     completion_guards:
       - when_conclusion_field_equals:
           status: success
@@ -600,6 +604,10 @@ steps:
           status_in: [CREATE_COMPLETE]
           match_conclusion_field: stack_id
         message_key: deploy_wait_create_complete
+      - when_conclusion_field_equals:
+          status: failed
+        required_conclusion_field: status_reason
+        message_key: deploy_report_failure_reason
     tools:
       include: []
       exclude:
diff --git a/src/iac_code/pipeline/selling/prompts/deploying.md b/src/iac_code/pipeline/selling/prompts/deploying.md
index 0d331afe..1ffb213c 100644
--- a/src/iac_code/pipeline/selling/prompts/deploying.md
+++ b/src/iac_code/pipeline/selling/prompts/deploying.md
@@ -46,8 +46,9 @@
 
 ## 错误处理
 - 模板校验失败 → 就地修复模板后重试（最多 5 轮）
-- 部署失败或等待超时 → 按技能的参数补全与 `ros_deploy` 恢复策略处理
-- 架构层面必须变更（如产品组合不可行）→ rollback_request 到 `architecture_planning`
+- 部署失败或等待超时 → 先把 `ros_deploy` 结果中的 `status` 与 `status_reason` 原文报告给用户，再按技能的参数补全与 `ros_deploy` 恢复策略重试
+- 恢复动作用尽仍不能创建成功 → 用 rollback_request 回到 `confirm_and_select` 重新选择方案/参数，或在架构层面必须变更（如产品组合不可行）时回到 `architecture_planning`
+- 确实无法恢复时才提交 `status: failed`，并同时填写 `status_reason`（ROS 原文）与 `error`（结论摘要）；不得用 `status: success` 或省略 `status_reason` 掩盖失败
 
 ## 注意事项
 - 不要读取项目文件或记忆，所需的上下文已在上方提供。
diff --git a/src/iac_code/pipeline/selling/skills/iac-aliyun-deploying/SKILL.md b/src/iac_code/pipeline/selling/skills/iac-aliyun-deploying/SKILL.md
index 48465a66..534111fe 100644
--- a/src/iac_code/pipeline/selling/skills/iac-aliyun-deploying/SKILL.md
+++ b/src/iac_code/pipeline/selling/skills/iac-aliyun-deploying/SKILL.md
@@ -24,6 +24,9 @@ conclusion_schema:
     error:
       type: string
       description: 失败原因（status 为 failed 时必填）
+    status_reason:
+      type: string
+      description: ROS 返回的失败原因原文，例如 CREATE_FAILED 的 status_reason（status 为 failed 时必填）
   allOf:
     - if:
         properties:
@@ -38,7 +41,7 @@ conclusion_schema:
             const: failed
         required: [status]
       then:
-        required: [error]
+        required: [error, status_reason]
 ---
 
 # 阿里云 ROS 部署技能
@@ -136,12 +139,19 @@ conclusion_schema:
 ## 错误处理
 
 ### 部署失败
+`ros_deploy` 返回 `is_success: false` 或 Stack 进入 `CREATE_FAILED` 等失败终态时，先把结果中的 `status`、`status_reason` 原文告知用户，再选择恢复动作；不得在未报告失败原因的情况下收口。
+
 分析错误原因：
 - 工具调用超时但已有 `stack_id`，且 Stack 仍在创建 → 调用 `ros_deploy` 的 `wait`
 - 权限/配额 → 告知用户处理
 - 模板/参数 → 修复后调用 `ros_deploy` 的 `continue_create`
 - `continue_create` 返回 `ContinueCreateStackValidationFailed` → 告知用户需要重建本步骤创建的失败 Stack，再调用 `ros_deploy` 的 `delete_and_create`
 
+恢复动作全部用尽仍无法创建成功时：
+- 需要更换架构 → 用 `rollback_request` 回到 `architecture_planning`
+- 需要用户重新选择方案或调整参数 → 用 `rollback_request` 回到 `confirm_and_select`
+- 均不适用时，返回 `status: failed`，并同时填写 `error`（结论摘要）和 `status_reason`（ROS 原文）。此时本步骤会被判定为失败终态，不得用 `status: success` 或省略 `status_reason` 掩盖失败。
+
 ### 删除并重建
 仅在 `continue_create` 返回 `ContinueCreateStackValidationFailed` 后使用 `delete_and_create`。调用时：
 - `stack_id` 指向本步骤创建的旧失败 Stack，不得使用通过查询发现的其他 Stack
diff --git a/tests/pipeline/engine/test_complete_step_tool.py b/tests/pipeline/engine/test_complete_step_tool.py
index b47ee908..fbf742fc 100644
--- a/tests/pipeline/engine/test_complete_step_tool.py
+++ b/tests/pipeline/engine/test_complete_step_tool.py
@@ -66,6 +66,50 @@ async def test_completion_guard_message_key_renders_translated_message(self, ste
         assert "deployment or cloud resource request" in result.content
         assert "intent_not_deployment_request" not in result.content
 
+    @pytest.mark.asyncio
+    async def test_failed_deployment_without_status_reason_is_rejected(self, step_config):
+        tool = CompleteStepTool(
+            step_config,
+            completion_guards=[
+                {
+                    "when_conclusion_field_equals": {"status": "failed"},
+                    "required_conclusion_field": "status_reason",
+                    "message_key": "deploy_report_failure_reason",
+                }
+            ],
+            completion_guard_state={},
+        )
+
+        result = await tool.execute(
+            tool_input={"conclusion": {"status": "failed", "error": "deploy failed"}},
+            context=ToolContext(),
+        )
+
+        assert result.is_error
+        assert "status_reason" in result.content
+        assert "deploy_report_failure_reason" not in result.content
+
+    @pytest.mark.asyncio
+    async def test_failed_deployment_with_status_reason_is_accepted(self, step_config):
+        tool = CompleteStepTool(
+            step_config,
+            completion_guards=[
+                {
+                    "when_conclusion_field_equals": {"status": "failed"},
+                    "required_conclusion_field": "status_reason",
+                    "message_key": "deploy_report_failure_reason",
+                }
+            ],
+            completion_guard_state={},
+        )
+
+        result = await tool.execute(
+            tool_input={"conclusion": {"status": "failed", "status_reason": "CREATE_FAILED: no stock"}},
+            context=ToolContext(),
+        )
+
+        assert not result.is_error
+
 
 class TestDynamicInputSchema:
     def test_schema_with_conclusion_schema(self):
diff --git a/tests/pipeline/engine/test_loader.py b/tests/pipeline/engine/test_loader.py
index 786586bd..f36dc0df 100644
--- a/tests/pipeline/engine/test_loader.py
+++ b/tests/pipeline/engine/test_loader.py
@@ -4,7 +4,7 @@
 
 import pytest
 
-from iac_code.pipeline.engine.loader import _parse_exit_condition, load_pipeline_dir
+from iac_code.pipeline.engine.loader import _parse_exit_condition, _parse_failure_condition, load_pipeline_dir
 
 
 def _write_pipeline(tmp_path: Path, yaml_content: str, prompts: dict[str, str] | None = None):
@@ -845,3 +845,40 @@ def test_missing_value_raises(self):
     def test_error_message_includes_step_id(self):
         with pytest.raises(ValueError, match="step_x"):
             _parse_exit_condition("wrong", "step_x")
+
+
+class TestParseFailureCondition:
+    def test_none_returns_none(self):
+        assert _parse_failure_condition(None, "step_x") is None
+
+    def test_valid_dict_returned_unchanged(self):
+        raw = {"field": "status", "value": "failed", "reason_fields": ["status_reason"]}
+        assert _parse_failure_condition(raw, "step_x") is raw
+
+    def test_non_dict_raises(self):
+        with pytest.raises(ValueError, match="must be a dict"):
+            _parse_failure_condition("wrong", "step_x")
+
+    def test_missing_value_raises(self):
+        with pytest.raises(ValueError, match="must be a dict"):
+            _parse_failure_condition({"field": "status"}, "step_x")
+
+    def test_unsupported_key_raises(self):
+        with pytest.raises(ValueError, match="unsupported keys: retry"):
+            _parse_failure_condition({"field": "status", "value": "failed", "retry": True}, "step_x")
+
+    def test_invalid_reason_fields_raises(self):
+        with pytest.raises(ValueError, match="reason_fields"):
+            _parse_failure_condition({"field": "status", "value": "failed", "reason_fields": "status_reason"}, "step_x")
+
+    def test_loads_failure_condition_from_yaml(self, tmp_path):
+        yaml_content = MINIMAL_YAML.replace(
+            "    skill: skill-x\n",
+            "    skill: skill-x\n    failure_condition:\n      field: status\n      value: failed\n",
+        )
+        _write_pipeline(tmp_path, yaml_content, {"step_a.md": "Do A", "step_b.md": "Do B with {intent}"})
+
+        loaded = load_pipeline_dir(tmp_path)
+
+        assert loaded.steps[0].failure_condition == {"field": "status", "value": "failed"}
+        assert loaded.steps[1].failure_condition is None
diff --git a/tests/pipeline/engine/test_pipeline_runner.py b/tests/pipeline/engine/test_pipeline_runner.py
index 18efe46f..91e8981b 100644
--- a/tests/pipeline/engine/test_pipeline_runner.py
+++ b/tests/pipeline/engine/test_pipeline_runner.py
@@ -4195,6 +4195,78 @@ async def mock_execute(step, context, session_id, *, user_message=None, **kwargs
         assert completed[0].data.get("early_exit") is None
 
 
+class TestFailureCondition:
+    @staticmethod
+    def _write_pipeline_with_failure_condition(tmp_path):
+        (tmp_path / "prompts").mkdir(exist_ok=True)
+        (tmp_path / "prompts" / "step1.md").write_text("Step1.", encoding="utf-8")
+        (tmp_path / "prompts" / "step2.md").write_text("Step2.", encoding="utf-8")
+        (tmp_path / "pipeline.yaml").write_text(
+            dedent("""\
+            name: test
+            context_dependencies:
+              intent: []
+              deployment: [intent]
+            max_rollbacks: 1
+            steps:
+              - id: intent_parsing
+                conclusion_field: intent
+                forward: deploying
+                prompt: prompts/step1.md
+              - id: deploying
+                conclusion_field: deployment
+                forward: null
+                prompt: prompts/step2.md
+                failure_condition:
+                  field: status
+                  value: failed
+                  reason_fields: [status_reason]
+        """),
+            encoding="utf-8",
+        )
+
+    @pytest.mark.asyncio
+    async def test_failed_deployment_conclusion_emits_step_failed(self, tmp_path):
+        self._write_pipeline_with_failure_condition(tmp_path)
+        runner = PipelineRunner(
+            pipeline_dir=tmp_path,
+            provider_manager=MagicMock(),
+            base_tool_registry=MagicMock(),
+            session_storage=FakeSessionStorage(),
+            session_id="test123",
+        )
+
+        async def mock_execute(step, context, session_id, *, user_message=None, **kwargs):
+            if step.step_id == "intent_parsing":
+                conclusion = {"requirement": "deploy"}
+                context.set_conclusion(step.conclusion_field, conclusion)
+                yield StepResult(step_id=step.step_id, status=StepStatus.COMPLETED, conclusion=conclusion)
+                return
+            conclusion = {"status": "failed", "status_reason": "CREATE_FAILED: quota exceeded"}
+            context.set_conclusion(step.conclusion_field, conclusion)
+            yield StepResult(
+                step_id=step.step_id,
+                status=StepStatus.FAILED,
+                conclusion=conclusion,
+                error="Step deploying reported a failed conclusion: status_reason=CREATE_FAILED: quota exceeded",
+            )
+
+        runner._step_executor.execute = mock_execute
+
+        events = [event async for event in runner.run("部署一个网站")]
+
+        failed = [e for e in events if isinstance(e, PipelineEvent) and e.type == PipelineEventType.STEP_FAILED]
+        assert len(failed) == 1
+        assert failed[0].step_id == "deploying"
+        assert "CREATE_FAILED: quota exceeded" in failed[0].data["error"]
+
+        completed = [
+            e for e in events if isinstance(e, PipelineEvent) and e.type == PipelineEventType.PIPELINE_COMPLETED
+        ]
+        assert len(completed) == 1
+        assert completed[0].data.get("failed") is True
+
+
 class TestInvalidRollbackTarget:
     """Regression: hallucinated rollback target must not crash the stream (P-C3)."""
 
diff --git a/tests/pipeline/engine/test_step_executor.py b/tests/pipeline/engine/test_step_executor.py
index 2d596efe..20ea1169 100644
--- a/tests/pipeline/engine/test_step_executor.py
+++ b/tests/pipeline/engine/test_step_executor.py
@@ -1,5 +1,6 @@
 import json
 import logging
+from dataclasses import replace
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import MagicMock, call, patch
@@ -353,6 +354,131 @@ async def test_detects_rollback_request(self, tmp_path):
         results = [e for e in collected if isinstance(e, StepResult)]
         assert results[0].rollback_request == ("spec_recommending", "cost_too_high")
 
+    @pytest.mark.asyncio
+    async def test_failure_condition_maps_conclusion_to_failed_step_result(self, tmp_path):
+        events = [
+            ToolUseStartEvent(tool_use_id="tu_1", name="complete_step"),
+            ToolUseEndEvent(
+                tool_use_id="tu_1",
+                name="complete_step",
+                input={
+                    "conclusion": {
+                        "status": "failed",
+                        "status_reason": "CREATE_FAILED: quota exceeded",
+                        "error": "deployment failed",
+                    }
+                },
+            ),
+            ToolResultEvent(tool_use_id="tu_1", tool_name="complete_step", result="ok"),
+        ]
+        fake_loop = _make_fake_agent_loop_class(events)
+        executor = _make_executor(tmp_path)
+        step = replace(
+            _make_step(),
+            failure_condition={
+                "field": "status",
+                "value": "failed",
+                "reason_fields": ["status_reason", "error"],
+            },
+        )
+        ctx = PipelineContext(SIMPLE_DEPS)
+
+        collected = []
+        with patch("iac_code.agent.agent_loop.AgentLoop", fake_loop):
+            async for event in executor.execute(step, ctx, "test_session"):
+                collected.append(event)
+
+        results = [e for e in collected if isinstance(e, StepResult)]
+        assert results[0].status == StepStatus.FAILED
+        assert "CREATE_FAILED: quota exceeded" in results[0].error
+        assert "deployment failed" in results[0].error
+        assert results[0].conclusion["status"] == "failed"
+        assert ctx.get_conclusion("intent")["status"] == "failed"
+
+    @pytest.mark.asyncio
+    async def test_failure_condition_keeps_completed_when_condition_not_met(self, tmp_path):
+        events = [
+            ToolUseStartEvent(tool_use_id="tu_1", name="complete_step"),
+            ToolUseEndEvent(
+                tool_use_id="tu_1",
+                name="complete_step",
+                input={"conclusion": {"status": "success", "stack_id": "stack-1"}},
+            ),
+            ToolResultEvent(tool_use_id="tu_1", tool_name="complete_step", result="ok"),
+        ]
+        fake_loop = _make_fake_agent_loop_class(events)
+        executor = _make_executor(tmp_path)
+        step = replace(_make_step(), failure_condition={"field": "status", "value": "failed"})
+        ctx = PipelineContext(SIMPLE_DEPS)
+
+        collected = []
+        with patch("iac_code.agent.agent_loop.AgentLoop", fake_loop):
+            async for event in executor.execute(step, ctx, "test_session"):
+                collected.append(event)
+
+        results = [e for e in collected if isinstance(e, StepResult)]
+        assert results[0].status == StepStatus.COMPLETED
+        assert results[0].error is None
+
+    @pytest.mark.asyncio
+    async def test_failure_condition_preserves_rollback_request(self, tmp_path):
+        events = [
+            ToolUseStartEvent(tool_use_id="tu_1", name="complete_step"),
+            ToolUseEndEvent(
+                tool_use_id="tu_1",
+                name="complete_step",
+                input={
+                    "conclusion": {"status": "failed", "status_reason": "CREATE_FAILED: no stock"},
+                    "rollback_request": {"target_step": "confirm_and_select", "reason": "no stock"},
+                },
+            ),
+            ToolResultEvent(tool_use_id="tu_1", tool_name="complete_step", result="ok"),
+        ]
+        fake_loop = _make_fake_agent_loop_class(events)
+        executor = _make_executor(tmp_path)
+        step = replace(
+            _make_step(),
+            failure_condition={"field": "status", "value": "failed", "reason_fields": ["status_reason"]},
+        )
+        ctx = PipelineContext(SIMPLE_DEPS)
+
+        collected = []
+        with patch("iac_code.agent.agent_loop.AgentLoop", fake_loop):
+            async for event in executor.execute(step, ctx, "test_session"):
+                collected.append(event)
+
+        results = [e for e in collected if isinstance(e, StepResult)]
+        assert results[0].status == StepStatus.COMPLETED
+        assert results[0].rollback_request == ("confirm_and_select", "no stock")
+
+    @pytest.mark.asyncio
+    async def test_failure_condition_without_reason_still_fails(self, tmp_path):
+        events = [
+            ToolUseStartEvent(tool_use_id="tu_1", name="complete_step"),
+            ToolUseEndEvent(
+                tool_use_id="tu_1",
+                name="complete_step",
+                input={"conclusion": {"status": "failed"}},
+            ),
+            ToolResultEvent(tool_use_id="tu_1", tool_name="complete_step", result="ok"),
+        ]
+        fake_loop = _make_fake_agent_loop_class(events)
+        executor = _make_executor(tmp_path)
+        step = replace(
+            _make_step(),
+            failure_condition={"field": "status", "value": "failed", "reason_fields": ["status_reason"]},
+        )
+        ctx = PipelineContext(SIMPLE_DEPS)
+
+        collected = []
+        with patch("iac_code.agent.agent_loop.AgentLoop", fake_loop):
+            async for event in executor.execute(step, ctx, "test_session"):
+                collected.append(event)
+
+        results = [e for e in collected if isinstance(e, StepResult)]
+        assert results[0].status == StepStatus.FAILED
+        assert results[0].error
+
     @pytest.mark.asyncio
     async def test_completion_guard_records_deep_copy_of_nested_tool_input(self, tmp_path):
         (tmp_path / "prompts").mkdir(exist_ok=True)
diff --git a/tests/pipeline/selling/skills/test_iac_aliyun_deploying_skill.py b/tests/pipeline/selling/skills/test_iac_aliyun_deploying_skill.py
index b362ce19..edb26839 100644
--- a/tests/pipeline/selling/skills/test_iac_aliyun_deploying_skill.py
+++ b/tests/pipeline/selling/skills/test_iac_aliyun_deploying_skill.py
@@ -62,11 +62,13 @@ def test_conclusion_schema_requires_stack_id_for_success_and_error_for_failed(se
         schema = fm["conclusion_schema"]
 
         jsonschema.validate({"status": "success", "stack_id": "stack-123"}, schema)
-        jsonschema.validate({"status": "failed", "error": "CREATE_FAILED"}, schema)
+        jsonschema.validate({"status": "failed", "error": "CREATE_FAILED", "status_reason": "CREATE_FAILED"}, schema)
         with pytest.raises(jsonschema.ValidationError):
             jsonschema.validate({"status": "success"}, schema)
         with pytest.raises(jsonschema.ValidationError):
             jsonschema.validate({"status": "failed"}, schema)
+        with pytest.raises(jsonschema.ValidationError):
+            jsonschema.validate({"status": "failed", "error": "CREATE_FAILED"}, schema)
 
 
 class TestSkillContentRosOnly:
diff --git a/tests/pipeline/selling/test_deploying_hook.py b/tests/pipeline/selling/test_deploying_hook.py
index 07df131d..014c2183 100644
--- a/tests/pipeline/selling/test_deploying_hook.py
+++ b/tests/pipeline/selling/test_deploying_hook.py
@@ -1,6 +1,11 @@
 from iac_code.pipeline.engine.context import PipelineContext
 from iac_code.pipeline.engine.ui_contract import encode_selected_candidate, parse_selected_candidate
-from iac_code.pipeline.selling.hooks.deploying import normalize_selected_plan, on_enter, resolve_selected_candidate
+from iac_code.pipeline.selling.hooks.deploying import (
+    normalize_selected_plan,
+    on_enter,
+    on_exit,
+    resolve_selected_candidate,
+)
 
 
 def _evaluated_candidates():
@@ -334,3 +339,30 @@ def test_on_enter_normalizes_selected_plan_in_context():
     selected_plan = context.get_conclusion("selected_plan")
     assert selected_plan["selection_valid"] is True
     assert selected_plan["selected_candidate"]["output_path"] == "templates/a.yml"
+
+
+def test_on_exit_backfills_status_reason_from_error_for_failed_deployment():
+    context = PipelineContext({"selected_plan": [], "evaluated_candidates": []})
+    conclusion = {"status": "failed", "error": "CREATE_FAILED: ECS quota exceeded"}
+
+    on_exit(context, conclusion)
+
+    assert conclusion["status_reason"] == "CREATE_FAILED: ECS quota exceeded"
+
+
+def test_on_exit_keeps_existing_status_reason():
+    context = PipelineContext({"selected_plan": [], "evaluated_candidates": []})
+    conclusion = {"status": "failed", "status_reason": "CREATE_FAILED: no stock", "error": "summary"}
+
+    on_exit(context, conclusion)
+
+    assert conclusion["status_reason"] == "CREATE_FAILED: no stock"
+
+
+def test_on_exit_ignores_successful_deployment():
+    context = PipelineContext({"selected_plan": [], "evaluated_candidates": []})
+    conclusion = {"status": "success", "stack_id": "stack-1"}
+
+    on_exit(context, conclusion)
+
+    assert "status_reason" not in conclusion
diff --git a/tests/pipeline/selling/test_deploying_prompt.py b/tests/pipeline/selling/test_deploying_prompt.py
index 207bc893..6ea1ee63 100644
--- a/tests/pipeline/selling/test_deploying_prompt.py
+++ b/tests/pipeline/selling/test_deploying_prompt.py
@@ -86,3 +86,45 @@ def test_deploying_renders_stack_outputs_after_complete_step() -> None:
     assert deploying_step.complete_step_terminal is False
     assert "`complete_step` 成功返回后" in prompt
     assert "`complete_step.conclusion.outputs` 渲染 Stack Outputs" in prompt
+
+
+def test_deploying_step_declares_failure_condition_and_failure_reason_guard() -> None:
+    loaded = load_pipeline_dir(_selling_dir())
+    deploying_step = next(step for step in loaded.steps if step.step_id == "deploying")
+
+    assert deploying_step.failure_condition == {
+        "field": "status",
+        "value": "failed",
+        "reason_fields": ["status_reason", "error"],
+    }
+
+    failure_guard = next(
+        guard
+        for guard in deploying_step.completion_guards
+        if guard.get("when_conclusion_field_equals", {}).get("status") == "failed"
+    )
+    assert failure_guard["required_conclusion_field"] == "status_reason"
+    assert failure_guard["message_key"] == "deploy_report_failure_reason"
+
+
+def test_deploying_prompt_requires_reporting_status_reason_and_recovery() -> None:
+    selling_dir = _selling_dir()
+    loaded = load_pipeline_dir(selling_dir)
+    deploying_step = next(step for step in loaded.steps if step.step_id == "deploying")
+    prompt = (selling_dir / deploying_step.prompt_file).read_text(encoding="utf-8")
+
+    assert "status_reason" in prompt
+    assert "rollback_request" in prompt
+    assert "confirm_and_select" in prompt
+
+
+def test_deploying_skill_requires_status_reason_for_failed_conclusion() -> None:
+    loaded = load_pipeline_dir(_selling_dir())
+    deploying_step = next(step for step in loaded.steps if step.step_id == "deploying")
+    schema = deploying_step.conclusion_schema
+
+    assert "status_reason" in schema["properties"]
+    failed_branch = next(
+        branch for branch in schema["allOf"] if branch["if"]["properties"]["status"]["const"] == "failed"
+    )
+    assert set(failed_branch["then"]["required"]) == {"error", "status_reason"}