fix(candidate): 对候选完整血缘图执行环检测

候选账本加载时构建跨 candidate_id 与 revision 的完整父引用图，并以深度优先遍历拒绝任意环，而不再只检查同一候选的未来 revision。

增加损坏账本 A→B→A 的失败关闭回归，证明循环血缘在调用 CandidateRepository 前即被拒绝，不产生外部写入副作用。

cyber_agent/application/candidate.py

@@ -148,7 +148,12 @@ class CandidateLedger(ApplicationModel):
         if len(operation_ids) != len(set(operation_ids)):
             raise ValueError("Candidate ledger contains duplicate operation IDs")
         known = set(identities)
+        parents_by_candidate: dict[
+            tuple[str, int], tuple[tuple[str, int], ...]
+        ] = {}
         for candidate in self.candidates:
+            candidate_key = (candidate.candidate_id, candidate.revision)
+            parent_keys: list[tuple[str, int]] = []
             for parent in candidate.parent_refs:
                 parent_key = (parent.candidate_id, parent.revision)
                 if parent_key not in known:
@@ -158,6 +163,25 @@ class CandidateLedger(ApplicationModel):
                     and parent.revision >= candidate.revision
                 ):
                     raise ValueError("Candidate lineage references a future revision")
+                parent_keys.append(parent_key)
+            parents_by_candidate[candidate_key] = tuple(parent_keys)
+
+        visited: set[tuple[str, int]] = set()
+        visiting: set[tuple[str, int]] = set()
+
+        def visit(candidate_key: tuple[str, int]) -> None:
+            if candidate_key in visiting:
+                raise ValueError("Candidate ledger contains a lineage cycle")
+            if candidate_key in visited:
+                return
+            visiting.add(candidate_key)
+            for parent_key in parents_by_candidate[candidate_key]:
+                visit(parent_key)
+            visiting.remove(candidate_key)
+            visited.add(candidate_key)
+
+        for candidate_key in parents_by_candidate:
+            visit(candidate_key)
         return self
 
     def candidate(self, pointer: CandidatePointer) -> CandidateRef:

tests/test_candidate_service.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 import tempfile
 import unittest
 from unittest.mock import patch
@@ -394,6 +395,41 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
         with self.assertRaisesRegex(TraceStoreCorruptionError, "candidate ledger"):
             await self.store.get_candidate_ledger("root-a")
 
+    async def test_cyclic_lineage_is_rejected_before_repository_side_effect(self):
+        first = await self.service.manage(
+            "root-a",
+            operation="create",
+            content={"text": "A"},
+            parent_refs=[],
+            effective_at_sequence=1,
+        )
+        raw = await self.store.get_candidate_ledger("root-a")
+        second = first.model_copy(update={
+            "candidate_id": "candidate-b",
+            "parent_refs": (CandidatePointer(
+                candidate_id=first.candidate_id,
+                revision=first.revision,
+            ),),
+        })
+        raw["candidates"][0]["parent_refs"] = [{
+            "candidate_id": second.candidate_id,
+            "revision": second.revision,
+        }]
+        raw["candidates"].append(second.model_dump(mode="json"))
+        ledger_path = self.store._get_candidate_ledger_file("root-a")
+        ledger_path.write_text(json.dumps(raw), encoding="utf-8")
+
+        put_calls = self.repository.put_calls
+        with self.assertRaisesRegex(ValidationError, "lineage cycle"):
+            await self.service.manage(
+                "root-a",
+                operation="create",
+                content={"text": "must not be persisted"},
+                parent_refs=[],
+                effective_at_sequence=2,
+            )
+        self.assertEqual(put_calls, self.repository.put_calls)
+
     async def test_review_adoption_is_idempotent_and_creates_rewind_barrier(self):
         await self._create_child("child-a")
         candidate = await self.service.manage(