fix(candidate): 封闭候选仓储与审批恢复的全部崩溃窗口

- 为候选命令增加可恢复审批状态，并只凭单一已批准调用、原 command_id 与输入哈希恢复\n- 区分仓储确定性拒绝与提交结果未知，保持幂等命令可重放\n- 对账已落盘 Tool Result，避免重复结果消息并恢复 Trace head\n- 在恢复完成前拒绝新消息、回溯、反思、压缩和停止写入口\n- 保持普通工具 execution_unknown 的失败关闭语义

cyber_agent/application/__init__.py

@@ -25,6 +25,7 @@ _LAZY_EXPORTS = {
     "CandidatePointer": ("cyber_agent.application.candidate", "CandidatePointer"),
     "CandidateRef": ("cyber_agent.application.candidate", "CandidateRef"),
     "CandidateRepository": ("cyber_agent.application.candidate", "CandidateRepository"),
+    "CandidateRepositoryRejected": ("cyber_agent.application.candidate", "CandidateRepositoryRejected"),
     "CandidateReviewAction": ("cyber_agent.application.candidate", "CandidateReviewAction"),
     "CandidateVersionRequest": ("cyber_agent.application.candidate", "CandidateVersionRequest"),
     "ContextMaterial": ("cyber_agent.application.ports", "ContextMaterial"),
@@ -65,6 +66,7 @@ __all__ = [
     "CandidatePointer",
     "CandidateRef",
     "CandidateRepository",
+    "CandidateRepositoryRejected",
     "CandidateReviewAction",
     "CandidateVersionRequest",
     "ContextMaterial",

cyber_agent/application/candidate.py

@@ -73,6 +73,10 @@ class CandidateAdoptionReceipt(ApplicationModel):
     committed: bool = True
 
 
+class CandidateRepositoryRejected(ValueError):
+    """Deterministic rejection proving that no candidate version was committed."""
+
+
 class CandidateReviewAction(ApplicationModel):
     action: Literal["adopt", "discard", "revise"]
     candidate_ref: CandidateRef

cyber_agent/application/candidate_service.py

@@ -16,6 +16,7 @@ from cyber_agent.application.candidate import (
     CandidatePointer,
     CandidateRef,
     CandidateRepository,
+    CandidateRepositoryRejected,
     CandidateReviewAction,
     CandidateVersionRequest,
     MAX_CANDIDATES_PER_ROOT,
@@ -64,6 +65,50 @@ class CandidateService:
             self._locks[key] = lock
         return lock
 
+    async def has_replayable_version_command(
+        self,
+        trace_id: str,
+        *,
+        command_id: str,
+        arguments: dict[str, Any],
+    ) -> bool:
+        """Verify the exact durable candidate command behind an orphaned call."""
+        try:
+            request = arguments.get("request")
+            if not isinstance(request, dict):
+                return False
+            operation = request.get("operation")
+            if operation not in {"create", "fork", "merge"}:
+                return False
+            parent_refs = [
+                CandidatePointer.model_validate(item)
+                for item in request.get("parent_refs", [])
+            ]
+            input_hash = sha256(canonical_json({
+                "operation": operation,
+                "content": request.get("content"),
+                "parent_refs": [
+                    item.model_dump(mode="json") for item in parent_refs
+                ],
+            }).encode("utf-8")).hexdigest()
+            command_hash = sha256(canonical_json({
+                "trace_id": trace_id,
+                "command_id": command_id,
+            }).encode("utf-8")).hexdigest()
+            operation_id = f"candidate:{command_hash}"
+            _trace, root = await self._require_owner_trace(trace_id)
+            ledger = await self._load_ledger(root.trace_id)
+        except Exception:
+            return False
+        return any(
+            item.operation_id == operation_id
+            and item.command_id == command_id
+            and item.operation == operation
+            and item.input_hash == input_hash
+            and item.status in {"pending", "completed", "failed"}
+            for item in ledger.operations
+        )
+
     async def manage(
         self,
         trace_id: str,
@@ -109,6 +154,7 @@ class CandidateService:
                 ),
                 None,
             )
+            recovering_pending_operation = existing_operation is not None
             if existing_operation is not None:
                 if existing_operation.input_hash != input_hash:
                     raise CandidateStateError(
@@ -190,11 +236,25 @@ class CandidateService:
                     parent_refs=tuple(parent_refs),
                     content=content,
                 )
-                artifact_ref = ArtifactRef.model_validate(
-                    await self.repository.merge(request)
-                    if operation == "merge"
-                    else await self.repository.put_version(request)
-                )
+                try:
+                    raw_artifact_ref = (
+                        await self.repository.merge(request)
+                        if operation == "merge"
+                        else await self.repository.put_version(request)
+                    )
+                except CandidateRepositoryRejected as exc:
+                    raise CandidateStateError(str(exc)) from exc
+                except Exception as exc:
+                    raise RecoverableToolExecutionError(
+                        "Candidate Repository execution outcome is unknown; "
+                        "the original idempotent command must be replayed"
+                    ) from exc
+                try:
+                    artifact_ref = ArtifactRef.model_validate(raw_artifact_ref)
+                except ValueError as exc:
+                    raise CandidateStateError(
+                        f"CandidateRepository returned an invalid ArtifactRef: {exc}"
+                    ) from exc
                 materials, issues = await resolve_artifact_refs(
                     [artifact_ref],
                     resolver=self.artifact_resolver,
@@ -203,6 +263,11 @@ class CandidateService:
                 )
                 if issues or len(materials) != 1:
                     reason = issues[0].reason if issues else "Artifact was not resolved"
+                    if any(issue.outcome == "unknown" for issue in issues):
+                        raise RecoverableToolExecutionError(
+                            "Candidate ArtifactResolver is temporarily unavailable; "
+                            "the original command must remain replayable"
+                        )
                     raise CandidateStateError(
                         f"CandidateRepository returned an invalid ArtifactRef: {reason}"
                     )
@@ -217,8 +282,27 @@ class CandidateService:
                     created_at_sequence=effective_at_sequence,
                 )
             except Exception as exc:
+                if isinstance(exc, RecoverableToolExecutionError):
+                    raise
+                # A recovered pending intent may already have committed in the
+                # application Repository. A transient Repository/Resolver error
+                # cannot prove failure, so keep the original command replayable.
+                if (
+                    recovering_pending_operation
+                    and not isinstance(exc, CandidateStateError)
+                ):
+                    raise RecoverableToolExecutionError(
+                        "Candidate recovery was interrupted before its durable "
+                        "ledger outcome could be established"
+                    ) from exc
                 failed = self._fail_operation(ledger, operation_id, str(exc))
-                await self._persist_ledger(root_trace_id, failed)
+                try:
+                    await self._persist_ledger(root_trace_id, failed)
+                except Exception as publish_exc:
+                    raise RecoverableToolExecutionError(
+                        "Candidate Repository result is durable but its failed "
+                        "ledger outcome must be replayed with the original command ID"
+                    ) from publish_exc
                 raise
             completed = self._complete_operation(
                 ledger,

cyber_agent/core/runner.py

@@ -1686,6 +1686,36 @@ class AgentRunner:
                 "Application runs do not allow caller-provided system messages"
             )
         if config.trace_id:
+            if self.trace_store and hasattr(
+                self.trace_store,
+                "get_tool_approval_batch",
+            ):
+                batch = await self.trace_store.get_tool_approval_batch(
+                    config.trace_id
+                )
+                if (
+                    batch is not None
+                    and batch.status == "executing"
+                    and await self._recover_candidate_approval_if_durable(
+                        config.trace_id,
+                        batch,
+                    )
+                ):
+                    batch = await self.trace_store.get_tool_approval_batch(
+                        config.trace_id
+                    )
+                if (
+                    batch is not None
+                    and batch.status == "recoverable_candidate_command"
+                    and (
+                        messages
+                        or config.after_sequence is not None
+                        or config.force_side_branch
+                    )
+                ):
+                    raise ValueError(
+                        "recoverable_candidate_command_must_resume_first"
+                    )
             return await self._prepare_existing_trace(config)
         else:
             return await self._prepare_new_trace(messages, config)
@@ -2017,6 +2047,19 @@ class AgentRunner:
             role = self.application_binding.role(snapshot.role_id)
             config.system_prompt = role.system_prompt
             config.skills = []
+        if (
+            config.approval_batch_id is None
+            and hasattr(self.trace_store, "get_tool_approval_batch")
+        ):
+            approval_batch = await self.trace_store.get_tool_approval_batch(
+                trace_obj.trace_id
+            )
+            if (
+                approval_batch is not None
+                and approval_batch.status == "recoverable_candidate_command"
+            ):
+                self._validate_recoverable_candidate_approval(approval_batch)
+                config.approval_batch_id = approval_batch.batch_id
         persisted_memory_identity = trace_obj.context.get(
             MEMORY_IDENTITY_CONTEXT_KEY
         )
@@ -2857,8 +2900,10 @@ class AgentRunner:
         batch = await self.trace_store.get_tool_approval_batch(trace.trace_id)
         if batch is None or batch.batch_id != config.approval_batch_id:
             raise RuntimeError("approved tool batch was not found")
-        if batch.status != "decided":
+        if batch.status not in {"decided", "recoverable_candidate_command"}:
             raise RuntimeError(f"tool approval batch is not executable: {batch.status}")
+        if batch.status == "recoverable_candidate_command":
+            self._validate_recoverable_candidate_approval(batch)
         if any(call.decision == "pending" for call in batch.calls):
             raise RuntimeError("tool approval batch still has pending decisions")
         if any(call.execution_status == "executing" for call in batch.calls):
@@ -2897,26 +2942,41 @@ class AgentRunner:
                     trace.trace_id,
                     batch,
                 )
