refactor(v1): reuse Pydantic adapters across validation

tests/v1/test_judges.py

@@ -9,6 +9,7 @@
 from pydantic import Field
 
 import verifiers.v1 as vf
+from verifiers.v1.envs.agentic_judge import ScoreConfig
 from verifiers.v1.graph import MessageNode
 from verifiers.v1.judge import Judge, JudgeResponse
 from verifiers.v1.types import AssistantMessage, UserMessage
@@ -480,10 +481,14 @@ def test_rubric_criteria_toml_and_json(tmp_path):
     toml = rubric_judge(tmp_path).criteria
     assert [c.name for c in toml] == ["mentions_paris", "is_polite"]
     assert [c.weight for c in toml] == [3.0, 1.0]
-    # JSON: both the {"criteria": [...]} object and a bare list parse to the same rubric
+    # JSON accepts a metadata-bearing object or a bare criteria list.
     items = [c.model_dump() for c in toml]
     assert (
-        rubric_judge(tmp_path, json.dumps({"criteria": items}), ".json").criteria
+        rubric_judge(
+            tmp_path,
+            json.dumps({"title": "Safety", "criteria": items}),
+            ".json",
+        ).criteria
         == toml
     )
     assert rubric_judge(tmp_path, json.dumps(items), ".json").criteria == toml
@@ -511,8 +516,23 @@ def test_rubric_rejects_bad_files(tmp_path):
         ).criteria
     # negative/NaN/inf weights would invert a criterion or corrupt the weighted mean
     for weight in (-1.0, float("nan"), float("inf")):
-        with pytest.raises(ValueError, match="negative or non-finite"):
+        with pytest.raises(
+            ValueError, match="greater than or equal to 0|finite number"
+        ):
             _ = rubric_judge(tmp_path, weights={"mentions_paris": weight}).criteria
+    with pytest.raises(ValueError, match="non-finite total"):
+        _ = rubric_judge(
+            tmp_path, weights={"mentions_paris": 1e308, "is_polite": 1e308}
+        ).criteria
+
+
+@pytest.mark.parametrize("weight", [float("nan"), float("inf"), float("-inf")])
+def test_judge_composition_weights_are_finite(weight):
+    with pytest.raises(ValueError, match="finite number"):
+        vf.JudgeConfig(weight=weight)
+    for field in ("task_weight", "judge_weight"):
+        with pytest.raises(ValueError, match="finite number"):
+            ScoreConfig.model_validate({field: weight})
 
 
 async def test_rubric_score(tmp_path, fake_judge_model):
@@ -569,7 +589,7 @@ async def off_menu(self, messages, *, trace=None, schema=None, parse=None, **s):
 
 
 def test_rubric_choices_validation(tmp_path):
-    with pytest.raises(ValueError, match="at least two"):
+    with pytest.raises(ValueError, match="at least 2"):
         _ = rubric_judge(
             tmp_path, body='[[criteria]]\nname = "x"\ntext = "t"\nchoices = ["only"]\n'
         ).criteria

verifiers/v1/cli/output.py

@@ -12,6 +12,7 @@
 
 import asyncio
 import json
+from functools import cache
 from pathlib import Path
 
 import tomli_w
@@ -29,6 +30,9 @@
 CONFIG_FILE = "config.toml"
 """Filename a run's resolved config is written to (re-runnable via `@ config.toml`)."""
 
+# Compiling an adapter is the expensive part; run output reuses only a few model classes.
+type_adapter = cache(TypeAdapter)
+
 
 def output_path(config: EvalConfig) -> Path:
     """Where this run writes: `outputs/<env>--<model>--<harness>/<uuid>` (or the explicit
@@ -72,7 +76,7 @@ def save_config(config: BaseModel, results_dir: Path) -> None:
 def write_episode(results_dir: Path, episode: Episode) -> None:
     """Serialize and append one rollout episode in the worker thread."""
     # Preserve fields declared by typed Trace subclasses nested in the episode.
-    data = TypeAdapter(type(episode)).dump_json(episode, exclude_none=True)
+    data = type_adapter(type(episode)).dump_json(episode, exclude_none=True)
     with (results_dir / TRACES_FILE).open("ab") as f:
         f.write(data + b"\n")
 
@@ -88,7 +92,7 @@ def read_episodes(results_dir: Path, trace_type: type) -> list[Episode]:
     `trace_type` (`Trace[WireTaskData, ...]` reads any taskset's file without
     importing it). A pre-episode line (one bare trace) is wrapped as a single-trace
     record, so both file generations read uniformly."""
-    trace_adapter = TypeAdapter(trace_type)
+    trace_adapter = type_adapter(trace_type)
     episodes: list[Episode] = []
     with (results_dir / TRACES_FILE).open(encoding="utf-8") as f:
         for line in f:

verifiers/v1/configs/judge.py

@@ -5,7 +5,7 @@
 from pathlib import Path
 from typing import Any
 
