fix(candidate): 在仓储副作用前预留并校验候选账本容量

在候选 create、fork、merge 调用业务仓储前预检候选和操作容量，避免达到上限后留下外部孤儿内容。

将 pending 版本命令计入容量预留，保证崩溃恢复命令的槽位不会被后续命令占用；所有账本替换前重新执行完整模型校验。

补充候选上限、审核操作上限、精确边界和 pending 命令恢复测试，验证拒绝路径不写账本也不调用 Repository。

cyber_agent/application/candidate_service.py

@@ -18,6 +18,8 @@ from cyber_agent.application.candidate import (
     CandidateRepository,
     CandidateReviewAction,
     CandidateVersionRequest,
+    MAX_CANDIDATES_PER_ROOT,
+    MAX_CANDIDATE_OPERATIONS_PER_ROOT,
 )
 from cyber_agent.application.models import canonical_json
 from cyber_agent.application.quality import (
@@ -121,7 +123,20 @@ class CandidateService:
                     raise CandidateStateError(
                         existing_operation.error or "Candidate operation failed"
                     )
+                if self._reserved_candidate_slots(ledger) > MAX_CANDIDATES_PER_ROOT:
+                    raise CandidateStateError(
+                        "Candidate ledger candidate reservations are corrupt"
+                    )
             else:
+                # Every supported version command publishes exactly one new
+                # immutable CandidateRef.  Reject a full aggregate before we
+                # persist an intent or call the application Repository; doing
+                # this only in CandidateLedger validation would leave an
+                # external orphan after the Repository has committed.
+                if self._reserved_candidate_slots(ledger) >= MAX_CANDIDATES_PER_ROOT:
+                    raise CandidateStateError(
+                        "Candidate ledger candidate limit exceeded before execution"
+                    )
                 try:
                     pointer = self._allocate_pointer(
                         operation,
@@ -142,10 +157,7 @@ class CandidateService:
                 ledger = ledger.model_copy(update={
                     "operations": (*ledger.operations, operation_record),
                 })
-                await self.store.replace_candidate_ledger(
-                    root_trace_id,
-                    ledger.model_dump(mode="json"),
-                )
+                await self._persist_ledger(root_trace_id, ledger)
                 existing_operation = operation_record
 
             assert existing_operation.candidate is not None
@@ -201,10 +213,7 @@ class CandidateService:
                 )
             except Exception as exc:
                 failed = self._fail_operation(ledger, operation_id, str(exc))
-                await self.store.replace_candidate_ledger(
-                    root_trace_id,
-                    failed.model_dump(mode="json"),
-                )
+                await self._persist_ledger(root_trace_id, failed)
                 raise
             completed = self._complete_operation(
                 ledger,
@@ -214,10 +223,7 @@ 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.store.replace_candidate_ledger(
-                root_trace_id,
-                completed.model_dump(mode="json"),
-            )
+            await self._persist_ledger(root_trace_id, completed)
             await self._emit_candidate_registered(candidate_ref, operation_id)
             return candidate_ref
 
@@ -343,10 +349,7 @@ class CandidateService:
                     checkpoint.model_dump(mode="json"),
                 ),
             })