-                tool_result = await self.tools.execute(
-                    call.tool_name,
-                    call.effective_arguments,
-                    uid=config.uid or "",
-                    context=self._build_tool_context(
-                        config=config,
-                        trace=trace,
-                        trace_id=trace.trace_id,
-                        goal_id=current_goal_id,
-                        goal_tree=goal_tree,
-                        sequence=sequence,
-                        head_sequence=head_seq,
+                try:
+                    tool_result = await self.tools.execute(
+                        call.tool_name,
+                        call.effective_arguments,
+                        uid=config.uid or "",
+                        context=self._build_tool_context(
+                            config=config,
+                            trace=trace,
+                            trace_id=trace.trace_id,
+                            goal_id=current_goal_id,
+                            goal_tree=goal_tree,
+                            sequence=sequence,
+                            head_sequence=head_seq,
+                            tool_call_id=call.tool_call_id,
+                            side_branch_ctx=side_branch_ctx,
+                            trigger_event=trigger_event,
+                        ),
+                        allowed_tool_names=runtime_tool_names,
                         tool_call_id=call.tool_call_id,
-                        side_branch_ctx=side_branch_ctx,
-                        trigger_event=trigger_event,
-                    ),
-                    allowed_tool_names=runtime_tool_names,
-                    tool_call_id=call.tool_call_id,
-                    approval_grant=approval_grant(batch, call),
-                )
+                        approval_grant=approval_grant(batch, call),
+                    )
+                except RecoverableToolExecutionError:
+                    if len(batch.calls) != 1 or call.tool_name != "manage_candidate":
+                        raise
+                    # CandidateService only raises this signal after its durable,
+                    # idempotent intent exists.  Preserve the original human grant
+                    # and call ID so a restart can publish the ledger terminal state.
+                    call.execution_status = "not_started"
+                    batch.status = "recoverable_candidate_command"
+                    batch.updated_at = datetime.now().isoformat()
+                    await self.trace_store.replace_tool_approval_batch(
+                        trace.trace_id,
+                        batch,
+                    )
+                    raise
             else:
                 raise RuntimeError(f"unsupported tool approval decision: {call.decision}")
 
