fix(candidate): 原子预留审核命令并恢复候选发布意图

父级审核首次落盘时同时写入 batch envelope 与全部 action pending operation，持久化预留容量，防止崩溃后兄弟命令抢占恢复槽位。

候选 Repository 已返回但账本发布失败时抛出可恢复执行信号，ToolRegistry 和并行 Runner 不再把原调用封成普通错误结果；续跑以原 tool_call_id 受控重放 manage_candidate。

重放继续经过人工确认门禁、生命周期独占门禁和参数哈希校验，不按相同载荷合并新的合法命令。

新增 2048 极限容量崩溃序列和真实 Registry/Runner 原命令恢复测试，证明兄弟命令无副作用、外部幂等键唯一且 pending intent 最终完成。

cyber_agent/application/candidate_service.py

@@ -29,6 +29,7 @@ from cyber_agent.application.quality import (
 from cyber_agent.core.agent_mode import require_mutable_trace_policy
 from cyber_agent.core.artifacts import ArtifactRef, ArtifactResolver, resolve_artifact_refs
 from cyber_agent.core.task_protocol import ensure_task_protocol, task_contract_ref
+from cyber_agent.tools.errors import RecoverableToolExecutionError
 
 
 class CandidateStateError(ValueError):
@@ -137,6 +138,10 @@ class CandidateService:
                     raise CandidateStateError(
                         "Candidate ledger candidate limit exceeded before execution"
                     )
+                if len(ledger.operations) >= MAX_CANDIDATE_OPERATIONS_PER_ROOT:
+                    raise CandidateStateError(
+                        "Candidate operation limit exceeded before execution"
+                    )
                 try:
                     pointer = self._allocate_pointer(
                         operation,
@@ -223,7 +228,13 @@ class CandidateService:
             # If this atomic publish fails after the Repository committed, the
             # pending operation remains recoverable. Re-execution uses the same
             # operation_id and the Repository's idempotency contract.
-            await self._persist_ledger(root_trace_id, completed)
+            try:
+                await self._persist_ledger(root_trace_id, completed)
+            except Exception as exc:
+                raise RecoverableToolExecutionError(
+                    "Candidate Repository result is durable but its ledger publish "
+                    "must be replayed with the original command ID"
+                ) from exc
             await self._emit_candidate_registered(candidate_ref, operation_id)
             return candidate_ref
 
@@ -522,46 +533,66 @@ class CandidateService:
                 ),
                 None,
             )
+            operations_to_reserve: list[CandidateOperationRecord] = []
             if batch_operation is not None:
                 if batch_operation.input_hash != batch_input_hash:
                     raise CandidateStateError(
                         "Candidate review batch payload changed during recovery"
                     )
             else:
-                existing_operation_ids = {
-                    item.operation_id for item in ledger.operations
-                }
-                new_action_operations = sum(
-                    self._review_operation_id(
-                        parent_trace_id,
-                        child_trace_id,
-                        action,
-                        command_id=command_id,
-                        effective_at_sequence=effective_at_sequence,
-                        action_index=action_index,
-                    ) not in existing_operation_ids
-                    for action_index, action in enumerate(actions)
+                batch_operation = CandidateOperationRecord(
+                    operation_id=batch_operation_id,
+                    command_id=command_id,
+                    operation="review_batch",
+                    input_hash=batch_input_hash,
+                )
+                operations_to_reserve.append(batch_operation)
+
+            existing_operation_ids = {
+                item.operation_id for item in ledger.operations
+            }
+            for action_index, action in enumerate(actions):
+                action_operation_id = self._review_operation_id(
+                    parent_trace_id,
+                    child_trace_id,
+                    action,
+                    command_id=command_id,
+                    effective_at_sequence=effective_at_sequence,
+                    action_index=action_index,
                 )
-                required_slots = 1 + new_action_operations
+                if action_operation_id in existing_operation_ids:
+                    continue
+                operations_to_reserve.append(CandidateOperationRecord(
+                    operation_id=action_operation_id,
+                    command_id=command_id,
+                    operation=action.action,
+                    input_hash=sha256(
+                        canonical_json(action).encode("utf-8")
+                    ).hexdigest(),
+                    candidate=CandidatePointer(
+                        candidate_id=action.candidate_ref.candidate_id,
+                        revision=action.candidate_ref.revision,
+                    ),
+                ))
+
+            if operations_to_reserve:
                 if (
-                    len(ledger.operations) + required_slots
+                    len(ledger.operations) + len(operations_to_reserve)
                     > MAX_CANDIDATE_OPERATIONS_PER_ROOT
                 ):
                     raise CandidateStateError(
                         "Candidate review operation limit exceeded before execution"
                     )
-                batch_operation = CandidateOperationRecord(
-                    operation_id=batch_operation_id,
-                    command_id=command_id,
-                    operation="review_batch",
-                    input_hash=batch_input_hash,
-                )
                 ledger = ledger.model_copy(update={
-                    "operations": (*ledger.operations, batch_operation),
+                    "operations": (
+                        *ledger.operations,
+                        *operations_to_reserve,
+                    ),
                 })
-                # Persist the complete ordered batch envelope before validating
-                # or executing any individual action.  A crash/replay can then
-                # neither append nor remove tail actions under the same call.
+                # Persist the ordered batch envelope and every action intent in
+                # one replacement.  Besides freezing completeness, this is the
+                # durable capacity reservation that keeps later commands from
+                # stealing slots needed by a crash replay.
                 await self._persist_ledger(root.trace_id, ledger)
             validation_by_key: dict[tuple[str, int], CandidateValidationRecord] = {}
             for raw in ledger.validations:

cyber_agent/core/runner.py

@@ -45,6 +45,7 @@ from cyber_agent.tools.approval import (
     approval_grant,
     tool_argument_hash,
 )
+from cyber_agent.tools.errors import RecoverableToolExecutionError
 from cyber_agent.tools.builtin.knowledge import KnowledgeConfig
 from cyber_agent.core.memory import (
     MEMORY_IDENTITY_CONTEXT_KEY,
@@ -3902,6 +3903,8 @@ class AgentRunner:
                                 allowed_tool_names=dispatch_allowlist,
                             )
                             return (tc, tool_args, tool_result)
+                        except RecoverableToolExecutionError:
+                            raise
                         except Exception as e:
                             import traceback
                             return (tc, tool_args, f"Error executing tool {tool_name}: {str(e)}\n{traceback.format_exc()}")
@@ -4809,9 +4812,10 @@ class AgentRunner:
 
         return cutoff
 
-    async def _replay_orphaned_review_call(
+    async def _replay_orphaned_lifecycle_call(
         self,
         *,
+        tool_name: str,
         trace: Trace,
         goal_tree: Optional[GoalTree],
         config: RunConfig,
@@ -4820,7 +4824,7 @@ class AgentRunner:
         sequence: int,
         head_sequence: int,
     ) -> Message:
-        """Replay one durable, idempotent review command before another LLM turn."""
+        """Replay one durable lifecycle command before another LLM turn."""
         function = tool_call.get("function") or {}
         raw_arguments = function.get("arguments", {})
         if isinstance(raw_arguments, str):
@@ -4828,26 +4832,26 @@ class AgentRunner:
                 arguments = json.loads(raw_arguments) if raw_arguments.strip() else {}
             except json.JSONDecodeError as exc:
                 raise RuntimeError(
-                    "Orphaned review command has invalid persisted arguments"
+                    "Orphaned lifecycle command has invalid persisted arguments"
                 ) from exc
         elif isinstance(raw_arguments, dict):
             arguments = dict(raw_arguments)
         else:
             raise RuntimeError(
-                "Orphaned review command has invalid persisted arguments"
+                "Orphaned lifecycle command has invalid persisted arguments"
             )
         tool_call_id = str(tool_call.get("id") or "")
         if not tool_call_id:
-            raise RuntimeError("Orphaned review command has no tool_call_id")
+            raise RuntimeError("Orphaned lifecycle command has no tool_call_id")
         fresh_trace = (
             await self.trace_store.get_trace(trace.trace_id)
             if self.trace_store
             else None
         )
         if fresh_trace is None:
-            raise RuntimeError("Orphaned review Trace no longer exists")
+            raise RuntimeError("Orphaned lifecycle Trace no longer exists")
         result = await self.tools.execute(
-            "review_task_result",
+            tool_name,
             arguments,
             uid=config.uid or "",
             context=self._build_tool_context(
@@ -4862,24 +4866,30 @@ class AgentRunner:
                 side_branch_ctx=None,
                 trigger_event=None,
             ),
-            allowed_tool_names={"review_task_result"},
+            allowed_tool_names={tool_name},
             tool_call_id=tool_call_id,
         )
-        result_text = (
-            result
-            if isinstance(result, str)
-            else json.dumps(result, ensure_ascii=False)
-        )
+        artifact_refs: list[dict[str, Any]] = []
+        if isinstance(result, dict):
+            artifact_refs = list(result.get("artifact_refs") or [])
+            raw_text = result.get("text")
+            result_text = (
+                raw_text
+                if isinstance(raw_text, str)
+                else json.dumps(result, ensure_ascii=False)
+            )
+        else:
+            result_text = str(result)
         try:
             payload = json.loads(result_text)
         except (json.JSONDecodeError, TypeError) as exc:
             raise RuntimeError(
-                "Orphaned review replay returned a non-JSON result"
+                "Orphaned lifecycle replay returned a non-JSON result"
             ) from exc
         if not isinstance(payload, dict) or payload.get("status") != "completed":
             reason = payload.get("error") if isinstance(payload, dict) else None
             raise RuntimeError(
-                "Orphaned review replay failed closed: "
+                "Orphaned lifecycle replay failed closed: "
                 + str(reason or "unexpected tool result")
             )
         return Message.create(
@@ -4890,8 +4900,9 @@ class AgentRunner:
             parent_sequence=head_sequence,
             tool_call_id=tool_call_id,
             content={
-                "tool_name": "review_task_result",
+                "tool_name": tool_name,
                 "result": result_text,
+                "artifact_refs": artifact_refs,
             },
         )
 
@@ -4909,8 +4920,8 @@ class AgentRunner:
         当 agent 被 stop/crash 中断时，可能有 assistant 的 tool_calls 没有对应的
         tool results（包括多 tool_call 部分完成的情况）。直接发给 LLM 会导致 400。
 
-        修复策略：幂等的 ``review_task_result`` 使用原 tool_call_id 在 LLM 前
-        受控重放；其他缺失调用插入合成的"中断通知"消息，而非裁剪。
+        修复策略：幂等的候选版本与父级审核命令使用原 tool_call_id
+        在 LLM 前受控重放；其他缺失调用插入合成的"中断通知"消息。
         - 普通工具：简短中断提示
         - agent/evaluate：包含 sub_trace_id、执行统计、continue_from 指引
 
@@ -4953,7 +4964,7 @@ class AgentRunner:
             assistant_msg, tc = tc_map[tc_id]
             tool_name = tc.get("function", {}).get("name", "unknown")
 
-            if tool_name == "review_task_result":
+            if tool_name in {"manage_candidate", "review_task_result"}:
                 assistant_calls = (
                     assistant_msg.content.get("tool_calls", [])
                     if isinstance(assistant_msg.content, dict)
@@ -4961,7 +4972,7 @@ class AgentRunner:
                 )
                 if len(assistant_calls) != 1:
                     raise RuntimeError(
-                        "Orphaned lifecycle-exclusive review was persisted in a mixed batch"
+                        "Orphaned lifecycle-exclusive command was persisted in a mixed batch"
                     )
                 needs_confirmation = (
                     not config.auto_execute_tools
@@ -4980,7 +4991,8 @@ class AgentRunner:
                         tool_calls=[tc],
                         config=config,
                     )
-                replayed = await self._replay_orphaned_review_call(
+                replayed = await self._replay_orphaned_lifecycle_call(
+                    tool_name=tool_name,
                     trace=trace,
                     goal_tree=goal_tree,
                     config=config,

cyber_agent/tools/errors.py

@@ -0,0 +1,6 @@
+"""Execution signals shared by the registry and the single Runner loop."""
+
+
+class RecoverableToolExecutionError(RuntimeError):
+    """A durable command must keep its assistant call orphaned for replay."""
+

cyber_agent/tools/registry.py

@@ -18,6 +18,7 @@ from copy import deepcopy
 from typing import Any, Callable, Dict, List, Optional
 
 from cyber_agent.tools.approval import tool_argument_hash
+from cyber_agent.tools.errors import RecoverableToolExecutionError
 from cyber_agent.tools.url_matcher import filter_by_url, match_url_with_patterns
 
 logger = logging.getLogger(__name__)
@@ -497,6 +498,13 @@ class ToolRegistry:
 
 			return json.dumps(result, ensure_ascii=False, indent=2)
 
+		except RecoverableToolExecutionError:
+			# The command already has a durable intent and may have committed an
+			# external idempotent side effect.  Returning a normal error result
+			# would close the original assistant call and lose its command ID.
+			stats.failure_count += 1
+			stats.total_duration += time.time() - start_time
+			raise
 		except Exception as e:
 			# 记录失败
 			stats.failure_count += 1

tests/test_candidate_service.py

@@ -27,6 +27,7 @@ from cyber_agent.application.candidate_service import (
     CandidateService,
     CandidateStateError,
 )
+from cyber_agent.core.runner import AgentRunner, RunConfig
 from cyber_agent.core.artifacts import (
     ArtifactRef,
     ValidationMaterial,
@@ -38,6 +39,8 @@ from cyber_agent.core.validation import (
     ValidationCheck,
     ValidationResult,
 )
+from cyber_agent.tools import get_tool_registry
+from cyber_agent.tools.errors import RecoverableToolExecutionError
 from cyber_agent.tools.registry import ToolRegistry
 from cyber_agent.trace.models import Message as TraceMessage, Trace
 from cyber_agent.trace.store import (
@@ -319,7 +322,10 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
             "replace_candidate_ledger",
             side_effect=fail_completed_publish,
         ):
-            with self.assertRaisesRegex(OSError, "finalize crash"):
+            with self.assertRaisesRegex(
+                RecoverableToolExecutionError,
+                "original command ID",
+            ):
                 await self.service.manage(
                     "root-a",
                     operation="create",
@@ -934,6 +940,111 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(MAX_CANDIDATE_OPERATIONS_PER_ROOT, len(completed.operations))
         self.assertEqual("discarded", completed.current_state(candidate))
 
+    async def test_review_batch_reserves_action_slot_across_crash_recovery(self):
+        await self._create_child("child-a")
+        candidate = await self.service.manage(
+            "child-a",
+            operation="create",
+            content={"text": "reserved review"},
+            parent_refs=[],
+            effective_at_sequence=4,
+        )
+        await self._record_validation("child-a", candidate)
+        ledger = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        target = MAX_CANDIDATE_OPERATIONS_PER_ROOT - 2
+        fillers = [
+            CandidateOperationRecord(
+                operation_id=f"review-reservation-filler-{index}",
+                operation="review_batch",
+                input_hash="2" * 64,
+                status="completed",
+            )
+            for index in range(target - len(ledger.operations))
+        ]
+        near_limit = CandidateLedger.model_validate({
+            **ledger.model_dump(mode="json"),
+            "operations": [
+                *[item.model_dump(mode="json") for item in ledger.operations],
+                *[item.model_dump(mode="json") for item in fillers],
+            ],
+        })
+        await self.store.replace_candidate_ledger(
+            "root-a",
+            near_limit.model_dump(mode="json"),
+        )
+        action = CandidateReviewAction(
+            action="adopt",
+            candidate_ref=candidate,
+            reason="reserve the complete review batch",
+        )
+        original_replace = self.store.replace_candidate_ledger
+        writes = 0
+
+        async def crash_after_batch_reservation(root_trace_id, raw_ledger):
+            nonlocal writes
+            writes += 1
+            if writes == 2:
+                raise OSError("crash after review reservation")
+            await original_replace(root_trace_id, raw_ledger)
+
+        with patch.object(
+            self.store,
+            "replace_candidate_ledger",
+            side_effect=crash_after_batch_reservation,
+        ):
+            with self.assertRaisesRegex(OSError, "review reservation"):
+                await self.service.apply_review_actions(
+                    "root-a",
+                    "child-a",
+                    report_refs=[candidate],
+                    actions=[action],
+                    effective_at_sequence=10,
+                    command_id="capacity-reserved-review",
+                )
+        reserved = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual(MAX_CANDIDATE_OPERATIONS_PER_ROOT, len(reserved.operations))
+        self.assertEqual(2, sum(
+            item.status == "pending" for item in reserved.operations
+        ))
+        repository_version_calls = self.repository.put_calls
+        with self.assertRaisesRegex(
+            CandidateStateError,
+            "operation limit exceeded before execution",
+        ):
+            await self.service.manage(
+                "child-a",
+                operation="create",
+                content={"text": "must not steal review capacity"},
+                parent_refs=[],
+                effective_at_sequence=11,
+                command_id="sibling-after-review-reservation",
+            )
+        self.assertEqual(repository_version_calls, self.repository.put_calls)
+
+        records = await self.service.apply_review_actions(
+            "root-a",
+            "child-a",
+            report_refs=[candidate],
+            actions=[action],
+            effective_at_sequence=999,
+            command_id="capacity-reserved-review",
+        )
+        self.assertEqual("adopted", records[0].state)
+        self.assertEqual(1, self.repository.adoption_calls)
+        completed = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual(MAX_CANDIDATE_OPERATIONS_PER_ROOT, len(completed.operations))
+        self.assertTrue(all(
+            item.status == "completed"
+            for item in completed.operations
+            if item.command_id == "capacity-reserved-review"
+        ))
+
     async def test_candidate_capacity_rejects_before_repository_or_intent_write(self):
         await self._create_child("child-a")
         candidate = await self.service.manage(
@@ -1026,7 +1137,10 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
             "replace_candidate_ledger",
             side_effect=crash_after_repository,
         ):
-            with self.assertRaisesRegex(OSError, "candidate publish"):
+            with self.assertRaisesRegex(
+                RecoverableToolExecutionError,
+                "original command ID",
+            ):
                 await self.service.manage(
                     "child-a",
                     operation="create",
@@ -1076,6 +1190,120 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(MAX_CANDIDATES_PER_ROOT, len(completed.candidates))
         self.assertEqual(recovered, completed.candidate(recovered))
 
+    async def test_runner_replays_pending_candidate_with_original_tool_call_id(self):
+        await self._create_child("child-a")
+        registry = get_tool_registry().clone(["manage_candidate"])
+        runner = AgentRunner(
+            trace_store=self.store,
+            tool_registry=registry,
+            llm_call=lambda **_kwargs: None,
+            application_binding=self.binding,
+            candidate_service=self.service,
+        )
+        config = RunConfig(
+            tools=["manage_candidate"],
+            tool_groups=[],
+            auto_execute_tools=True,
+        )
+        request = {
+            "request": {
+                "operation": "create",
+                "content": {"text": "recover with the original command"},
+                "parent_refs": [],
+            },
+        }
+        assistant = TraceMessage.create(
+            trace_id="child-a",
+            role="assistant",
+            sequence=4,
+            content={
+                "text": "",
+                "tool_calls": [{
+                    "id": "recoverable-candidate-call",
+                    "type": "function",
+                    "function": {
+                        "name": "manage_candidate",
+                        "arguments": json.dumps(request),
+                    },
+                }],
+            },
+        )
+        await self.store.add_message(assistant)
+        await self.store.update_trace(
+            "child-a",
+            head_sequence=4,
+            last_sequence=4,
+        )
+        trace = await self.store.get_trace("child-a")
+        original_replace = self.store.replace_candidate_ledger
+        writes = 0
+
+        async def fail_first_publish(root_trace_id, raw_ledger):
+            nonlocal writes
+            writes += 1
+            if writes == 2:
+                raise OSError("candidate publish interrupted")
+            await original_replace(root_trace_id, raw_ledger)
+
+        with patch.object(
+            self.store,
+            "replace_candidate_ledger",
+            side_effect=fail_first_publish,
+        ):
+            with self.assertRaises(RecoverableToolExecutionError):
+                await registry.execute(
+                    "manage_candidate",
+                    request,
+                    context=runner._build_tool_context(
+                        config=config,
+                        trace=trace,
+                        trace_id=trace.trace_id,
+                        goal_id=None,
+                        goal_tree=None,
+                        sequence=5,
+                        head_sequence=4,
+                        tool_call_id="recoverable-candidate-call",
+                        side_branch_ctx=None,
+                        trigger_event=None,
+                    ),
+                    allowed_tool_names={"manage_candidate"},
+                    tool_call_id="recoverable-candidate-call",
+                )
+        pending = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual("pending", pending.operations[-1].status)
+        self.assertEqual(1, len(self.repository.contents))
+        self.assertFalse(any(
+            item.role == "tool"
+            for item in await self.store.get_trace_messages("child-a")
+        ))
+
+        healed, next_sequence = await runner._heal_orphaned_tool_calls(
+            [assistant],
+            await self.store.get_trace("child-a"),
+            None,
+            config,
+            5,
+        )
+        self.assertEqual(6, next_sequence)
+        self.assertEqual("tool", healed[-1].role)
+        self.assertEqual(
+            "recoverable-candidate-call",
+            healed[-1].tool_call_id,
+        )
+        self.assertEqual(1, len(self.repository.contents))
+        completed = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        matching = [
+            item for item in completed.operations
+            if item.command_id == "recoverable-candidate-call"
+        ]
+        self.assertEqual(1, len(matching))
+        self.assertEqual("completed", matching[0].status)
+        self.assertEqual(1, len(completed.candidates))
+
     async def test_descendant_adoption_blocks_ancestor_rewind_without_mapping(self):
         await self._create_child("child-a")
         await self.store.create_trace(Trace(