fix(candidate): 绑定持久化命令身份并封闭回溯与质量副作用窗口

候选创建和父级采用改用真实 tool_call_id 作为稳定命令身份：同一调用崩溃重放保持幂等，不同调用即使载荷相同也不会误合并。

候选报告、验证、血缘与父级决策统一校验 owner Trace 当前消息祖先链，回溯后旧分支的未来候选和生命周期状态不能在新分支复活。

同时把确定性质量检查预算预留移到 Provider 调用之前，限制检查点可变字段并重新经过模型校验，候选验证哈希只绑定精确候选序列，保证 A/B2 局部修订不污染已通过的 A。

新增相同载荷不同调用、崩溃重放、回溯失效、预算耗尽零 Provider 调用、真实 Runner 创建与采用命令身份等回归测试。

cyber_agent/application/candidate.py

@@ -117,6 +117,10 @@ class CandidateLifecycleRecord(ApplicationModel):
 
 class CandidateOperationRecord(ApplicationModel):
     operation_id: str = Field(min_length=1, max_length=300)
+    # Stable framework command identity.  Candidate tool commands persist the
+    # assistant tool_call_id here so retries of one call are idempotent while
+    # 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"
     ]

cyber_agent/application/candidate_service.py

@@ -36,7 +36,7 @@ class CandidateStateError(ValueError):
 class CandidateService:
     """Validate, persist, and recover candidate version operations."""
 
-    _locks: "WeakValueDictionary[str, asyncio.Lock]" = WeakValueDictionary()
+    _locks: "WeakValueDictionary[tuple[asyncio.AbstractEventLoop, int, str], asyncio.Lock]" = WeakValueDictionary()
 
     def __init__(
         self,
@@ -53,12 +53,12 @@ class CandidateService:
         self.artifact_resolver = artifact_resolver
         self.event_service = event_service
 
-    @classmethod
-    def _lock_for(cls, root_trace_id: str) -> asyncio.Lock:
-        lock = cls._locks.get(root_trace_id)
+    def _lock_for(self, root_trace_id: str) -> asyncio.Lock:
+        key = (asyncio.get_running_loop(), id(self.store), root_trace_id)
+        lock = self._locks.get(key)
         if lock is None:
             lock = asyncio.Lock()
-            cls._locks[root_trace_id] = lock
+            self._locks[key] = lock
         return lock
 
     async def manage(
@@ -69,6 +69,7 @@ class CandidateService:
         content: Any,
         parent_refs: list[CandidatePointer],
         effective_at_sequence: int,
+        command_id: str | None = None,
     ) -> CandidateRef:
         trace, root = await self._require_owner_trace(trace_id)
         root_trace_id = root.trace_id
@@ -80,12 +81,18 @@ 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.
+        if command_id is not None and (not command_id or len(command_id) > 300):
+            raise CandidateStateError(
+                "Candidate command_id must be a non-empty string up to 300 chars"
+            )
+        # A persisted tool_call_id identifies one durable command.  Direct
+        # internal callers that do not have an assistant call identity retain
+        # a sequence-scoped fallback, but the public tool always supplies it.
+        command_identity = command_id or f"sequence:{effective_at_sequence}"
         command_hash = sha256(
             canonical_json({
                 "trace_id": trace_id,
-                "input_hash": input_hash,
+                "command_id": command_identity,
             }).encode("utf-8")
         ).hexdigest()
         operation_id = f"candidate:{command_hash}"
@@ -127,6 +134,7 @@ class CandidateService:
                     raise CandidateStateError(str(exc)) from exc
                 operation_record = CandidateOperationRecord(
                     operation_id=operation_id,
+                    command_id=command_id,
                     operation=operation,
                     input_hash=input_hash,
                     candidate=pointer,
@@ -145,6 +153,8 @@ class CandidateService:
             try:
                 parents = [ledger.candidate(item) for item in parent_refs]
                 self._validate_parents(trace, root, parents)
+                for parent in parents:
+                    await self._assert_candidate_active(ledger, parent)
                 state = ensure_task_protocol(trace.context)
                 request = CandidateVersionRequest(
                     operation_id=operation_id,
@@ -215,6 +225,8 @@ class CandidateService:
         self,
         trace_id: str,
         refs: list[CandidateRef],
+        *,
+        active_head_sequence: int | None = None,
     ) -> None:
         trace, root = await self._require_owner_trace(trace_id)
         async with self._lock_for(root.trace_id):
@@ -226,7 +238,17 @@ class CandidateService:
                         "TaskReport CandidateRef does not match the root ledger"
                     )
                 self._validate_candidate_owner(persisted, trace, root)
-                if ledger.current_state(ref) != "proposed":
+                await self._assert_candidate_active(
+                    ledger,
+                    persisted,
+                    active_head_sequence=active_head_sequence,
+                )
+                if await self._current_state(
+                    ledger,
+                    ref,
+                    active_trace_id=trace_id,
+                    active_head_sequence=active_head_sequence,
+                ) != "proposed":
                     raise CandidateStateError(
                         "TaskReport may only reference proposed candidates"
                     )
@@ -246,6 +268,7 @@ class CandidateService:
                     "CandidateRef does not match the root ledger"
                 )
             self._validate_candidate_owner(persisted, trace, root)