@@ -2999,6 +3059,23 @@ class AgentRunner:
         config.approval_batch_id = None
         return history, sequence, head_seq
 
+    @staticmethod
+    def _validate_recoverable_candidate_approval(batch: Any) -> None:
+        """Accept only the exact approved candidate command safe for replay."""
+        if len(batch.calls) != 1:
+            raise RuntimeError(
+                "recoverable candidate approval must contain exactly one call"
+            )
+        call = batch.calls[0]
+        if (
+            call.tool_name != "manage_candidate"
+            or call.decision not in {"approved", "auto_approved"}
+            or call.execution_status != "not_started"
+        ):
+            raise RuntimeError(
+                "recoverable candidate approval does not contain a replayable grant"
+            )
+
     async def _mark_approval_execution_unknown(self, trace_id: str) -> bool:
         """Fail closed if an approved batch stopped after execution began."""
         if not self.trace_store or not hasattr(
@@ -3008,6 +3085,8 @@ class AgentRunner:
         batch = await self.trace_store.get_tool_approval_batch(trace_id)
         if batch is None or batch.status != "executing":
             return False
+        if await self._recover_candidate_approval_if_durable(trace_id, batch):
+            return False
         batch.status = "execution_unknown"
         batch.updated_at = datetime.now().isoformat()
         for call in batch.calls:
@@ -3024,6 +3103,61 @@ class AgentRunner:
         )
         return True
 
