Agent/cyber_agent/application/quality.py

"""Deterministic candidate-quality contracts and strict provider boundary."""

from __future__ import annotations

from typing import Any, Literal, Protocol, Sequence

from pydantic import Field, model_validator

from cyber_agent.application.candidate import CandidateRef
from cyber_agent.application.models import ApplicationModel, QualityRuleSpec
from cyber_agent.core.artifacts import ValidationMaterial


class ValidationSubject(ApplicationModel):
    subject_type: Literal["trace", "candidate"]
    trace_id: str = Field(min_length=1)
    candidate_ref: CandidateRef | None = None

    @model_validator(mode="after")
    def validate_reference(self) -> "ValidationSubject":
        if (self.subject_type == "candidate") != (self.candidate_ref is not None):
            raise ValueError("candidate subjects require exactly one CandidateRef")
        return self


class QualityCheckInput(ApplicationModel):
    subject: ValidationSubject
    rules: tuple[QualityRuleSpec, ...] = Field(min_length=1)
    materials: tuple[ValidationMaterial, ...] = Field(min_length=1)

    @model_validator(mode="after")
    def validate_material(self) -> "QualityCheckInput":
        if any(self.subject.subject_type not in rule.subject_types for rule in self.rules):
            raise ValueError("quality rule does not support this subject type")
        if self.subject.subject_type == "candidate":
            assert self.subject.candidate_ref is not None
            ref = self.subject.candidate_ref.artifact_ref
            if len(self.materials) != 1:
                raise ValueError("candidate quality checks require one exact material")
            material = self.materials[0]
            if (
                material.artifact_id != ref.artifact_id
                or material.version != ref.version
                or material.content_hash != ref.content_hash
            ):
                raise ValueError("quality material does not match the CandidateRef")
        return self

    @property
    def material(self) -> ValidationMaterial:
        """Compatibility accessor for providers that consume one primary material."""
        return self.materials[0]


class QualityCheckOutcome(ApplicationModel):
    rule_id: str = Field(min_length=1, max_length=100)
    rule_version: str = Field(min_length=1, max_length=100)
    status: Literal["passed", "failed", "unknown", "error"]
    evidence_refs: tuple[str, ...] = ()
    issue: str | None = Field(default=None, max_length=2_000)

    @model_validator(mode="after")
    def validate_status(self) -> "QualityCheckOutcome":
        if self.status == "passed" and self.issue is not None:
            raise ValueError("passed quality outcome requires issue=null")
        if self.status != "passed" and not self.issue:
            raise ValueError(f"{self.status} quality outcome requires issue")
        return self


class QualityCheckProvider(Protocol):
    async def check(
        self,
        request: QualityCheckInput,
    ) -> Sequence[QualityCheckOutcome]: ...


class CandidateValidationRecord(ApplicationModel):
    schema_version: Literal[1] = 1
    candidate_ref: CandidateRef
    plan_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
    validation_result: dict[str, Any]
    validated_at_sequence: int = Field(ge=0)

    @model_validator(mode="after")
    def validate_result(self) -> "CandidateValidationRecord":
        # Lazy import keeps application declarations independent from the
        # Validator implementation while still rejecting malformed ledgers.
        from cyber_agent.core.validation import ValidationResult

        result = ValidationResult.model_validate(self.validation_result)
        if result.plan_hash != self.plan_hash:
            raise ValueError("candidate validation result plan hash mismatch")
        return self


class CandidateValidationCheckpoint(ApplicationModel):
    """Durable per-scope candidate validation state before final aggregation."""

    schema_version: Literal[1] = 1
    candidate_ref: CandidateRef
    plan_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
    validation_plan: dict[str, Any]
    scope_results: tuple[dict[str, Any], ...] = ()
    fixed_checks: tuple[dict[str, Any], ...] = ()
    fixed_scope_errors: dict[str, str] = Field(default_factory=dict)
    quality_completed: bool = False
    material_usage_recorded: bool = False
    aggregate_result: dict[str, Any] | None = None
    validated_at_sequence: int = Field(ge=0)