Agent/cyber_agent/application/candidate.py

"""Application-owned candidate content behind framework-owned immutable references."""

from __future__ import annotations

from typing import Any, Literal, Protocol, runtime_checkable

from pydantic import Field, model_validator

from cyber_agent.application.models import ApplicationModel, ApplicationRef, canonical_json
from cyber_agent.core.artifacts import ArtifactRef
from cyber_agent.core.task_contract import TaskContractRef


MAX_CANDIDATES_PER_ROOT = 256
MAX_CANDIDATE_OPERATIONS_PER_ROOT = 2_048


class CandidatePointer(ApplicationModel):
    candidate_id: str = Field(min_length=1, max_length=200)
    revision: int = Field(ge=1)


class CandidateRef(CandidatePointer):
    schema_version: Literal[1] = 1
    application_ref: ApplicationRef
    root_trace_id: str = Field(min_length=1)
    owner_trace_id: str = Field(min_length=1)
    task_contract_ref: TaskContractRef
    artifact_ref: ArtifactRef
    parent_refs: tuple[CandidatePointer, ...] = ()
    created_at_sequence: int = Field(ge=0)

    @model_validator(mode="after")
    def validate_parent_refs(self) -> "CandidateRef":
        identities = [
            (item.candidate_id, item.revision)
            for item in self.parent_refs
        ]
        if len(identities) != len(set(identities)):
            raise ValueError("CandidateRef parent_refs must be unique")
        if (self.candidate_id, self.revision) in set(identities):
            raise ValueError("CandidateRef cannot reference itself")
        return self


class CandidateVersionRequest(ApplicationModel):
    operation_id: str = Field(min_length=1, max_length=300)
    application_ref: ApplicationRef
    root_trace_id: str = Field(min_length=1)
    owner_trace_id: str = Field(min_length=1)
    uid: str | None = None
    candidate_id: str = Field(min_length=1, max_length=200)
    revision: int = Field(ge=1)
    task_contract_ref: TaskContractRef
    parent_refs: tuple[CandidatePointer, ...] = ()
    content: Any

    @model_validator(mode="after")
    def validate_content(self) -> "CandidateVersionRequest":
        canonical_json(self.content)
        return self


class CandidateAdoptionRequest(ApplicationModel):
    operation_id: str = Field(min_length=1, max_length=300)
    candidate_ref: CandidateRef


class CandidateAdoptionReceipt(ApplicationModel):
    operation_id: str = Field(min_length=1, max_length=300)
    external_id: str = Field(min_length=1, max_length=500)
    committed: bool = True


class CandidateReviewAction(ApplicationModel):
    action: Literal["adopt", "discard", "revise"]
    candidate_ref: CandidateRef
    reason: str = Field(min_length=1, max_length=2_000)


@runtime_checkable
class CandidateRepository(Protocol):
    async def put_version(
        self,
        request: CandidateVersionRequest,
    ) -> ArtifactRef: ...

    async def load_version(self, candidate_ref: CandidateRef) -> Any: ...

    async def merge(
        self,
        request: CandidateVersionRequest,
    ) -> ArtifactRef: ...

    async def commit_adoption(
        self,
        request: CandidateAdoptionRequest,
    ) -> CandidateAdoptionReceipt: ...


class CandidateLifecycleRecord(ApplicationModel):
    operation_id: str = Field(min_length=1, max_length=300)
    candidate: CandidatePointer
    state: Literal[
        "proposed",
        "adoption_pending",
        "adopted",
        "discarded",
        "adoption_failed",
    ]
    effective_at_sequence: int = Field(ge=0)
    source_trace_id: str | None = None
    reason: str | None = Field(default=None, max_length=2_000)
    receipt: CandidateAdoptionReceipt | None = None


class CandidateOperationRecord(ApplicationModel):
    operation_id: str = Field(min_length=1, max_length=300)
    operation: Literal[
        "create", "fork", "merge", "adopt", "discard", "revise"
    ]
    input_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
    status: Literal["pending", "completed", "failed"] = "pending"
    candidate: CandidatePointer | None = None
    error: str | None = Field(default=None, max_length=2_000)


class CandidateLedger(ApplicationModel):
    schema_version: Literal[1] = 1
    candidates: tuple[CandidateRef, ...] = ()
    lifecycle: tuple[CandidateLifecycleRecord, ...] = ()
    operations: tuple[CandidateOperationRecord, ...] = ()
    validations: tuple[dict[str, Any], ...] = ()

    @model_validator(mode="after")
    def validate_ledger(self) -> "CandidateLedger":
        if len(self.candidates) > MAX_CANDIDATES_PER_ROOT:
            raise ValueError("Candidate ledger candidate limit exceeded")
        if len(self.operations) > MAX_CANDIDATE_OPERATIONS_PER_ROOT:
            raise ValueError("Candidate ledger operation limit exceeded")
        identities = [
            (item.candidate_id, item.revision)
            for item in self.candidates
        ]
        if len(identities) != len(set(identities)):
            raise ValueError("Candidate ledger contains duplicate revisions")
        operation_ids = [item.operation_id for item in self.operations]
        if len(operation_ids) != len(set(operation_ids)):
            raise ValueError("Candidate ledger contains duplicate operation IDs")
        known = set(identities)
        for candidate in self.candidates:
            for parent in candidate.parent_refs:
                parent_key = (parent.candidate_id, parent.revision)
                if parent_key not in known:
                    raise ValueError("Candidate ledger contains an unknown parent")
                if (
                    parent.candidate_id == candidate.candidate_id
                    and parent.revision >= candidate.revision
                ):
                    raise ValueError("Candidate lineage references a future revision")
        return self

    def candidate(self, pointer: CandidatePointer) -> CandidateRef:
        for item in self.candidates:
            if (
                item.candidate_id == pointer.candidate_id
                and item.revision == pointer.revision
            ):
                return item
        raise ValueError(
            f"Candidate revision not found: {pointer.candidate_id}@{pointer.revision}"
        )

    def current_state(self, pointer: CandidatePointer) -> str | None:
        state = None
        for item in self.lifecycle:
            if (
                item.candidate.candidate_id == pointer.candidate_id
                and item.candidate.revision == pointer.revision
            ):
                state = item.state
        return state