fix(pipeline): 诊断卡死的候选子步骤，禁止静默换新候选掩盖失败

Session 7be8bcd32aa44fd399f3e5bf7ec7db7d 中候选 096430be 的
cost_estimating 子步骤长时间 conclusion=null / status=working，流水线
既未诊断也未收敛，而是直接开启新的 evaluate_candidates，由一个缩减模板
的新候选顶替，原失败被完全掩盖，交付方案被静默改写。

三处根因分别修复：

1. 子步骤卡死不可观测。SubPipelineSpec 新增 sub_step_stall_timeout_s /
   sub_step_stall_retries，SubPipelineExecutor 用看门狗按"相邻事件间隔"
   判定卡死（而非总时长），卡死时以 SubStepStalled 记录显式失败原因、
   卡死秒数与重试信息。等待用户回答与流水线暂停期间窗口作废，避免把
   合法等待误判为卡死。

2. 卡死后不重试原子步骤。重试预算内优先原地重试同一子步骤，候选状态保持
   running、已完成子步骤结论保留；预算耗尽才让候选进入终态。

3. 候选无终态、失败无原因。取消候选时记录 cancel_reason 并落终态条目
   （status 仍为 failed 以兼容 restore/replay 的三态路由，terminal_reason
   区分 canceled / superseded）；原地重启不写终态。聚合阶段不再产出无原因
   的 {failed: True}，缺失结果时补 CandidateOutcomeMissing 原因。

selling 流水线 evaluate_candidate 启用 600s 阈值 + 1 次原地重试。

Aone: 84817164

Author: ruanzhengxin-zhuxing

src/iac_code/pipeline/engine/loader.py

@@ -245,10 +245,28 @@ def _parse_sub_pipelines(
             max_rollbacks=sub_raw.get("max_rollbacks", 5),
             iterate_over=sub_raw.get("iterate_over", ""),
             context_fields_from_parent=sub_raw.get("context_fields_from_parent", []),
+            sub_step_stall_timeout_s=_parse_sub_step_stall_timeout(sub_name, sub_raw.get("sub_step_stall_timeout_s")),
+            sub_step_stall_retries=_parse_sub_step_stall_retries(sub_name, sub_raw.get("sub_step_stall_retries", 1)),
         )
     return result
 
 
+def _parse_sub_step_stall_timeout(sub_name: str, raw: object) -> float | None:
+    if raw is None:
+        return None
+    if not isinstance(raw, (int, float)) or isinstance(raw, bool) or raw <= 0:
+        raise ValueError(
+            f"sub_pipelines.{sub_name}.sub_step_stall_timeout_s must be a positive number or omitted, got {raw!r}"
+        )
+    return float(raw)
+
+
+def _parse_sub_step_stall_retries(sub_name: str, raw: object) -> int:
+    if not isinstance(raw, int) or isinstance(raw, bool) or raw < 0:
+        raise ValueError(f"sub_pipelines.{sub_name}.sub_step_stall_retries must be a non-negative integer, got {raw!r}")
+    return raw
+
+
 def _parse_steps(raw_steps: list[dict]) -> list[StepSpec]:
     steps: list[StepSpec] = []
     for raw in raw_steps:

src/iac_code/pipeline/engine/pipeline_runner.py

