fix(candidate): 固化父级审核动作批次完整性

在执行任何候选分项或 Repository 副作用前，先以 review tool_call_id 持久化有序 candidate_actions 批次信封和完整载荷哈希。

重放时严格比较整个批次，删除尾项、新增尾项、换序或改单项内容都会在副作用前失败；分项仍用批次位置维持可恢复的独立幂等操作。

新增双候选删除、追加、换序和 adoption finalize 崩溃恢复测试。

cyber_agent/application/candidate.py

@@ -122,7 +122,8 @@ class CandidateOperationRecord(ApplicationModel):
     # two distinct calls with identical arguments remain distinct commands.
     command_id: str | None = Field(default=None, min_length=1, max_length=300)
     operation: Literal[
-        "create", "fork", "merge", "adopt", "discard", "revise"
+        "create", "fork", "merge", "adopt", "discard", "revise",
+        "review_batch",
     ]
     input_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
     status: Literal["pending", "completed", "failed"] = "pending"

cyber_agent/application/candidate_service.py

@@ -511,6 +511,44 @@ class CandidateService:
             raise CandidateStateError("Candidate review actions must be unique")
         async with self._lock_for(root.trace_id):
             ledger = await self._load_ledger(root.trace_id)
+            batch_operation_id = self._review_batch_operation_id(
+                parent_trace_id,
+                child_trace_id,
+                command_id=command_id,
+                effective_at_sequence=effective_at_sequence,
+            )
+            batch_input_hash = sha256(canonical_json([
+                item.model_dump(mode="json") for item in actions
+            ]).encode("utf-8")).hexdigest()
+            batch_operation = next(
+                (
+                    item for item in ledger.operations
+                    if item.operation_id == batch_operation_id
+                ),
+                None,
+            )
+            if batch_operation is not None:
+                if batch_operation.input_hash != batch_input_hash:
+                    raise CandidateStateError(
+                        "Candidate review batch payload changed during recovery"
+                    )
+            else:
+                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),
+                })
+                # 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.
+                await self.store.replace_candidate_ledger(
+                    root.trace_id,
+                    ledger.model_dump(mode="json"),
+                )
             validation_by_key: dict[tuple[str, int], CandidateValidationRecord] = {}
             for raw in ledger.validations:
                 record = CandidateValidationRecord.model_validate(raw)
@@ -583,6 +621,16 @@ class CandidateService:
                     action_index=action_index,
                 )
                 applied.append(record)
+            operations = tuple(
+                item.model_copy(update={"status": "completed", "error": None})
+                if item.operation_id == batch_operation_id else item
+                for item in ledger.operations
+            )
+            ledger = ledger.model_copy(update={"operations": operations})
+            await self.store.replace_candidate_ledger(
+                root.trace_id,
+                ledger.model_dump(mode="json"),
+            )
             return applied
 
     async def _apply_review_action(
@@ -746,6 +794,21 @@ class CandidateService:
         ).hexdigest()
         return f"candidate-review:{command_hash}"
 
+    @staticmethod
+    def _review_batch_operation_id(
+        parent_trace_id: str,
+        child_trace_id: str,
+        *,
+        command_id: str | None,
+        effective_at_sequence: int,
+    ) -> str:
+        command_hash = sha256(canonical_json({
+            "parent_trace_id": parent_trace_id,
+            "child_trace_id": child_trace_id,
+            "command_id": command_id or f"sequence:{effective_at_sequence}",
+        }).encode("utf-8")).hexdigest()
+        return f"candidate-review-batch:{command_hash}"
+
     async def _emit_candidate_registered(
         self,
         candidate: CandidateRef,

tests/test_candidate_service.py

@@ -652,6 +652,14 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
             effective_at_sequence=4,
         )
         await self._record_validation("child-a", candidate)
+        peer = await self.service.manage(
+            "child-a",
+            operation="create",
+            content={"text": "peer"},
+            parent_refs=[],
+            effective_at_sequence=5,
+        )
+        await self._record_validation("child-a", peer)
         action = CandidateReviewAction(
             action="adopt",
             candidate_ref=candidate,
@@ -660,7 +668,7 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
         first = await self.service.apply_review_actions(
             "root-a",
             "child-a",
-            report_refs=[candidate],
+            report_refs=[candidate, peer],
             actions=[action],
             effective_at_sequence=10,
             command_id="adoption-review-call",
@@ -668,7 +676,7 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
         recovered = await self.service.apply_review_actions(
             "root-a",
             "child-a",
-            report_refs=[candidate],
+            report_refs=[candidate, peer],
             actions=[action],
             effective_at_sequence=999,
             command_id="adoption-review-call",
@@ -678,12 +686,12 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(1, len(self.repository.adoptions))
         with self.assertRaisesRegex(
             CandidateStateError,
-            "review payload changed",
+            "review batch payload changed",
         ):
             await self.service.apply_review_actions(
                 "root-a",
                 "child-a",
-                report_refs=[candidate],
+                report_refs=[candidate, peer],
                 actions=[action.model_copy(update={
                     "reason": "tampered decision payload",
                 })],
@@ -691,6 +699,74 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
                 command_id="adoption-review-call",
             )
         self.assertEqual(1, self.repository.adoption_calls)
+        with self.assertRaisesRegex(
+            CandidateStateError,
+            "review batch payload changed",
+        ):
+            await self.service.apply_review_actions(
+                "root-a",
+                "child-a",
+                report_refs=[candidate, peer],
+                actions=[
+                    action,
+                    CandidateReviewAction(
+                        action="discard",
+                        candidate_ref=peer,
+                        reason="injected tail action",
+                    ),
+                ],
+                effective_at_sequence=1_001,
+                command_id="adoption-review-call",
+            )
+        self.assertEqual("proposed", (
+            CandidateLedger.model_validate(
+                await self.store.get_candidate_ledger("root-a")
+            ).current_state(peer)
+        ))
+
+    async def test_review_batch_replay_rejects_deleted_and_reordered_actions(self):
+        await self._create_child("child-a")
+        candidates = []
+        for sequence, text in ((4, "A"), (5, "B")):
+            candidate = await self.service.manage(
+                "child-a",
+                operation="create",
+                content={"text": text},
+                parent_refs=[],
+                effective_at_sequence=sequence,
+            )
+            await self._record_validation("child-a", candidate)
+            candidates.append(candidate)
+        actions = [
+            CandidateReviewAction(
+                action="discard",
+                candidate_ref=candidate,
+                reason=f"discard {candidate.candidate_id}",
+            )
+            for candidate in candidates
+        ]
+        await self.service.apply_review_actions(
+            "root-a",
+            "child-a",
+            report_refs=candidates,
+            actions=actions,
+            effective_at_sequence=10,
+            command_id="two-action-review",
+        )
+
+        for altered in (actions[:1], list(reversed(actions))):
+            with self.assertRaisesRegex(
+                CandidateStateError,
+                "review batch payload changed",
+            ):
+                await self.service.apply_review_actions(
+                    "root-a",
+                    "child-a",
+                    report_refs=candidates,
+                    actions=altered,
+                    effective_at_sequence=999,
+                    command_id="two-action-review",
+                )
 
     async def test_descendant_adoption_blocks_ancestor_rewind_without_mapping(self):
         await self._create_child("child-a")
@@ -790,7 +866,7 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
         async def fail_finalize(root_trace_id, ledger):
             nonlocal calls
             calls += 1
-            if calls == 2:
+            if calls == 3:
                 raise OSError("crash before framework finalize")
             await original_replace(root_trace_id, ledger)