feat(validation): add policy and evidence ports

agent/agent/orchestration/evidence.py

@@ -0,0 +1,93 @@
+"""Guarded V2 evidence queries scoped to a frozen validation snapshot."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Protocol
+
+from .models import ArtifactRef
+
+
+class EvidenceError(RuntimeError):
+    pass
+
+
+class EvidenceOwnershipError(EvidenceError):
+    pass
+
+
+class EvidenceBudgetExceeded(EvidenceError):
+    pass
+
+
+class EvidenceProviderError(EvidenceError):
+    pass
+
+
+@dataclass(frozen=True)
+class EvidenceQuery:
+    root_trace_id: str
+    task_id: str
+    attempt_id: str
+    snapshot_id: str
+    validation_id: str
+    query: str
+    limit: int = 10
+
+    def __post_init__(self) -> None:
+        for name in (
+            "root_trace_id",
+            "task_id",
+            "attempt_id",
+            "snapshot_id",
+            "validation_id",
+            "query",
+        ):
+            value = getattr(self, name)
+            if not isinstance(value, str) or not value.strip():
+                raise ValueError(f"{name} must be a non-empty string")
+        if (
+            isinstance(self.limit, bool)
+            or not isinstance(self.limit, int)
+            or self.limit < 1
+        ):
+            raise ValueError("limit must be a positive integer")
+
+
+@dataclass(frozen=True)
+class EvidenceProviderResult:
+    items: List[Dict[str, Any]] = field(default_factory=list)
+    evidence_refs: List[ArtifactRef] = field(default_factory=list)
+
+
+@dataclass(frozen=True)
+class EvidenceResponse:
+    items: List[Dict[str, Any]]
+    evidence_refs: List[ArtifactRef]
+    queries_used: int
+    queries_remaining: int
+
+
+class EvidenceProvider(Protocol):
+    async def query(self, request: EvidenceQuery) -> EvidenceProviderResult: ...
+
+
+class EvidencePort(Protocol):
+    async def query(
+        self,
+        actor_context: Dict[str, Any],
+        request: EvidenceQuery,
+    ) -> EvidenceResponse: ...
+
+
+__all__ = [
+    "EvidenceBudgetExceeded",
+    "EvidenceError",
+    "EvidenceOwnershipError",
+    "EvidencePort",
+    "EvidenceProvider",
+    "EvidenceProviderError",
+    "EvidenceProviderResult",
+    "EvidenceQuery",
+    "EvidenceResponse",
+]

agent/agent/orchestration/validation_policy.py

@@ -0,0 +1,167 @@
+"""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, 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
+
+
+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",
+]