-from pydantic import BaseModel, SerializeAsAny
+from pydantic import BaseModel, FiniteFloat, SerializeAsAny
 
 from verifiers.v1.clients import BaseClientConfig
 from verifiers.v1.types import ID, SamplingConfig
@@ -17,7 +17,7 @@ class JudgeConfig(BaseClientConfig):
     """Plugin id; empty for a judge called directly by task code."""
     name: str = ""
     """Reward key override for a plugged judge."""
-    weight: float = 1.0
+    weight: FiniteFloat = 1.0
     model: str = "openai/gpt-5.4-nano"
     sampling: SamplingConfig = SamplingConfig()
     prompt: Path | None = None

verifiers/v1/envs/agentic_judge/env.py

@@ -14,14 +14,18 @@
 """
 
 import json
-import math
 import re
-import tomllib
 from pathlib import Path
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import FiniteFloat
 
 import verifiers.v1 as vf
+from verifiers.v1.judges.rubric import (
+    Criterion,
+    RubricVerdicts,
+    load_criteria,
+    score_verdicts,
+)
 from verifiers.v1.utils.compile import validate_pairing
 
 VERDICT_FILE = "/tmp/verdict.json"
@@ -38,29 +42,6 @@
 {prompt}"""
 
 
-class Criterion(BaseModel):
-    """One rubric criterion — the plugged rubric judge's format, mirrored so the
-    same `criteria` files grade both judges."""
-
-    name: str
-    """Key for the criterion's metric (`judge/<name>`)."""
-    text: str
-    weight: float = 1.0
-    """The criterion's share of the reward."""
-    choices: list[str] = Field(default_factory=lambda: ["no", "yes"])
-    """Allowed answers, ordered **worst → best**: the first scores 0.0, the last 1.0, the rest
-    evenly spaced by rank. Default `["no", "yes"]` is a binary check. Needs >= 2, no duplicates."""
-
-    @field_validator("choices")
-    @classmethod
-    def _check_choices(cls, v: list[str]) -> list[str]:
-        if len(v) < 2:
-            raise ValueError(f"`choices` needs at least two options, got {v}")
-        if len(set(v)) != len(v):
-            raise ValueError(f"`choices` has duplicate options: {v}")
-        return v
-
-
 SOLVED = Criterion(
     name="solved",
     text="The task is fully solved: what the task asked for is achieved, and "
@@ -210,7 +191,7 @@ async def finalize(self, trace: vf.Trace, runtime: vf.Runtime) -> None:
                 f"the judge wrote no verdict to {VERDICT_FILE}; its final act must "
                 'be writing {"verdicts": [{"name", "reason", "verdict"}, ...]} there'
             ) from e
-        trace.info["verdict"] = json.loads(raw)
+        trace.info["verdict"] = RubricVerdicts.model_validate_json(raw).model_dump()
 
 
 class TextFile(vf.BaseConfig):
@@ -259,38 +240,16 @@ def build_hint(self) -> str | None:
     def criteria(self) -> list[Criterion]:
         if self.rubric is None:
             return [SOLVED]
-        text = self.rubric.read_text(encoding="utf-8")
-        data = (
-            tomllib.loads(text)
-            if self.rubric.suffix.lower() == ".toml"
-            else json.loads(text)
-        )
-        items = data.get("criteria", []) if isinstance(data, dict) else data
-        criteria = [Criterion.model_validate(item) for item in items]
-        if not criteria:
-            raise ValueError(f"rubric file '{self.rubric}' lists no criteria")
-        names = [criterion.name for criterion in criteria]
-        if len(set(names)) != len(names):
-            raise ValueError(
-                f"rubric file '{self.rubric}' has duplicate criterion names"
-            )
-        if bad := [c.name for c in criteria if not 0 <= c.weight < math.inf]:
-            raise ValueError(
-                f"rubric '{self.rubric}' has negative or non-finite criterion "
-                f"weights: {bad}"
-            )
-        if sum(criterion.weight for criterion in criteria) <= 0:
-            raise ValueError(f"rubric '{self.rubric}' has no positive criterion weight")
-        return criteria
+        return load_criteria(self.rubric)
 
 
 class ScoreConfig(vf.BaseConfig):
     """How the judge's verdict composes with the taskset's own rewards on the
     solver's trace. Judge-only by default."""
 
-    task_weight: float = 0.0
+    task_weight: FiniteFloat = 0.0
     """Scale applied to the taskset's own rewards; 1 keeps them next to the verdict."""
-    judge_weight: float = 1.0
+    judge_weight: FiniteFloat = 1.0
     """Weight of the judge's verdict in the solver's reward."""
 
 
@@ -363,37 +322,9 @@ async def finalize(self, task: vf.Task, episode: vf.Episode) -> None:
         if "judge" not in by_agent:
             return
         solution, verdict = by_agent["solver"], by_agent["judge"]
