script-build-and-framework-tmp/agent/agent/orchestration/validation_policy.py

"""Minimal V2 validation planning and deterministic rule execution."""

from __future__ import annotations

import inspect
from dataclasses import asdict, dataclass, field
from typing import Any, Awaitable, Dict, List, Mapping, Optional, Protocol, Sequence, Union

from .models import (
    ArtifactRef,
    ArtifactSnapshot,
    TaskAttempt,
    TaskRecord,
    ValidationMode,
    ValidationPlan,
    ValidationVerdict,
    json_values,
)


@dataclass(frozen=True)
class ValidationContext:
    root_trace_id: str
    task: TaskRecord
    attempt: TaskAttempt
    snapshot: ArtifactSnapshot
    prior_validation_count: int = 0
    context_documents: Mapping[str, Any] = field(default_factory=dict)


class ValidationPolicy(Protocol):
    def plan(self, context: ValidationContext) -> ValidationPlan: ...


@dataclass(frozen=True)
class DefaultValidationPolicy:
    """Preserves V1: every attempt receives a full independent Agent Validator."""

    validator_preset: str = "validator"

    def __post_init__(self) -> None:
        if not self.validator_preset.strip():
            raise ValueError("validator_preset cannot be empty")

    def plan(self, context: ValidationContext) -> ValidationPlan:
        del context
        return ValidationPlan(
            mode=ValidationMode.AGENT,
            validator_preset=self.validator_preset,
        )


@dataclass(frozen=True)
class DeterministicRuleResult:
    rule_id: str
    verdict: ValidationVerdict
    reason: str
    evidence_refs: List[ArtifactRef] = field(default_factory=list)
    error: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        return json_values(asdict(self))


@dataclass(frozen=True)
class DeterministicValidationResult:
    verdict: ValidationVerdict
    rule_results: List[DeterministicRuleResult] = field(default_factory=list)

    @property
    def errors(self) -> List[str]:
        return [item.error for item in self.rule_results if item.error]

    def to_dict(self) -> Dict[str, Any]:
        return json_values(asdict(self))


RuleOutcome = Union[
    DeterministicRuleResult,
    Awaitable[DeterministicRuleResult],
]


class DeterministicRule(Protocol):
    rule_id: str

    def evaluate(self, context: ValidationContext) -> RuleOutcome: ...


class DeterministicValidator(Protocol):
    async def validate(
        self,
        context: ValidationContext,
        plan: ValidationPlan,
    ) -> DeterministicValidationResult: ...


class RuleBasedDeterministicValidator:
    """Runs named rules in plan order and converts rule failures to inconclusive."""

    def __init__(self, rules: Sequence[DeterministicRule]) -> None:
        self._rules: Dict[str, DeterministicRule] = {}
        for rule in rules:
            rule_id = rule.rule_id.strip()
            if not rule_id:
                raise ValueError("Deterministic rule_id cannot be empty")
            if rule_id in self._rules:
                raise ValueError(f"Duplicate deterministic rule_id: {rule_id}")
            self._rules[rule_id] = rule

    async def validate(
        self,
        context: ValidationContext,
        plan: ValidationPlan,
    ) -> DeterministicValidationResult:
        if plan.mode != ValidationMode.DETERMINISTIC:
            raise ValueError("RuleBasedDeterministicValidator requires a deterministic plan")
        if not plan.rule_ids:
            return DeterministicValidationResult(ValidationVerdict.INCONCLUSIVE)

        results: List[DeterministicRuleResult] = []
        for rule_id in plan.rule_ids:
            rule = self._rules.get(rule_id)
            if rule is None:
                results.append(_rule_error(rule_id, "Deterministic rule is not registered"))
                continue
            try:
                outcome = rule.evaluate(context)
                result = await outcome if inspect.isawaitable(outcome) else outcome
                if result.rule_id != rule_id:
                    raise ValueError(
                        f"Rule returned id {result.rule_id!r}, expected {rule_id!r}"
                    )
                results.append(result)
            except Exception as exc:
                results.append(_rule_error(rule_id, f"{type(exc).__name__}: {exc}"))

        verdicts = {item.verdict for item in results}
        if ValidationVerdict.FAILED in verdicts:
            verdict = ValidationVerdict.FAILED
        elif verdicts == {ValidationVerdict.PASSED}:
            verdict = ValidationVerdict.PASSED
        else:
            verdict = ValidationVerdict.INCONCLUSIVE
        return DeterministicValidationResult(verdict, results)


def _rule_error(rule_id: str, error: str) -> DeterministicRuleResult:
    return DeterministicRuleResult(
        rule_id=rule_id,
        verdict=ValidationVerdict.INCONCLUSIVE,
        reason="Rule could not produce a verdict",
        error=error,
    )


__all__ = [
    "DefaultValidationPolicy",
    "DeterministicRule",
    "DeterministicRuleResult",
    "DeterministicValidationResult",
    "DeterministicValidator",
    "RuleBasedDeterministicValidator",
    "ValidationContext",
    "ValidationMode",
    "ValidationPlan",
    "ValidationPolicy",
]