+            await self._assert_candidate_active(ledger, persisted)
         materials, issues = await resolve_artifact_refs(
             [candidate_ref.artifact_ref],
             resolver=self.artifact_resolver,
@@ -270,6 +293,7 @@ class CandidateService:
             ledger = await self._load_ledger(root.trace_id)
             persisted = ledger.candidate(candidate_ref)
             self._validate_candidate_owner(persisted, trace, root)
+            await self._assert_candidate_active(ledger, persisted)
             for raw in ledger.validations:
                 record = CandidateValidationRecord.model_validate(raw)
                 if (
@@ -298,6 +322,7 @@ class CandidateService:
                     "Candidate validation references an altered CandidateRef"
                 )
             self._validate_candidate_owner(persisted, trace, root)
+            await self._assert_candidate_active(ledger, persisted)
             for raw in ledger.validation_checkpoints:
                 checkpoint = CandidateValidationCheckpoint.model_validate(raw)
                 if (
@@ -331,6 +356,20 @@ class CandidateService:
         plan_hash: str,
         **updates: Any,
     ) -> CandidateValidationCheckpoint:
+        mutable_fields = {
+            "scope_results",
+            "fixed_checks",
+            "fixed_scope_errors",
+            "quality_completed",
+            "material_usage_recorded",
+            "aggregate_result",
+        }
+        unknown_fields = set(updates) - mutable_fields
+        if unknown_fields:
+            raise CandidateStateError(
+                "Candidate validation checkpoint fields are immutable: "
+                f"{', '.join(sorted(unknown_fields))}"
+            )
         trace, root = await self._require_bound_trace(trace_id)
         async with self._lock_for(root.trace_id):
             ledger = await self._load_ledger(root.trace_id)
@@ -348,7 +387,10 @@ class CandidateService:
                     checkpoint.candidate_ref == candidate_ref
                     and checkpoint.plan_hash == plan_hash
                 ):
-                    checkpoint = checkpoint.model_copy(update=updates)
+                    checkpoint = CandidateValidationCheckpoint.model_validate({
+                        **checkpoint.model_dump(mode="json"),
+                        **updates,
+                    })
                     result = checkpoint
                 checkpoints.append(checkpoint.model_dump(mode="json"))
             if result is None:
@@ -376,6 +418,7 @@ class CandidateService:
                     "Candidate validation references an altered CandidateRef"
                 )
             self._validate_candidate_owner(persisted, trace, root)
+            await self._assert_candidate_active(ledger, persisted)
             retained = []
             for raw in ledger.validations:
                 item = CandidateValidationRecord.model_validate(raw)
@@ -438,6 +481,7 @@ class CandidateService:
         report_refs: list[CandidateRef],
         actions: list[CandidateReviewAction],
         effective_at_sequence: int,
+        command_id: str | None = None,
     ) -> list[CandidateLifecycleRecord]:
         """Apply parent-owned candidate decisions behind one root lock."""
         if not actions:
@@ -485,6 +529,7 @@ class CandidateService:
                     raise CandidateStateError(
                         "Candidate action references another Task or altered revision"
                     )
