fix(candidate): 用审核批次位置隔离命令身份与载荷完整性

父级候选动作的 operation_id 只绑定持久化 review tool_call_id 和动作在批次中的固定位置，不再包含可变 action 内容。

完整 action 继续只进入 input_hash；同一审核调用被改 reason、目标或动作时稳定命中原操作并失败关闭，多候选批次仍拥有互不冲突的操作身份。

新增采用成功后同 call_id 篡改重放的回归测试，确认仓储写回仍严格为一次。

cyber_agent/application/candidate_service.py

@@ -519,7 +519,7 @@ class CandidateService:
                     record.candidate_ref.revision,
                 )
                 validation_by_key[key] = record
-            for action, key in zip(actions, action_keys):
+            for action_index, (action, key) in enumerate(zip(actions, action_keys)):
                 if key not in report_by_key or report_by_key[key] != action.candidate_ref:
                     raise CandidateStateError(
                         "Candidate action must reference the exact child TaskReport"
@@ -555,6 +555,7 @@ class CandidateService:
                     action,
                     command_id=command_id,
                     effective_at_sequence=effective_at_sequence,
+                    action_index=action_index,
                 )
                 same_completed_operation = any(
                     item.operation_id == expected_operation_id
@@ -570,7 +571,7 @@ class CandidateService:
                     )
 
             applied: list[CandidateLifecycleRecord] = []
-            for action in actions:
+            for action_index, action in enumerate(actions):
                 ledger, record = await self._apply_review_action(
                     ledger,
                     action,
@@ -579,6 +580,7 @@ class CandidateService:
                     root_trace_id=root.trace_id,
                     effective_at_sequence=effective_at_sequence,
                     command_id=command_id,
+                    action_index=action_index,
                 )
                 applied.append(record)
             return applied
@@ -593,6 +595,7 @@ class CandidateService:
         root_trace_id: str,
         effective_at_sequence: int,
         command_id: str | None,
+        action_index: int,
     ) -> tuple[CandidateLedger, CandidateLifecycleRecord]:
         pointer = CandidatePointer(
             candidate_id=action.candidate_ref.candidate_id,
@@ -604,6 +607,7 @@ class CandidateService:
             action,
             command_id=command_id,
             effective_at_sequence=effective_at_sequence,
+            action_index=action_index,
         )
         input_hash = sha256(
             canonical_json(action).encode("utf-8")
@@ -729,16 +733,15 @@ class CandidateService:
         *,
         command_id: str | None = None,
         effective_at_sequence: int = 0,
+        action_index: int = 0,
     ) -> str:
-        action_hash = sha256(
-            canonical_json(action).encode("utf-8")
-        ).hexdigest()
+        del action
         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}",
-                "action_hash": action_hash,
+                "action_index": action_index,
             }).encode("utf-8")
         ).hexdigest()
         return f"candidate-review:{command_hash}"

tests/test_candidate_service.py

@@ -676,6 +676,21 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(first, recovered)
         self.assertEqual(1, self.repository.adoption_calls)
         self.assertEqual(1, len(self.repository.adoptions))
+        with self.assertRaisesRegex(
+            CandidateStateError,
+            "review payload changed",
+        ):
+            await self.service.apply_review_actions(
+                "root-a",
+                "child-a",
+                report_refs=[candidate],
+                actions=[action.model_copy(update={
+                    "reason": "tampered decision payload",
+                })],
+                effective_at_sequence=1_000,
+                command_id="adoption-review-call",
+            )
+        self.assertEqual(1, self.repository.adoption_calls)
 
     async def test_descendant_adoption_blocks_ancestor_rewind_without_mapping(self):
         await self._create_child("child-a")