fix(candidate): 使用稳定命令身份恢复仓储副作用

候选 create/fork/merge 的 operation ID 攉由 Trace 与规范化命令内容生成，不再依赖本次 sequence；Repository 成功而账本最终发布失败后，即使模型在新 sequence 重试，也会复用同一候选指针和幂等 key，不再产生孤儿版本。

父级 adopt/discard/revise 同样改用 parent、child 与规范化 action 的稳定身份；adoption 已外部提交但 TaskReview 尚未落盘时，重试会读取同一 completed operation 并继续协议收敛，外部采用只发生一次。

同时限制并严格校验候选 validation/checkpoint 集合，补充 finalize 崩溃、新 sequence 重试、采用恢复和完整账本校验测试。

cyber_agent/application/candidate.py

@@ -13,6 +13,7 @@ from cyber_agent.core.task_contract import TaskContractRef
 
 MAX_CANDIDATES_PER_ROOT = 256
 MAX_CANDIDATE_OPERATIONS_PER_ROOT = 2_048
+MAX_CANDIDATE_VALIDATIONS_PER_ROOT = 2_048
 
 
 class CandidatePointer(ApplicationModel):
@@ -139,6 +140,10 @@ class CandidateLedger(ApplicationModel):
             raise ValueError("Candidate ledger candidate limit exceeded")
         if len(self.operations) > MAX_CANDIDATE_OPERATIONS_PER_ROOT:
             raise ValueError("Candidate ledger operation limit exceeded")