-            await self.store.replace_candidate_ledger(
-                root.trace_id,
-                updated.model_dump(mode="json"),
-            )
+            await self._persist_ledger(root.trace_id, updated)
             return checkpoint
 
     async def update_validation_checkpoint(
@@ -398,10 +401,7 @@ class CandidateService:
             updated = ledger.model_copy(update={
                 "validation_checkpoints": tuple(checkpoints),
             })
-            await self.store.replace_candidate_ledger(
-                root.trace_id,
-                updated.model_dump(mode="json"),
-            )
+            await self._persist_ledger(root.trace_id, updated)
             return result
 
     async def record_validation(
@@ -450,10 +450,7 @@ class CandidateService:
             updated = updated.model_copy(update={
                 "validation_checkpoints": tuple(checkpoints),
             })
-            await self.store.replace_candidate_ledger(
-                root.trace_id,
-                updated.model_dump(mode="json"),
-            )
+            await self._persist_ledger(root.trace_id, updated)
             if self.event_service is not None:
                 await self.event_service.emit_after_commit(
                     source_trace_id=record.candidate_ref.owner_trace_id,
@@ -531,6 +528,28 @@ class CandidateService:
                         "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)
+                )
+                required_slots = 1 + new_action_operations
+                if (
+                    len(ledger.operations) + required_slots
+                    > 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,
@@ -543,10 +562,7 @@ class CandidateService:
                 # 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"),
-                )
+                await self._persist_ledger(root.trace_id, ledger)
             validation_by_key: dict[tuple[str, int], CandidateValidationRecord] = {}
             for raw in ledger.validations:
                 record = CandidateValidationRecord.model_validate(raw)
@@ -625,10 +641,7 @@ class CandidateService:
                 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"),
-            )
+            await self._persist_ledger(root.trace_id, ledger)
             return applied
 
     async def _apply_review_action(
@@ -698,10 +711,7 @@ class CandidateService:
                 ),
             )
             ledger = self._finish_review_operation(ledger, operation_id, record)
-            await self.store.replace_candidate_ledger(
-                root_trace_id,
-                ledger.model_dump(mode="json"),
-            )
+            await self._persist_ledger(root_trace_id, ledger)
             await self._emit_lifecycle(record, root_trace_id)
             return ledger, record
 
@@ -717,10 +727,7 @@ class CandidateService:
             ledger = ledger.model_copy(update={
                 "lifecycle": (*ledger.lifecycle, pending),
             })
-            await self.store.replace_candidate_ledger(
-                root_trace_id,
-                ledger.model_dump(mode="json"),
-            )
+            await self._persist_ledger(root_trace_id, ledger)
             await self._emit_lifecycle(pending, root_trace_id)
         try:
             receipt = CandidateAdoptionReceipt.model_validate(
@@ -746,10 +753,7 @@ class CandidateService:
                 failed,
                 error=str(exc),
             )