+    async def _recover_candidate_approval_if_durable(
+        self,
+        trace_id: str,
+        batch: Any,
+    ) -> bool:
+        """Recover only an exact approved candidate command with a durable intent."""
+        if self.candidate_service is None or len(batch.calls) != 1:
+            return False
+        call = batch.calls[0]
+        if (
+            call.tool_name != "manage_candidate"
+            or call.decision not in {"approved", "auto_approved"}
+            or call.execution_status not in {"executing", "execution_unknown"}
+        ):
+            return False
+        messages = await self.trace_store.get_trace_messages(trace_id)
+        persisted_results = [
+            message
+            for message in messages
+            if message.role == "tool"
+            and message.tool_call_id == call.tool_call_id
+        ]
+        if persisted_results:
+            if len(persisted_results) != 1:
+                return False
+            result_message = persisted_results[0]
+            content = result_message.content
+            if (
+                result_message.parent_sequence != batch.assistant_sequence
+                or not isinstance(content, dict)
+                or content.get("tool_name") != "manage_candidate"
+            ):
+                return False
+            call.result_message_id = result_message.message_id
+            call.execution_status = "executed"
+            batch.status = "completed"
+            batch.updated_at = datetime.now().isoformat()
+            await self.trace_store.replace_tool_approval_batch(trace_id, batch)
+            await self.trace_store.update_trace(
+                trace_id,
+                head_sequence=result_message.sequence,
+            )
+            return True
+        if not await self.candidate_service.has_replayable_version_command(
+            trace_id,
+            command_id=call.tool_call_id,
+            arguments=call.effective_arguments,
+        ):
+            return False
+        call.execution_status = "not_started"
+        batch.status = "recoverable_candidate_command"
+        batch.updated_at = datetime.now().isoformat()
+        await self.trace_store.replace_tool_approval_batch(trace_id, batch)
+        return True
+
     async def _agent_loop(
         self,
         trace: Trace,
@@ -4886,7 +5020,16 @@ class AgentRunner:
             raise RuntimeError(
                 "Orphaned lifecycle replay returned a non-JSON result"
             ) from exc
-        if not isinstance(payload, dict) or payload.get("status") != "completed":
+        replay_completed = (
+            isinstance(payload, dict)
+            and payload.get("status") == "completed"
+        )
+        replay_failed_safely = (
+            tool_name == "manage_candidate"
+            and isinstance(payload, dict)
+            and payload.get("status") == "failed"
+        )
+        if not replay_completed and not replay_failed_safely:
             reason = payload.get("error") if isinstance(payload, dict) else None
             raise RuntimeError(
                 "Orphaned lifecycle replay failed closed: "

cyber_agent/tools/approval.py

@@ -76,6 +76,7 @@ class ToolApprovalBatchV1(_StrictModel):
         "pending",
         "decided",
         "executing",
+        "recoverable_candidate_command",
         "completed",
         "cancelled",
         "execution_unknown",

cyber_agent/trace/run_api.py

@@ -405,6 +405,28 @@ async def _restore_runner_and_config(
                 status_code=409,
                 detail=f"Failed to restore project environment: {project_name}",
             ) from exc
+    if runner.trace_store and hasattr(
+        runner.trace_store,
+        "get_tool_approval_batch",
+    ):
+        batch = await runner.trace_store.get_tool_approval_batch(trace.trace_id)
+        if (
+            batch is not None
+            and batch.status == "executing"
+            and await runner._recover_candidate_approval_if_durable(
+                trace.trace_id,
+                batch,
+            )
+        ):
+            batch = await runner.trace_store.get_tool_approval_batch(trace.trace_id)
+        if batch is not None and batch.status == "recoverable_candidate_command":
+            if after_sequence is not None or force_side_branch:
+                raise HTTPException(
+                    status_code=409,
+                    detail="recoverable_candidate_command_must_resume_first",
+                )
+            if approval_batch_id is None:
+                config.approval_batch_id = batch.batch_id
     return runner, config
 
 
@@ -696,6 +718,21 @@ async def run_trace(trace_id: str, req: TraceRunRequest):
         base_runner=runner,
         after_sequence=after_sequence,
     )
+    approval_batch = (
+        await runner.trace_store.get_tool_approval_batch(trace_id)
+        if runner.trace_store
+        and hasattr(runner.trace_store, "get_tool_approval_batch")
+        else None
+    )
+    if (
+        approval_batch is not None
+        and approval_batch.status == "recoverable_candidate_command"
+        and req.messages
+    ):
+        raise HTTPException(
+            status_code=409,
+            detail="recoverable_candidate_command_must_resume_first",
+        )
 
     # 验证 trace 状态
     if runner.trace_store:
@@ -897,6 +934,19 @@ async def stop_trace(trace_id: str):
                 trace=trace,
                 base_runner=runner,
             )
