Agent/cyber_agent/application/candidate_service.py

"""Candidate command service with one root-level consistency boundary."""

from __future__ import annotations

import asyncio
from hashlib import sha256
from typing import Any
from weakref import WeakValueDictionary

from cyber_agent.application.candidate import (
    CandidateAdoptionReceipt,
    CandidateAdoptionRequest,
    CandidateLedger,
    CandidateLifecycleRecord,
    CandidateOperationRecord,
    CandidatePointer,
    CandidateRef,
    CandidateRepository,
    CandidateRepositoryRejected,
    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 (
    CandidateValidationCheckpoint,
    CandidateValidationRecord,
)
from cyber_agent.core.agent_mode import require_mutable_trace_policy
from cyber_agent.core.artifacts import ArtifactRef, ArtifactResolver, resolve_artifact_refs
from cyber_agent.core.task_protocol import ensure_task_protocol, task_contract_ref
from cyber_agent.tools.errors import RecoverableToolExecutionError


class CandidateStateError(ValueError):
    pass


class CandidateService:
    """Validate, persist, and recover candidate version operations."""

    _locks: "WeakValueDictionary[tuple[asyncio.AbstractEventLoop, int, str], asyncio.Lock]" = WeakValueDictionary()

    def __init__(
        self,
        *,
        store: Any,
        application_binding: Any,
        repository: CandidateRepository,
        artifact_resolver: ArtifactResolver,
        event_service: Any = None,
    ) -> None:
        self.store = store
        self.binding = application_binding
        self.repository = repository
        self.artifact_resolver = artifact_resolver
        self.event_service = event_service

    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()
            self._locks[key] = lock
        return lock

    async def has_replayable_version_command(
        self,
        trace_id: str,
        *,
        command_id: str,
        arguments: dict[str, Any],
    ) -> bool:
        """Verify the exact durable candidate command behind an orphaned call."""
        try:
            request = arguments.get("request")
            if not isinstance(request, dict):
                return False
            operation = request.get("operation")
            if operation not in {"create", "fork", "merge"}:
                return False
            parent_refs = [
                CandidatePointer.model_validate(item)
                for item in request.get("parent_refs", [])
            ]
            input_hash = sha256(canonical_json({
                "operation": operation,
                "content": request.get("content"),
                "parent_refs": [
                    item.model_dump(mode="json") for item in parent_refs
                ],
            }).encode("utf-8")).hexdigest()
            command_hash = sha256(canonical_json({
                "trace_id": trace_id,
                "command_id": command_id,
            }).encode("utf-8")).hexdigest()
            operation_id = f"candidate:{command_hash}"
            _trace, root = await self._require_owner_trace(trace_id)
            ledger = await self._load_ledger(root.trace_id)
        except Exception:
            return False
        return any(
            item.operation_id == operation_id
            and item.command_id == command_id
            and item.operation == operation
            and item.input_hash == input_hash
            and item.status in {"pending", "completed", "failed"}
            for item in ledger.operations
        )

    async def manage(
        self,
        trace_id: str,
        *,
        operation: str,
        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
        input_payload = {
            "operation": operation,
            "content": content,
            "parent_refs": [item.model_dump(mode="json") for item in parent_refs],
        }
        input_hash = sha256(
            canonical_json(input_payload).encode("utf-8")
        ).hexdigest()
        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,
                "command_id": command_identity,
            }).encode("utf-8")
        ).hexdigest()
        operation_id = f"candidate:{command_hash}"
        async with self._lock_for(root_trace_id):
            ledger = await self._load_ledger(root_trace_id)
            existing_operation = next(
                (
                    item
                    for item in ledger.operations
                    if item.operation_id == operation_id
                ),
                None,
            )
            recovering_pending_operation = existing_operation is not None
            if existing_operation is not None:
                if existing_operation.input_hash != input_hash:
                    raise CandidateStateError(
                        "Candidate operation payload changed during recovery"
                    )
                if existing_operation.status == "completed":
                    if existing_operation.candidate is None:
                        raise CandidateStateError(
                            "Completed candidate operation has no result"
                        )
                    return ledger.candidate(existing_operation.candidate)
                if existing_operation.status == "failed":
                    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"
                    )
                if len(ledger.operations) >= MAX_CANDIDATE_OPERATIONS_PER_ROOT:
                    raise CandidateStateError(
                        "Candidate operation limit exceeded before execution"
                    )
                try:
                    pointer = self._allocate_pointer(
                        operation,
                        trace_id,
                        effective_at_sequence,
                        parent_refs,
                        ledger,
                    )
                except ValueError as exc:
                    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,
                )
                ledger = ledger.model_copy(update={
                    "operations": (*ledger.operations, operation_record),
                })
                await self._persist_ledger(root_trace_id, ledger)
                existing_operation = operation_record

            assert existing_operation.candidate is not None
            pointer = existing_operation.candidate
            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,
                    application_ref=self.binding.application_ref,
                    root_trace_id=root_trace_id,
                    owner_trace_id=trace_id,
                    uid=trace.uid,
                    candidate_id=pointer.candidate_id,
                    revision=pointer.revision,
                    task_contract_ref=task_contract_ref(
                        state,
                        root_task_anchor_hash=trace.context.get(
                            "root_task_anchor_hash"
                        ),
                    ),
                    parent_refs=tuple(parent_refs),
                    content=content,
                )
                try:
                    raw_artifact_ref = (
                        await self.repository.merge(request)
                        if operation == "merge"
                        else await self.repository.put_version(request)
                    )
                except CandidateRepositoryRejected as exc:
                    raise CandidateStateError(str(exc)) from exc
                except Exception as exc:
                    raise RecoverableToolExecutionError(
                        "Candidate Repository execution outcome is unknown; "
                        "the original idempotent command must be replayed"
                    ) from exc
                try:
                    artifact_ref = ArtifactRef.model_validate(raw_artifact_ref)
                except ValueError as exc:
                    raise CandidateStateError(
                        f"CandidateRepository returned an invalid ArtifactRef: {exc}"
                    ) from exc
                materials, issues = await resolve_artifact_refs(
                    [artifact_ref],
                    resolver=self.artifact_resolver,
                    root_trace_id=root_trace_id,
                    uid=trace.uid,
                )
                if issues or len(materials) != 1:
                    reason = issues[0].reason if issues else "Artifact was not resolved"
                    if any(issue.outcome == "unknown" for issue in issues):
                        raise RecoverableToolExecutionError(
                            "Candidate ArtifactResolver is temporarily unavailable; "
                            "the original command must remain replayable"
                        )
                    raise CandidateStateError(
                        f"CandidateRepository returned an invalid ArtifactRef: {reason}"
                    )
                candidate_ref = CandidateRef(
                    **pointer.model_dump(mode="json"),
                    application_ref=self.binding.application_ref,
                    root_trace_id=root_trace_id,
                    owner_trace_id=trace_id,
                    task_contract_ref=request.task_contract_ref,
                    artifact_ref=artifact_ref,
                    parent_refs=tuple(parent_refs),
                    created_at_sequence=effective_at_sequence,
                )
            except Exception as exc:
                if isinstance(exc, RecoverableToolExecutionError):
                    raise
                # A recovered pending intent may already have committed in the
                # application Repository. A transient Repository/Resolver error
                # cannot prove failure, so keep the original command replayable.
                if (
                    recovering_pending_operation
                    and not isinstance(exc, CandidateStateError)
                ):
                    raise RecoverableToolExecutionError(
                        "Candidate recovery was interrupted before its durable "
                        "ledger outcome could be established"
                    ) from exc
                failed = self._fail_operation(ledger, operation_id, str(exc))
                try:
                    await self._persist_ledger(root_trace_id, failed)
                except Exception as publish_exc:
                    raise RecoverableToolExecutionError(
                        "Candidate Repository result is durable but its failed "
                        "ledger outcome must be replayed with the original command ID"
                    ) from publish_exc
                raise
            completed = self._complete_operation(
                ledger,
                operation_id,
                candidate_ref,
            )
            # 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.
            try:
                await self._persist_ledger(root_trace_id, completed)
            except Exception as exc:
                raise RecoverableToolExecutionError(
                    "Candidate Repository result is durable but its ledger publish "
                    "must be replayed with the original command ID"
                ) from exc
            await self._emit_candidate_registered(candidate_ref, operation_id)
            return candidate_ref

    async def validate_report_refs(
        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):
            ledger = await self._load_ledger(root.trace_id)
            for ref in refs:
                persisted = ledger.candidate(ref)
                if persisted != ref:
                    raise CandidateStateError(
                        "TaskReport CandidateRef does not match the root ledger"
                    )
                self._validate_candidate_owner(persisted, trace, root)
                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"
                    )

    async def resolve_for_validation(
        self,
        trace_id: str,
        candidate_ref: CandidateRef,
    ):
        """Resolve one exact report-owned candidate through the trusted resolver."""
        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)
            persisted = ledger.candidate(candidate_ref)
            if persisted != candidate_ref:
                raise CandidateStateError(
                    "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,
            root_trace_id=root.trace_id,
            uid=trace.uid,
        )
        if issues or len(materials) != 1:
            reason = issues[0].reason if issues else "Artifact was not resolved"
            raise CandidateStateError(
                f"Candidate ArtifactRef cannot be validated: {reason}"
            )
        return materials[0]

    async def cached_validation(
        self,
        trace_id: str,
        candidate_ref: CandidateRef,
        plan_hash: str,
    ) -> CandidateValidationRecord | None:
        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)
            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 (
                    record.candidate_ref == candidate_ref
                    and record.plan_hash == plan_hash
                ):
                    return record
        return None

    async def begin_validation_checkpoint(
        self,
        trace_id: str,
        candidate_ref: CandidateRef,
        *,
        plan: dict[str, Any],
        plan_hash: str,
        validated_at_sequence: int,
        material_usage_recorded: bool,
    ) -> CandidateValidationCheckpoint:
        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)
            persisted = ledger.candidate(candidate_ref)
            if persisted != candidate_ref:
                raise CandidateStateError(
                    "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 (
                    checkpoint.candidate_ref == candidate_ref
                    and checkpoint.plan_hash == plan_hash
                ):
                    return checkpoint
            checkpoint = CandidateValidationCheckpoint(
                candidate_ref=candidate_ref,
                plan_hash=plan_hash,
                validation_plan=plan,
                validated_at_sequence=validated_at_sequence,
                material_usage_recorded=material_usage_recorded,
            )
            updated = ledger.model_copy(update={
                "validation_checkpoints": (
                    *ledger.validation_checkpoints,
                    checkpoint.model_dump(mode="json"),
                ),
            })
            await self._persist_ledger(root.trace_id, updated)
            return checkpoint

    async def update_validation_checkpoint(
        self,
        trace_id: str,
        candidate_ref: CandidateRef,
        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)
            persisted = ledger.candidate(candidate_ref)
            if persisted != candidate_ref:
                raise CandidateStateError(
                    "Candidate validation references an altered CandidateRef"
                )
            self._validate_candidate_owner(persisted, trace, root)
            checkpoints = []
            result = None
            for raw in ledger.validation_checkpoints:
                checkpoint = CandidateValidationCheckpoint.model_validate(raw)
                if (
                    checkpoint.candidate_ref == candidate_ref
                    and checkpoint.plan_hash == plan_hash
                ):
                    checkpoint = CandidateValidationCheckpoint.model_validate({
                        **checkpoint.model_dump(mode="json"),
                        **updates,
                    })
                    result = checkpoint
                checkpoints.append(checkpoint.model_dump(mode="json"))
            if result is None:
                raise CandidateStateError("Candidate validation checkpoint not found")
            updated = ledger.model_copy(update={
                "validation_checkpoints": tuple(checkpoints),
            })
            await self._persist_ledger(root.trace_id, updated)
            return result

    async def record_validation(
        self,
        trace_id: str,
        record: CandidateValidationRecord,
    ) -> None:
        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)
            persisted = ledger.candidate(record.candidate_ref)
            if persisted != record.candidate_ref:
                raise CandidateStateError(
                    "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)
                if (
                    item.candidate_ref == record.candidate_ref
                    and item.plan_hash == record.plan_hash
                ):
                    continue
                retained.append(raw)
            updated = ledger.model_copy(update={
                "validations": (
                    *retained,
                    record.model_dump(mode="json"),
                ),
            })
            checkpoints = []
            for raw in updated.validation_checkpoints:
                checkpoint = CandidateValidationCheckpoint.model_validate(raw)
                if (
                    checkpoint.candidate_ref == record.candidate_ref
                    and checkpoint.plan_hash == record.plan_hash
                ):
                    result = record.validation_result
                    checkpoint = checkpoint.model_copy(update={
                        "scope_results": tuple(result.get("scope_results", [])),
                        "aggregate_result": result,
                    })
                checkpoints.append(checkpoint.model_dump(mode="json"))
            updated = updated.model_copy(update={
                "validation_checkpoints": tuple(checkpoints),
            })
            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,
                    event_type="validation.completed",
                    event_key=(
                        "validation.completed:candidate:"
                        f"{record.candidate_ref.candidate_id}:"
                        f"{record.candidate_ref.revision}:{record.plan_hash}"
                    ),
                    effective_at_sequence=record.validated_at_sequence,
                    payload={
                        "subject_type": "candidate",
                        "candidate_ref": record.candidate_ref.model_dump(
                            mode="json"
                        ),
                        "validation_result": record.validation_result,
                    },
                )

    async def apply_review_actions(
        self,
        parent_trace_id: str,
        child_trace_id: str,
        *,
        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."""
        parent = await self.store.get_trace(parent_trace_id)
        child = await self.store.get_trace(child_trace_id)
        if parent is None or child is None:
            raise CandidateStateError("Parent or child Trace not found")
        if child.parent_trace_id != parent_trace_id or child.uid != parent.uid:
            raise CandidateStateError("Candidate review requires a direct child")
        _parent, root = await self._require_bound_trace(parent_trace_id)
        child_root = child.context.get("root_trace_id") or child.trace_id
        if (
            child_root != root.trace_id
            or child.context.get("application_ref")
            != self.binding.application_ref.model_dump(mode="json")
        ):
            raise CandidateStateError("Candidate review application/root mismatch")
        report_by_key = {
            (item.candidate_id, item.revision): item for item in report_refs
        }
        action_keys = [
            (item.candidate_ref.candidate_id, item.candidate_ref.revision)
            for item in actions
        ]
        if len(action_keys) != len(set(action_keys)):
            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,
            )
            operations_to_reserve: list[CandidateOperationRecord] = []
            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,
                )
                operations_to_reserve.append(batch_operation)

            existing_operation_ids = {
                item.operation_id for item in ledger.operations
            }
            for action_index, action in enumerate(actions):
                action_operation_id = 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,
                )
                if action_operation_id in existing_operation_ids:
                    continue
                operations_to_reserve.append(CandidateOperationRecord(
                    operation_id=action_operation_id,
                    command_id=command_id,
                    operation=action.action,
                    input_hash=sha256(
                        canonical_json(action).encode("utf-8")
                    ).hexdigest(),
                    candidate=CandidatePointer(
                        candidate_id=action.candidate_ref.candidate_id,
                        revision=action.candidate_ref.revision,
                    ),
                ))

            if operations_to_reserve:
                if (
                    len(ledger.operations) + len(operations_to_reserve)
                    > MAX_CANDIDATE_OPERATIONS_PER_ROOT
                ):
                    raise CandidateStateError(
                        "Candidate review operation limit exceeded before execution"
                    )
                ledger = ledger.model_copy(update={
                    "operations": (
                        *ledger.operations,
                        *operations_to_reserve,
                    ),
                })
                # Persist the ordered batch envelope and every action intent in
                # one replacement.  Besides freezing completeness, this is the
                # durable capacity reservation that keeps later commands from
                # stealing slots needed by a crash replay.
                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)
                key = (
                    record.candidate_ref.candidate_id,
                    record.candidate_ref.revision,
                )
                validation_by_key[key] = record
            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"
                    )
                persisted = ledger.candidate(action.candidate_ref)
                if persisted != action.candidate_ref or persisted.owner_trace_id != child_trace_id:
                    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(
                        "Candidate action requires framework validation"
                    )
                from cyber_agent.core.validation import ValidationResult

                outcome = ValidationResult.model_validate(
                    record.validation_result
                ).outcome
                if action.action == "adopt" and outcome != "passed":
                    raise CandidateStateError(
                        "Only a passed candidate revision can be adopted"
                    )
                if action.action == "revise" and outcome == "passed":
                    raise CandidateStateError(
                        "Candidate revise requires a non-passed validation"
                    )
                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,
                    action_index=action_index,
                )
                same_completed_operation = any(
                    item.operation_id == expected_operation_id
                    and item.status == "completed"
                    for item in ledger.operations
                )
                if (
                    state not in {"proposed", "adoption_pending"}
                    and not same_completed_operation
                ):
                    raise CandidateStateError(
                        f"Candidate already has terminal state: {state}"
                    )

            applied: list[CandidateLifecycleRecord] = []
            for action_index, action in enumerate(actions):
                ledger, record = await self._apply_review_action(
                    ledger,
                    action,
                    parent_trace_id=parent_trace_id,
                    child_trace_id=child_trace_id,
                    root_trace_id=root.trace_id,
                    effective_at_sequence=effective_at_sequence,
                    command_id=command_id,
                    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._persist_ledger(root.trace_id, ledger)
            return applied

    async def _apply_review_action(
        self,
        ledger: CandidateLedger,
        action: CandidateReviewAction,
        *,
        parent_trace_id: str,
        child_trace_id: str,
        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,
            revision=action.candidate_ref.revision,
        )
        operation_id = 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,
        )
        input_hash = sha256(
            canonical_json(action).encode("utf-8")
        ).hexdigest()
        existing = next(
            (item for item in ledger.operations if item.operation_id == operation_id),
            None,
        )
        if existing is not None:
            if existing.input_hash != input_hash:
                raise CandidateStateError("Candidate review payload changed during recovery")
            if existing.status == "completed":
                record = next(
                    item for item in reversed(ledger.lifecycle)
                    if item.operation_id == operation_id
                )
                return ledger, record
            if existing.status == "failed":
                raise CandidateStateError(existing.error or "Candidate review failed")
        else:
            operation = CandidateOperationRecord(
                operation_id=operation_id,
                command_id=command_id,
                operation=action.action,
                input_hash=input_hash,
                candidate=pointer,
            )
            ledger = ledger.model_copy(update={
                "operations": (*ledger.operations, operation),
            })
        if action.action != "adopt":
            record = CandidateLifecycleRecord(
                operation_id=operation_id,
                candidate=pointer,
                state="discarded",
                effective_at_sequence=effective_at_sequence,
                source_trace_id=parent_trace_id,
                reason=(
                    f"retry_from=output: {action.reason}"
                    if action.action == "revise"
                    else action.reason
                ),
            )
            ledger = self._finish_review_operation(ledger, operation_id, record)
            await self._persist_ledger(root_trace_id, ledger)
            await self._emit_lifecycle(record, root_trace_id)
            return ledger, record

        pending = CandidateLifecycleRecord(
            operation_id=operation_id,
            candidate=pointer,
            state="adoption_pending",
            effective_at_sequence=effective_at_sequence,
            source_trace_id=parent_trace_id,
            reason=action.reason,
        )
        if ledger.current_state(pointer) != "adoption_pending":
            ledger = ledger.model_copy(update={
                "lifecycle": (*ledger.lifecycle, pending),
            })
            await self._persist_ledger(root_trace_id, ledger)
            await self._emit_lifecycle(pending, root_trace_id)
        try:
            receipt = CandidateAdoptionReceipt.model_validate(
                await self.repository.commit_adoption(CandidateAdoptionRequest(
                    operation_id=operation_id,
                    candidate_ref=action.candidate_ref,
                ))
            )
            if receipt.operation_id != operation_id or not receipt.committed:
                raise CandidateStateError("Candidate adoption receipt is invalid")
        except Exception as exc:
            failed = CandidateLifecycleRecord(
                operation_id=operation_id,
                candidate=pointer,
                state="adoption_failed",
                effective_at_sequence=effective_at_sequence,
                source_trace_id=parent_trace_id,
                reason=str(exc)[:2_000],
            )
            ledger = self._finish_review_operation(
                ledger,
                operation_id,
                failed,
                error=str(exc),
            )
            await self._persist_ledger(root_trace_id, ledger)
            await self._emit_lifecycle(failed, root_trace_id)
            raise
        adopted = CandidateLifecycleRecord(
            operation_id=operation_id,
            candidate=pointer,
            state="adopted",
            effective_at_sequence=effective_at_sequence,
            source_trace_id=parent_trace_id,
            reason=action.reason,
            receipt=receipt,
        )
        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._persist_ledger(root_trace_id, ledger)
        await self._emit_lifecycle(adopted, root_trace_id)
        return ledger, adopted

    @staticmethod
    def _review_operation_id(
        parent_trace_id: str,
        child_trace_id: str,
        action: CandidateReviewAction,
        *,
        command_id: str | None = None,
        effective_at_sequence: int = 0,
        action_index: int = 0,
    ) -> str:
        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_index": action_index,
            }).encode("utf-8")
        ).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,
        operation_id: str,
    ) -> None:
        if self.event_service is None:
            return
        await self.event_service.emit_after_commit(
            source_trace_id=candidate.owner_trace_id,
            event_type="candidate.version_registered",
            event_key=(
                f"candidate.version_registered:{candidate.candidate_id}:"
                f"{candidate.revision}"
            ),
            effective_at_sequence=candidate.created_at_sequence,
            payload={"candidate_ref": candidate.model_dump(mode="json")},
        )
        await self.event_service.emit_after_commit(
            source_trace_id=candidate.owner_trace_id,
            event_type="candidate.lifecycle_changed",
            event_key=(
                f"candidate.lifecycle_changed:{operation_id}:proposed"
            ),
            effective_at_sequence=candidate.created_at_sequence,
            payload={"lifecycle": CandidateLifecycleRecord(
                operation_id=operation_id,
                candidate=CandidatePointer(
                    candidate_id=candidate.candidate_id,
                    revision=candidate.revision,
                ),
                state="proposed",
                effective_at_sequence=candidate.created_at_sequence,
                source_trace_id=candidate.owner_trace_id,
            ).model_dump(mode="json")},
        )

    async def _emit_lifecycle(
        self,
        lifecycle: CandidateLifecycleRecord,
        root_trace_id: str,
    ) -> None:
        del root_trace_id
        if self.event_service is None or lifecycle.source_trace_id is None:
            return
        await self.event_service.emit_after_commit(
            source_trace_id=lifecycle.source_trace_id,
            event_type="candidate.lifecycle_changed",
            event_key=(
                f"candidate.lifecycle_changed:{lifecycle.operation_id}:"
                f"{lifecycle.state}"
            ),
            effective_at_sequence=lifecycle.effective_at_sequence,
            payload={"lifecycle": lifecycle.model_dump(mode="json")},
        )

    @staticmethod
    def _finish_review_operation(
        ledger: CandidateLedger,
        operation_id: str,
        lifecycle: CandidateLifecycleRecord,
        *,
        error: str | None = None,
    ) -> CandidateLedger:
        operations = tuple(
            item.model_copy(update={
                "status": "failed" if error else "completed",
                "error": error[:2_000] if error else None,
            })
            if item.operation_id == operation_id else item
            for item in ledger.operations
        )
        return ledger.model_copy(update={
            "operations": operations,
            "lifecycle": (*ledger.lifecycle, lifecycle),
        })

    async def assert_rewind_allowed(
        self,
        trace_id: str,
        after_sequence: int,
    ) -> None:
        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)
            for lifecycle in ledger.lifecycle:
                if lifecycle.state != "adopted":
                    continue
                candidate = ledger.candidate(lifecycle.candidate)
                ancestor_ids: set[str] = set()
                for start_trace_id in {
                    lifecycle.source_trace_id,
                    candidate.owner_trace_id,
                }:
                    current_trace_id = start_trace_id
                    while current_trace_id and current_trace_id not in ancestor_ids:
                        ancestor_ids.add(current_trace_id)
                        current = await self.store.get_trace(current_trace_id)
                        current_trace_id = (
                            current.parent_trace_id if current is not None else None
                        )
                crosses_review = (
                    lifecycle.source_trace_id == trace_id
                    and lifecycle.effective_at_sequence > after_sequence
                )
                crosses_version = (
                    candidate.owner_trace_id == trace_id
                    and candidate.created_at_sequence > after_sequence
                )
                incomparable_ancestor = (
                    trace_id in ancestor_ids
                    and trace_id not in {
                        lifecycle.source_trace_id,
                        candidate.owner_trace_id,
                    }
                )
                if crosses_review or crosses_version or incomparable_ancestor:
                    raise CandidateStateError(
                        "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)
        if state.get("task_report") is not None:
            raise CandidateStateError("Candidates cannot change after TaskReport submission")
        if state.get("pending_reviews") or state.get("next_actions"):
            raise CandidateStateError("Resolve protocol lifecycle work before candidates")
        return trace, root

    async def _require_bound_trace(self, trace_id: str):
        trace = await self.store.get_trace(trace_id)
        if trace is None:
            raise CandidateStateError(f"Trace not found: {trace_id}")
        policy = require_mutable_trace_policy(trace.context)
        if not policy.requires_task_progress:
            raise CandidateStateError(
                "Candidates require Recursive revision 3"
            )
        application_ref = trace.context.get("application_ref")
        if application_ref != self.binding.application_ref.model_dump(mode="json"):
            raise CandidateStateError("Candidate Trace application binding mismatch")
        root_trace_id = trace.context.get("root_trace_id") or trace.trace_id
        root = trace if trace.trace_id == root_trace_id else await self.store.get_trace(
            root_trace_id
        )
        if root is None or root.uid != trace.uid:
            raise CandidateStateError("Candidate root or owner mismatch")
        if root.context.get("application_ref") != application_ref:
            raise CandidateStateError("Candidate root application mismatch")
        return trace, root

    async def _load_ledger(self, root_trace_id: str) -> CandidateLedger:
        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,
        trace_id: str,
        sequence: int,
        parents: list[CandidatePointer],
        ledger: CandidateLedger,
    ) -> CandidatePointer:
        if operation == "create":
            if parents:
                raise CandidateStateError("create does not accept parent_refs")
            digest = sha256(f"{trace_id}:{sequence}".encode("utf-8")).hexdigest()
            return CandidatePointer(candidate_id=f"cand_{digest[:24]}", revision=1)
        if operation == "fork":
            if len(parents) != 1:
                raise CandidateStateError("fork requires exactly one parent")
            parent = ledger.candidate(parents[0])
            revisions = [
                item.revision
                for item in ledger.candidates
                if item.candidate_id == parent.candidate_id
            ]
            return CandidatePointer(
                candidate_id=parent.candidate_id,
                revision=max(revisions) + 1,
            )
        if operation == "merge":
            if len(parents) < 2:
                raise CandidateStateError("merge requires at least two parents")
            digest = sha256(f"{trace_id}:{sequence}:merge".encode("utf-8")).hexdigest()
            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)

    def _validate_candidate_owner(self, candidate, trace, root) -> None:
        if (
            candidate.application_ref != self.binding.application_ref
            or candidate.root_trace_id != root.trace_id
            or candidate.owner_trace_id != trace.trace_id
        ):
            raise CandidateStateError(
                "Candidate belongs to another application, root, or Task"
            )

    @staticmethod
    def _complete_operation(
        ledger: CandidateLedger,
        operation_id: str,
        candidate: CandidateRef,
    ) -> CandidateLedger:
        operations = tuple(
            item.model_copy(update={"status": "completed", "error": None})
            if item.operation_id == operation_id
            else item
            for item in ledger.operations
        )
        lifecycle = (*ledger.lifecycle, CandidateLifecycleRecord(
            operation_id=operation_id,
            candidate=CandidatePointer(
                candidate_id=candidate.candidate_id,
                revision=candidate.revision,
            ),
            state="proposed",
            effective_at_sequence=candidate.created_at_sequence,
            source_trace_id=candidate.owner_trace_id,
        ))
        return CandidateLedger(
            candidates=(*ledger.candidates, candidate),
            lifecycle=lifecycle,
            operations=operations,
            validations=ledger.validations,
            validation_checkpoints=ledger.validation_checkpoints,
        )

    @staticmethod
    def _fail_operation(
        ledger: CandidateLedger,
        operation_id: str,
        error: str,
    ) -> CandidateLedger:
        operations = tuple(
            item.model_copy(update={
                "status": "failed",
                "error": error[:2_000],
            })
            if item.operation_id == operation_id
            else item
            for item in ledger.operations
        )
        return ledger.model_copy(update={"operations": operations})