@@ -3197,6 +3197,11 @@ def _cancel_candidate_task(
 
         parent_step_id = parent_step_id or self.state_machine.current_step.step_id
         candidate_name = state.get("name", "")
+        # A candidate that is about to restart in place keeps running; every other
+        # cancellation reason must leave a terminal, diagnosable state behind so a
+        # replacement candidate cannot mask why this one stopped.
+        if reason != "candidate_restart":
+            state["cancel_reason"] = reason
         if not state.get("_candidate_cancelled_observed", False):
             state["_candidate_cancelled_observed"] = True
             self._observability.candidate_cancelled(
@@ -4605,6 +4610,50 @@ async def save_candidate_failed(i: int, state: dict[str, Any]) -> None:
             failed_by_index[i] = dict(entry)
             await self._save_running(step.step_id, reason="parallel candidate failed")
 
+        def save_candidate_cancelled_sync(i: int, state: dict[str, Any], reason: str) -> None:
+            """Persist a terminal entry for a cancelled candidate, preserving the reason.
+
+            Runs synchronously because the owning task is already cancelled. The
+            status stays ``failed`` so restore/replay keep routing on the three
+            known states; ``terminal_reason`` carries why it stopped.
+            """
+            active_attempt_id = state.get("active_attempt_id")
+            if active_attempt_id:
+                self._mark_attempt_status(active_attempt_id, "failed")
+            terminal_reason = "superseded" if reason == "hard_interrupt_parent_rollback" else "canceled"
+            failure = public_error(
+                message=_("Candidate {index} was {reason} before completing sub-step {sub_step}.").format(
+                    index=i + 1,
+                    reason=terminal_reason,
+                    sub_step=state.get("current_sub_step") or "?",
+                ),
+                error_type="CandidateCancelled",
+                extra_details={
+                    "cancel_reason": reason,
+                    "terminal_reason": terminal_reason,
+                    "sub_step_id": state.get("current_sub_step") or None,
+                },
+            )
+            entry = {
+                "status": "failed",
+                "terminal_reason": terminal_reason,
+                "candidate": candidates[i],
+                "sub_pipeline_id": state.get("sub_pipeline_id") or f"{sub_spec.name}_candidate_{i}",
+                "state_machine": state.get("state_machine"),
+                "context": state.get("context"),
+                "current_sub_step": state.get("current_sub_step", ""),
+                "current_index": state.get("current_index"),
+                "active_attempt_id": active_attempt_id,
+                "transcript_id": state.get("transcript_id"),
+                "conclusions": state.get("conclusions", {}),
+                "step_conclusions": state.get("step_conclusions", {}),
+                "error": state.get("error") or failure.summary,
+                "error_details": state.get("error_details") or failure.details,
+            }
+            self._execution.setdefault("candidates", {})[str(i)] = entry
+            failed_by_index[i] = dict(entry)
+            self._save_running_sync(step.step_id, reason="parallel candidate cancelled")
+
         async def put_candidate_event(candidate_index: int, event: Any) -> None:
             nonlocal event_sequence
             event_sequence += 1
@@ -4781,6 +4830,12 @@ async def record_sub_step_state(payload: dict[str, Any]) -> None:
                     await put_candidate_event(i, event)
             except asyncio.CancelledError:
                 logger.debug("Candidate %d cancelled", i)
+                cancel_reason = state.get("cancel_reason")
+                if isinstance(cancel_reason, str) and i not in conclusions_by_index and i not in failed_by_index:
+                    try:
+                        save_candidate_cancelled_sync(i, state, cancel_reason)
+                    except PipelineStatePersistenceError:
+                        logger.warning("Failed to persist cancelled candidate %d terminal state", i)
             except PipelineStatePersistenceError as exc:
                 await put_candidate_event(i, exc)
             except SessionBackupBlocked as exc:
@@ -4964,9 +5019,27 @@ async def record_sub_step_state(payload: dict[str, Any]) -> None:
                     result["error"] = restored["error"]
                 if restored.get("error_details") is not None:
                     result["error_details"] = restored["error_details"]
+                if restored.get("terminal_reason") is not None:
+                    result["terminal_reason"] = restored["terminal_reason"]
                 aggregated.append(result)
             else:
-                aggregated.append({"candidate": candidate, "failed": True})
+                # Never aggregate a reason-less failure: a replacement candidate must
+                # not be able to silently take over from an undiagnosed one.
+                failure = public_error(
+                    message=_("Candidate {index} ended without a result or a recorded failure reason.").format(
+                        index=i + 1
+                    ),
+                    error_type="CandidateOutcomeMissing",
+                    extra_details={"parent_step_id": step.step_id, "candidate_index": i},
+                )
+                aggregated.append(
+                    {
+                        "candidate": candidate,
+                        "failed": True,
+                        "error": failure.summary,
+                        "error_details": failure.details,
+                    }
+                )
 
         self.context.set_conclusion(step.conclusion_field, aggregated)
         candidate_success_count = sum(1 for item in aggregated if isinstance(item, dict) and not item.get("failed"))

src/iac_code/pipeline/engine/step_spec.py

@@ -117,6 +117,13 @@ class SubPipelineSpec:
     max_rollbacks: int
     iterate_over: str
     context_fields_from_parent: list[str] = field(default_factory=list)
+    # Seconds a sub-step may go without emitting any event before it is treated
+    # as stalled. ``None`` disables stall detection.
+    sub_step_stall_timeout_s: float | None = None
+    # How many times a stalled sub-step is retried in place before the candidate
+    # is failed. Retrying the same sub-step is preferred over replacing the
+    # candidate so the original failure stays visible.
+    sub_step_stall_retries: int = 1
 
 
 @dataclass