+            batch = (
+                await runner.trace_store.get_tool_approval_batch(trace_id)
+                if hasattr(runner.trace_store, "get_tool_approval_batch")
+                else None
+            )
+            if (
+                batch is not None
+                and batch.status == "recoverable_candidate_command"
+            ):
+                raise HTTPException(
+                    status_code=409,
+                    detail="recoverable_candidate_command_must_resume_first",
+                )
         if trace and trace.status == "waiting_confirmation":
             batch = (
                 await runner.trace_store.get_tool_approval_batch(trace_id)
@@ -1182,7 +1232,12 @@ async def reconcile_traces():
                 trace.status == "running"
                 or (
                     batch is not None
-                    and batch.status in {"pending", "decided", "executing"}
+                    and batch.status in {
+                        "pending",
+                        "decided",
+                        "executing",
+                        "recoverable_candidate_command",
+                    }
                 )
             )
             if is_application_trace and needs_recovery_action:
@@ -1223,8 +1278,39 @@ async def reconcile_traces():
                     )
                     count += 1
                 continue
-            if batch and batch.status == "decided" and trace.status in {
-                "running", "stopped", "waiting_confirmation",
+            if batch and batch.status == "executing":
+                if await trace_runner._recover_candidate_approval_if_durable(
+                    tid,
+                    batch,
+                ):
+                    batch = await trace_runner.trace_store.get_tool_approval_batch(
+                        tid
+                    )
+                else:
+                    batch.status = "execution_unknown"
+                    batch.updated_at = datetime.now().isoformat()
+                    for call in batch.calls:
+                        if call.execution_status == "executing":
+                            call.execution_status = "execution_unknown"
+                    await trace_runner.trace_store.replace_tool_approval_batch(
+                        tid,
+                        batch,
+                    )
+                    await trace_runner.trace_store.update_trace(
+                        tid,
+                        status="failed",
+                        error_message=(
+                            "Tool execution outcome is unknown after process restart; "
+                            "automatic retry was refused"
+                        ),
+                    )
+                    count += 1
+                    continue
+            if batch and batch.status in {
+                "decided",
+                "recoverable_candidate_command",
+            } and trace.status in {
+                "running", "stopped", "waiting_confirmation", "failed",
             }:
                 logger.info(
                     "[Reconciliation] Resuming safely-decided approval batch for %s",
@@ -1252,23 +1338,6 @@ async def reconcile_traces():
                 _running_tasks[tid] = task
                 count += 1
                 continue
-            if batch and batch.status == "executing":
-                batch.status = "execution_unknown"
-                batch.updated_at = datetime.now().isoformat()
-                for call in batch.calls:
-                    if call.execution_status == "executing":
-                        call.execution_status = "execution_unknown"
-                await trace_runner.trace_store.replace_tool_approval_batch(tid, batch)
-                await trace_runner.trace_store.update_trace(
-                    tid,
-                    status="failed",
-                    error_message=(
-                        "Tool execution outcome is unknown after process restart; "
-                        "automatic retry was refused"
-                    ),
-                )
-                count += 1
-                continue
             if trace.status == "running":
                 logger.info(f"[Reconciliation] Fixing trace {tid}: running -> stopped")
                 await trace_runner.trace_store.update_trace(

frontend/react-template/src/api/traceApi.ts

@@ -43,7 +43,14 @@ export interface ToolApprovalBatchV1 {
   trace_id: string;
   assistant_message_id: string;
   assistant_sequence: number;
-  status: "pending" | "decided" | "executing" | "completed" | "cancelled" | "execution_unknown";
+  status:
+    | "pending"
+    | "decided"
+    | "executing"
+    | "recoverable_candidate_command"
+    | "completed"
+    | "cancelled"
+    | "execution_unknown";
   calls: ToolApprovalCallV1[];
   created_at: string;
   updated_at: string;