+                await self._assert_candidate_active(ledger, persisted)
                 record = validation_by_key.get(key)
                 if record is None:
                     raise CandidateStateError(
@@ -503,11 +548,13 @@ class CandidateService:
                     raise CandidateStateError(
                         "Candidate revise requires a non-passed validation"
                     )
-                state = ledger.current_state(action.candidate_ref)
+                state = await self._current_state(ledger, action.candidate_ref)
                 expected_operation_id = self._review_operation_id(
                     parent_trace_id,
                     child_trace_id,
                     action,
+                    command_id=command_id,
+                    effective_at_sequence=effective_at_sequence,
                 )
                 same_completed_operation = any(
                     item.operation_id == expected_operation_id
@@ -531,6 +578,7 @@ class CandidateService:
                     child_trace_id=child_trace_id,
                     root_trace_id=root.trace_id,
                     effective_at_sequence=effective_at_sequence,
+                    command_id=command_id,
                 )
                 applied.append(record)
             return applied
@@ -544,6 +592,7 @@ class CandidateService:
         child_trace_id: str,
         root_trace_id: str,
         effective_at_sequence: int,
+        command_id: str | None,
     ) -> tuple[CandidateLedger, CandidateLifecycleRecord]:
         pointer = CandidatePointer(
             candidate_id=action.candidate_ref.candidate_id,
@@ -553,6 +602,8 @@ class CandidateService:
             parent_trace_id,
             child_trace_id,
             action,
+            command_id=command_id,
+            effective_at_sequence=effective_at_sequence,
         )
         input_hash = sha256(
             canonical_json(action).encode("utf-8")
@@ -575,6 +626,7 @@ class CandidateService:
         else:
             operation = CandidateOperationRecord(
                 operation_id=operation_id,
+                command_id=command_id,
                 operation=action.action,
                 input_hash=input_hash,
                 candidate=pointer,
@@ -674,6 +726,9 @@ class CandidateService:
         parent_trace_id: str,
         child_trace_id: str,
         action: CandidateReviewAction,
+        *,
+        command_id: str | None = None,
+        effective_at_sequence: int = 0,
     ) -> str:
         action_hash = sha256(
             canonical_json(action).encode("utf-8")
@@ -682,6 +737,7 @@ class CandidateService:
             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,
             }).encode("utf-8")
         ).hexdigest()
@@ -807,6 +863,106 @@ class CandidateService:
                         "Cannot rewind across a committed candidate adoption"
                     )
 
+    @staticmethod
+    def _candidate_operation(
+        ledger: CandidateLedger,
+        candidate: CandidateRef,
+    ) -> CandidateOperationRecord:
+        for operation in ledger.operations:
+            if (
+                operation.operation in {"create", "fork", "merge"}
+                and operation.candidate is not None
+                and operation.candidate.candidate_id == candidate.candidate_id
+                and operation.candidate.revision == candidate.revision
+            ):
+                return operation
+        raise CandidateStateError(
+            "Candidate has no persisted creation command"
+        )
+
+    async def _sequence_is_active(
+        self,
+        trace_id: str,
+        sequence: int,
+        *,
+        command_id: str | None,
+        allow_untracked: bool,
+        head_sequence: int | None = None,
+    ) -> bool:
+        trace = await self.store.get_trace(trace_id)
+        if trace is None:
+            return False
+        all_messages = await self.store.get_trace_messages(trace_id)
+        if not all_messages:
+            # CandidateService has a direct internal port for tests and
+            # controlled adapters.  Production tool commands are always bound
+            # to a persisted assistant tool_call_id and never take this path.
+            return allow_untracked
+        path = await self.store.get_main_path_messages(
+            trace_id,
+            trace.head_sequence if head_sequence is None else head_sequence,
+        )
+        message = next((item for item in path if item.sequence == sequence), None)
+        if message is None:
+            return False
+        if command_id is not None:
+            return message.role == "tool" and message.tool_call_id == command_id
+        return True
+
+    async def _assert_candidate_active(
+        self,
+        ledger: CandidateLedger,
+        candidate: CandidateRef,
+        *,
+        active_head_sequence: int | None = None,
+    ) -> None:
+        operation = self._candidate_operation(ledger, candidate)
+        if not await self._sequence_is_active(
+            candidate.owner_trace_id,
+            candidate.created_at_sequence,
+            command_id=operation.command_id,
+            allow_untracked=operation.command_id is None,
+            head_sequence=active_head_sequence,
+        ):
+            raise CandidateStateError(
+                "Candidate is not on the current owner Trace branch"
+            )
+
+    async def _current_state(
+        self,
+        ledger: CandidateLedger,
+        pointer: CandidatePointer,
+        *,
+        active_trace_id: str | None = None,
+        active_head_sequence: int | None = None,
+    ) -> str | None:
+        candidate = ledger.candidate(pointer)
+        creation = self._candidate_operation(ledger, candidate)
+        state = None
+        operations = {item.operation_id: item for item in ledger.operations}
+        for lifecycle in ledger.lifecycle:
+            if (
+                lifecycle.candidate.candidate_id != pointer.candidate_id
+                or lifecycle.candidate.revision != pointer.revision
+                or lifecycle.source_trace_id is None
+            ):
+                continue
+            operation = operations.get(lifecycle.operation_id)
+            command_id = operation.command_id if operation is not None else None
+            if await self._sequence_is_active(
+                lifecycle.source_trace_id,
+                lifecycle.effective_at_sequence,
+                command_id=command_id,
+                allow_untracked=creation.command_id is None,
+                head_sequence=(
+                    active_head_sequence
+                    if lifecycle.source_trace_id == active_trace_id
+                    else None
+                ),
+            ):
+                state = lifecycle.state
+        return state
+
     async def _require_owner_trace(self, trace_id: str):
         trace, root = await self._require_bound_trace(trace_id)
         state = ensure_task_protocol(trace.context)