+        if len(self.validations) > MAX_CANDIDATE_VALIDATIONS_PER_ROOT:
+            raise ValueError("Candidate ledger validation limit exceeded")
+        if len(self.validation_checkpoints) > MAX_CANDIDATE_VALIDATIONS_PER_ROOT:
+            raise ValueError("Candidate ledger validation checkpoint limit exceeded")
         identities = [
             (item.candidate_id, item.revision)
             for item in self.candidates
@@ -148,11 +153,41 @@ class CandidateLedger(ApplicationModel):
         operation_ids = [item.operation_id for item in self.operations]
         if len(operation_ids) != len(set(operation_ids)):
             raise ValueError("Candidate ledger contains duplicate operation IDs")
-        from cyber_agent.application.quality import CandidateValidationCheckpoint
+        from cyber_agent.application.quality import (
+            CandidateValidationCheckpoint,
+            CandidateValidationRecord,
+        )
+
+        candidate_by_key = {
+            (item.candidate_id, item.revision): item
+            for item in self.candidates
+        }
+        validation_keys = []
+        for raw in self.validations:
+            validation = CandidateValidationRecord.model_validate(raw)
+            key = (
+                validation.candidate_ref.candidate_id,
+                validation.candidate_ref.revision,
+            )
+            if candidate_by_key.get(key) != validation.candidate_ref:
+                raise ValueError(
+                    "Candidate ledger validation references an altered candidate"
+                )
+            validation_keys.append((key, validation.plan_hash))
+        if len(validation_keys) != len(set(validation_keys)):
+            raise ValueError("Candidate ledger contains duplicate validations")
 
         checkpoint_keys = []
         for raw in self.validation_checkpoints:
             checkpoint = CandidateValidationCheckpoint.model_validate(raw)
+            candidate_key = (
+                checkpoint.candidate_ref.candidate_id,
+                checkpoint.candidate_ref.revision,
+            )
+            if candidate_by_key.get(candidate_key) != checkpoint.candidate_ref:
+                raise ValueError(
+                    "Candidate ledger checkpoint references an altered candidate"
+                )
             checkpoint_keys.append((
                 checkpoint.candidate_ref.candidate_id,
                 checkpoint.candidate_ref.revision,

cyber_agent/application/candidate_service.py

@@ -72,7 +72,6 @@ class CandidateService:
     ) -> CandidateRef:
         trace, root = await self._require_owner_trace(trace_id)
         root_trace_id = root.trace_id
-        operation_id = f"candidate:{trace_id}:{effective_at_sequence}"
         input_payload = {
             "operation": operation,
             "content": content,
@@ -81,6 +80,15 @@ class CandidateService:
         input_hash = sha256(
             canonical_json(input_payload).encode("utf-8")
         ).hexdigest()
+        # The command identity must survive a process crash and a subsequent
+        # model retry. Sequence numbers identify attempts, not durable intent.
+        command_hash = sha256(
+            canonical_json({
+                "trace_id": trace_id,
+                "input_hash": input_hash,
+            }).encode("utf-8")
+        ).hexdigest()
+        operation_id = f"candidate:{command_hash}"
         async with self._lock_for(root_trace_id):
             ledger = await self._load_ledger(root_trace_id)
             existing_operation = next(
@@ -200,7 +208,7 @@ class CandidateService:
                 root_trace_id,
                 completed.model_dump(mode="json"),
             )
-            await self._emit_candidate_registered(candidate_ref)
+            await self._emit_candidate_registered(candidate_ref, operation_id)
             return candidate_ref
 
     async def validate_report_refs(
@@ -496,9 +504,10 @@ class CandidateService:
                         "Candidate revise requires a non-passed validation"
                     )
                 state = ledger.current_state(action.candidate_ref)
-                expected_operation_id = (
-                    f"candidate-review:{parent_trace_id}:{effective_at_sequence}:"
-                    f"{key[0]}:{key[1]}:{action.action}"
+                expected_operation_id = self._review_operation_id(
+                    parent_trace_id,
+                    child_trace_id,
+                    action,
                 )
                 same_completed_operation = any(
                     item.operation_id == expected_operation_id
@@ -519,6 +528,7 @@ class CandidateService:
                     ledger,
                     action,
                     parent_trace_id=parent_trace_id,
+                    child_trace_id=child_trace_id,
                     root_trace_id=root.trace_id,
                     effective_at_sequence=effective_at_sequence,
                 )
@@ -531,6 +541,7 @@ class CandidateService:
         action: CandidateReviewAction,
         *,
         parent_trace_id: str,
+        child_trace_id: str,
         root_trace_id: str,
         effective_at_sequence: int,
     ) -> tuple[CandidateLedger, CandidateLifecycleRecord]:
@@ -538,9 +549,10 @@ class CandidateService:
             candidate_id=action.candidate_ref.candidate_id,
             revision=action.candidate_ref.revision,
         )
-        operation_id = (
-            f"candidate-review:{parent_trace_id}:{effective_at_sequence}:"
-            f"{pointer.candidate_id}:{pointer.revision}:{action.action}"
+        operation_id = self._review_operation_id(
+            parent_trace_id,
+            child_trace_id,
+            action,
         )
         input_hash = sha256(
             canonical_json(action).encode("utf-8")
@@ -657,7 +669,29 @@ class CandidateService:
         await self._emit_lifecycle(adopted, root_trace_id)
         return ledger, adopted
 
-    async def _emit_candidate_registered(self, candidate: CandidateRef) -> None:
+    @staticmethod
+    def _review_operation_id(
+        parent_trace_id: str,
+        child_trace_id: str,
+        action: CandidateReviewAction,
+    ) -> str:
+        action_hash = sha256(
+            canonical_json(action).encode("utf-8")
+        ).hexdigest()
+        command_hash = sha256(
+            canonical_json({
+                "parent_trace_id": parent_trace_id,
+                "child_trace_id": child_trace_id,
+                "action_hash": action_hash,
+            }).encode("utf-8")
+        ).hexdigest()
+        return f"candidate-review:{command_hash}"
+
+    async def _emit_candidate_registered(
+        self,
+        candidate: CandidateRef,
+        operation_id: str,
+    ) -> None:
         if self.event_service is None:
             return
         await self.event_service.emit_after_commit(
@@ -674,15 +708,11 @@ class CandidateService:
             source_trace_id=candidate.owner_trace_id,
             event_type="candidate.lifecycle_changed",
             event_key=(
-                f"candidate.lifecycle_changed:candidate:"
-                f"{candidate.owner_trace_id}:{candidate.created_at_sequence}:proposed"
+                f"candidate.lifecycle_changed:{operation_id}:proposed"
             ),
             effective_at_sequence=candidate.created_at_sequence,
             payload={"lifecycle": CandidateLifecycleRecord(
-                operation_id=(
-                    f"candidate:{candidate.owner_trace_id}:"
-                    f"{candidate.created_at_sequence}"
-                ),
+                operation_id=operation_id,
                 candidate=CandidatePointer(
                     candidate_id=candidate.candidate_id,
                     revision=candidate.revision,

tests/test_candidate_service.py

@@ -300,6 +300,55 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
             for item in ledger.candidates
         ))
 
+    async def test_finalize_crash_retries_same_semantic_command_identity(self):
+        original_replace = self.store.replace_candidate_ledger
+        writes = 0
+
+        async def fail_completed_publish(root_trace_id, ledger):
+            nonlocal writes
+            writes += 1
+            if writes == 2:
+                raise OSError("simulated ledger finalize crash")
+            await original_replace(root_trace_id, ledger)
+
+        with patch.object(
+            self.store,
+            "replace_candidate_ledger",
+            side_effect=fail_completed_publish,
+        ):
+            with self.assertRaisesRegex(OSError, "finalize crash"):
+                await self.service.manage(
+                    "root-a",
+                    operation="create",
+                    content={"text": "survives retry"},
+                    parent_refs=[],
+                    effective_at_sequence=2,
+                )
+
+        pending = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual("pending", pending.operations[0].status)
+        operation_id = pending.operations[0].operation_id
+        recovered = await self.service.manage(
+            "root-a",
+            operation="create",
+            content={"text": "survives retry"},
+            parent_refs=[],
+            effective_at_sequence=99,
+        )
+        ledger = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual(operation_id, ledger.operations[0].operation_id)
+        self.assertEqual("completed", ledger.operations[0].status)
+        self.assertEqual(1, len(ledger.candidates))
+        self.assertEqual(1, len(self.repository.contents))
+        self.assertEqual(
+            recovered,
+            ledger.candidate(ledger.operations[0].candidate),
+        )
+
     async def test_cross_root_and_cross_task_parent_access_is_rejected(self):
         first = await self.service.manage(
             "root-a",
@@ -478,6 +527,39 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
                 effective_at_sequence=10,
             )
 
+    async def test_adoption_retry_uses_same_command_after_parent_sequence_changes(self):
+        await self._create_child("child-a")
+        candidate = await self.service.manage(
+            "child-a",
+            operation="create",
+            content={"text": "publish once"},
+            parent_refs=[],
+            effective_at_sequence=4,
+        )
+        await self._record_validation("child-a", candidate)
+        action = CandidateReviewAction(
+            action="adopt",
+            candidate_ref=candidate,
+            reason="stable parent decision",
+        )
+        first = await self.service.apply_review_actions(
+            "root-a",
+            "child-a",
+            report_refs=[candidate],
+            actions=[action],
+            effective_at_sequence=10,
+        )
+        recovered = await self.service.apply_review_actions(
+            "root-a",
+            "child-a",
+            report_refs=[candidate],
+            actions=[action],
+            effective_at_sequence=999,
+        )
+        self.assertEqual(first, recovered)
+        self.assertEqual(1, self.repository.adoption_calls)
+        self.assertEqual(1, len(self.repository.adoptions))
+
     async def test_descendant_adoption_blocks_ancestor_rewind_without_mapping(self):
         await self._create_child("child-a")
         await self.store.create_trace(Trace(