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,
    CandidateReviewAction,
    CandidateVersionRequest,
)
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


class CandidateStateError(ValueError):
    pass


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

    _locks: "WeakValueDictionary[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

    @classmethod
    def _lock_for(cls, root_trace_id: str) -> asyncio.Lock:
        lock = cls._locks.get(root_trace_id)
        if lock is None:
            lock = asyncio.Lock()
            cls._locks[root_trace_id] = lock
        return lock

    async def manage(
        self,
        trace_id: str,
        *,
        operation: str,
        content: Any,
        parent_refs: list[CandidatePointer],
        effective_at_sequence: int,
    ) -> 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()
        # The command identity must survive a process crash and a subsequent
        # model retry. Sequence numbers identify attempts, not durable intent.
        command_hash = sha256(
            canonical_json({
                "trace_id": trace_id,
                "input_hash": input_hash,
            }).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,
            )
            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"
                    )
            else:
                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,
                    operation=operation,
                    input_hash=input_hash,
                    candidate=pointer,
                )
                ledger = ledger.model_copy(update={
                    "operations": (*ledger.operations, operation_record),
                })
                await self.store.replace_candidate_ledger(
                    root_trace_id,
                    ledger.model_dump(mode="json"),
                )
                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)
                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,
                )
                artifact_ref = ArtifactRef.model_validate(
                    await self.repository.merge(request)
                    if operation == "merge"
                    else await self.repository.put_version(request)
                )
                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"
                    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:
                failed = self._fail_operation(ledger, operation_id, str(exc))
                await self.store.replace_candidate_ledger(
                    root_trace_id,
                    failed.model_dump(mode="json"),
                )
                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.
            await self.store.replace_candidate_ledger(
                root_trace_id,
                completed.model_dump(mode="json"),
            )
            await self._emit_candidate_registered(candidate_ref, operation_id)
            return candidate_ref

    async def validate_report_refs(
        self,
        trace_id: str,
        refs: list[CandidateRef],
    ) -> 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)
                if ledger.current_state(ref) != "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)
        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)
            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)
            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.store.replace_candidate_ledger(
                root.trace_id,
                updated.model_dump(mode="json"),
            )
            return checkpoint

    async def update_validation_checkpoint(
        self,
        trace_id: str,
        candidate_ref: CandidateRef,
        plan_hash: str,
        **updates: Any,
    ) -> 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)
            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 = checkpoint.model_copy(update=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.store.replace_candidate_ledger(
                root.trace_id,
                updated.model_dump(mode="json"),
            )
            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)
            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.store.replace_candidate_ledger(
                root.trace_id,
                updated.model_dump(mode="json"),
            )
            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,
    ) -> list[CandidateLifecycleRecord]:
        """Apply parent-owned candidate decisions behind one root lock."""
        if not actions:
            return []
        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)
            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, key in 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"
                    )
                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 = ledger.current_state(action.candidate_ref)
                expected_operation_id = self._review_operation_id(
                    parent_trace_id,
                    child_trace_id,
                    action,
                )
                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 in 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,
                )
                applied.append(record)
            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,
    ) -> 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,
        )
        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,
                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.store.replace_candidate_ledger(
                root_trace_id,
                ledger.model_dump(mode="json"),
            )
            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.store.replace_candidate_ledger(
                root_trace_id,
                ledger.model_dump(mode="json"),
            )
            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.store.replace_candidate_ledger(
                root_trace_id,
                ledger.model_dump(mode="json"),
            )
            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.store.replace_candidate_ledger(
            root_trace_id,
            ledger.model_dump(mode="json"),
        )
        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,
    ) -> str:
        action_hash = sha256(
            canonical_json(action).encode("utf-8")
        ).hexdigest()
        command_hash = sha256(
            canonical_json({
                "parent_trace_id": parent_trace_id,
                "child_trace_id": child_trace_id,
                "action_hash": action_hash,
            }).encode("utf-8")
        ).hexdigest()
        return f"candidate-review:{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"
                    )

    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 {})

    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}")

    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})