-        data = verdict.info.get("verdict")
-        if not isinstance(data, dict) or not isinstance(data.get("verdicts"), list):
-            raise TypeError(
-                f"no verdicts on the judge's trace (expected {VERDICT_FILE} with a "
-                '"verdicts" list)'
-            )
+        verdicts = RubricVerdicts.model_validate(verdict.info.get("verdict")).verdicts
         criteria = self.config.task.criteria()
-        by_criterion = {c.name: c for c in criteria}
-        answers: dict[str, str] = {}
-        for entry in data["verdicts"]:
-            if not isinstance(entry, dict):
-                raise TypeError(f"verdict entry {entry!r} is not an object")
-            name = str(entry.get("name"))
-            if name in answers:
-                # Contradictory duplicates must not collapse to whichever came last.
-                raise ValueError(f"judge answered criterion {name!r} more than once")
-            answers[name] = str(entry.get("verdict"))
-        if sorted(answers) != sorted(by_criterion):
-            raise ValueError(
-                f"judge verdicts name {sorted(answers)}, expected the rubric's "
-                f"{sorted(by_criterion)}"
-            )
-        scores: dict[str, float] = {}
-        for name, answer in answers.items():
-            choices = by_criterion[name].choices
-            # An off-menu answer is a judge failure, not a zero score.
-            if answer not in choices:
-                raise ValueError(
-                    f"judge answered {answer!r} for '{name}', expected one of {choices}"
-                )
-            scores[name] = choices.index(answer) / (len(choices) - 1)
+        scores = score_verdicts(verdicts, criteria, "the rubric's")
         for criterion in criteria:
             solution.record_metric(f"judge/{criterion.name}", scores[criterion.name])
         if self.config.score.task_weight != 1.0:

verifiers/v1/harnesses/rlm/harness.py

@@ -6,7 +6,7 @@
 import shlex
 from typing import Literal
 
-from pydantic import Field, model_validator
+from pydantic import Field, PositiveInt, model_validator
 
 from verifiers.v1.clients import ModelContext
 from verifiers.v1.configs.harness import HarnessConfig
@@ -38,26 +38,18 @@ class RLMHarnessConfig(HarnessConfig):
     builtin_skills: list[BuiltinSkill] = Field(default_factory=list)
     """Built-in rlm skills to enable (RLM_SKILLS), e.g. `["edit"]`; empty enables none.
     The tool set is fixed (ipython); the base `skills` field takes SKILL.md paths."""
-    summarize_at_tokens: int | tuple[int, int] | None = None
+    summarize_at_tokens: PositiveInt | tuple[PositiveInt, PositiveInt] | None = None
     """Auto-compaction threshold (RLM_SUMMARIZE_AT_TOKENS): compact the context once it grows
     past this many tokens. An int is a fixed threshold; a `(lo, hi)` pair draws a per-group
     threshold (seeded by the task index, so a task's rollouts share one draw and tasks vary).
     `None` disables auto-compaction; ints must be positive."""
 
     @model_validator(mode="after")
-    def validate_limits(self) -> "RLMHarnessConfig":
+    def validate_range(self) -> "RLMHarnessConfig":
         value = self.summarize_at_tokens
-        if isinstance(value, tuple):
-            lo, hi = value
-            if lo <= 0 or hi <= 0:
-                raise ValueError("`summarize_at_tokens` range bounds must be positive.")
-            if lo > hi:
-                raise ValueError(
-                    "`summarize_at_tokens` range must be (lo, hi) with lo <= hi."
-                )
-        elif value is not None and value <= 0:
+        if isinstance(value, tuple) and value[0] > value[1]:
             raise ValueError(
-                "`summarize_at_tokens` must be positive, or None to disable."
+                "`summarize_at_tokens` range must be (lo, hi) with lo <= hi."
             )
         return self
 

verifiers/v1/interception/server.py

@@ -30,7 +30,7 @@
 from typing import Literal
 
 from aiohttp import web
-from pydantic import TypeAdapter, ValidationError
+from pydantic import ValidationError
 from pydantic_core import PydanticSerializationError, from_json, to_json
 
 from verifiers.v1 import graph
@@ -737,7 +737,7 @@ async def handle_state_get(self, request: web.Request) -> web.Response:
         state = session.trace.state
         return web.Response(
             # TypeAdapter emits UTF-8 bytes directly, avoiding a JSON str copy in aiohttp.
-            body=TypeAdapter(type(state)).dump_json(state),
+            body=session.state_adapter.dump_json(state),
             content_type="application/json",
             charset="utf-8",
         )
@@ -768,7 +768,7 @@ async def handle_state_put(self, request: web.Request) -> web.Response:
         state_cls = type(session.trace.state)
         raw = await request.read()
         try:
-            new_state = state_cls.model_validate_json(raw)
+            new_state = session.state_adapter.validate_json(raw)
         except ValidationError as e:
             # Reject malformed, over-nested, or mismatched state before it enters the shared channel.
             logger.warning("state PUT rejected: id=%s %s", session.trace.id, e)