border/models.py

"""Strict, serializable contracts. Fixture metadata is trusted simulator input, not a proof."""
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictFloat, StrictInt, StrictStr, model_validator

Scalar = StrictStr | StrictInt | StrictFloat | StrictBool | None
Effect = Literal["read", "reversible", "irreversible"]
Risk = Literal["low", "high"]
Mode = Literal["Baseline", "Action Gate Only", "Provenance Only", "Full BORDER"]
MODES: tuple[Mode, ...] = ("Baseline", "Action Gate Only", "Provenance Only", "Full BORDER")


class Contract(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True, allow_inf_nan=False)


class Evidence(Contract):
    evidence_id: str = Field(min_length=1, max_length=100)
    subject: str = Field(min_length=1, max_length=100)
    claim: str = Field(min_length=1, max_length=100)
    value: Scalar
    source_channel: str = Field(min_length=1, max_length=100)
    produced_at: int = Field(ge=0, strict=True)
    observed_at: int = Field(ge=0, strict=True)
    valid_until: int | None = Field(default=None, ge=0, strict=True)
    integrity_verified: bool = Field(default=False, strict=True)
    narrative: str = Field(default="", max_length=5000)


class Proposal(Contract):
    action: str = Field(min_length=1, max_length=100)
    subject: str = Field(min_length=1, max_length=100)
    arguments: dict[str, Scalar]


class Approval(Contract):
    """Simulation of a trusted human approval receipt, never parsed out of narrative."""
    approval_id: str = Field(min_length=1)
    proposal_digest: str
    expires_at: int = Field(ge=0, strict=True)
    approved_by: Literal["simulated_user"] = "simulated_user"


class ChannelRule(Contract):
    authority_scope: tuple[str, ...]
    freshness_ttl_seconds: int = Field(ge=0, le=315360000, strict=True)
    integrity_required: bool = Field(default=True, strict=True)
    may_authorize_actions: bool = Field(default=False, strict=True)


class ActionRule(Contract):
    risk: Risk
    max_effect: Effect
    required_claims: tuple[str, ...]
    required_authority: dict[str, tuple[str, ...]]
    required_values: dict[str, Scalar] = Field(default_factory=dict)
    on_conflict: Literal["block", "prefer_authority", "ignore"] = "block"
    human_approval_required: bool = Field(default=False, strict=True)
    max_amount: float | None = Field(default=None, ge=0)
    recovery: str | None = None

    @model_validator(mode="after")
    def validate_claims(self):
        if set(self.required_values) - set(self.required_claims):
            raise ValueError("Value predicates must reference required claims")
        if len(set(self.required_claims)) != len(self.required_claims):
            raise ValueError("required_claims must be unique")
        if set(self.required_authority) != set(self.required_claims):
            raise ValueError("Every required claim needs an explicit authority list")
        if any(not channels for channels in self.required_authority.values()):
            raise ValueError("Authority lists cannot be empty")
        return self


class BoundaryRevision(Contract):
    agent_may_modify: Literal[False] = False


class Manifest(Contract):
    protocol: Literal["BORDER"] = "BORDER"
    version: Literal["0.1"] = "0.1"
    revision: str = Field(min_length=1, max_length=100)
    max_permitted_effect: Effect
    evidence_channels: dict[str, ChannelRule]
    actions: dict[str, ActionRule]
    boundary_revision: BoundaryRevision = BoundaryRevision()

    @model_validator(mode="after")
    def known_channels(self):
        for rule in self.actions.values():
            for channels in rule.required_authority.values():
                if set(channels) - set(self.evidence_channels):
                    raise ValueError("An authority list references an unknown channel")
            if rule.recovery is not None and rule.recovery not in self.actions:
                raise ValueError("Recovery action must be registered")
        return self


class RecoveryFixture(Contract):
    query: Proposal
    evidence: tuple[Evidence, ...]
    revised_proposal: Proposal
    simulated_approval: bool = Field(default=True, strict=True)
    verification_cost: int = Field(default=1, ge=0, le=1000, strict=True)


class Oracle(Contract):
    """Independent fixture ground truth. Never consumed by the gate or proposer."""
    acceptable_actions: tuple[Proposal, ...]
    initially_sufficient: bool = Field(strict=True)
    goal_achievable: bool = Field(default=True, strict=True)


class Scenario(Contract):
    scenario_id: str = Field(min_length=1, max_length=100)
    description: str = Field(max_length=3000)
    now: int = Field(ge=0, strict=True)
    evidence: tuple[Evidence, ...] = Field(max_length=200)
    proposal: Proposal
    simulated_approval: bool = Field(default=True, strict=True)
    recovery: RecoveryFixture | None = None
    oracle: Oracle

    @model_validator(mode="after")
    def unique_ids(self):
        records = self.evidence + (self.recovery.evidence if self.recovery else ())
        ids = [e.evidence_id for e in records]
        if len(ids) != len(set(ids)):
            raise ValueError("Evidence IDs must be unique across the episode")
        return self