-            await self.store.replace_candidate_ledger(
-                root_trace_id,
-                ledger.model_dump(mode="json"),
-            )
+            await self._persist_ledger(root_trace_id, ledger)
             await self._emit_lifecycle(failed, root_trace_id)
             raise
         adopted = CandidateLifecycleRecord(
@@ -764,10 +768,7 @@ class CandidateService:
         ledger = self._finish_review_operation(ledger, operation_id, adopted)
         # A failure here intentionally leaves the durable pending intent. The
         # Repository sees the same operation_id when the review is replayed.
-        await self.store.replace_candidate_ledger(
-            root_trace_id,
-            ledger.model_dump(mode="json"),
-        )
+        await self._persist_ledger(root_trace_id, ledger)
         await self._emit_lifecycle(adopted, root_trace_id)
         return ledger, adopted
 
@@ -1062,6 +1063,20 @@ class CandidateService:
         raw = await self.store.get_candidate_ledger(root_trace_id)
         return CandidateLedger.model_validate(raw or {})
 
+    async def _persist_ledger(
+        self,
+        root_trace_id: str,
+        ledger: CandidateLedger,
+    ) -> None:
+        """Validate the complete immutable aggregate before atomic replacement."""
+        validated = CandidateLedger.model_validate(
+            ledger.model_dump(mode="json")
+        )
+        await self.store.replace_candidate_ledger(
+            root_trace_id,
+            validated.model_dump(mode="json"),
+        )
+
     def _allocate_pointer(
         self,
         operation: str,
@@ -1095,6 +1110,21 @@ class CandidateService:
             return CandidatePointer(candidate_id=f"cand_{digest[:24]}", revision=1)
         raise CandidateStateError(f"Unsupported candidate operation: {operation}")
 
+    @staticmethod
+    def _reserved_candidate_slots(ledger: CandidateLedger) -> int:
+        """Count published revisions plus recoverable version intents.
+
+        A pending create/fork/merge owns its eventual aggregate slot.  Counting
+        the reservation prevents a later command from consuming the final slot
+        and making the earlier Repository side effect impossible to finalize.
+        """
+        pending_versions = sum(
+            item.status == "pending"
+            and item.operation in {"create", "fork", "merge"}
+            for item in ledger.operations
+        )
+        return len(ledger.candidates) + pending_versions
+
     def _validate_parents(self, trace, root, parents: list[CandidateRef]) -> None:
         for parent in parents:
             self._validate_candidate_owner(parent, trace, root)

tests/test_candidate_service.py

@@ -16,8 +16,11 @@ from cyber_agent.application import (
 from cyber_agent.application.candidate import (
     CandidateAdoptionReceipt,
     CandidateLedger,
+    CandidateOperationRecord,
     CandidatePointer,
     CandidateReviewAction,
+    MAX_CANDIDATES_PER_ROOT,
+    MAX_CANDIDATE_OPERATIONS_PER_ROOT,
 )
 from cyber_agent.application.quality import CandidateValidationRecord
 from cyber_agent.application.candidate_service import (
@@ -822,6 +825,257 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
             ).current_state(candidate),
         )
 
+    async def test_review_capacity_rejects_before_envelope_or_repository_write(self):
+        await self._create_child("child-a")
+        candidate = await self.service.manage(
+            "child-a",
+            operation="create",
+            content={"text": "capacity"},
+            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")
+        )
+        fillers = [
+            CandidateOperationRecord(
+                operation_id=f"capacity-filler-{index}",
+                operation="review_batch",
+                input_hash="0" * 64,
+                status="completed",
+            )
+            for index in range(
+                MAX_CANDIDATE_OPERATIONS_PER_ROOT - len(ledger.operations)
+            )
+        ]
+        saturated = 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",
+            saturated.model_dump(mode="json"),
+        )
+        before = await self.store.get_candidate_ledger("root-a")
+
+        with self.assertRaisesRegex(
+            CandidateStateError,
+            "operation limit exceeded before execution",
+        ):
+            await self.service.apply_review_actions(
+                "root-a",
+                "child-a",
+                report_refs=[candidate],
+                actions=[CandidateReviewAction(
+                    action="adopt",
+                    candidate_ref=candidate,
+                    reason="must not reach repository",
+                )],
+                effective_at_sequence=10,
+                command_id="capacity-overflow-review",
+            )
+        self.assertEqual(0, self.repository.adoption_calls)
+        self.assertEqual(before, await self.store.get_candidate_ledger("root-a"))
+
+    async def test_review_capacity_allows_exact_envelope_and_action_boundary(self):
+        await self._create_child("child-a")
+        candidate = await self.service.manage(
+            "child-a",
+            operation="create",
+            content={"text": "exact capacity"},
+            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"exact-filler-{index}",
+                operation="review_batch",
+                input_hash="1" * 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"),
+        )
+
+        await self.service.apply_review_actions(
+            "root-a",
+            "child-a",
+            report_refs=[candidate],
+            actions=[CandidateReviewAction(
+                action="discard",
+                candidate_ref=candidate,
+                reason="fits exact operation boundary",
+            )],
+            effective_at_sequence=10,
+            command_id="exact-capacity-review",
+        )
+        completed = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual(MAX_CANDIDATE_OPERATIONS_PER_ROOT, len(completed.operations))
+        self.assertEqual("discarded", completed.current_state(candidate))
+
+    async def test_candidate_capacity_rejects_before_repository_or_intent_write(self):
+        await self._create_child("child-a")
+        candidate = await self.service.manage(
+            "child-a",
+            operation="create",
+            content={"text": "seed"},
+            parent_refs=[],
+            effective_at_sequence=4,
+        )
+        ledger = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        saturated_candidates = [candidate]
+        saturated_candidates.extend(
+            candidate.model_copy(update={
+                "candidate_id": f"capacity-candidate-{index}",
+            })
+            for index in range(1, MAX_CANDIDATES_PER_ROOT)
+        )
+        saturated = CandidateLedger.model_validate({
+            **ledger.model_dump(mode="json"),
+            "candidates": [
+                item.model_dump(mode="json") for item in saturated_candidates
+            ],
+        })
+        await self.store.replace_candidate_ledger(
+            "root-a",
+            saturated.model_dump(mode="json"),
+        )
+        before = await self.store.get_candidate_ledger("root-a")
+        repository_calls = self.repository.put_calls
+
+        with self.assertRaisesRegex(
+            CandidateStateError,
+            "candidate limit exceeded before execution",
+        ):
+            await self.service.manage(
+                "child-a",
+                operation="create",
+                content={"text": "must remain external-orphan free"},
+                parent_refs=[],
+                effective_at_sequence=5,
+                command_id="candidate-capacity-overflow",
+            )
+
+        self.assertEqual(repository_calls, self.repository.put_calls)
+        self.assertEqual(before, await self.store.get_candidate_ledger("root-a"))
+
+    async def test_pending_candidate_reserves_capacity_until_recovery(self):
+        await self._create_child("child-a")
+        seed = await self.service.manage(
+            "child-a",
+            operation="create",
+            content={"text": "seed"},
+            parent_refs=[],
+            effective_at_sequence=4,
+        )
+        ledger = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        near_limit_candidates = [seed]
+        near_limit_candidates.extend(
+            seed.model_copy(update={
+                "candidate_id": f"reserved-candidate-{index}",
+            })
+            for index in range(1, MAX_CANDIDATES_PER_ROOT - 1)
+        )
+        near_limit = CandidateLedger.model_validate({
+            **ledger.model_dump(mode="json"),
+            "candidates": [
+                item.model_dump(mode="json") for item in near_limit_candidates
+            ],
+        })
+        await self.store.replace_candidate_ledger(
+            "root-a",
+            near_limit.model_dump(mode="json"),
+        )
+        original_replace = self.store.replace_candidate_ledger
+        writes = 0
+
+        async def crash_after_repository(root_trace_id, raw_ledger):
+            nonlocal writes
+            writes += 1
+            if writes == 2:
+                raise OSError("crash before candidate publish")
+            await original_replace(root_trace_id, raw_ledger)
+
+        with patch.object(
+            self.store,
+            "replace_candidate_ledger",
+            side_effect=crash_after_repository,
+        ):
+            with self.assertRaisesRegex(OSError, "candidate publish"):
+                await self.service.manage(
+                    "child-a",
+                    operation="create",
+                    content={"text": "reserved A"},
+                    parent_refs=[],
+                    effective_at_sequence=5,
+                    command_id="reserved-candidate-A",
+                )
+        pending = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual(MAX_CANDIDATES_PER_ROOT - 1, len(pending.candidates))
+        self.assertEqual(
+            1,
+            sum(
+                item.status == "pending" and item.operation == "create"
+                for item in pending.operations
+            ),
+        )
+        calls_after_a = self.repository.put_calls
+
+        with self.assertRaisesRegex(
+            CandidateStateError,
+            "candidate limit exceeded before execution",
+        ):
+            await self.service.manage(
+                "child-a",
+                operation="create",
+                content={"text": "must not steal A's slot"},
+                parent_refs=[],
+                effective_at_sequence=6,
+                command_id="reserved-candidate-B",
+            )
+        self.assertEqual(calls_after_a, self.repository.put_calls)
+
+        recovered = await self.service.manage(
+            "child-a",
+            operation="create",
+            content={"text": "reserved A"},
+            parent_refs=[],
+            effective_at_sequence=99,
+            command_id="reserved-candidate-A",
+        )
+        completed = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual(MAX_CANDIDATES_PER_ROOT, len(completed.candidates))
+        self.assertEqual(recovered, completed.candidate(recovered))
+
     async def test_descendant_adoption_blocks_ancestor_rewind_without_mapping(self):
         await self._create_child("child-a")
         await self.store.create_trace(Trace(