cyber_agent/core/runner.py

@@ -759,7 +759,11 @@ class AgentRunner:
             root_task_anchor=root_anchor,
             task_report=task_report,
             candidate_output=candidate_output,
-            evaluated_head_sequence=evaluated.head_sequence or evaluated.last_sequence,
+            evaluated_head_sequence=(
+                candidate_ref.created_at_sequence
+                if candidate_ref is not None
+                else (evaluated.head_sequence or evaluated.last_sequence)
+            ),
             materials=materials,
             material_issues=material_issues,
             model_by_scope=model_by_scope,
@@ -1198,6 +1202,18 @@ class AgentRunner:
             rules=rules,
             materials=materials,
         )
+        try:
+            # Reserve/record the idempotent quality call before crossing the
+            # provider boundary.  Exhausted budgets therefore fail closed with
+            # zero provider side effects; provider errors still consume the
+            # attempted call exactly once.
+            await self.record_recursive_validation_usage(
+                subject.trace_id,
+                tool_calls=len(rules),
+                operation_id=usage_operation_id,
+            )
+        except Exception as exc:
+            return provider_error(f"Quality validation budget failed: {exc}")
         try:
             raw = await asyncio.wait_for(
                 provider.check(request),
@@ -1229,14 +1245,6 @@ class AgentRunner:
             )
             if outcome.status == "error":
                 errors[rule.scope] = outcome.issue or "Quality check failed"
-        try:
-            await self.record_recursive_validation_usage(
-                subject.trace_id,
-                tool_calls=len(rules),
-                operation_id=usage_operation_id,
-            )
-        except Exception as exc:
-            return provider_error(f"Quality validation budget failed: {exc}")
         return checks, errors
 
     async def dream(
@@ -2748,6 +2756,7 @@ class AgentRunner:
                         goal_id=current_goal_id,
                         goal_tree=goal_tree,
                         sequence=sequence,
+                        head_sequence=head_seq,
                         tool_call_id=call.tool_call_id,
                         side_branch_ctx=side_branch_ctx,
                         trigger_event=trigger_event,
@@ -3776,6 +3785,7 @@ class AgentRunner:
                                     goal_id=current_goal_id,
                                     goal_tree=goal_tree,
                                     sequence=sequence,
+                                    head_sequence=head_seq,
                                     tool_call_id=tc["id"],
                                     side_branch_ctx=side_branch_ctx,
                                     trigger_event=trigger_event_for_tool,
@@ -3926,6 +3936,7 @@ class AgentRunner:
                                     goal_id=current_goal_id,
                                     goal_tree=goal_tree,
                                     sequence=sequence,
+                                    head_sequence=head_seq,
                                     tool_call_id=tc["id"],
                                     side_branch_ctx=side_branch_ctx,
                                     trigger_event=trigger_event_for_tool,
@@ -5656,7 +5667,7 @@ class AgentRunner:
     ) -> List[Dict]:
         """按 Trace 持久化模式和当前协议状态生成工具 Schema。
 
-        主循环把结果交给 LLM；Recursive revision 2 只允许受治理的本地
+        主循环把结果交给 LLM；Structured Recursive 只允许受治理的本地
         ``task_brief`` 委托，Legacy 和 Recursive revision 1 继续使用 ``task``。
         """
         tool_names = (
@@ -5723,6 +5734,7 @@ class AgentRunner:
         goal_id: Optional[str],
         goal_tree: Optional[GoalTree],
         sequence: int,
+        head_sequence: int | None = None,
         side_branch_ctx: Optional[SideBranchContext],
         trigger_event: Optional[str],
         tool_call_id: str | None = None,
@@ -5752,6 +5764,7 @@ class AgentRunner:
                 else None
             ),
             "sequence": sequence,
