openenv/echo_env

1"""Severity policy: the only place a check acquires a severity.2 3The policy is a versioned data artifact (`policies/severity-v1.json`), independent of4code. It maps every known check id — including reserved hub/statistical ids — to a5lane and severity, and bounds what tolerances an author may declare.6"""7 8import json9from importlib.resources import files10 11from pydantic import BaseModel, ConfigDict, model_validator12 13from .report import CheckResult14from .types import CheckStatus, Lane, Level, Severity, Verdict15 16 17class PolicyError(Exception):18    """Internal policy error (unknown check id, missing policy file). CLI exit code 3."""19 20 21class PolicyEntry(BaseModel):22    """One check id's lane and severity under this policy version."""23 24    model_config = ConfigDict(extra="forbid")25 26    check_id: str27    level: Level28    lane: Lane29    severity: Severity30 31 32class DeclarationBounds(BaseModel):33    """34    Ceilings on author-declared manifest values.35 36    Authors declare tolerances in the manifest; the policy bounds what is declarable;37    reports carry the declared values so hubs can apply stricter ceilings.38    """39 40    model_config = ConfigDict(extra="forbid")41 42    max_oracle_tolerance: float43    min_floor_margin: float44    max_variance_tolerance: float45    max_episode_timeout_s: float46 47 48class SeverityPolicy(BaseModel):49    """A versioned severity policy: every known check id, exactly one lane each."""50 51    model_config = ConfigDict(extra="forbid")52 53    policy_version: str54    entries: list[PolicyEntry]55    bounds: DeclarationBounds56 57    @model_validator(mode="after")58    def _unique_check_ids(self) -> "SeverityPolicy":59        ids = [entry.check_id for entry in self.entries]60        if len(ids) != len(set(ids)):61            raise ValueError("duplicate check ids in policy")62        return self63 64    def entries_for_lane(self, lane: Lane) -> dict[str, PolicyEntry]:65        """66        Return the applicable entries keyed by check id.67 68        Hub-lane entries are filtered out entirely for `lane=LOCAL` — an author never69        sees a check they cannot red-to-green. The hub lane is a superset: operators70        run the local checks plus the hub-only ones.71 72        Args:73            lane ([`~openenv.validation.types.Lane`]):74                The lane the run executes in.75 76        Returns:77            `dict[str, PolicyEntry]`: applicable entries keyed by check id.78        """79        if lane is Lane.LOCAL:80            applicable = [e for e in self.entries if e.lane is Lane.LOCAL]81        else:82            applicable = list(self.entries)83        return {e.check_id: e for e in applicable}84 85 86def load_policy(version: str = "v1") -> SeverityPolicy:87    """88    Load a committed severity policy by version.89 90    Args:91        version (`str`, *optional*, defaults to `"v1"`):92            The policy version to load.93 94    Returns:95        [`~openenv.validation.policy.SeverityPolicy`]: the parsed policy.96    """97    resource = files("openenv.validation").joinpath(98        "policies", f"severity-{version}.json"99    )100    try:101        raw = resource.read_text()102    except (FileNotFoundError, OSError) as exc:103        raise PolicyError(f"no severity policy for version {version!r}") from exc104    return SeverityPolicy.model_validate(json.loads(raw))105 106 107def apply_policy(108    results: list[CheckResult], policy: SeverityPolicy, lane: Lane109) -> Verdict:110    """111    Map check results to the run verdict. The only severity-assigning code path.112 113    FAIL if any policy-fail check FAILed; ERROR results fail closed; WARN if the run114    contains a SKIP or the only findings carry warn/advisory severity. Results with115    ids unknown to the policy (or outside the run's lane) are an internal error.116 117    Args:118        results (`list` of [`~openenv.validation.report.CheckResult`]):119            Grader outputs for this run.120        policy ([`~openenv.validation.policy.SeverityPolicy`]):121            The pinned policy version.122        lane ([`~openenv.validation.types.Lane`]):123            The lane the run executes in.124 125    Returns:126        [`~openenv.validation.types.Verdict`]: the overall verdict.127    """128    entries = policy.entries_for_lane(lane)129    classified_results = []130    for result in results:131        entry = entries.get(result.check_id)132        if entry is None:133            raise PolicyError(134                f"check id {result.check_id!r} is unknown to policy "135                f"{policy.policy_version!r} in lane {lane.value!r}"136            )137        classified_results.append((result, entry))138 139    verdict = Verdict.PASS140    for result, entry in classified_results:141        if result.status is CheckStatus.ERROR:142            return Verdict.FAIL143        if result.status is CheckStatus.FAIL:144            if entry.severity is Severity.FAIL:145                return Verdict.FAIL146            verdict = Verdict.WARN147        if result.status is CheckStatus.SKIP:148            verdict = Verdict.WARN149    return verdict150