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)
    material: ValidationMaterial

    @model_validator(mode="after")
    def validate_material(self) -> "QualityCheckInput":
        if self.subject.subject_type != "candidate" or self.subject.candidate_ref is None:
            raise ValueError("quality checks currently require a candidate subject")
        ref = self.subject.candidate_ref.artifact_ref
        if (
            self.material.artifact_id != ref.artifact_id
            or self.material.version != ref.version
            or self.material.content_hash != ref.content_hash
        ):
            raise ValueError("quality material does not match the CandidateRef")
        return self


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