+            "head_sequence": head_sequence,
             "tool_call_id": tool_call_id,
             "side_branch": {
                 "type": side_branch_ctx.type,

cyber_agent/tools/builtin/candidate.py

@@ -43,11 +43,17 @@ async def manage_candidate(
         return {"status": "failed", "error": f"Invalid candidate request: {exc}"}
     service = context.get("candidate_service")
     trace_id = context.get("trace_id")
+    command_id = context.get("tool_call_id")
     if service is None or not trace_id:
         return {
             "status": "failed",
             "error": "CandidateService is unavailable for this application",
         }
+    if not command_id:
+        return {
+            "status": "failed",
+            "error": "A persisted tool_call_id is required to manage candidates",
+        }
     try:
         candidate = await service.manage(
             trace_id,
@@ -55,6 +61,7 @@ async def manage_candidate(
             content=request.content,
             parent_refs=request.parent_refs,
             effective_at_sequence=int(context.get("sequence", 0) or 0),
+            command_id=str(command_id),
         )
     except (CandidateStateError, ValueError) as exc:
         return {"status": "failed", "error": str(exc)}

cyber_agent/tools/builtin/task_protocol.py

@@ -182,6 +182,7 @@ async def _submit_task_report_locked(
             await candidate_service.validate_report_refs(
                 trace_id,
                 parsed_submission.candidate_refs,
+                active_head_sequence=context.get("head_sequence"),
             )
         except ValueError as exc:
             return _error(f"Invalid TaskReport CandidateRefs: {exc}")
@@ -436,6 +437,10 @@ async def _review_task_result_locked(
         candidate_service = context.get("candidate_service")
         if candidate_service is None:
             return _error("Candidate actions require the bound CandidateService")
+        if not context.get("tool_call_id"):
+            return _error(
+                "Candidate actions require a persisted review tool_call_id"
+            )
         try:
             await candidate_service.apply_review_actions(
                 parent_trace_id,
@@ -443,6 +448,7 @@ async def _review_task_result_locked(
                 report_refs=report.candidate_refs,
                 actions=review.candidate_actions,
                 effective_at_sequence=reviewed_at_sequence,
+                command_id=str(context["tool_call_id"]),
             )
         except ValueError as exc:
             return _error(f"Invalid candidate action: {exc}")

tests/test_application_reference.py

@@ -248,6 +248,14 @@ class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
                 await store.get_candidate_ledger(root.trace_id)
             )
             self.assertEqual("adopted", ledger.current_state(ledger.candidates[0]))
+            self.assertIn(
+                "candidate-create",
+                {item.command_id for item in ledger.operations},
+            )
+            self.assertIn(
+                "adopt-review",
+                {item.command_id for item in ledger.operations},
+            )
 
     async def test_local_candidate_retry_preserves_fact_and_peer_validation(self):
         with tempfile.TemporaryDirectory() as temp_dir:
@@ -320,6 +328,7 @@ class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
             await store.update_trace(root.trace_id, head_sequence=2)
 
             async def execute_tool(trace_id, tool_name, arguments, sequence):
+                tool_call_id = f"reference-{tool_name}-{sequence}"
                 result = await binding.tool_registry.execute(
                     tool_name,
                     arguments,
@@ -328,12 +337,24 @@ class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
                         "store": store,
                         "trace_id": trace_id,
                         "sequence": sequence,
+                        "tool_call_id": tool_call_id,
                         "task_protocol_service": runner.task_protocol_service,
                         "candidate_service": runner.candidate_service,
                         "event_service": runner.event_service,
                     },
                     allowed_tool_names={tool_name},
                 )
+                if tool_name == "manage_candidate":
+                    trace = await store.get_trace(trace_id)
+                    await store.add_message(Message.create(
+                        trace_id=trace_id,
+                        role="tool",
+                        sequence=sequence,
+                        parent_sequence=(trace.head_sequence or None),
+                        tool_call_id=tool_call_id,
+                        content={"tool_name": tool_name, "result": "registered"},
+                    ))
+                    await store.update_trace(trace_id, head_sequence=sequence)
                 if isinstance(result, str):
                     return json.loads(result)
                 text = result.get("text") if isinstance(result, dict) else None
@@ -515,8 +536,16 @@ class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
             self.assertEqual(1, llm_calls)
             frozen_a = validation_a.result.model_dump(mode="json")
 
-            await runner._mark_trace_stopped(root.trace_id, root.head_sequence)
-            await runner._mark_trace_stopped(writer.trace_id, writer.head_sequence)
+            fresh_root = await store.get_trace(root.trace_id)
+            fresh_writer = await store.get_trace(writer.trace_id)
+            await runner._mark_trace_stopped(
+                root.trace_id,
+                fresh_root.head_sequence,
+            )
+            await runner._mark_trace_stopped(
+                writer.trace_id,
+                fresh_writer.head_sequence,
+            )
             restarted_components = build_reference_components(
                 artifacts=components.artifacts,
                 candidates=components.candidates,

tests/test_candidate_quality.py

@@ -1,7 +1,7 @@
 import json
 import tempfile
 import unittest
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 from cyber_agent.application import (
     AgentApplication,
@@ -13,7 +13,10 @@ from cyber_agent.application import (
     QualityRuleSpec,
 )
 from cyber_agent.application.candidate import CandidateLedger
-from cyber_agent.application.quality import QualityCheckOutcome
+from cyber_agent.application.quality import (
+    QualityCheckOutcome,
+    ValidationSubject,
+)
 from cyber_agent.core.artifacts import (
     ArtifactRef,
     ValidationMaterial,
@@ -315,6 +318,35 @@ class CandidateQualityTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(0, self.quality.calls)
         self.assertEqual(1, self.llm_calls)
 
+    async def test_exhausted_quality_budget_blocks_provider_before_call(self):
+        candidate = await self._candidate("finished", 2)
+        material = await self.repository.resolve(
+            candidate.artifact_ref,
+            candidate.root_trace_id,
+            self.trace.uid,
+        )
+        subject = ValidationSubject(
+            subject_type="candidate",
+            trace_id=self.trace.trace_id,
+            candidate_ref=candidate,
+        )
+        before = self.quality.calls
+        with patch.object(
+            self.runner,
+            "record_recursive_validation_usage",
+            new=AsyncMock(side_effect=ValueError("budget exhausted")),
+        ):
+            checks, errors = await self.runner._run_quality_check_batch(
+                subject,
+                (material,),
+                self.binding.application.quality_rules,
+                "quality-budget-exhausted",
+            )
+
+        self.assertEqual(before, self.quality.calls)
+        self.assertIn("output", errors)
+        self.assertTrue(all(item.status == "unknown" for item in checks.values()))
+
     async def test_candidate_scope_checkpoint_survives_final_publish_crash(self):
         candidate = await self._candidate("finished", 2)
         original = self.runner.candidate_service.record_validation

tests/test_candidate_service.py

@@ -36,7 +36,7 @@ from cyber_agent.core.validation import (
     ValidationResult,
 )
 from cyber_agent.tools.registry import ToolRegistry
-from cyber_agent.trace.models import Trace
+from cyber_agent.trace.models import Message as TraceMessage, Trace
 from cyber_agent.trace.store import (
     FileSystemTraceStore,
     TraceStoreCorruptionError,
@@ -323,6 +323,7 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
                     content={"text": "survives retry"},
                     parent_refs=[],
                     effective_at_sequence=2,
+                    command_id="candidate-crash-call",
                 )
 
         pending = CandidateLedger.model_validate(
@@ -336,6 +337,7 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
             content={"text": "survives retry"},
             parent_refs=[],
             effective_at_sequence=99,
+            command_id="candidate-crash-call",
         )
         ledger = CandidateLedger.model_validate(
             await self.store.get_candidate_ledger("root-a")
@@ -349,6 +351,119 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
             ledger.candidate(ledger.operations[0].candidate),
         )
 
+    async def test_distinct_tool_calls_with_identical_payload_create_versions(self):
+        first = await self.service.manage(
+            "root-a",
+            operation="create",
+            content={"text": "same payload"},
+            parent_refs=[],
+            effective_at_sequence=10,
+            command_id="candidate-call-one",
+        )
+        await self.store.add_message(TraceMessage.create(
+            trace_id="root-a",
+            role="tool",
+            sequence=10,
+            tool_call_id="candidate-call-one",
+            content={"tool_name": "manage_candidate", "result": "registered"},
+        ))
+        await self.store.update_trace("root-a", head_sequence=10)
+        second = await self.service.manage(
+            "root-a",
+            operation="create",
+            content={"text": "same payload"},
+            parent_refs=[],
+            effective_at_sequence=20,
+            command_id="candidate-call-two",
+        )
+        await self.store.add_message(TraceMessage.create(
+            trace_id="root-a",
+            role="tool",
+            sequence=20,
+            parent_sequence=10,
+            tool_call_id="candidate-call-two",
+            content={"tool_name": "manage_candidate", "result": "registered"},
+        ))
+        await self.store.update_trace("root-a", head_sequence=20)
+
+        self.assertNotEqual(first, second)
+        self.assertEqual(2, self.repository.put_calls)
+        ledger = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual(2, len(ledger.candidates))
+        self.assertEqual(2, len(ledger.operations))
+
+    async def test_rewind_makes_future_branch_candidate_unreportable(self):
+        await self.store.add_message(TraceMessage.create(
+            trace_id="root-a",
+            role="user",
+            sequence=1,
+            content="before candidate",
+        ))
+        await self.store.update_trace("root-a", head_sequence=1)
+        candidate = await self.service.manage(
+            "root-a",
+            operation="create",
+            content={"text": "future branch"},
+            parent_refs=[],
+            effective_at_sequence=10,
+            command_id="future-candidate-call",
+        )
+        await self.store.add_message(TraceMessage.create(
+            trace_id="root-a",
+            role="tool",
+            sequence=10,
+            parent_sequence=1,
+            tool_call_id="future-candidate-call",
+            content={"tool_name": "manage_candidate", "result": "registered"},
+        ))
+        await self.store.update_trace("root-a", head_sequence=10)
+        await self.service.validate_report_refs("root-a", [candidate])
+
+        await self.store.update_trace("root-a", head_sequence=1)
+        with self.assertRaisesRegex(
+            CandidateStateError,
+            "current owner Trace branch",
+        ):
+            await self.service.validate_report_refs("root-a", [candidate])
+
+    async def test_checkpoint_update_cannot_mutate_identity_or_sequence(self):
+        candidate = await self.service.manage(
+            "root-a",
+            operation="create",
+            content={"text": "checkpoint"},
+            parent_refs=[],
+            effective_at_sequence=3,
+        )
+        plan_hash = "a" * 64
+        await self.service.begin_validation_checkpoint(
+            "root-a",
+            candidate,
+            plan={"checks": []},
+            plan_hash=plan_hash,
+            validated_at_sequence=3,
+            material_usage_recorded=False,
+        )
+
+        with self.assertRaisesRegex(
+            CandidateStateError,
+            "fields are immutable",
+        ):
+            await self.service.update_validation_checkpoint(
+                "root-a",
+                candidate,
+                plan_hash,
+                validated_at_sequence=-1,
+            )
+        ledger = CandidateLedger.model_validate(
+            await self.store.get_candidate_ledger("root-a")
+        )
+        self.assertEqual(
+            3,
+            ledger.validation_checkpoints[0]["validated_at_sequence"],
+        )
+
     async def test_cross_root_and_cross_task_parent_access_is_rejected(self):
         first = await self.service.manage(
             "root-a",
@@ -548,6 +663,7 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
             report_refs=[candidate],
             actions=[action],
             effective_at_sequence=10,
+            command_id="adoption-review-call",
         )
         recovered = await self.service.apply_review_actions(
             "root-a",
@@ -555,6 +671,7 @@ class CandidateServiceTest(unittest.IsolatedAsyncioTestCase):
             report_refs=[candidate],
             actions=[action],
             effective_at_sequence=999,
+            command_id="adoption-review-call",
         )
         self.assertEqual(first, recovered)
         self.assertEqual(1, self.repository.adoption_calls)