From ce9eb900bc6fddd9d8bc7a0bf793cf2c18b95b4b Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Fri, 31 Jul 2026 15:02:37 -0700 Subject: [PATCH 01/25] feat: build one client per rollout, share renderers process-wide Co-Authored-By: Claude Fable 5 --- docs/v1/agent.md | 17 ++-- tests/v1/conftest.py | 21 ++-- tests/v1/test_e2e.py | 2 +- verifiers/v1/agent.py | 38 +++----- verifiers/v1/cli/eval/runner.py | 8 +- verifiers/v1/clients/__init__.py | 9 +- verifiers/v1/clients/client.py | 29 +++++- verifiers/v1/clients/eval.py | 39 +++++--- verifiers/v1/clients/train.py | 97 +++++++++++++------ .../{clients/config.py => configs/client.py} | 33 ++----- verifiers/v1/env.py | 23 +---- verifiers/v1/gepa/runner.py | 13 +-- verifiers/v1/harnesses/claude_code/harness.py | 3 - verifiers/v1/interception/server.py | 6 +- verifiers/v1/judge.py | 2 +- verifiers/v1/legacy.py | 2 +- verifiers/v1/rollout.py | 26 ++++- verifiers/v1/serve/client.py | 2 +- verifiers/v1/serve/server.py | 27 ++---- verifiers/v1/serve/types.py | 2 +- verifiers/v1/session.py | 5 +- 21 files changed, 213 insertions(+), 191 deletions(-) rename verifiers/v1/{clients/config.py => configs/client.py} (74%) diff --git a/docs/v1/agent.md b/docs/v1/agent.md index 7cb42b31f0..8c254d9bf6 100644 --- a/docs/v1/agent.md +++ b/docs/v1/agent.md @@ -9,9 +9,7 @@ async with vf.make_agent(vf.AgentConfig(model="z-ai/glm-5.2")) as solver: trace = await solver.run(vf.Task(vf.TaskData(prompt="What is 2+2?"))) ``` -Every run is a standard rollout producing a `vf.Trace`. By default, the agent is self-contained: its context owns the model client and shared interception server, while each run owns its runtime and any per-run interception machinery. - -Exiting the context closes an agent-owned client, so create a new agent for later runs; injected clients remain caller-owned. +Every run is a standard rollout producing a `vf.Trace`. By default, the agent is self-contained: its context owns a shared interception server, while each rollout owns its model client, its runtime, and any per-run interception machinery. ## Interactions @@ -55,19 +53,16 @@ async with InterceptionServer() as server: ... ``` -### Client +The caller is responsible for correctly handling the lifecycle of such borrowed resources: they must be live for every run placed on them, and the agent never tears them down. -Pass `client=` to reuse an existing client — agents on the same endpoint should share a single `Client` (one connection pool). +Model clients are not borrowed: the endpoint is config (`AgentConfig.client`), and every rollout builds and closes its own `Client`. One rollout's connection pool, retries, and connection state are therefore never shared with another's, and in-flight capacity scales with the number of rollouts instead of being capped by one pool. ```python -client = vf.resolve_client(vf.EvalClientConfig()) - -solver = vf.make_agent(vf.AgentConfig(model="z-ai/glm-5.2"), client=client) -judge = vf.make_agent(vf.AgentConfig(model="openai/gpt-5.4-mini"), client=client) +solver = vf.make_agent( + vf.AgentConfig(model="z-ai/glm-5.2", client=vf.EvalClientConfig()) +) ``` -The caller is responsible for correctly handling the lifecycle of such borrowed resources: they must be live for every run placed on them, and the agent never tears them down. - ## Trace A `Trace` holds all information on a single agent's rollout: the message graph, model calls, usage, timing, the rewards, metrics, and errors it recorded. Whatever a run did, the trace is the artifact you store, chain, and train on. diff --git a/tests/v1/conftest.py b/tests/v1/conftest.py index 0e8f4a2ae8..c629be897a 100644 --- a/tests/v1/conftest.py +++ b/tests/v1/conftest.py @@ -216,20 +216,17 @@ async def _run(taskset: str, **kwargs) -> list[Trace]: @pytest.fixture async def live_ctx(): - """A live `ModelContext` (the e2e default model + endpoint, provider-default - sampling) for driving `Agent` directly — the agent-surface counterpart of `run_v1`.""" - from verifiers.v1.clients import EvalClientConfig, ModelContext, resolve_client + """The e2e `ModelContext` (default model + endpoint config, provider-default sampling) + for driving `Agent` directly — the agent-surface counterpart of `run_v1`.""" + from verifiers.v1.clients import EvalClientConfig, ModelContext from verifiers.v1.types import SamplingConfig - client = resolve_client(EvalClientConfig()) - try: - yield ModelContext( - model="deepseek/deepseek-v4-flash", - client=client, - sampling=SamplingConfig(max_tokens=2048), - ) - finally: - await client.close() + # Endpoint config only — each rollout builds and closes its own client. + yield ModelContext( + model="deepseek/deepseek-v4-flash", + client=EvalClientConfig(), + sampling=SamplingConfig(max_tokens=2048), + ) @pytest.fixture diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index 2eb418807f..351c2702af 100644 --- a/tests/v1/test_e2e.py +++ b/tests/v1/test_e2e.py @@ -177,8 +177,8 @@ async def test_interaction(live_ctx): harness=NullHarnessConfig(id="null"), model=live_ctx.model, sampling=live_ctx.sampling, + client=live_ctx.client, ), - client=live_ctx.client, ) task = vf.Task( vf.TaskData( diff --git a/verifiers/v1/agent.py b/verifiers/v1/agent.py index 71478d88c7..89ce074239 100644 --- a/verifiers/v1/agent.py +++ b/verifiers/v1/agent.py @@ -15,10 +15,8 @@ from typing import Self from verifiers.v1.clients import ( - Client, EvalClientConfig, ModelContext, - resolve_client, ) from verifiers.v1.configs.agent import AgentConfig, TimeoutConfig from verifiers.v1.harness import Harness @@ -227,10 +225,10 @@ async def close(self) -> Trace: class Agent: """A configured harness + model + runtime policy, runnable on any task. - Built from an `AgentConfig` alone; `client=`/`interception=` inject live - resources to borrow — agents on one endpoint should share one `Client`, and a - live `Interception`'s owner keeps its lifecycle. The config's `runtime` is a - *policy*: each `run` provisions a fresh box from it, resolved + Built from an `AgentConfig` alone; `interception=` injects a live resource to + borrow — its owner keeps the lifecycle. The endpoint stays config: each rollout + builds and closes its own `Client`, so an agent holds no transport. The config's + `runtime` is a *policy*: each `run` provisions a fresh box from it, resolved per task; `run(runtime=...)` places the run into an existing box instead (borrowed boxes are never started or torn down by the run).""" @@ -238,7 +236,6 @@ def __init__( self, config: AgentConfig, *, - client: Client | None = None, interception: Interception | None = None, ) -> None: from verifiers.v1.utils.loaders import harness_config_type, load_harness @@ -258,12 +255,9 @@ def __init__( config = config.model_copy(update={"sampling": Sampling()}) self.config = config self.harness = load_harness(config.harness) - self._owns_client = client is None - if self._owns_client: - client = resolve_client(config.client or EvalClientConfig()) self.ctx = ModelContext( model=config.model, - client=client, + client=config.client or EvalClientConfig(), sampling=config.sampling, ) self._closed = False @@ -305,22 +299,14 @@ async def __aenter__(self) -> Self: # A failed __aenter__ gets no __aexit__ from `async with`: unwind # here, or the agent stays "already entered" forever. self._entered, self._server = False, None - if self._owns_client: - self._closed = True - await self.ctx.client.close() raise return self async def __aexit__(self, *exc) -> None: self._entered = False server, self._server = self._server, None - try: - if server is not None: - await server.__aexit__(*exc) - finally: - if self._owns_client: - self._closed = True - await self.ctx.client.close() + if server is not None: + await server.__aexit__(*exc) def _interception_for( self, run_is_local: bool, task: Task, shared_tools: Mapping @@ -593,7 +579,6 @@ def __init__( self, config: AgentConfig, *, - client: Client, interception: Interception | None, name: str, shared_tools: Mapping[str, SharedToolServer], @@ -604,7 +589,7 @@ def __init__( on_discard: Callable[[Trace], None] | None, warned_resources: set, ) -> None: - super().__init__(config, client=client, interception=interception) + super().__init__(config, interception=interception) # Resource warnings dedupe env-wide, not per episode. self._warned_resources = warned_resources self._name = name @@ -701,12 +686,11 @@ def remember(current: Trace) -> None: def make_agent( config: AgentConfig, *, - client: Client | None = None, interception: Interception | None = None, ) -> Agent: - """The agent for a config; `client`/`interception` inject live resources to - borrow, everything else comes from the config.""" - return Agent(config, client=client, interception=interception) + """The agent for a config; `interception` injects a live resource to borrow, + everything else comes from the config.""" + return Agent(config, interception=interception) MakeAgent = Callable[[str, AgentConfig], Agent] diff --git a/verifiers/v1/cli/eval/runner.py b/verifiers/v1/cli/eval/runner.py index c7b98ed696..73f9e96e8a 100644 --- a/verifiers/v1/cli/eval/runner.py +++ b/verifiers/v1/cli/eval/runner.py @@ -14,7 +14,7 @@ output_path, save_config, ) -from verifiers.v1.clients import ModelContext, resolve_client +from verifiers.v1.clients import ModelContext from verifiers.v1.configs.cli.eval import EvalConfig from verifiers.v1.env import Env, RunSlot from verifiers.v1.episode import Episode @@ -26,7 +26,6 @@ async def run_eval(env: Env, config: EvalConfig) -> list[Episode]: logger.info("eval config:\n%s", config.model_dump_json(indent=2)) - client = resolve_client(config.client) taskset = env.taskset if config.num_tasks is None and taskset.INFINITE: raise ValueError( @@ -36,7 +35,9 @@ async def run_eval(env: Env, config: EvalConfig) -> list[Episode]: if config.num_tasks is not None: selected = selected.head(config.num_tasks) tasks = list(selected) - ctx = ModelContext(client=client, model=config.model, sampling=config.sampling) + ctx = ModelContext( + client=config.client, model=config.model, sampling=config.sampling + ) semaphore = ( asyncio.Semaphore(config.max_concurrent) if config.max_concurrent else None ) @@ -107,7 +108,6 @@ async def on_complete(episode: Episode) -> None: push_state.started = True await asyncio.to_thread(push_traces, episodes, config, push_state) - await client.close() return episodes diff --git a/verifiers/v1/clients/__init__.py b/verifiers/v1/clients/__init__.py index ab38dc7249..0d2f8fcc19 100644 --- a/verifiers/v1/clients/__init__.py +++ b/verifiers/v1/clients/__init__.py @@ -1,13 +1,12 @@ -from verifiers.v1.clients.client import Client, ModelContext -from verifiers.v1.clients.config import ( +from verifiers.v1.clients.client import Client, ModelContext, resolve_client +from verifiers.v1.clients.eval import EvalClient +from verifiers.v1.clients.train import TrainClient +from verifiers.v1.configs.client import ( BaseClientConfig, ClientConfig, EvalClientConfig, TrainClientConfig, - resolve_client, ) -from verifiers.v1.clients.eval import EvalClient -from verifiers.v1.clients.train import TrainClient __all__ = [ "BaseClientConfig", diff --git a/verifiers/v1/clients/client.py b/verifiers/v1/clients/client.py index 5940f49f26..2b2600239e 100644 --- a/verifiers/v1/clients/client.py +++ b/verifiers/v1/clients/client.py @@ -5,6 +5,11 @@ from collections.abc import AsyncIterator, Awaitable, Callable, Mapping from dataclasses import dataclass, field +from verifiers.v1.configs.client import ( + BaseClientConfig, + ClientConfig, + TrainClientConfig, +) from verifiers.v1.dialects import Dialect from verifiers.v1.graph import PendingTurn from verifiers.v1.types import Response, Sampling, SamplingConfig @@ -78,10 +83,30 @@ async def close(self) -> None: pass +def resolve_client(config: BaseClientConfig) -> Client: + """The client for `config` — built per rollout, so each owns its own transport. + + Imported locally: both clients build themselves from a config, and importing them + here at module scope would cycle back through this module.""" + if isinstance(config, TrainClientConfig): + # The renderer calls a vLLM `/inference/v1/generate` engine through the OpenAI SDK. + from verifiers.v1.clients.train import TrainClient + + return TrainClient(config) + # The proxy is a raw httpx forwarder; the dialect supplies the auth scheme + upstream path. + from verifiers.v1.clients.eval import EvalClient + + return EvalClient(config) + + @dataclass(frozen=True) class ModelContext: - """Client, model, and sampling settings for one rollout.""" + """What a run samples with: model, sampling settings, and the endpoint. + + `client` is the endpoint *config*, not a live client — every rollout builds (and closes) + its own from it, so no transport, connection pool, or mutable client state is shared + between rollouts. The live client lives on the rollout's `RolloutSession`.""" model: str - client: Client + client: ClientConfig sampling: Sampling = field(default_factory=Sampling) diff --git a/verifiers/v1/clients/eval.py b/verifiers/v1/clients/eval.py index 2951bb23a8..42efdda7d8 100644 --- a/verifiers/v1/clients/eval.py +++ b/verifiers/v1/clients/eval.py @@ -20,6 +20,7 @@ from pydantic_core import from_json, to_json from verifiers.v1.clients.client import SESSION_ID_HEADER, Client, RelayReply +from verifiers.v1.configs.client import BaseClientConfig, resolve_api_key from verifiers.v1.dialects import Dialect from verifiers.v1.errors import model_error from verifiers.v1.graph import PendingTurn @@ -64,23 +65,35 @@ class EvalClient(Client): """Relay native JSON to the provider and parse a copy for the trace.""" - def __init__( - self, base_url: str, api_key: str, headers: dict[str, str] | None = None - ) -> None: - self.base_url = base_url.rstrip("/") - self.api_key = api_key + def __init__(self, config: BaseClientConfig) -> None: + self.base_url = config.base_url.rstrip("/") + self.api_key = resolve_api_key(config) # Keep endpoint headers separate so they can override intercepted request headers before # the dialect's provider authentication is applied. - self.headers = dict(headers or {}) + self.headers = dict(config.headers or {}) # No timeout: agentic completions are slow and the rollout timeout is the real backstop. - # Build full URLs ourselves (base_url + dialect.upstream_path) rather than relying on - # httpx base-url joining, which drops the base path for a leading-slash request path. - # Match V1's default concurrency while retaining HTTPX's 20-idle keepalive bound. + # Build full URLs ourselves (`_url`) rather than relying on httpx base-url joining, + # which drops the base path for a leading-slash request path. + # One client per rollout, so this pool serves ONE rollout: its turns are sequential, + # and the headroom covers a harness SDK retrying while the first attempt drains. + # Sizing per rollout (rather than a shared cap) makes in-flight capacity scale with + # rollout count instead of silently ceiling it. self.http = httpx.AsyncClient( timeout=None, - limits=httpx.Limits(max_connections=128, max_keepalive_connections=20), + limits=httpx.Limits(max_connections=4, max_keepalive_connections=2), ) + def _url(self, path: str) -> str: + """Join `base_url` with a dialect path without duplicating the API version segment. + An Anthropic-style absolute path (`/v1/messages`) against a base that already ends in + `/v1` would otherwise request `/v1/v1/messages`; a relative one (`/chat/completions`) + keeps the base as-is.""" + head = path.split("/")[1] if path.startswith("/") else "" + base = self.base_url + if head and base.endswith(f"/{head}"): + base = base[: -len(head) - 1] + return base + path + async def get_response( self, dialect: Dialect, @@ -92,7 +105,7 @@ async def get_response( headers: Mapping[str, str] | None = None, ) -> Response: resp = await self._request( - self.base_url + dialect.upstream_path, + self._url(dialect.upstream_path), dialect.apply_overrides(body, model, sampling_args), self._headers(dialect, headers, session_id), ) @@ -193,7 +206,7 @@ async def relay( # Relay complete SSE events so the interception server can safely insert keepalives # between them. Error responses are mapped before any event is handed back. resp = await self._request( - self.base_url + dialect.upstream_path, + self._url(dialect.upstream_path), dialect.apply_overrides(body, model, sampling_args), self._headers(dialect, headers, session_id), stream=True, @@ -228,7 +241,7 @@ async def relay_aux( ) -> dict: # A side request (e.g. count_tokens): relay its native JSON and return the provider JSON. resp = await self._request( - self.base_url + route, + self._url(route), body, self._headers(dialect, headers, None), ) diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index 785f05a923..0f998dc25e 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -8,15 +8,18 @@ needs a running vLLM engine. """ +import asyncio import json +import threading from collections.abc import Mapping from typing import Any -from openai import AsyncOpenAI, OpenAIError +from openai import OpenAIError from renderers import OverlongPromptError as RendererOverlongPromptError from renderers import RenderedTokens, RendererConfig from verifiers.v1.clients.client import SESSION_ID_HEADER, Client +from verifiers.v1.configs.client import TrainClientConfig, build_async_openai from verifiers.v1.dialects import FINISH_REASONS, ChatDialect, Dialect, parse_tools from verifiers.v1.dialects.chat import message_to_wire from verifiers.v1.errors import OverlongPromptError, model_error @@ -179,41 +182,79 @@ def _has_multimodal_content(messages) -> bool: return False +_RENDERER_SLOTS = 8 +"""Independent tokenizer copies per (model, renderer config), shared by every rollout in the +process. Sized as a constant rather than a knob: a renderer is held only for the duration of +one render call (tens of ms) while a turn takes seconds, so the rollouts rendering at any +instant are far fewer than the rollouts in flight — a handful of slots absorbs the overlap at +any concurrency. Each slot is a full tokenizer (~75-95 MB), so this is also the process's +tokenizer memory bound; sharing per rollout instead would scale it with `--max-concurrent`.""" + +_RENDERER_POOLS: dict[str, Any] = {} +_RENDERER_POOLS_LOCK = threading.Lock() + + +async def shared_renderer_pool( + renderer_model: str, + config: RendererConfig | None, + *, + chat_template_kwargs: Mapping[str, Any] | None = None, +): + """The process-wide `RendererPool` for this (model, config, template kwargs). + + Renderers carry no rollout state — the pool hands one out per render and takes it back — + so a pool is shared rather than owned by a client. Building one loads `_RENDERER_SLOTS` + tokenizers (seconds), so it happens on a thread and behind a lock: concurrent first + callers wait for one build instead of each loading a duplicate set.""" + key = json.dumps( + [ + renderer_model, + config.model_dump(mode="json") if config is not None else None, + dict(chat_template_kwargs) if chat_template_kwargs else None, + ], + sort_keys=True, + default=str, + ) + if (pool := _RENDERER_POOLS.get(key)) is not None: + return pool + + def build(): + with _RENDERER_POOLS_LOCK: + if key not in _RENDERER_POOLS: + from renderers import create_renderer_pool + + pool_kwargs: dict[str, Any] = {"size": _RENDERER_SLOTS} + if chat_template_kwargs: + pool_kwargs["chat_template_kwargs"] = chat_template_kwargs + _RENDERER_POOLS[key] = create_renderer_pool( + renderer_model, config, **pool_kwargs + ) + return _RENDERER_POOLS[key] + + return await asyncio.to_thread(build) + + class TrainClient(Client): - """Renders prompts to token ids and calls a vLLM `/inference/v1/generate` engine.""" + """Renders prompts to token ids and calls a vLLM `/inference/v1/generate` engine. - def __init__( - self, - openai: AsyncOpenAI, - pool_size: int = 1, - config: RendererConfig | None = None, - renderer_model_name: str | None = None, - ) -> None: - self.openai = openai - self.pool_size = pool_size + One client per rollout: it owns its engine connection and borrows the process-wide + renderer pool (`shared_renderer_pool`) for each render.""" + + def __init__(self, config: TrainClientConfig) -> None: self.config = config - self.renderer_model_name = renderer_model_name - self._pool = None + self.openai = build_async_openai(config) - def _renderer_pool( + async def _renderer_pool( self, model: str, *, chat_template_kwargs: Mapping[str, Any] | None = None, ): - renderer_model = self.renderer_model_name or model - if self._pool is None: - from renderers import create_renderer_pool - - pool_kwargs: dict[str, Any] = {"size": self.pool_size} - if chat_template_kwargs: - pool_kwargs["chat_template_kwargs"] = chat_template_kwargs - self._pool = create_renderer_pool( - renderer_model, - self.config, - **pool_kwargs, - ) - return self._pool + return await shared_renderer_pool( + self.config.renderer_model_name or model, + self.config.renderer, + chat_template_kwargs=chat_template_kwargs, + ) async def get_response( self, @@ -259,7 +300,7 @@ async def get_response( ) chat_template_kwargs = sampling_params.pop("chat_template_kwargs", None) sampling_params.update(raw_sampling) - renderer = self._renderer_pool( + renderer = await self._renderer_pool( model, chat_template_kwargs=chat_template_kwargs, ) diff --git a/verifiers/v1/clients/config.py b/verifiers/v1/configs/client.py similarity index 74% rename from verifiers/v1/clients/config.py rename to verifiers/v1/configs/client.py index d8783ee656..b92dbe2376 100644 --- a/verifiers/v1/clients/config.py +++ b/verifiers/v1/configs/client.py @@ -1,11 +1,12 @@ -"""Client configs: describe an OpenAI-compatible endpoint and resolve it to a Client. +"""Client configs: describe an OpenAI-compatible endpoint. A `BaseClientConfig` is an OpenAI-compatible endpoint (base_url + API-key env var -+ extra headers) that `resolve_client` turns into a `Client`. The default Prime -endpoint, API key, and team fall back to the active Prime CLI config, so direct -`uv run eval` calls behave like `prime eval`. Both the eval entrypoint (its model -client) and in-env LLM calls (e.g. a judge reward) build clients from these. -`ClientConfig` is the CLI-selectable discriminated union (eval | train). ++ extra headers); `clients.resolve_client` turns one into a live `Client`, and every +rollout does so for itself. The default Prime endpoint, API key, and team fall back to +the active Prime CLI config, so direct `uv run eval` calls behave like `prime eval`. +Both the eval entrypoint (its model client) and in-env LLM calls (e.g. a judge reward) +build clients from these. `ClientConfig` is the CLI-selectable discriminated union +(eval | train). """ import os @@ -18,9 +19,6 @@ from renderers import RendererConfig from verifiers.utils.client_utils import load_prime_config -from verifiers.v1.clients.client import Client -from verifiers.v1.clients.eval import EvalClient -from verifiers.v1.clients.train import TrainClient DEFAULT_PRIME_INFERENCE_URL = "https://api.pinference.ai/api/v1" PRIME_INFERENCE_HOST = "pinference.ai" @@ -74,8 +72,6 @@ class TrainClientConfig(BaseClientConfig): `None` auto-resolves from the model — which falls back to the default renderer (no tool support) for models not in the renderer map, so set it explicitly for fine-tunes / tool-using envs.""" - pool_size: int = 1 - """Renderer slots shared across concurrent rollouts (client-side tokenization).""" renderer_model_name: str | None = None """Model the tokenizer/renderer pool is built for. Pin to the base model so a LoRA adapter name (served only for sampling) never drives tokenizer loading. Falls back to @@ -110,18 +106,3 @@ def build_async_openai(config: BaseClientConfig) -> AsyncOpenAI: api_key=resolve_api_key(config), default_headers=config.headers or None, ) - - -def resolve_client(config: BaseClientConfig) -> Client: - if isinstance(config, TrainClientConfig): - # The renderer calls a vLLM `/inference/v1/generate` engine through the OpenAI SDK. - return TrainClient( - build_async_openai(config), - pool_size=config.pool_size, - config=config.renderer, - renderer_model_name=config.renderer_model_name, - ) - # The proxy is a raw httpx forwarder; the dialect supplies the auth scheme + upstream path. - return EvalClient( - config.base_url, resolve_api_key(config), headers=config.headers or None - ) diff --git a/verifiers/v1/env.py b/verifiers/v1/env.py index df921e0308..acc3e6c1fa 100644 --- a/verifiers/v1/env.py +++ b/verifiers/v1/env.py @@ -13,7 +13,7 @@ ) from verifiers.v1.agent import Agent, Agents, _EpisodeAgent -from verifiers.v1.clients import Client, ClientConfig, ModelContext, resolve_client +from verifiers.v1.clients import ModelContext from verifiers.v1.configs.agent import AgentConfig from verifiers.v1.configs.env import ( EnvConfig, @@ -138,8 +138,6 @@ def __init__(self, config: ConfigT) -> None: # Serving resources, live only inside `serving()`; the env's agents borrow them. self._shared_tools: dict[str, SharedToolServer] = {} self._interception: Interception | None = None - # Clients for endpoint-pinning roles, cached by config, closed with serving(). - self._agent_clients: dict[str, Client] = {} # Resource warnings dedupe env-wide (agents are per-episode). self._warned_resources: set = set() @@ -203,7 +201,9 @@ def _episode_agents( gate = asyncio.Semaphore(limit) if limit else None def make(name: str, spec: AgentConfig) -> Agent: - # Unpinned fields fall back to the run's ctx / the taskset's harness. + # Unpinned fields fall back to the run's ctx / the taskset's harness. The + # endpoint resolves as config, not a live client: each of the seat's rollouts + # builds its own from it. resolved = spec.model_copy( update={ "harness": spec.harness @@ -213,13 +213,11 @@ def make(name: str, spec: AgentConfig) -> Agent: "sampling": spec.sampling if spec.sampling is not None else ctx.sampling, + "client": spec.client if spec.client is not None else ctx.client, } ) return _EpisodeAgent( resolved, - client=self._client_for(spec.client) - if spec.client is not None - else ctx.client, interception=self._interception, name=name, shared_tools=self._shared_tools, @@ -234,13 +232,6 @@ def make(name: str, spec: AgentConfig) -> Agent: agents = Agents(self.config, make) return agents - def _client_for(self, config: ClientConfig) -> Client: - """Resolve (and cache by config) an agent-pinned endpoint's client.""" - key = config.model_dump_json() - if key not in self._agent_clients: - self._agent_clients[key] = resolve_client(config) - return self._agent_clients[key] - async def run_episode( self, task: Task, @@ -371,10 +362,6 @@ async def serving(self): finally: self._shared_tools = {} self._interception = None - clients, self._agent_clients = self._agent_clients, {} - for client in clients.values(): - with contextlib.suppress(Exception): - await client.close() def _runs_local(self) -> bool: """Whether every role's runtime policy is local (any remote role means tunnels).""" diff --git a/verifiers/v1/gepa/runner.py b/verifiers/v1/gepa/runner.py index f4139158d0..e0cfafba9c 100644 --- a/verifiers/v1/gepa/runner.py +++ b/verifiers/v1/gepa/runner.py @@ -12,7 +12,7 @@ from gepa.core.result import GEPAResult from verifiers.v1.cli.output import append_episode, output_path, save_config -from verifiers.v1.clients import ModelContext, resolve_client +from verifiers.v1.clients import ModelContext from verifiers.v1.env import Env from verifiers.v1.episode import Episode from verifiers.v1.gepa.adapter import GEPAAdapter @@ -72,12 +72,11 @@ async def on_complete(episode: Episode) -> None: if run_dir is not None: await append_episode(run_dir, episode, write_lock) - # The client opens an httpx pool at construction, so build it inside the try that closes it — - # a failure while building ctx/reflection_lm must not leak the pool. - client = None try: - client = resolve_client(config.client) - ctx = ModelContext(client=client, model=config.model, sampling=config.sampling) + # The endpoint stays config: each rollout builds and closes its own client. + ctx = ModelContext( + client=config.client, model=config.model, sampling=config.sampling + ) reflection_lm = build_reflection_lm(config) serving = env.serving() loop.run_until_complete(serving.__aenter__()) @@ -125,6 +124,4 @@ async def on_complete(episode: Episode) -> None: finally: loop.run_until_complete(serving.__aexit__(None, None, None)) finally: - if client is not None: - loop.run_until_complete(client.close()) loop.close() diff --git a/verifiers/v1/harnesses/claude_code/harness.py b/verifiers/v1/harnesses/claude_code/harness.py index 2c8d07b824..679ec398d9 100644 --- a/verifiers/v1/harnesses/claude_code/harness.py +++ b/verifiers/v1/harnesses/claude_code/harness.py @@ -63,9 +63,6 @@ async def launch( data: TaskData, ) -> ProgramResult: system_prompt, instruction = self.resolve_text_prompt(data) - if ctx.client.base_url == "https://api.pinference.ai/api/v1": - # remove the /v1 from pinference - ctx.client.base_url = ctx.client.base_url.removesuffix("/v1") env = { **self.config.resolved_env, # Claude appends /v1/messages; give it the interception root, not the model endpoint. diff --git a/verifiers/v1/interception/server.py b/verifiers/v1/interception/server.py index 32870067a2..b3582d680b 100644 --- a/verifiers/v1/interception/server.py +++ b/verifiers/v1/interception/server.py @@ -416,7 +416,7 @@ def serve(response: Response) -> web.Response: upstream_request = dialect.apply_overrides( body, session.ctx.model, session.ctx.sampling ) - call_response = await session.ctx.client.get_response( + call_response = await session.client.get_response( dialect, body, session.ctx.model, @@ -543,7 +543,7 @@ async def _stream( upstream_request = dialect.apply_overrides( body, session.ctx.model, session.ctx.sampling ) - reply = await session.ctx.client.relay( + reply = await session.client.relay( dialect, body, session.ctx.model, @@ -692,7 +692,7 @@ async def handle_aux( session.adopt(asyncio.current_task()) logger.debug("intercept aux %s: id=%s", route, session.trace.id) try: - result = await session.ctx.client.relay_aux( + result = await session.client.relay_aux( dialect, route, await request.json(), headers=request.headers ) except RolloutError as e: diff --git a/verifiers/v1/judge.py b/verifiers/v1/judge.py index 9c60708495..c1957e3909 100644 --- a/verifiers/v1/judge.py +++ b/verifiers/v1/judge.py @@ -55,7 +55,7 @@ async def correct(self, trace) -> float: from pydantic import BaseModel from typing_extensions import TypeVar -from verifiers.v1.clients.config import build_async_openai +from verifiers.v1.configs.client import build_async_openai from verifiers.v1.configs.judge import ( JudgeConfig, judge_key, diff --git a/verifiers/v1/legacy.py b/verifiers/v1/legacy.py index fbdeb835ef..9a87c72c5e 100644 --- a/verifiers/v1/legacy.py +++ b/verifiers/v1/legacy.py @@ -23,8 +23,8 @@ from pydantic import ValidationError from verifiers.v1 import graph -from verifiers.v1.clients.config import ClientConfig, TrainClientConfig from verifiers.v1.configs.agent import AgentConfig +from verifiers.v1.configs.client import ClientConfig, TrainClientConfig from verifiers.v1.episode import Episode from verifiers.v1.serve.server import EnvServer from verifiers.v1.serve.types import ( diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index c0a1794e08..46c236e458 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -25,7 +25,7 @@ from contextlib import AsyncExitStack, asynccontextmanager from dataclasses import dataclass -from verifiers.v1.clients import ModelContext +from verifiers.v1.clients import ModelContext, resolve_client from verifiers.v1.configs.agent import AgentConfig from verifiers.v1.dialects import parse_message from verifiers.v1.errors import ( @@ -164,8 +164,16 @@ def __init__( ) if on_trace is not None: on_trace(self.trace) + # This rollout's own client, closed with the rollout: its connection pool serves + # exactly one trajectory, so capacity scales with rollouts in flight and neither + # connection state nor an aborted rollout's sockets leak into anyone else's. + self.client = resolve_client(ctx.client) self._session = RolloutSession( - ctx, self.trace, discover_decorated(task, "stop"), limits or RolloutLimits() + ctx, + self.client, + self.trace, + discover_decorated(task, "stop"), + limits or RolloutLimits(), ) self._stack = AsyncExitStack() self._failed = False @@ -387,13 +395,15 @@ async def step(self, messages: Messages | None = None) -> bool: return self.ok and trace.num_turns > turns_before async def abort(self) -> None: - """Free everything this run holds — the entered servers and an owned - runtime — without finalizing or scoring: the escape path when an exception + """Free everything this run holds — the entered servers, its client, and an + owned runtime — without finalizing or scoring: the escape path when an exception (a cancellation mid-setup, a lifetime bug raised to the caller) means the driver will never reach `close()`. Safe after a partial `close()`.""" self._closed = True with contextlib.suppress(Exception): await self._stack.aclose() + with contextlib.suppress(Exception): + await self.client.close() if self.runtime is not None: with contextlib.suppress(Exception): await self.harness.cleanup(self.trace, self.runtime) @@ -472,6 +482,14 @@ async def close(self) -> Trace: logger.warning( "runtime teardown failed (rollout %s)", trace.id, exc_info=True ) + # The rollout's own transport: nothing outside it holds a reference, and + # scoring is done, so the connection pool goes with the trajectory. + try: + await self.client.close() + except Exception: + logger.warning( + "client teardown failed (rollout %s)", trace.id, exc_info=True + ) logger.info( "rollout done: id=%s task=%s reward=%.3f turns=%d stop=%s", trace.id, diff --git a/verifiers/v1/serve/client.py b/verifiers/v1/serve/client.py index 385ab35bd1..2d9bb176f6 100644 --- a/verifiers/v1/serve/client.py +++ b/verifiers/v1/serve/client.py @@ -19,7 +19,7 @@ import zmq import zmq.asyncio -from verifiers.v1.clients.config import ClientConfig +from verifiers.v1.configs.client import ClientConfig from verifiers.v1.episode import WireEpisode from verifiers.v1.serve.types import ( BaseRequest, diff --git a/verifiers/v1/serve/server.py b/verifiers/v1/serve/server.py index 4164138a0f..840e0661ec 100644 --- a/verifiers/v1/serve/server.py +++ b/verifiers/v1/serve/server.py @@ -1,5 +1,4 @@ import asyncio -import contextlib import logging import msgpack @@ -8,9 +7,8 @@ from verifiers.utils.process_utils import use_threading_tqdm_lock from verifiers.utils.serve_utils import msgpack_encoder -from verifiers.v1.clients import ModelContext, resolve_client -from verifiers.v1.clients.client import Client -from verifiers.v1.clients.config import ClientConfig +from verifiers.v1.clients import ModelContext +from verifiers.v1.configs.client import ClientConfig from verifiers.v1.configs.env import EnvConfig from verifiers.v1.serve.types import ( BaseResponse, @@ -57,10 +55,6 @@ def __init__( self.requires_group_scoring = False # This worker's episode bound (`--max-concurrent`), spanning requests. self._gate = asyncio.Semaphore(max_concurrent) if max_concurrent else None - self._clients: dict[ - tuple[str, str], Client - ] = {} # (client_config, model) -> Client - self.ctx = zmq.asyncio.Context() self.frontend = self.ctx.socket(zmq.ROUTER) self.frontend.setsockopt(zmq.ROUTER_MANDATORY, 1) @@ -102,19 +96,13 @@ def _build_task(self, task_data: dict | None) -> Task: data = self.data_cls.model_validate(task_data) return self.task_cls(data, self.env.config.taskset.task) - def _client(self, client_config: ClientConfig, model: str) -> Client: - """Cache clients because renderer initialization builds a tokenizer pool.""" - key = (client_config.model_dump_json(), model) - if key not in self._clients: - self._clients[key] = resolve_client(client_config) - return self._clients[key] - def _context( self, client_config: ClientConfig, model: str, sampling: SamplingConfig ) -> ModelContext: - return ModelContext( - client=self._client(client_config, model), model=model, sampling=sampling - ) + """The request's sampling context. No client is built or cached here — each + rollout constructs its own from `client_config` and closes it, so a request's + endpoint (and a training run's changing model) leaves nothing behind.""" + return ModelContext(client=client_config, model=model, sampling=sampling) def serving(self): """The env's serving resources, entered for the server's lifetime so they're @@ -212,9 +200,6 @@ async def run(self) -> None: finally: for task in tasks: task.cancel() - for client in self._clients.values(): - with contextlib.suppress(Exception): - await client.close() self.frontend.close() self.ctx.term() logger.info("EnvServer down: taskset=%s", self.taskset_id) diff --git a/verifiers/v1/serve/types.py b/verifiers/v1/serve/types.py index a5e23364a8..5bdd22942d 100644 --- a/verifiers/v1/serve/types.py +++ b/verifiers/v1/serve/types.py @@ -2,7 +2,7 @@ from pydantic import BaseModel, Field, SerializeAsAny, model_validator -from verifiers.v1.clients.config import ClientConfig +from verifiers.v1.configs.client import ClientConfig from verifiers.v1.episode import WireEpisode from verifiers.v1.task import WireTaskData # noqa: F401 (docstring reference) from verifiers.v1.trace import WireTrace diff --git a/verifiers/v1/session.py b/verifiers/v1/session.py index 4dac9299c7..53bc298697 100644 --- a/verifiers/v1/session.py +++ b/verifiers/v1/session.py @@ -13,7 +13,7 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING -from verifiers.v1.clients import ModelContext +from verifiers.v1.clients import Client, ModelContext from verifiers.v1.trace import Trace if TYPE_CHECKING: @@ -61,6 +61,9 @@ def reached(self, trace: Trace) -> str | None: @dataclass class RolloutSession: ctx: ModelContext + client: Client + """This rollout's own client, built and closed by the rollout — the server calls it to + serve each intercepted turn. One per rollout, so no transport is shared between them.""" trace: Trace stops: list[Callable[[Trace], Awaitable[bool]]] = field(default_factory=list) limits: RolloutLimits = field(default_factory=RolloutLimits) From 643c21f7a0396e38f98634a176d303390f70370c Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Fri, 31 Jul 2026 15:17:54 -0700 Subject: [PATCH 02/25] feat: match OpenAI SDK transport defaults, no client retries Co-Authored-By: Claude Fable 5 --- verifiers/v1/clients/client.py | 8 ------- verifiers/v1/clients/eval.py | 35 ++++++++++++++++------------- verifiers/v1/clients/train.py | 41 +++++++++++++--------------------- verifiers/v1/configs/client.py | 13 +++++++++++ verifiers/v1/legacy.py | 3 ++- verifiers/v1/rollout.py | 5 ----- verifiers/v1/session.py | 2 -- 7 files changed, 50 insertions(+), 57 deletions(-) diff --git a/verifiers/v1/clients/client.py b/verifiers/v1/clients/client.py index 2b2600239e..6925eca751 100644 --- a/verifiers/v1/clients/client.py +++ b/verifiers/v1/clients/client.py @@ -89,11 +89,9 @@ def resolve_client(config: BaseClientConfig) -> Client: Imported locally: both clients build themselves from a config, and importing them here at module scope would cycle back through this module.""" if isinstance(config, TrainClientConfig): - # The renderer calls a vLLM `/inference/v1/generate` engine through the OpenAI SDK. from verifiers.v1.clients.train import TrainClient return TrainClient(config) - # The proxy is a raw httpx forwarder; the dialect supplies the auth scheme + upstream path. from verifiers.v1.clients.eval import EvalClient return EvalClient(config) @@ -101,12 +99,6 @@ def resolve_client(config: BaseClientConfig) -> Client: @dataclass(frozen=True) class ModelContext: - """What a run samples with: model, sampling settings, and the endpoint. - - `client` is the endpoint *config*, not a live client — every rollout builds (and closes) - its own from it, so no transport, connection pool, or mutable client state is shared - between rollouts. The live client lives on the rollout's `RolloutSession`.""" - model: str client: ClientConfig sampling: Sampling = field(default_factory=Sampling) diff --git a/verifiers/v1/clients/eval.py b/verifiers/v1/clients/eval.py index 42efdda7d8..ba2f9765ff 100644 --- a/verifiers/v1/clients/eval.py +++ b/verifiers/v1/clients/eval.py @@ -20,7 +20,12 @@ from pydantic_core import from_json, to_json from verifiers.v1.clients.client import SESSION_ID_HEADER, Client, RelayReply -from verifiers.v1.configs.client import BaseClientConfig, resolve_api_key +from verifiers.v1.configs.client import ( + DEFAULT_LIMITS, + DEFAULT_TIMEOUT, + BaseClientConfig, + resolve_api_key, +) from verifiers.v1.dialects import Dialect from verifiers.v1.errors import model_error from verifiers.v1.graph import PendingTurn @@ -58,6 +63,8 @@ "signature-input", } ) + + # Atomic so one CRLF cannot backtrack into two line endings and split an event mid-field. _SSE_EVENT_END = re.compile(rb"(?>\r\n|\r|\n){2}") @@ -71,17 +78,12 @@ def __init__(self, config: BaseClientConfig) -> None: # Keep endpoint headers separate so they can override intercepted request headers before # the dialect's provider authentication is applied. self.headers = dict(config.headers or {}) - # No timeout: agentic completions are slow and the rollout timeout is the real backstop. - # Build full URLs ourselves (`_url`) rather than relying on httpx base-url joining, - # which drops the base path for a leading-slash request path. - # One client per rollout, so this pool serves ONE rollout: its turns are sequential, - # and the headroom covers a harness SDK retrying while the first attempt drains. - # Sizing per rollout (rather than a shared cap) makes in-flight capacity scale with - # rollout count instead of silently ceiling it. - self.http = httpx.AsyncClient( - timeout=None, - limits=httpx.Limits(max_connections=4, max_keepalive_connections=2), - ) + # Timeout and limits mirror the OpenAI SDK's defaults (see configs.client), so relayed + # and rendered turns behave alike and match the SDK on the harness's side. The limits + # are no longer a shared ceiling — one client per rollout means in-flight capacity + # scales with rollout count. Full URLs are built here (`_url`) rather than by httpx + # base-url joining, which drops the base path for a leading-slash request path. + self.http = httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, limits=DEFAULT_LIMITS) def _url(self, path: str) -> str: """Join `base_url` with a dialect path without duplicating the API version segment. @@ -157,6 +159,12 @@ async def _request( *, stream: bool = False, ) -> httpx.Response: + """POST `body` upstream, once. The client never retries: a failure surfaces with the + provider's own status so the harness SDK can retry 5xx/429 and not 4xx, and the + framework's own retry surfaces (`AgentConfig.retries`, the interception layer's replay + and coalescing) stay the only ones — a silent attempt here would hide a failure from + the trace and double up with theirs. An empty/HTML body (say a 404 from a base_url + missing `/v1`) keeps its text rather than becoming an information-free ProviderError.""" headers.setdefault("content-type", "application/json") request = self.http.build_request( "POST", @@ -176,9 +184,6 @@ async def _request( try: response.raise_for_status() except httpx.HTTPStatusError as e: - # relay the provider's status (and body) so the harness SDK retries 5xx/429 and not - # 4xx; an empty/HTML body (e.g. a 404 from a base_url missing `/v1`) would otherwise - # make an information-free ProviderError raise model_error( f"upstream {e.response.status_code}: {e.response.text}", status_code=e.response.status_code, diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index 0f998dc25e..f6692da476 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -182,7 +182,7 @@ def _has_multimodal_content(messages) -> bool: return False -_RENDERER_SLOTS = 8 +RENDERER_SLOTS = 8 """Independent tokenizer copies per (model, renderer config), shared by every rollout in the process. Sized as a constant rather than a knob: a renderer is held only for the duration of one render call (tens of ms) while a turn takes seconds, so the rollouts rendering at any @@ -190,7 +190,7 @@ def _has_multimodal_content(messages) -> bool: any concurrency. Each slot is a full tokenizer (~75-95 MB), so this is also the process's tokenizer memory bound; sharing per rollout instead would scale it with `--max-concurrent`.""" -_RENDERER_POOLS: dict[str, Any] = {} +_RENDERER_POOLS: dict[tuple[str, str | None, str | None], Any] = {} _RENDERER_POOLS_LOCK = threading.Lock() @@ -203,17 +203,17 @@ async def shared_renderer_pool( """The process-wide `RendererPool` for this (model, config, template kwargs). Renderers carry no rollout state — the pool hands one out per render and takes it back — - so a pool is shared rather than owned by a client. Building one loads `_RENDERER_SLOTS` + so a pool is shared rather than owned by a client. Building one loads `RENDERER_SLOTS` tokenizers (seconds), so it happens on a thread and behind a lock: concurrent first callers wait for one build instead of each loading a duplicate set.""" - key = json.dumps( - [ - renderer_model, - config.model_dump(mode="json") if config is not None else None, - dict(chat_template_kwargs) if chat_template_kwargs else None, - ], - sort_keys=True, - default=str, + # Same key shape as v0's `RendererClient._shared_pools`: renderers owns config + # resolution, so we only separate pools whose construction inputs differ. + key = ( + renderer_model, + config.model_dump_json() if config is not None else None, + json.dumps(dict(chat_template_kwargs), sort_keys=True) + if chat_template_kwargs + else None, ) if (pool := _RENDERER_POOLS.get(key)) is not None: return pool @@ -223,7 +223,7 @@ def build(): if key not in _RENDERER_POOLS: from renderers import create_renderer_pool - pool_kwargs: dict[str, Any] = {"size": _RENDERER_SLOTS} + pool_kwargs: dict[str, Any] = {"size": RENDERER_SLOTS} if chat_template_kwargs: pool_kwargs["chat_template_kwargs"] = chat_template_kwargs _RENDERER_POOLS[key] = create_renderer_pool( @@ -244,18 +244,6 @@ def __init__(self, config: TrainClientConfig) -> None: self.config = config self.openai = build_async_openai(config) - async def _renderer_pool( - self, - model: str, - *, - chat_template_kwargs: Mapping[str, Any] | None = None, - ): - return await shared_renderer_pool( - self.config.renderer_model_name or model, - self.config.renderer, - chat_template_kwargs=chat_template_kwargs, - ) - async def get_response( self, dialect: Dialect, @@ -300,8 +288,9 @@ async def get_response( ) chat_template_kwargs = sampling_params.pop("chat_template_kwargs", None) sampling_params.update(raw_sampling) - renderer = await self._renderer_pool( - model, + renderer = await shared_renderer_pool( + self.config.renderer_model_name or model, + self.config.renderer, chat_template_kwargs=chat_template_kwargs, ) bridged_turn: PendingTurn | None = None diff --git a/verifiers/v1/configs/client.py b/verifiers/v1/configs/client.py index b92dbe2376..73b409a2d1 100644 --- a/verifiers/v1/configs/client.py +++ b/verifiers/v1/configs/client.py @@ -13,6 +13,7 @@ from typing import Annotated, Literal from urllib.parse import urlparse +import httpx from openai import AsyncOpenAI from pydantic import Field, model_validator from pydantic_config import BaseConfig @@ -21,6 +22,15 @@ from verifiers.utils.client_utils import load_prime_config DEFAULT_PRIME_INFERENCE_URL = "https://api.pinference.ai/api/v1" + +# Transport settings shared by every client, mirroring the OpenAI SDK's own defaults so a +# rollout behaves the same whether its turns are relayed (eval) or rendered (train) — and +# the same as the SDK the harness itself is using on the other side of the interception. +DEFAULT_TIMEOUT = httpx.Timeout(connect=5.0, read=600.0, write=600.0, pool=600.0) +DEFAULT_LIMITS = httpx.Limits(max_connections=1000, max_keepalive_connections=100) +MAX_RETRIES = 0 +"""No client-side retries: a failed call surfaces to the harness SDK and the trace instead of +being silently reattempted, so the framework's retry surfaces stay the only ones.""" PRIME_INFERENCE_HOST = "pinference.ai" PRIME_TEAM_ID_HEADER = "X-Prime-Team-ID" @@ -105,4 +115,7 @@ def build_async_openai(config: BaseClientConfig) -> AsyncOpenAI: base_url=config.base_url, api_key=resolve_api_key(config), default_headers=config.headers or None, + timeout=DEFAULT_TIMEOUT, + max_retries=MAX_RETRIES, + http_client=httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, limits=DEFAULT_LIMITS), ) diff --git a/verifiers/v1/legacy.py b/verifiers/v1/legacy.py index 9a87c72c5e..27f9c05c28 100644 --- a/verifiers/v1/legacy.py +++ b/verifiers/v1/legacy.py @@ -23,6 +23,7 @@ from pydantic import ValidationError from verifiers.v1 import graph +from verifiers.v1.clients.train import RENDERER_SLOTS from verifiers.v1.configs.agent import AgentConfig from verifiers.v1.configs.client import ClientConfig, TrainClientConfig from verifiers.v1.episode import Episode @@ -406,7 +407,7 @@ def _v0_client(self, client_config: ClientConfig, model: str): client_type="renderer", renderer_config=client_config.renderer, renderer_model_name=renderer_model, - renderer_pool_size=client_config.pool_size, + renderer_pool_size=RENDERER_SLOTS, api_base_url=client_config.base_url, api_key_var=client_config.api_key_var, extra_headers=dict(client_config.headers or {}), diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index 46c236e458..10c8a5b7f9 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -164,9 +164,6 @@ def __init__( ) if on_trace is not None: on_trace(self.trace) - # This rollout's own client, closed with the rollout: its connection pool serves - # exactly one trajectory, so capacity scales with rollouts in flight and neither - # connection state nor an aborted rollout's sockets leak into anyone else's. self.client = resolve_client(ctx.client) self._session = RolloutSession( ctx, @@ -482,8 +479,6 @@ async def close(self) -> Trace: logger.warning( "runtime teardown failed (rollout %s)", trace.id, exc_info=True ) - # The rollout's own transport: nothing outside it holds a reference, and - # scoring is done, so the connection pool goes with the trajectory. try: await self.client.close() except Exception: diff --git a/verifiers/v1/session.py b/verifiers/v1/session.py index 53bc298697..3aa65f56ba 100644 --- a/verifiers/v1/session.py +++ b/verifiers/v1/session.py @@ -62,8 +62,6 @@ def reached(self, trace: Trace) -> str | None: class RolloutSession: ctx: ModelContext client: Client - """This rollout's own client, built and closed by the rollout — the server calls it to - serve each intercepted turn. One per rollout, so no transport is shared between them.""" trace: Trace stops: list[Callable[[Trace], Awaitable[bool]]] = field(default_factory=list) limits: RolloutLimits = field(default_factory=RolloutLimits) From 52a82721faa026e11c5986ba40c5a2a484a41203 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Fri, 31 Jul 2026 23:10:19 +0000 Subject: [PATCH 03/25] feat: elastic renderer pool, sized by a multiplex knob The fixed 8-tokenizer pool was built lazily behind a threading lock, one asyncio.to_thread per caller: every rollout arriving before the build queued a task on the default executor and blocked there. At 2048 concurrent rollouts the first quartile of model calls averaged 23.4s against 3.6s for the rest, and the convoy starved the same executor episode writes use. ElasticRendererPool warms one tokenizer when a client is built and grows one per `multiplex` concurrent rollouts, mirroring ElasticInterceptionPool. Each slot is a size=1 RendererPool, so the lock that makes concurrent renders safe (and the thread offload) comes from renderers itself. The v0 bridge keeps its own static pool at the v0 default of 1, as on main. Co-Authored-By: Claude Opus 5 --- verifiers/v1/clients/client.py | 4 - verifiers/v1/clients/eval.py | 14 +- verifiers/v1/clients/train.py | 304 ++++++++++++++++++++++----------- verifiers/v1/configs/client.py | 6 + verifiers/v1/legacy.py | 2 - 5 files changed, 210 insertions(+), 120 deletions(-) diff --git a/verifiers/v1/clients/client.py b/verifiers/v1/clients/client.py index 6925eca751..faf6e644ef 100644 --- a/verifiers/v1/clients/client.py +++ b/verifiers/v1/clients/client.py @@ -84,10 +84,6 @@ async def close(self) -> None: def resolve_client(config: BaseClientConfig) -> Client: - """The client for `config` — built per rollout, so each owns its own transport. - - Imported locally: both clients build themselves from a config, and importing them - here at module scope would cycle back through this module.""" if isinstance(config, TrainClientConfig): from verifiers.v1.clients.train import TrainClient diff --git a/verifiers/v1/clients/eval.py b/verifiers/v1/clients/eval.py index ba2f9765ff..8770d3c8f9 100644 --- a/verifiers/v1/clients/eval.py +++ b/verifiers/v1/clients/eval.py @@ -83,7 +83,7 @@ def __init__(self, config: BaseClientConfig) -> None: # are no longer a shared ceiling — one client per rollout means in-flight capacity # scales with rollout count. Full URLs are built here (`_url`) rather than by httpx # base-url joining, which drops the base path for a leading-slash request path. - self.http = httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, limits=DEFAULT_LIMITS) + self.client = httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, limits=DEFAULT_LIMITS) def _url(self, path: str) -> str: """Join `base_url` with a dialect path without duplicating the API version segment. @@ -159,21 +159,15 @@ async def _request( *, stream: bool = False, ) -> httpx.Response: - """POST `body` upstream, once. The client never retries: a failure surfaces with the - provider's own status so the harness SDK can retry 5xx/429 and not 4xx, and the - framework's own retry surfaces (`AgentConfig.retries`, the interception layer's replay - and coalescing) stay the only ones — a silent attempt here would hide a failure from - the trace and double up with theirs. An empty/HTML body (say a 404 from a base_url - missing `/v1`) keeps its text rather than becoming an information-free ProviderError.""" headers.setdefault("content-type", "application/json") - request = self.http.build_request( + request = self.client.build_request( "POST", url, content=to_json(body, inf_nan_mode="null"), headers=headers, ) try: - response = await self.http.send(request, stream=stream) + response = await self.client.send(request, stream=stream) except httpx.TimeoutException as e: raise model_error(str(e), status_code=504) from e except httpx.HTTPError as e: @@ -253,4 +247,4 @@ async def relay_aux( return from_json(resp.content) async def close(self) -> None: - await self.http.aclose() + await self.client.aclose() diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index f6692da476..8012e3cd60 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -9,9 +9,12 @@ """ import asyncio +import contextlib import json -import threading -from collections.abc import Mapping +import logging +from collections.abc import AsyncIterator, Mapping +from contextlib import asynccontextmanager +from dataclasses import dataclass from typing import Any from openai import OpenAIError @@ -36,6 +39,8 @@ Usage, ) +logger = logging.getLogger(__name__) + def tool_to_wire(tool: Tool) -> dict: function: dict = { @@ -182,67 +187,156 @@ def _has_multimodal_content(messages) -> bool: return False -RENDERER_SLOTS = 8 -"""Independent tokenizer copies per (model, renderer config), shared by every rollout in the -process. Sized as a constant rather than a knob: a renderer is held only for the duration of -one render call (tens of ms) while a turn takes seconds, so the rollouts rendering at any -instant are far fewer than the rollouts in flight — a handful of slots absorbs the overlap at -any concurrency. Each slot is a full tokenizer (~75-95 MB), so this is also the process's -tokenizer memory bound; sharing per rollout instead would scale it with `--max-concurrent`.""" - -_RENDERER_POOLS: dict[tuple[str, str | None, str | None], Any] = {} -_RENDERER_POOLS_LOCK = threading.Lock() - - -async def shared_renderer_pool( - renderer_model: str, - config: RendererConfig | None, - *, - chat_template_kwargs: Mapping[str, Any] | None = None, -): - """The process-wide `RendererPool` for this (model, config, template kwargs). - - Renderers carry no rollout state — the pool hands one out per render and takes it back — - so a pool is shared rather than owned by a client. Building one loads `RENDERER_SLOTS` - tokenizers (seconds), so it happens on a thread and behind a lock: concurrent first - callers wait for one build instead of each loading a duplicate set.""" - # Same key shape as v0's `RendererClient._shared_pools`: renderers owns config - # resolution, so we only separate pools whose construction inputs differ. - key = ( - renderer_model, - config.model_dump_json() if config is not None else None, - json.dumps(dict(chat_template_kwargs), sort_keys=True) - if chat_template_kwargs - else None, - ) - if (pool := _RENDERER_POOLS.get(key)) is not None: - return pool +@dataclass +class _RendererSlot: + """One tokenizer, and the rollouts currently sharing it. A `size=1` pool rather than a + bare renderer: it carries the lock that makes concurrent renders on one tokenizer safe, + and `renderers` offloads its work to a thread only for a pool.""" - def build(): - with _RENDERER_POOLS_LOCK: - if key not in _RENDERER_POOLS: - from renderers import create_renderer_pool + renderer: Any + load: int = 0 - pool_kwargs: dict[str, Any] = {"size": RENDERER_SLOTS} - if chat_template_kwargs: - pool_kwargs["chat_template_kwargs"] = chat_template_kwargs - _RENDERER_POOLS[key] = create_renderer_pool( - renderer_model, config, **pool_kwargs - ) - return _RENDERER_POOLS[key] - return await asyncio.to_thread(build) +class ElasticRendererPool: + """Renderers grown on demand: one warmed up front, then `multiplex` rollouts per + tokenizer — the renderer-side counterpart to `ElasticInterceptionPool`. + + Sizing a pool up front means paying for tokenizers a run may never need while every + rollout that arrives before the build waits on all of them. Warming one and growing + from there costs a single tokenizer at startup, and only a run that actually reaches + `multiplex` concurrent rollouts pays for a second. + + Renderers carry no rollout state, so a pool is keyed by what builds it — `shared` hands + every client with the same (model, config, template kwargs, multiplex) the same pool. + With one client per rollout, owning one each would put a tokenizer behind every rollout.""" + + _shared: dict[tuple, "ElasticRendererPool"] = {} + + def __init__( + self, + renderer_model: str, + config: RendererConfig | None, + *, + chat_template_kwargs: Mapping[str, Any] | None = None, + multiplex: int, + ) -> None: + self.renderer_model = renderer_model + self.config = config + self.chat_template_kwargs = chat_template_kwargs + self.multiplex = multiplex + self.slots: list[_RendererSlot] = [] + self._lock = asyncio.Lock() + self._warm_task: asyncio.Task[_RendererSlot] | None = None + + @classmethod + def shared( + cls, + renderer_model: str, + config: RendererConfig | None, + *, + chat_template_kwargs: Mapping[str, Any] | None = None, + multiplex: int, + ) -> "ElasticRendererPool": + """The process-wide pool for these construction inputs, warming its first renderer + on the way. Same key shape as v0's `RendererClient._shared_pools`: renderers owns + config resolution, so only pools whose build inputs differ are kept apart.""" + key = ( + renderer_model, + config.model_dump_json() if config is not None else None, + json.dumps(dict(chat_template_kwargs), sort_keys=True) + if chat_template_kwargs + else None, + multiplex, + ) + pool = cls._shared.get(key) + if pool is None: + pool = cls._shared[key] = cls( + renderer_model, + config, + chat_template_kwargs=chat_template_kwargs, + multiplex=multiplex, + ) + pool.warm() + return pool + + def warm(self) -> None: + """Start building the first renderer, if nothing has yet. Called when a client is + built so the tokenizer loads while the rollout is still provisioning, rather than + in front of its first turn. A no-op off the event loop (tests, sync construction) — + `acquire` builds on demand anyway.""" + if self._warm_task is not None or self.slots: + return + try: + self._warm_task = asyncio.get_running_loop().create_task(self._grow()) + except RuntimeError: + pass + + async def _grow(self) -> _RendererSlot: + """Load one more tokenizer, on a thread — `create_renderer_pool` is seconds of + blocking work. Callers hold `_lock`, so exactly one grows at a time.""" + from renderers import create_renderer_pool + + kwargs: dict[str, Any] = {"size": 1} + if self.chat_template_kwargs: + kwargs["chat_template_kwargs"] = self.chat_template_kwargs + renderer = await asyncio.to_thread( + create_renderer_pool, self.renderer_model, self.config, **kwargs + ) + slot = _RendererSlot(renderer) + self.slots.append(slot) + logger.info( + "renderer pool: %d renderer(s), multiplex=%d", + len(self.slots), + self.multiplex, + ) + return slot + + @asynccontextmanager + async def acquire(self) -> AsyncIterator[Any]: + """A renderer to render this turn with, growing the pool when every one already + carries `multiplex` rollouts. The slot is held for the turn, so `load` counts + rollouts in flight rather than renders in progress.""" + if self._warm_task is not None: + # Shielded: a cancelled acquire must not cancel the build every other rollout + # is waiting on. A failed warm falls through to growing under the lock, where + # the error reaches a caller instead of vanishing into a stray task. + with contextlib.suppress(Exception): + await asyncio.shield(self._warm_task) + self._warm_task = None + async with self._lock: + slot = next((s for s in self.slots if s.load < self.multiplex), None) + if slot is None: + slot = await self._grow() + slot.load += 1 + try: + yield slot.renderer + finally: + slot.load -= 1 class TrainClient(Client): """Renders prompts to token ids and calls a vLLM `/inference/v1/generate` engine. - One client per rollout: it owns its engine connection and borrows the process-wide - renderer pool (`shared_renderer_pool`) for each render.""" + One client per rollout: it owns its engine connection and takes a slot on the elastic + renderer pool for each turn. Building the client warms the pool's first tokenizer, so + the load happens while the rollout provisions rather than in front of its first turn. + The pool itself is shared across clients — see `ElasticRendererPool`.""" def __init__(self, config: TrainClientConfig) -> None: self.config = config - self.openai = build_async_openai(config) + self.client = build_async_openai(config) + # The per-request model is only known at call time; a config that pins the renderer + # model can warm now, which is every training run (prime-rl always pins it). + if config.renderer_model_name is not None: + self._pool_for(config.renderer_model_name) + + def _pool_for(self, renderer_model: str, chat_template_kwargs=None): + return ElasticRendererPool.shared( + renderer_model, + self.config.renderer, + chat_template_kwargs=chat_template_kwargs, + multiplex=self.config.multiplex, + ) async def get_response( self, @@ -288,64 +382,66 @@ async def get_response( ) chat_template_kwargs = sampling_params.pop("chat_template_kwargs", None) sampling_params.update(raw_sampling) - renderer = await shared_renderer_pool( + pool = self._pool_for( self.config.renderer_model_name or model, - self.config.renderer, chat_template_kwargs=chat_template_kwargs, ) bridged_turn: PendingTurn | None = None - # Only build the (O(context)) previous-turn token ids once the cheap guards pass — a - # multimodal prompt or a tail that isn't a clean `[tool*, user?]` extension can't bridge. - can_bridge = ( - turn is not None - and not _has_multimodal_content(prompt) - and _is_valid_incremental_tail(wire_messages) - ) - previous_ids = turn.previous_token_ids() if can_bridge else None - if previous_ids is not None: - previous_prompt_ids, previous_completion_ids = previous_ids - - def bridge(): - return renderer.bridge_to_next_turn( - previous_prompt_ids, - previous_completion_ids, - wire_messages, + async with pool.acquire() as renderer: + # Only build the (O(context)) previous-turn token ids once the cheap guards pass — a + # multimodal prompt or a tail that isn't a clean `[tool*, user?]` extension can't bridge. + can_bridge = ( + turn is not None + and not _has_multimodal_content(prompt) + and _is_valid_incremental_tail(wire_messages) + ) + previous_ids = turn.previous_token_ids() if can_bridge else None + if previous_ids is not None: + previous_prompt_ids, previous_completion_ids = previous_ids + + def bridge(): + return renderer.bridge_to_next_turn( + previous_prompt_ids, + previous_completion_ids, + wire_messages, + tools=wire_tools, + ) + + bridged = await _maybe_offload(renderer, bridge) + if bridged is not None: + prompt_ids = bridged.token_ids + multi_modal_data = bridged.multi_modal_data + prompt_attribution = bridged + bridged_turn = turn + sampling_params["routed_experts_prompt_start"] = max( + len(previous_prompt_ids) + len(previous_completion_ids) - 1, + 0, + ) + + # Bridged prompt ids bypass rendering; only fallback needs the full wire prompt. + if prompt_ids is None: + wire_messages = [message_to_wire(m) for m in prompt] + + try: + result = await generate( + client=self.client, + renderer=renderer, + messages=wire_messages, + model=model, + prompt_ids=prompt_ids, + multi_modal_data=multi_modal_data, + prompt_attribution=prompt_attribution, tools=wire_tools, + sampling_params=sampling_params, + extra_headers={SESSION_ID_HEADER: session_id} + if session_id + else None, ) - - bridged = await _maybe_offload(renderer, bridge) - if bridged is not None: - prompt_ids = bridged.token_ids - multi_modal_data = bridged.multi_modal_data - prompt_attribution = bridged - bridged_turn = turn - sampling_params["routed_experts_prompt_start"] = max( - len(previous_prompt_ids) + len(previous_completion_ids) - 1, - 0, - ) - - # Bridged prompt ids bypass rendering; only fallback needs the full wire prompt. - if prompt_ids is None: - wire_messages = [message_to_wire(m) for m in prompt] - - try: - result = await generate( - client=self.openai, - renderer=renderer, - messages=wire_messages, - model=model, - prompt_ids=prompt_ids, - multi_modal_data=multi_modal_data, - prompt_attribution=prompt_attribution, - tools=wire_tools, - sampling_params=sampling_params, - extra_headers={SESSION_ID_HEADER: session_id} if session_id else None, - ) - except RendererOverlongPromptError as e: - raise OverlongPromptError(str(e)) from e - except OpenAIError as e: - raise model_error(e) from e + except RendererOverlongPromptError as e: + raise OverlongPromptError(str(e)) from e + except OpenAIError as e: + raise model_error(e) from e response = response_from_generate(result, model, bridged_turn) # No provider response to relay (we generated), so serialize one for the program; the # interception server hands `Response.raw` back regardless of client. @@ -353,4 +449,4 @@ def bridge(): return response async def close(self) -> None: - await self.openai.close() + await self.client.close() diff --git a/verifiers/v1/configs/client.py b/verifiers/v1/configs/client.py index 73b409a2d1..0e7127a9de 100644 --- a/verifiers/v1/configs/client.py +++ b/verifiers/v1/configs/client.py @@ -86,6 +86,12 @@ class TrainClientConfig(BaseClientConfig): """Model the tokenizer/renderer pool is built for. Pin to the base model so a LoRA adapter name (served only for sampling) never drives tokenizer loading. Falls back to the per-request model when None.""" + multiplex: int = Field(256, ge=1) + """Rollouts that share one renderer. The pool warms one and grows on demand, so N + concurrent rollouts hold ~N/multiplex tokenizers instead of a fixed set. A renderer is + held only for the render itself (milliseconds) while a turn takes seconds, so one + absorbs many rollouts; the default keeps 2048 concurrent rollouts at 8 tokenizers. + Lower it when rendering is the slow part (very long prompts), at ~75-95 MB each.""" # Discriminated union for a CLI-selectable client (`--client.type eval|train`). diff --git a/verifiers/v1/legacy.py b/verifiers/v1/legacy.py index 27f9c05c28..8a792c005f 100644 --- a/verifiers/v1/legacy.py +++ b/verifiers/v1/legacy.py @@ -23,7 +23,6 @@ from pydantic import ValidationError from verifiers.v1 import graph -from verifiers.v1.clients.train import RENDERER_SLOTS from verifiers.v1.configs.agent import AgentConfig from verifiers.v1.configs.client import ClientConfig, TrainClientConfig from verifiers.v1.episode import Episode @@ -407,7 +406,6 @@ def _v0_client(self, client_config: ClientConfig, model: str): client_type="renderer", renderer_config=client_config.renderer, renderer_model_name=renderer_model, - renderer_pool_size=RENDERER_SLOTS, api_base_url=client_config.base_url, api_key_var=client_config.api_key_var, extra_headers=dict(client_config.headers or {}), From fcdc59cd3a842cd96e1a825511d345828be91d21 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 01:59:28 +0000 Subject: [PATCH 04/25] docs: drop the model-client paragraph from Borrowed Resources Co-Authored-By: Claude Opus 5 --- docs/v1/agent.md | 2 -- 1 file changed, 2 deletions(-) diff --git a/docs/v1/agent.md b/docs/v1/agent.md index 600e89d47f..c92ab7105d 100644 --- a/docs/v1/agent.md +++ b/docs/v1/agent.md @@ -56,8 +56,6 @@ async with InterceptionServer() as server: The caller is responsible for correctly handling the lifecycle of such borrowed resources: they must be live for every run placed on them, and the agent never tears them down. -Model clients are not borrowed: the endpoint is config (`AgentConfig.client`), and every rollout builds and closes its own `Client`. One rollout's connection pool, retries, and connection state are therefore never shared with another's, and in-flight capacity scales with the number of rollouts instead of being capped by one pool. - ```python solver = vf.make_agent( vf.AgentConfig(model="z-ai/glm-5.2", client=vf.EvalClientConfig()) From 2d80d69426fb4b0777ffa472034eaa3055ae023e Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 02:00:43 +0000 Subject: [PATCH 05/25] chore: drop transport-settings comment from EvalClient Co-Authored-By: Claude Opus 5 --- verifiers/v1/clients/eval.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/verifiers/v1/clients/eval.py b/verifiers/v1/clients/eval.py index 8770d3c8f9..2dbf374a4e 100644 --- a/verifiers/v1/clients/eval.py +++ b/verifiers/v1/clients/eval.py @@ -78,11 +78,6 @@ def __init__(self, config: BaseClientConfig) -> None: # Keep endpoint headers separate so they can override intercepted request headers before # the dialect's provider authentication is applied. self.headers = dict(config.headers or {}) - # Timeout and limits mirror the OpenAI SDK's defaults (see configs.client), so relayed - # and rendered turns behave alike and match the SDK on the harness's side. The limits - # are no longer a shared ceiling — one client per rollout means in-flight capacity - # scales with rollout count. Full URLs are built here (`_url`) rather than by httpx - # base-url joining, which drops the base path for a leading-slash request path. self.client = httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, limits=DEFAULT_LIMITS) def _url(self, path: str) -> str: From 13e0def9758e3cc06415842545425cbefe362aa0 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 02:10:40 +0000 Subject: [PATCH 06/25] fix: dedup only version-shaped segments in the upstream URL join MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The join stripped any leading path segment the base happened to end with, so a base ending in `/chat` would swallow `/chat/completions`. The invariant is narrower — don't repeat the API version — so gate the dedup on a `v\d+` segment. Verified live: pinference serves Anthropic messages at `/api/v1/messages` and 404s `/api/v1/v1/messages`; bare-origin bases (`https://api.anthropic.com`) join unchanged. Co-Authored-By: Claude Opus 5 --- verifiers/v1/clients/eval.py | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/verifiers/v1/clients/eval.py b/verifiers/v1/clients/eval.py index 2dbf374a4e..64fa810623 100644 --- a/verifiers/v1/clients/eval.py +++ b/verifiers/v1/clients/eval.py @@ -68,6 +68,9 @@ # Atomic so one CRLF cannot backtrack into two line endings and split an event mid-field. _SSE_EVENT_END = re.compile(rb"(?>\r\n|\r|\n){2}") +# An API version path segment (`v1`, `v2`, ...) — the only kind `_url` dedups. +_VERSION_SEGMENT = re.compile(r"v\d+") + class EvalClient(Client): """Relay native JSON to the provider and parse a copy for the trace.""" @@ -82,12 +85,17 @@ def __init__(self, config: BaseClientConfig) -> None: def _url(self, path: str) -> str: """Join `base_url` with a dialect path without duplicating the API version segment. - An Anthropic-style absolute path (`/v1/messages`) against a base that already ends in - `/v1` would otherwise request `/v1/v1/messages`; a relative one (`/chat/completions`) - keeps the base as-is.""" + + Dialect paths keep their provider's convention — Anthropic puts the version in the + path (`/v1/messages`, bare-origin base), OpenAI puts it in the base (`.../v1` + + `/chat/completions`) — while `base_url` may be either shape. The one collision is a + version-in-path dialect against a version-in-base URL (`.../api/v1` + `/v1/messages` + would request `/v1/v1/messages`), so drop the path's version segment when the base + already ends with it. Only version-shaped segments dedup: a base genuinely ending in + `/chat` must not swallow `/chat/completions`.""" head = path.split("/")[1] if path.startswith("/") else "" base = self.base_url - if head and base.endswith(f"/{head}"): + if _VERSION_SEGMENT.fullmatch(head) and base.endswith(f"/{head}"): base = base[: -len(head) - 1] return base + path From 32933a74a8ec8f4c17d275713aea95b5af837700 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 02:34:17 +0000 Subject: [PATCH 07/25] refactor: lift the upstream URL join into clients.client.join_url Co-Authored-By: Claude Opus 5 --- verifiers/v1/clients/client.py | 22 ++++++++++++++++++++++ verifiers/v1/clients/eval.py | 27 ++++----------------------- 2 files changed, 26 insertions(+), 23 deletions(-) diff --git a/verifiers/v1/clients/client.py b/verifiers/v1/clients/client.py index faf6e644ef..79c2f4db1a 100644 --- a/verifiers/v1/clients/client.py +++ b/verifiers/v1/clients/client.py @@ -1,6 +1,7 @@ """Client interfaces for model inference and relay.""" import logging +import re from abc import ABC, abstractmethod from collections.abc import AsyncIterator, Awaitable, Callable, Mapping from dataclasses import dataclass, field @@ -16,6 +17,27 @@ logger = logging.getLogger(__name__) +# An API version path segment (`v1`, `v2`, ...) — the only kind `join_url` dedups. +_VERSION_SEGMENT = re.compile(r"v\d+") + + +def join_url(base_url: str, path: str) -> str: + """Join `base_url` with a dialect path without duplicating the API version segment. + + Dialect paths keep their provider's convention — Anthropic puts the version in the + path (`/v1/messages`, bare-origin base), OpenAI puts it in the base (`.../v1` + + `/chat/completions`) — while `base_url` may be either shape. The one collision is a + version-in-path dialect against a version-in-base URL (`.../api/v1` + `/v1/messages` + would request `/v1/v1/messages`), so drop the path's version segment when the base + already ends with it. Only version-shaped segments dedup: a base genuinely ending in + `/chat` must not swallow `/chat/completions`.""" + head = path.split("/")[1] if path.startswith("/") else "" + base = base_url.rstrip("/") + if _VERSION_SEGMENT.fullmatch(head) and base.endswith(f"/{head}"): + base = base[: -len(head) - 1] + return base + path + + SESSION_ID_HEADER = "X-Session-ID" """Per-rollout routing header. Every turn of one rollout sends the same value (the trace id), so a session-affinity router (e.g. vLLM's ``consistent_hash`` policy keyed on its diff --git a/verifiers/v1/clients/eval.py b/verifiers/v1/clients/eval.py index 64fa810623..ee3cb63c1f 100644 --- a/verifiers/v1/clients/eval.py +++ b/verifiers/v1/clients/eval.py @@ -19,7 +19,7 @@ from pydantic import ValidationError from pydantic_core import from_json, to_json -from verifiers.v1.clients.client import SESSION_ID_HEADER, Client, RelayReply +from verifiers.v1.clients.client import SESSION_ID_HEADER, Client, RelayReply, join_url from verifiers.v1.configs.client import ( DEFAULT_LIMITS, DEFAULT_TIMEOUT, @@ -68,9 +68,6 @@ # Atomic so one CRLF cannot backtrack into two line endings and split an event mid-field. _SSE_EVENT_END = re.compile(rb"(?>\r\n|\r|\n){2}") -# An API version path segment (`v1`, `v2`, ...) — the only kind `_url` dedups. -_VERSION_SEGMENT = re.compile(r"v\d+") - class EvalClient(Client): """Relay native JSON to the provider and parse a copy for the trace.""" @@ -83,22 +80,6 @@ def __init__(self, config: BaseClientConfig) -> None: self.headers = dict(config.headers or {}) self.client = httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, limits=DEFAULT_LIMITS) - def _url(self, path: str) -> str: - """Join `base_url` with a dialect path without duplicating the API version segment. - - Dialect paths keep their provider's convention — Anthropic puts the version in the - path (`/v1/messages`, bare-origin base), OpenAI puts it in the base (`.../v1` + - `/chat/completions`) — while `base_url` may be either shape. The one collision is a - version-in-path dialect against a version-in-base URL (`.../api/v1` + `/v1/messages` - would request `/v1/v1/messages`), so drop the path's version segment when the base - already ends with it. Only version-shaped segments dedup: a base genuinely ending in - `/chat` must not swallow `/chat/completions`.""" - head = path.split("/")[1] if path.startswith("/") else "" - base = self.base_url - if _VERSION_SEGMENT.fullmatch(head) and base.endswith(f"/{head}"): - base = base[: -len(head) - 1] - return base + path - async def get_response( self, dialect: Dialect, @@ -110,7 +91,7 @@ async def get_response( headers: Mapping[str, str] | None = None, ) -> Response: resp = await self._request( - self._url(dialect.upstream_path), + join_url(self.base_url, dialect.upstream_path), dialect.apply_overrides(body, model, sampling_args), self._headers(dialect, headers, session_id), ) @@ -208,7 +189,7 @@ async def relay( # Relay complete SSE events so the interception server can safely insert keepalives # between them. Error responses are mapped before any event is handed back. resp = await self._request( - self._url(dialect.upstream_path), + join_url(self.base_url, dialect.upstream_path), dialect.apply_overrides(body, model, sampling_args), self._headers(dialect, headers, session_id), stream=True, @@ -243,7 +224,7 @@ async def relay_aux( ) -> dict: # A side request (e.g. count_tokens): relay its native JSON and return the provider JSON. resp = await self._request( - self._url(route), + join_url(self.base_url, route), body, self._headers(dialect, headers, None), ) From 292aeca7b2eabb4e8fd64b8e726f60a4e056cea9 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 02:35:06 +0000 Subject: [PATCH 08/25] chore: public VERSION_SEGMENT, trim the join_url docstring Co-Authored-By: Claude Opus 5 --- verifiers/v1/clients/client.py | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) diff --git a/verifiers/v1/clients/client.py b/verifiers/v1/clients/client.py index 79c2f4db1a..171ecd382e 100644 --- a/verifiers/v1/clients/client.py +++ b/verifiers/v1/clients/client.py @@ -18,22 +18,16 @@ logger = logging.getLogger(__name__) # An API version path segment (`v1`, `v2`, ...) — the only kind `join_url` dedups. -_VERSION_SEGMENT = re.compile(r"v\d+") +VERSION_SEGMENT = re.compile(r"v\d+") def join_url(base_url: str, path: str) -> str: - """Join `base_url` with a dialect path without duplicating the API version segment. - - Dialect paths keep their provider's convention — Anthropic puts the version in the - path (`/v1/messages`, bare-origin base), OpenAI puts it in the base (`.../v1` + - `/chat/completions`) — while `base_url` may be either shape. The one collision is a - version-in-path dialect against a version-in-base URL (`.../api/v1` + `/v1/messages` - would request `/v1/v1/messages`), so drop the path's version segment when the base - already ends with it. Only version-shaped segments dedup: a base genuinely ending in - `/chat` must not swallow `/chat/completions`.""" + """Join `base_url` with a dialect path without repeating the API version segment: + `.../api/v1` + `/v1/messages` -> `.../api/v1/messages`. Only version-shaped segments + dedup, so a base ending in `/chat` doesn't swallow `/chat/completions`.""" head = path.split("/")[1] if path.startswith("/") else "" base = base_url.rstrip("/") - if _VERSION_SEGMENT.fullmatch(head) and base.endswith(f"/{head}"): + if VERSION_SEGMENT.fullmatch(head) and base.endswith(f"/{head}"): base = base[: -len(head) - 1] return base + path From cedc939838f9057181ef77c3367833c7370b23c9 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 02:38:10 +0000 Subject: [PATCH 09/25] refactor: move transport settings and build_async_openai to clients.client Timeouts, limits, and the retry policy are client behavior, not endpoint schema; configs.client keeps only the config classes and API-key resolution. Co-Authored-By: Claude Opus 5 --- verifiers/v1/clients/client.py | 27 +++++++++++++++++++++++++++ verifiers/v1/clients/eval.py | 10 ++++++---- verifiers/v1/clients/train.py | 4 ++-- verifiers/v1/configs/client.py | 23 ----------------------- verifiers/v1/judge.py | 2 +- 5 files changed, 36 insertions(+), 30 deletions(-) diff --git a/verifiers/v1/clients/client.py b/verifiers/v1/clients/client.py index 171ecd382e..96892319ed 100644 --- a/verifiers/v1/clients/client.py +++ b/verifiers/v1/clients/client.py @@ -6,10 +6,14 @@ from collections.abc import AsyncIterator, Awaitable, Callable, Mapping from dataclasses import dataclass, field +import httpx +from openai import AsyncOpenAI + from verifiers.v1.configs.client import ( BaseClientConfig, ClientConfig, TrainClientConfig, + resolve_api_key, ) from verifiers.v1.dialects import Dialect from verifiers.v1.graph import PendingTurn @@ -17,6 +21,29 @@ logger = logging.getLogger(__name__) +# Transport settings shared by every client, mirroring the OpenAI SDK's own defaults so a +# rollout behaves the same whether its turns are relayed (eval) or rendered (train) — and +# the same as the SDK the harness itself is using on the other side of the interception. +DEFAULT_TIMEOUT = httpx.Timeout(connect=5.0, read=600.0, write=600.0, pool=600.0) +DEFAULT_LIMITS = httpx.Limits(max_connections=1000, max_keepalive_connections=100) +MAX_RETRIES = 0 +"""No client-side retries: a failed call surfaces to the harness SDK and the trace instead of +being silently reattempted, so the framework's retry surfaces stay the only ones.""" + + +def build_async_openai(config: BaseClientConfig) -> AsyncOpenAI: + """An `AsyncOpenAI` for `config` (resolved key + extra headers) — for in-env model calls + (e.g. a judge) and the training client's engine connection.""" + return AsyncOpenAI( + base_url=config.base_url, + api_key=resolve_api_key(config), + default_headers=config.headers or None, + timeout=DEFAULT_TIMEOUT, + max_retries=MAX_RETRIES, + http_client=httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, limits=DEFAULT_LIMITS), + ) + + # An API version path segment (`v1`, `v2`, ...) — the only kind `join_url` dedups. VERSION_SEGMENT = re.compile(r"v\d+") diff --git a/verifiers/v1/clients/eval.py b/verifiers/v1/clients/eval.py index ee3cb63c1f..eba407f9bb 100644 --- a/verifiers/v1/clients/eval.py +++ b/verifiers/v1/clients/eval.py @@ -19,13 +19,15 @@ from pydantic import ValidationError from pydantic_core import from_json, to_json -from verifiers.v1.clients.client import SESSION_ID_HEADER, Client, RelayReply, join_url -from verifiers.v1.configs.client import ( +from verifiers.v1.clients.client import ( DEFAULT_LIMITS, DEFAULT_TIMEOUT, - BaseClientConfig, - resolve_api_key, + SESSION_ID_HEADER, + Client, + RelayReply, + join_url, ) +from verifiers.v1.configs.client import BaseClientConfig, resolve_api_key from verifiers.v1.dialects import Dialect from verifiers.v1.errors import model_error from verifiers.v1.graph import PendingTurn diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index 8012e3cd60..8eb02ef0cf 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -21,8 +21,8 @@ from renderers import OverlongPromptError as RendererOverlongPromptError from renderers import RenderedTokens, RendererConfig -from verifiers.v1.clients.client import SESSION_ID_HEADER, Client -from verifiers.v1.configs.client import TrainClientConfig, build_async_openai +from verifiers.v1.clients.client import SESSION_ID_HEADER, Client, build_async_openai +from verifiers.v1.configs.client import TrainClientConfig from verifiers.v1.dialects import FINISH_REASONS, ChatDialect, Dialect, parse_tools from verifiers.v1.dialects.chat import message_to_wire from verifiers.v1.errors import OverlongPromptError, model_error diff --git a/verifiers/v1/configs/client.py b/verifiers/v1/configs/client.py index 0e7127a9de..f1211d7854 100644 --- a/verifiers/v1/configs/client.py +++ b/verifiers/v1/configs/client.py @@ -13,8 +13,6 @@ from typing import Annotated, Literal from urllib.parse import urlparse -import httpx -from openai import AsyncOpenAI from pydantic import Field, model_validator from pydantic_config import BaseConfig from renderers import RendererConfig @@ -23,14 +21,6 @@ DEFAULT_PRIME_INFERENCE_URL = "https://api.pinference.ai/api/v1" -# Transport settings shared by every client, mirroring the OpenAI SDK's own defaults so a -# rollout behaves the same whether its turns are relayed (eval) or rendered (train) — and -# the same as the SDK the harness itself is using on the other side of the interception. -DEFAULT_TIMEOUT = httpx.Timeout(connect=5.0, read=600.0, write=600.0, pool=600.0) -DEFAULT_LIMITS = httpx.Limits(max_connections=1000, max_keepalive_connections=100) -MAX_RETRIES = 0 -"""No client-side retries: a failed call surfaces to the harness SDK and the trace instead of -being silently reattempted, so the framework's retry surfaces stay the only ones.""" PRIME_INFERENCE_HOST = "pinference.ai" PRIME_TEAM_ID_HEADER = "X-Prime-Team-ID" @@ -112,16 +102,3 @@ def resolve_api_key(config: BaseClientConfig) -> str: ): api_key = load_prime_config().get("api_key") return api_key or "EMPTY" - - -def build_async_openai(config: BaseClientConfig) -> AsyncOpenAI: - """An `AsyncOpenAI` for `config` (resolved key + extra headers) — for in-env model calls - (e.g. a judge) and the training client's engine connection.""" - return AsyncOpenAI( - base_url=config.base_url, - api_key=resolve_api_key(config), - default_headers=config.headers or None, - timeout=DEFAULT_TIMEOUT, - max_retries=MAX_RETRIES, - http_client=httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, limits=DEFAULT_LIMITS), - ) diff --git a/verifiers/v1/judge.py b/verifiers/v1/judge.py index c1957e3909..645c90ec71 100644 --- a/verifiers/v1/judge.py +++ b/verifiers/v1/judge.py @@ -55,7 +55,7 @@ async def correct(self, trace) -> float: from pydantic import BaseModel from typing_extensions import TypeVar -from verifiers.v1.configs.client import build_async_openai +from verifiers.v1.clients.client import build_async_openai from verifiers.v1.configs.judge import ( JudgeConfig, judge_key, From 30c071f3d7176e266a686d5b0d3f4568a353608f Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 02:38:44 +0000 Subject: [PATCH 10/25] chore: drop seat-resolution comment in Env Co-Authored-By: Claude Opus 5 --- verifiers/v1/env.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/verifiers/v1/env.py b/verifiers/v1/env.py index acc3e6c1fa..fa2a706093 100644 --- a/verifiers/v1/env.py +++ b/verifiers/v1/env.py @@ -201,9 +201,6 @@ def _episode_agents( gate = asyncio.Semaphore(limit) if limit else None def make(name: str, spec: AgentConfig) -> Agent: - # Unpinned fields fall back to the run's ctx / the taskset's harness. The - # endpoint resolves as config, not a live client: each of the seat's rollouts - # builds its own from it. resolved = spec.model_copy( update={ "harness": spec.harness From b4914745637f6611f2b35e623e777e22e2cbeeab Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 02:46:40 +0000 Subject: [PATCH 11/25] refactor: client utils into clients.base, multiplex out of the pool key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `multiplex` is scaling policy, not a build input — keying the shared renderer pool on it would duplicate a model's tokenizers for configs that differ only there; the first client's value wins instead. Also restores one-line docstrings on ModelContext and RolloutSession.client. Co-Authored-By: Claude Opus 5 --- verifiers/v1/clients/base.py | 44 +++++++++++++++++++++++++++++++++ verifiers/v1/clients/client.py | 45 ++-------------------------------- verifiers/v1/clients/eval.py | 10 ++------ verifiers/v1/clients/train.py | 12 +++++---- verifiers/v1/judge.py | 2 +- verifiers/v1/session.py | 1 + 6 files changed, 57 insertions(+), 57 deletions(-) create mode 100644 verifiers/v1/clients/base.py diff --git a/verifiers/v1/clients/base.py b/verifiers/v1/clients/base.py new file mode 100644 index 0000000000..eefe54d4c6 --- /dev/null +++ b/verifiers/v1/clients/base.py @@ -0,0 +1,44 @@ +"""Shared client plumbing: transport defaults and URL/key utilities.""" + +import re + +import httpx +from openai import AsyncOpenAI + +from verifiers.v1.configs.client import BaseClientConfig, resolve_api_key + +# Transport settings shared by every client, mirroring the OpenAI SDK's own defaults so a +# rollout behaves the same whether its turns are relayed (eval) or rendered (train) — and +# the same as the SDK the harness itself is using on the other side of the interception. +DEFAULT_TIMEOUT = httpx.Timeout(connect=5.0, read=600.0, write=600.0, pool=600.0) +DEFAULT_LIMITS = httpx.Limits(max_connections=1000, max_keepalive_connections=100) +MAX_RETRIES = 0 +"""No client-side retries: a failed call surfaces to the harness SDK and the trace instead of +being silently reattempted, so the framework's retry surfaces stay the only ones.""" + +# An API version path segment (`v1`, `v2`, ...) — the only kind `join_url` dedups. +VERSION_SEGMENT = re.compile(r"v\d+") + + +def build_async_openai(config: BaseClientConfig) -> AsyncOpenAI: + """An `AsyncOpenAI` for `config` (resolved key + extra headers) — for in-env model calls + (e.g. a judge) and the training client's engine connection.""" + return AsyncOpenAI( + base_url=config.base_url, + api_key=resolve_api_key(config), + default_headers=config.headers or None, + timeout=DEFAULT_TIMEOUT, + max_retries=MAX_RETRIES, + http_client=httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, limits=DEFAULT_LIMITS), + ) + + +def join_url(base_url: str, path: str) -> str: + """Join `base_url` with a dialect path without repeating the API version segment: + `.../api/v1` + `/v1/messages` -> `.../api/v1/messages`. Only version-shaped segments + dedup, so a base ending in `/chat` doesn't swallow `/chat/completions`.""" + head = path.split("/")[1] if path.startswith("/") else "" + base = base_url.rstrip("/") + if VERSION_SEGMENT.fullmatch(head) and base.endswith(f"/{head}"): + base = base[: -len(head) - 1] + return base + path diff --git a/verifiers/v1/clients/client.py b/verifiers/v1/clients/client.py index 96892319ed..715348c2ca 100644 --- a/verifiers/v1/clients/client.py +++ b/verifiers/v1/clients/client.py @@ -1,19 +1,14 @@ """Client interfaces for model inference and relay.""" import logging -import re from abc import ABC, abstractmethod from collections.abc import AsyncIterator, Awaitable, Callable, Mapping from dataclasses import dataclass, field -import httpx -from openai import AsyncOpenAI - from verifiers.v1.configs.client import ( BaseClientConfig, ClientConfig, TrainClientConfig, - resolve_api_key, ) from verifiers.v1.dialects import Dialect from verifiers.v1.graph import PendingTurn @@ -21,44 +16,6 @@ logger = logging.getLogger(__name__) -# Transport settings shared by every client, mirroring the OpenAI SDK's own defaults so a -# rollout behaves the same whether its turns are relayed (eval) or rendered (train) — and -# the same as the SDK the harness itself is using on the other side of the interception. -DEFAULT_TIMEOUT = httpx.Timeout(connect=5.0, read=600.0, write=600.0, pool=600.0) -DEFAULT_LIMITS = httpx.Limits(max_connections=1000, max_keepalive_connections=100) -MAX_RETRIES = 0 -"""No client-side retries: a failed call surfaces to the harness SDK and the trace instead of -being silently reattempted, so the framework's retry surfaces stay the only ones.""" - - -def build_async_openai(config: BaseClientConfig) -> AsyncOpenAI: - """An `AsyncOpenAI` for `config` (resolved key + extra headers) — for in-env model calls - (e.g. a judge) and the training client's engine connection.""" - return AsyncOpenAI( - base_url=config.base_url, - api_key=resolve_api_key(config), - default_headers=config.headers or None, - timeout=DEFAULT_TIMEOUT, - max_retries=MAX_RETRIES, - http_client=httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, limits=DEFAULT_LIMITS), - ) - - -# An API version path segment (`v1`, `v2`, ...) — the only kind `join_url` dedups. -VERSION_SEGMENT = re.compile(r"v\d+") - - -def join_url(base_url: str, path: str) -> str: - """Join `base_url` with a dialect path without repeating the API version segment: - `.../api/v1` + `/v1/messages` -> `.../api/v1/messages`. Only version-shaped segments - dedup, so a base ending in `/chat` doesn't swallow `/chat/completions`.""" - head = path.split("/")[1] if path.startswith("/") else "" - base = base_url.rstrip("/") - if VERSION_SEGMENT.fullmatch(head) and base.endswith(f"/{head}"): - base = base[: -len(head) - 1] - return base + path - - SESSION_ID_HEADER = "X-Session-ID" """Per-rollout routing header. Every turn of one rollout sends the same value (the trace id), so a session-affinity router (e.g. vLLM's ``consistent_hash`` policy keyed on its @@ -138,6 +95,8 @@ def resolve_client(config: BaseClientConfig) -> Client: @dataclass(frozen=True) class ModelContext: + """Model, endpoint config, and sampling for one rollout.""" + model: str client: ClientConfig sampling: Sampling = field(default_factory=Sampling) diff --git a/verifiers/v1/clients/eval.py b/verifiers/v1/clients/eval.py index eba407f9bb..622706eba5 100644 --- a/verifiers/v1/clients/eval.py +++ b/verifiers/v1/clients/eval.py @@ -19,14 +19,8 @@ from pydantic import ValidationError from pydantic_core import from_json, to_json -from verifiers.v1.clients.client import ( - DEFAULT_LIMITS, - DEFAULT_TIMEOUT, - SESSION_ID_HEADER, - Client, - RelayReply, - join_url, -) +from verifiers.v1.clients.base import DEFAULT_LIMITS, DEFAULT_TIMEOUT, join_url +from verifiers.v1.clients.client import SESSION_ID_HEADER, Client, RelayReply from verifiers.v1.configs.client import BaseClientConfig, resolve_api_key from verifiers.v1.dialects import Dialect from verifiers.v1.errors import model_error diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index 8eb02ef0cf..ae9c01da5e 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -21,7 +21,8 @@ from renderers import OverlongPromptError as RendererOverlongPromptError from renderers import RenderedTokens, RendererConfig -from verifiers.v1.clients.client import SESSION_ID_HEADER, Client, build_async_openai +from verifiers.v1.clients.base import build_async_openai +from verifiers.v1.clients.client import SESSION_ID_HEADER, Client from verifiers.v1.configs.client import TrainClientConfig from verifiers.v1.dialects import FINISH_REASONS, ChatDialect, Dialect, parse_tools from verifiers.v1.dialects.chat import message_to_wire @@ -237,16 +238,17 @@ def shared( chat_template_kwargs: Mapping[str, Any] | None = None, multiplex: int, ) -> "ElasticRendererPool": - """The process-wide pool for these construction inputs, warming its first renderer - on the way. Same key shape as v0's `RendererClient._shared_pools`: renderers owns - config resolution, so only pools whose build inputs differ are kept apart.""" + """The process-wide pool for these build inputs, warming its first renderer on + the way. Same key shape as v0's `RendererClient._shared_pools`: renderers owns + config resolution, so only pools whose build inputs differ are kept apart. + `multiplex` is policy, not a build input — the first client's value wins, so + configs differing only there share one tokenizer set.""" key = ( renderer_model, config.model_dump_json() if config is not None else None, json.dumps(dict(chat_template_kwargs), sort_keys=True) if chat_template_kwargs else None, - multiplex, ) pool = cls._shared.get(key) if pool is None: diff --git a/verifiers/v1/judge.py b/verifiers/v1/judge.py index 645c90ec71..cbd0fddab4 100644 --- a/verifiers/v1/judge.py +++ b/verifiers/v1/judge.py @@ -55,7 +55,7 @@ async def correct(self, trace) -> float: from pydantic import BaseModel from typing_extensions import TypeVar -from verifiers.v1.clients.client import build_async_openai +from verifiers.v1.clients.base import build_async_openai from verifiers.v1.configs.judge import ( JudgeConfig, judge_key, diff --git a/verifiers/v1/session.py b/verifiers/v1/session.py index 3aa65f56ba..be1f33872b 100644 --- a/verifiers/v1/session.py +++ b/verifiers/v1/session.py @@ -62,6 +62,7 @@ def reached(self, trace: Trace) -> str | None: class RolloutSession: ctx: ModelContext client: Client + """The rollout's own live client, resolved from `ctx.client` and closed with the rollout.""" trace: Trace stops: list[Callable[[Trace], Awaitable[bool]]] = field(default_factory=list) limits: RolloutLimits = field(default_factory=RolloutLimits) From 69c0a5001752f44d7c5c1ffc7e90ac711b2199d0 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 02:58:08 +0000 Subject: [PATCH 12/25] fix: annotate ElasticRendererPool._shared as ClassVar (RUF012) Co-Authored-By: Claude Opus 5 --- verifiers/v1/clients/train.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index ae9c01da5e..449d403081 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -15,7 +15,7 @@ from collections.abc import AsyncIterator, Mapping from contextlib import asynccontextmanager from dataclasses import dataclass -from typing import Any +from typing import Any, ClassVar from openai import OpenAIError from renderers import OverlongPromptError as RendererOverlongPromptError @@ -211,7 +211,7 @@ class ElasticRendererPool: every client with the same (model, config, template kwargs, multiplex) the same pool. With one client per rollout, owning one each would put a tokenizer behind every rollout.""" - _shared: dict[tuple, "ElasticRendererPool"] = {} + _shared: ClassVar[dict[tuple, "ElasticRendererPool"]] = {} def __init__( self, From a0d057a0191f83e37a468ec5dbd5acae5ca53435 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 03:02:31 +0000 Subject: [PATCH 13/25] fix: recover from a warm task cancelled by a dead event loop The pool outlives event loops (process-wide _shared), but its warm task is loop-bound: an asyncio.run() that builds a TrainClient and exits before any turn leaves the task cancelled. acquire()'s suppress(Exception) doesn't catch CancelledError and skipped clearing _warm_task, so every later acquire on a fresh loop re-raised forever. Distinguish the two cancellations: a cancelled warm task falls through to rebuilding under the lock; a cancelled acquire still re-raises its own cancellation. Co-Authored-By: Claude Opus 5 --- verifiers/v1/clients/train.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index 449d403081..fefdd62ac8 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -302,8 +302,16 @@ async def acquire(self) -> AsyncIterator[Any]: # Shielded: a cancelled acquire must not cancel the build every other rollout # is waiting on. A failed warm falls through to growing under the lock, where # the error reaches a caller instead of vanishing into a stray task. - with contextlib.suppress(Exception): + try: await asyncio.shield(self._warm_task) + except asyncio.CancelledError: + # The pool outlives event loops, and a loop's shutdown cancels a warm + # task it never awaited — recover by rebuilding under the lock. Only + # re-raise when it was THIS acquire that got cancelled. + if not self._warm_task.cancelled(): + raise + except Exception: + pass self._warm_task = None async with self._lock: slot = next((s for s in self.slots if s.load < self.multiplex), None) From ae0da58dae3fbb00ace56d0f7a52a7d231d0709d Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 03:03:35 +0000 Subject: [PATCH 14/25] chore: log a failed warm at debug, drop unused contextlib import Co-Authored-By: Claude Opus 5 --- verifiers/v1/clients/train.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index fefdd62ac8..6b819a5e9f 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -9,7 +9,6 @@ """ import asyncio -import contextlib import json import logging from collections.abc import AsyncIterator, Mapping @@ -311,7 +310,9 @@ async def acquire(self) -> AsyncIterator[Any]: if not self._warm_task.cancelled(): raise except Exception: - pass + logger.debug( + "renderer warm failed - rebuilding under the lock", exc_info=True + ) self._warm_task = None async with self._lock: slot = next((s for s in self.slots if s.load < self.multiplex), None) From 0b25aa2db2580ce2e614be94f7cbece106d5f8db Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 03:10:19 +0000 Subject: [PATCH 15/25] docs: give the client example its own section MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The endpoint is config, not a borrowed resource — its example was stranded under Borrowed Resources after the client= parameter went away. Co-Authored-By: Claude Opus 5 --- docs/v1/agent.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/v1/agent.md b/docs/v1/agent.md index c92ab7105d..232b75b155 100644 --- a/docs/v1/agent.md +++ b/docs/v1/agent.md @@ -56,6 +56,10 @@ async with InterceptionServer() as server: The caller is responsible for correctly handling the lifecycle of such borrowed resources: they must be live for every run placed on them, and the agent never tears them down. +## Client + +The model endpoint is not a borrowed resource — it is config. Set `AgentConfig.client`; each rollout builds and closes its own `Client` from it. + ```python solver = vf.make_agent( vf.AgentConfig(model="z-ai/glm-5.2", client=vf.EvalClientConfig()) From f36f6cf9a4cfdda4a99c9c3a532d1df3ece656cf Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Sat, 1 Aug 2026 03:10:34 +0000 Subject: [PATCH 16/25] chore: shorten the transport-defaults comment Co-Authored-By: Claude Opus 5 --- verifiers/v1/clients/base.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/verifiers/v1/clients/base.py b/verifiers/v1/clients/base.py index eefe54d4c6..f1172c5704 100644 --- a/verifiers/v1/clients/base.py +++ b/verifiers/v1/clients/base.py @@ -7,9 +7,7 @@ from verifiers.v1.configs.client import BaseClientConfig, resolve_api_key -# Transport settings shared by every client, mirroring the OpenAI SDK's own defaults so a -# rollout behaves the same whether its turns are relayed (eval) or rendered (train) — and -# the same as the SDK the harness itself is using on the other side of the interception. +# Mirrors the OAI SDK defaults DEFAULT_TIMEOUT = httpx.Timeout(connect=5.0, read=600.0, write=600.0, pool=600.0) DEFAULT_LIMITS = httpx.Limits(max_connections=1000, max_keepalive_connections=100) MAX_RETRIES = 0 From 339fb37f59dc0496a72b9ca9a77b0fa7c569205d Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Fri, 31 Jul 2026 20:36:40 -0700 Subject: [PATCH 17/25] feat: elastic renderer pool, aligned with the interception pool Co-Authored-By: Claude Fable 5 --- verifiers/v1/clients/train.py | 57 ++++++++++++++++++++--------------- 1 file changed, 33 insertions(+), 24 deletions(-) diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index 6b819a5e9f..663da209a6 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -188,7 +188,7 @@ def _has_multimodal_content(messages) -> bool: @dataclass -class _RendererSlot: +class RendererSlot: """One tokenizer, and the rollouts currently sharing it. A `size=1` pool rather than a bare renderer: it carries the lock that makes concurrent renders on one tokenizer safe, and `renderers` offloads its work to a thread only for a pool.""" @@ -211,6 +211,11 @@ class ElasticRendererPool: With one client per rollout, owning one each would put a tokenizer behind every rollout.""" _shared: ClassVar[dict[tuple, "ElasticRendererPool"]] = {} + """Where the interception pool has an owner (the env constructs it and tears it down, + injecting it into agents), a renderer pool has none — clients are built per rollout with + no injection channel, and the pool must outlive event loops and clients alike. So sharing + is a registry on the type (v0's `RendererClient._shared_pools`, the same shape) and there + is no `start`/`stop`: a renderer is pure memory, nothing holds a socket or tunnel.""" def __init__( self, @@ -224,9 +229,9 @@ def __init__( self.config = config self.chat_template_kwargs = chat_template_kwargs self.multiplex = multiplex - self.slots: list[_RendererSlot] = [] + self.renderers: list[RendererSlot] = [] self._lock = asyncio.Lock() - self._warm_task: asyncio.Task[_RendererSlot] | None = None + self._warm_task: asyncio.Task[RendererSlot] | None = None @classmethod def shared( @@ -257,24 +262,29 @@ def shared( chat_template_kwargs=chat_template_kwargs, multiplex=multiplex, ) - pool.warm() + pool._warm() return pool - def warm(self) -> None: - """Start building the first renderer, if nothing has yet. Called when a client is - built so the tokenizer loads while the rollout is still provisioning, rather than - in front of its first turn. A no-op off the event loop (tests, sync construction) — + def _warm(self) -> None: + """Start building the first renderer, if nothing has yet — the registry-shaped + counterpart of the interception pool's `start()`, run when a client is built so + the tokenizer loads while the rollout is still provisioning rather than in front + of its first turn. A no-op off the event loop (tests, sync construction) — `acquire` builds on demand anyway.""" - if self._warm_task is not None or self.slots: + if self._warm_task is not None or self.renderers: return try: - self._warm_task = asyncio.get_running_loop().create_task(self._grow()) + self._warm_task = asyncio.get_running_loop().create_task(self._renderer()) except RuntimeError: pass - async def _grow(self) -> _RendererSlot: - """Load one more tokenizer, on a thread — `create_renderer_pool` is seconds of - blocking work. Callers hold `_lock`, so exactly one grows at a time.""" + async def _renderer(self) -> RendererSlot: + """A renderer with spare capacity — reuse one under `multiplex`, else load one + more tokenizer on a thread (`create_renderer_pool` is seconds of blocking work). + Acquires hold `_lock`; the warm task runs before they reach this path.""" + for slot in self.renderers: + if slot.load < self.multiplex: + return slot from renderers import create_renderer_pool kwargs: dict[str, Any] = {"size": 1} @@ -283,11 +293,11 @@ async def _grow(self) -> _RendererSlot: renderer = await asyncio.to_thread( create_renderer_pool, self.renderer_model, self.config, **kwargs ) - slot = _RendererSlot(renderer) - self.slots.append(slot) + slot = RendererSlot(renderer) + self.renderers.append(slot) logger.info( "renderer pool: %d renderer(s), multiplex=%d", - len(self.slots), + len(self.renderers), self.multiplex, ) return slot @@ -299,14 +309,15 @@ async def acquire(self) -> AsyncIterator[Any]: rollouts in flight rather than renders in progress.""" if self._warm_task is not None: # Shielded: a cancelled acquire must not cancel the build every other rollout - # is waiting on. A failed warm falls through to growing under the lock, where - # the error reaches a caller instead of vanishing into a stray task. + # is waiting on. A failed warm falls through to `_renderer()` under the lock, + # where the error reaches a caller instead of vanishing into a stray task. try: await asyncio.shield(self._warm_task) except asyncio.CancelledError: - # The pool outlives event loops, and a loop's shutdown cancels a warm - # task it never awaited — recover by rebuilding under the lock. Only - # re-raise when it was THIS acquire that got cancelled. + # Unlike the interception pool's warm task (cancelled by its owner's + # `stop()`), this one can outlive its event loop — a loop shutdown + # cancels it unawaited. Rebuild under the lock; re-raise only when it + # was THIS acquire that got cancelled. if not self._warm_task.cancelled(): raise except Exception: @@ -315,9 +326,7 @@ async def acquire(self) -> AsyncIterator[Any]: ) self._warm_task = None async with self._lock: - slot = next((s for s in self.slots if s.load < self.multiplex), None) - if slot is None: - slot = await self._grow() + slot = await self._renderer() slot.load += 1 try: yield slot.renderer From d3d27b1fecc8d06ce599b33263069f503c0d1068 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Fri, 31 Jul 2026 20:38:34 -0700 Subject: [PATCH 18/25] chore: no underscore methods on ElasticRendererPool Co-Authored-By: Claude Fable 5 --- verifiers/v1/clients/train.py | 29 ++++++++++++----------------- 1 file changed, 12 insertions(+), 17 deletions(-) diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index 663da209a6..82f50039e0 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -262,23 +262,18 @@ def shared( chat_template_kwargs=chat_template_kwargs, multiplex=multiplex, ) - pool._warm() + # Warm the first renderer — the registry-shaped counterpart of the interception + # pool's `start()`: a client is built while its rollout is still provisioning, so + # the tokenizer loads now rather than in front of the first turn. A no-op off the + # event loop (tests, sync construction) — `acquire` builds on demand anyway. + if pool._warm_task is None and not pool.renderers: + try: + pool._warm_task = asyncio.get_running_loop().create_task(pool.grow()) + except RuntimeError: + pass return pool - def _warm(self) -> None: - """Start building the first renderer, if nothing has yet — the registry-shaped - counterpart of the interception pool's `start()`, run when a client is built so - the tokenizer loads while the rollout is still provisioning rather than in front - of its first turn. A no-op off the event loop (tests, sync construction) — - `acquire` builds on demand anyway.""" - if self._warm_task is not None or self.renderers: - return - try: - self._warm_task = asyncio.get_running_loop().create_task(self._renderer()) - except RuntimeError: - pass - - async def _renderer(self) -> RendererSlot: + async def grow(self) -> RendererSlot: """A renderer with spare capacity — reuse one under `multiplex`, else load one more tokenizer on a thread (`create_renderer_pool` is seconds of blocking work). Acquires hold `_lock`; the warm task runs before they reach this path.""" @@ -309,7 +304,7 @@ async def acquire(self) -> AsyncIterator[Any]: rollouts in flight rather than renders in progress.""" if self._warm_task is not None: # Shielded: a cancelled acquire must not cancel the build every other rollout - # is waiting on. A failed warm falls through to `_renderer()` under the lock, + # is waiting on. A failed warm falls through to `grow()` under the lock, # where the error reaches a caller instead of vanishing into a stray task. try: await asyncio.shield(self._warm_task) @@ -326,7 +321,7 @@ async def acquire(self) -> AsyncIterator[Any]: ) self._warm_task = None async with self._lock: - slot = await self._renderer() + slot = await self.grow() slot.load += 1 try: yield slot.renderer From 62ba5899fcc3f14f7ae07e4cde4e8662e55b030a Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Mon, 3 Aug 2026 09:25:41 -0700 Subject: [PATCH 19/25] feat: bare renderers with slot-owned locking Co-Authored-By: Claude Fable 5 --- verifiers/v1/clients/base.py | 2 - verifiers/v1/clients/eval.py | 14 +----- verifiers/v1/clients/train.py | 89 +++++++++++++++++++---------------- 3 files changed, 49 insertions(+), 56 deletions(-) diff --git a/verifiers/v1/clients/base.py b/verifiers/v1/clients/base.py index f1172c5704..e5624e25af 100644 --- a/verifiers/v1/clients/base.py +++ b/verifiers/v1/clients/base.py @@ -19,8 +19,6 @@ def build_async_openai(config: BaseClientConfig) -> AsyncOpenAI: - """An `AsyncOpenAI` for `config` (resolved key + extra headers) — for in-env model calls - (e.g. a judge) and the training client's engine connection.""" return AsyncOpenAI( base_url=config.base_url, api_key=resolve_api_key(config), diff --git a/verifiers/v1/clients/eval.py b/verifiers/v1/clients/eval.py index 622706eba5..51ec652a57 100644 --- a/verifiers/v1/clients/eval.py +++ b/verifiers/v1/clients/eval.py @@ -1,16 +1,4 @@ -"""The eval client: relay the program's native request to the provider. - -`EvalClient` (the default) is a thin `httpx` forwarder: it sends the program's request body -without a typed round-trip, mutating only what the eval owns (model + sampling, via the dialect's -`apply_overrides`). Eligible end-to-end request headers are forwarded too; rollout auth, body -framing, and connection headers are replaced. The provider response is parsed into a vf -`Response` for the trace, while its full JSON object stays on `Response.raw` for the interception -server to return. - -The transport is provider-agnostic: the dialect supplies the upstream path + auth headers, so a -new wire format (incl. non-OpenAI providers like Anthropic) is just a new `Dialect` — no client -change. Endpoint config (base url, api key, billing headers) comes from the client config. -""" +"""The eval client: proxies harness-native request to the provider.""" import re from collections.abc import Mapping diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index 82f50039e0..8e009fe5de 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -1,24 +1,17 @@ -"""Renderer client: client-side tokenization via the `renderers` package. - -A drop-in alternative to the chat-completions client: instead of sending messages -as JSON text, it renders them to token ids with a HF chat template and calls a -vLLM `/inference/v1/generate` engine, so every response carries token ids + -sampling logprobs (recorded on the trace's per-turn `tokens`) for training. It -reuses the chat client's wire translation (message/tool shapes are the same), and -needs a running vLLM engine. -""" +"""Train client: intercepts + renders prompts to tokens for training using `renderers`.""" import asyncio import json import logging -from collections.abc import AsyncIterator, Mapping +import threading +from collections.abc import AsyncIterator, Callable, Mapping from contextlib import asynccontextmanager -from dataclasses import dataclass -from typing import Any, ClassVar +from dataclasses import dataclass, field +from typing import Any, ClassVar, TypeVar from openai import OpenAIError from renderers import OverlongPromptError as RendererOverlongPromptError -from renderers import RenderedTokens, RendererConfig +from renderers import RenderedTokens, Renderer, RendererConfig from verifiers.v1.clients.base import build_async_openai from verifiers.v1.clients.client import SESSION_ID_HEADER, Client @@ -41,6 +34,8 @@ logger = logging.getLogger(__name__) +T = TypeVar("T") + def tool_to_wire(tool: Tool) -> dict: function: dict = { @@ -189,26 +184,25 @@ def _has_multimodal_content(messages) -> bool: @dataclass class RendererSlot: - """One tokenizer, and the rollouts currently sharing it. A `size=1` pool rather than a - bare renderer: it carries the lock that makes concurrent renders on one tokenizer safe, - and `renderers` offloads its work to a thread only for a pool.""" + """One renderer, the rollouts currently holding it, and the lock that makes it safe: + encoding mutates a fast tokenizer's truncation/padding state, so `run` serializes the + slot's encode-side work (render, bridge) on a thread. Decode-side work (`generate`'s + response parsing) is a pure read and needs neither the lock nor the hop.""" - renderer: Any + renderer: Renderer load: int = 0 + lock: threading.Lock = field(default_factory=threading.Lock) + async def run(self, fn: Callable[[], T]) -> T: + def locked() -> T: + with self.lock: + return fn() -class ElasticRendererPool: - """Renderers grown on demand: one warmed up front, then `multiplex` rollouts per - tokenizer — the renderer-side counterpart to `ElasticInterceptionPool`. + return await asyncio.to_thread(locked) - Sizing a pool up front means paying for tokenizers a run may never need while every - rollout that arrives before the build waits on all of them. Warming one and growing - from there costs a single tokenizer at startup, and only a run that actually reaches - `multiplex` concurrent rollouts pays for a second. - Renderers carry no rollout state, so a pool is keyed by what builds it — `shared` hands - every client with the same (model, config, template kwargs, multiplex) the same pool. - With one client per rollout, owning one each would put a tokenizer behind every rollout.""" +class ElasticRendererPool: + """Process-shared renderers pool, multiplexed and auto-growing.""" _shared: ClassVar[dict[tuple, "ElasticRendererPool"]] = {} """Where the interception pool has an owner (the env constructs it and tears it down, @@ -280,15 +274,17 @@ async def grow(self) -> RendererSlot: for slot in self.renderers: if slot.load < self.multiplex: return slot - from renderers import create_renderer_pool + from renderers import create_renderer + from renderers.base import load_tokenizer + + def build(): + return create_renderer( + load_tokenizer(self.renderer_model), + self.config, + chat_template_kwargs=self.chat_template_kwargs, + ) - kwargs: dict[str, Any] = {"size": 1} - if self.chat_template_kwargs: - kwargs["chat_template_kwargs"] = self.chat_template_kwargs - renderer = await asyncio.to_thread( - create_renderer_pool, self.renderer_model, self.config, **kwargs - ) - slot = RendererSlot(renderer) + slot = RendererSlot(await asyncio.to_thread(build)) self.renderers.append(slot) logger.info( "renderer pool: %d renderer(s), multiplex=%d", @@ -298,7 +294,7 @@ async def grow(self) -> RendererSlot: return slot @asynccontextmanager - async def acquire(self) -> AsyncIterator[Any]: + async def acquire(self) -> AsyncIterator[RendererSlot]: """A renderer to render this turn with, growing the pool when every one already carries `multiplex` rollouts. The slot is held for the turn, so `load` counts rollouts in flight rather than renders in progress.""" @@ -324,7 +320,7 @@ async def acquire(self) -> AsyncIterator[Any]: slot = await self.grow() slot.load += 1 try: - yield slot.renderer + yield slot finally: slot.load -= 1 @@ -382,7 +378,7 @@ async def get_response( tools = parse_tools(body.get("tools")) else: prompt, tools = dialect.parse_request(body) - from renderers.client import _maybe_offload, generate + from renderers.client import generate wire_tools = [tool_to_wire(t) for t in tools] if tools else None wire_messages = ( @@ -403,7 +399,8 @@ async def get_response( ) bridged_turn: PendingTurn | None = None - async with pool.acquire() as renderer: + async with pool.acquire() as slot: + renderer = slot.renderer # Only build the (O(context)) previous-turn token ids once the cheap guards pass — a # multimodal prompt or a tail that isn't a clean `[tool*, user?]` extension can't bridge. can_bridge = ( @@ -423,7 +420,7 @@ def bridge(): tools=wire_tools, ) - bridged = await _maybe_offload(renderer, bridge) + bridged = await slot.run(bridge) if bridged is not None: prompt_ids = bridged.token_ids multi_modal_data = bridged.multi_modal_data @@ -434,9 +431,19 @@ def bridge(): 0, ) - # Bridged prompt ids bypass rendering; only fallback needs the full wire prompt. + # Render here (encode-side, so through the slot) rather than inside `generate`: + # handed prebuilt prompt_ids, generate's own renderer touches are decode-side + # and stop-id reads, safe on a bare renderer without lock or thread hop. if prompt_ids is None: wire_messages = [message_to_wire(m) for m in prompt] + rendered = await slot.run( + lambda: renderer.render( + wire_messages, tools=wire_tools, add_generation_prompt=True + ) + ) + prompt_ids = rendered.token_ids + multi_modal_data = rendered.multi_modal_data + prompt_attribution = rendered try: result = await generate( From 480aa7a765c39b60ab9b3acb72cf40a2db0fcf4d Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Mon, 3 Aug 2026 09:26:11 -0700 Subject: [PATCH 20/25] chore: name the pool registry _pools Co-Authored-By: Claude Fable 5 --- verifiers/v1/clients/train.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index 8e009fe5de..a14cdd6011 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -204,7 +204,7 @@ def locked() -> T: class ElasticRendererPool: """Process-shared renderers pool, multiplexed and auto-growing.""" - _shared: ClassVar[dict[tuple, "ElasticRendererPool"]] = {} + _pools: ClassVar[dict[tuple, "ElasticRendererPool"]] = {} """Where the interception pool has an owner (the env constructs it and tears it down, injecting it into agents), a renderer pool has none — clients are built per rollout with no injection channel, and the pool must outlive event loops and clients alike. So sharing @@ -248,9 +248,9 @@ def shared( if chat_template_kwargs else None, ) - pool = cls._shared.get(key) + pool = cls._pools.get(key) if pool is None: - pool = cls._shared[key] = cls( + pool = cls._pools[key] = cls( renderer_model, config, chat_template_kwargs=chat_template_kwargs, From 7780331faf32be4b79296f33dea6aeeb984b54fc Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Mon, 3 Aug 2026 09:31:23 -0700 Subject: [PATCH 21/25] feat: process-shared renderer list, pools as cheap views Co-Authored-By: Claude Fable 5 --- verifiers/v1/clients/train.py | 155 ++++++++++++++-------------------- 1 file changed, 65 insertions(+), 90 deletions(-) diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index a14cdd6011..d2a8d5dad6 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -202,14 +202,22 @@ def locked() -> T: class ElasticRendererPool: - """Process-shared renderers pool, multiplexed and auto-growing.""" - - _pools: ClassVar[dict[tuple, "ElasticRendererPool"]] = {} - """Where the interception pool has an owner (the env constructs it and tears it down, - injecting it into agents), a renderer pool has none — clients are built per rollout with - no injection channel, and the pool must outlive event loops and clients alike. So sharing - is a registry on the type (v0's `RendererClient._shared_pools`, the same shape) and there - is no `start`/`stop`: a renderer is pure memory, nothing holds a socket or tunnel.""" + """Process-shared renderers, multiplexed and auto-growing. A pool object is a cheap + per-client view: the renderers themselves are the shared state, keyed by what builds + them, so every client with the same build inputs works one list.""" + + _renderers: ClassVar[dict[tuple, list[RendererSlot]]] = {} + """The process's renderers, keyed by build inputs. Where the interception pool has an + owner (the env constructs it and tears it down, injecting it into agents), renderers + have none — clients are built per rollout with no injection channel, and the tokenizers + must outlive event loops and clients alike. So the shared state lives on the type (v0's + `RendererClient._shared_pools`, the same shape) and there is no `start`/`stop`: a + renderer is pure memory, nothing holds a socket or tunnel.""" + + _locks: ClassVar[dict[tuple, tuple[asyncio.AbstractEventLoop, asyncio.Lock]]] = {} + """Per-key single-flight lock for `grow`. An asyncio lock binds to the loop that first + awaits it while the renderers outlive loops, so each key keeps (loop, lock) and a loop + change mints a fresh lock — nothing from a dead loop can still hold it.""" def __init__( self, @@ -223,102 +231,69 @@ def __init__( self.config = config self.chat_template_kwargs = chat_template_kwargs self.multiplex = multiplex - self.renderers: list[RendererSlot] = [] - self._lock = asyncio.Lock() - self._warm_task: asyncio.Task[RendererSlot] | None = None - - @classmethod - def shared( - cls, - renderer_model: str, - config: RendererConfig | None, - *, - chat_template_kwargs: Mapping[str, Any] | None = None, - multiplex: int, - ) -> "ElasticRendererPool": - """The process-wide pool for these build inputs, warming its first renderer on - the way. Same key shape as v0's `RendererClient._shared_pools`: renderers owns - config resolution, so only pools whose build inputs differ are kept apart. - `multiplex` is policy, not a build input — the first client's value wins, so - configs differing only there share one tokenizer set.""" - key = ( + self.key = ( renderer_model, config.model_dump_json() if config is not None else None, json.dumps(dict(chat_template_kwargs), sort_keys=True) if chat_template_kwargs else None, ) - pool = cls._pools.get(key) - if pool is None: - pool = cls._pools[key] = cls( - renderer_model, - config, - chat_template_kwargs=chat_template_kwargs, - multiplex=multiplex, - ) - # Warm the first renderer — the registry-shaped counterpart of the interception - # pool's `start()`: a client is built while its rollout is still provisioning, so - # the tokenizer loads now rather than in front of the first turn. A no-op off the - # event loop (tests, sync construction) — `acquire` builds on demand anyway. - if pool._warm_task is None and not pool.renderers: - try: - pool._warm_task = asyncio.get_running_loop().create_task(pool.grow()) - except RuntimeError: - pass - return pool + self.renderers = self._renderers.setdefault(self.key, []) + + def warm(self) -> None: + """Start building the first renderer if none exists — the counterpart of the + interception pool's `start()`: a client is built while its rollout is still + provisioning, so the tokenizer loads now rather than in front of the first + turn. A no-op off the event loop (tests, sync construction) — `acquire` + builds on demand anyway.""" + if self.renderers: + return + try: + task = asyncio.get_running_loop().create_task(self.grow()) + except RuntimeError: + return + # A failed warm is not an event: acquire retries the build and surfaces the error. + task.add_done_callback(lambda t: t.cancelled() or t.exception()) async def grow(self) -> RendererSlot: """A renderer with spare capacity — reuse one under `multiplex`, else load one - more tokenizer on a thread (`create_renderer_pool` is seconds of blocking work). - Acquires hold `_lock`; the warm task runs before they reach this path.""" - for slot in self.renderers: - if slot.load < self.multiplex: - return slot - from renderers import create_renderer - from renderers.base import load_tokenizer - - def build(): - return create_renderer( - load_tokenizer(self.renderer_model), - self.config, - chat_template_kwargs=self.chat_template_kwargs, - ) + more tokenizer on a thread (`create_renderer` is seconds of blocking work). + Single-flight per key: every caller serializes on the key's lock, so concurrent + cold acquires (and warms) wait for one build instead of stacking tokenizers.""" + loop = asyncio.get_running_loop() + bound = self._locks.get(self.key) + if bound is None or bound[0] is not loop: + bound = self._locks[self.key] = (loop, asyncio.Lock()) + async with bound[1]: + for slot in self.renderers: + if slot.load < self.multiplex: + return slot + from renderers import create_renderer + from renderers.base import load_tokenizer + + def build(): + return create_renderer( + load_tokenizer(self.renderer_model), + self.config, + chat_template_kwargs=self.chat_template_kwargs, + ) - slot = RendererSlot(await asyncio.to_thread(build)) - self.renderers.append(slot) - logger.info( - "renderer pool: %d renderer(s), multiplex=%d", - len(self.renderers), - self.multiplex, - ) - return slot + slot = RendererSlot(await asyncio.to_thread(build)) + self.renderers.append(slot) + logger.info( + "renderer pool: %d renderer(s), multiplex=%d", + len(self.renderers), + self.multiplex, + ) + return slot @asynccontextmanager async def acquire(self) -> AsyncIterator[RendererSlot]: """A renderer to render this turn with, growing the pool when every one already carries `multiplex` rollouts. The slot is held for the turn, so `load` counts rollouts in flight rather than renders in progress.""" - if self._warm_task is not None: - # Shielded: a cancelled acquire must not cancel the build every other rollout - # is waiting on. A failed warm falls through to `grow()` under the lock, - # where the error reaches a caller instead of vanishing into a stray task. - try: - await asyncio.shield(self._warm_task) - except asyncio.CancelledError: - # Unlike the interception pool's warm task (cancelled by its owner's - # `stop()`), this one can outlive its event loop — a loop shutdown - # cancels it unawaited. Rebuild under the lock; re-raise only when it - # was THIS acquire that got cancelled. - if not self._warm_task.cancelled(): - raise - except Exception: - logger.debug( - "renderer warm failed - rebuilding under the lock", exc_info=True - ) - self._warm_task = None - async with self._lock: - slot = await self.grow() - slot.load += 1 + slot = await self.grow() + slot.load += 1 try: yield slot finally: @@ -339,10 +314,10 @@ def __init__(self, config: TrainClientConfig) -> None: # The per-request model is only known at call time; a config that pins the renderer # model can warm now, which is every training run (prime-rl always pins it). if config.renderer_model_name is not None: - self._pool_for(config.renderer_model_name) + self._pool_for(config.renderer_model_name).warm() def _pool_for(self, renderer_model: str, chat_template_kwargs=None): - return ElasticRendererPool.shared( + return ElasticRendererPool( renderer_model, self.config.renderer, chat_template_kwargs=chat_template_kwargs, From 6449686152b1edde911f71a72b95f9a8876092a0 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Mon, 3 Aug 2026 09:32:04 -0700 Subject: [PATCH 22/25] chore: inline _pool_for at its call sites Co-Authored-By: Claude Fable 5 --- verifiers/v1/clients/train.py | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index d2a8d5dad6..f01405d450 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -314,15 +314,11 @@ def __init__(self, config: TrainClientConfig) -> None: # The per-request model is only known at call time; a config that pins the renderer # model can warm now, which is every training run (prime-rl always pins it). if config.renderer_model_name is not None: - self._pool_for(config.renderer_model_name).warm() - - def _pool_for(self, renderer_model: str, chat_template_kwargs=None): - return ElasticRendererPool( - renderer_model, - self.config.renderer, - chat_template_kwargs=chat_template_kwargs, - multiplex=self.config.multiplex, - ) + ElasticRendererPool( + config.renderer_model_name, + config.renderer, + multiplex=config.multiplex, + ).warm() async def get_response( self, @@ -368,9 +364,11 @@ async def get_response( ) chat_template_kwargs = sampling_params.pop("chat_template_kwargs", None) sampling_params.update(raw_sampling) - pool = self._pool_for( + pool = ElasticRendererPool( self.config.renderer_model_name or model, + self.config.renderer, chat_template_kwargs=chat_template_kwargs, + multiplex=self.config.multiplex, ) bridged_turn: PendingTurn | None = None From 4e8e91cc97ff3dac9f7a467624cdac06ca03db82 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Mon, 3 Aug 2026 09:37:58 -0700 Subject: [PATCH 23/25] chore: trim client docstrings, fix multiplex doc, drop dead code Co-Authored-By: Claude Fable 5 --- verifiers/v1/clients/base.py | 4 ++-- verifiers/v1/clients/client.py | 22 +++++------------ verifiers/v1/clients/eval.py | 11 ++++----- verifiers/v1/clients/train.py | 43 ++++++++++++++-------------------- verifiers/v1/configs/client.py | 9 ++++--- 5 files changed, 33 insertions(+), 56 deletions(-) diff --git a/verifiers/v1/clients/base.py b/verifiers/v1/clients/base.py index e5624e25af..41317f430d 100644 --- a/verifiers/v1/clients/base.py +++ b/verifiers/v1/clients/base.py @@ -11,8 +11,8 @@ DEFAULT_TIMEOUT = httpx.Timeout(connect=5.0, read=600.0, write=600.0, pool=600.0) DEFAULT_LIMITS = httpx.Limits(max_connections=1000, max_keepalive_connections=100) MAX_RETRIES = 0 -"""No client-side retries: a failed call surfaces to the harness SDK and the trace instead of -being silently reattempted, so the framework's retry surfaces stay the only ones.""" +"""No client-side retries: failures surface to the harness SDK and the trace instead of +being silently reattempted.""" # An API version path segment (`v1`, `v2`, ...) — the only kind `join_url` dedups. VERSION_SEGMENT = re.compile(r"v\d+") diff --git a/verifiers/v1/clients/client.py b/verifiers/v1/clients/client.py index 715348c2ca..283490ac2b 100644 --- a/verifiers/v1/clients/client.py +++ b/verifiers/v1/clients/client.py @@ -1,6 +1,5 @@ """Client interfaces for model inference and relay.""" -import logging from abc import ABC, abstractmethod from collections.abc import AsyncIterator, Awaitable, Callable, Mapping from dataclasses import dataclass, field @@ -14,14 +13,9 @@ from verifiers.v1.graph import PendingTurn from verifiers.v1.types import Response, Sampling, SamplingConfig -logger = logging.getLogger(__name__) - SESSION_ID_HEADER = "X-Session-ID" -"""Per-rollout routing header. Every turn of one rollout sends the same value (the trace id), -so a session-affinity router (e.g. vLLM's ``consistent_hash`` policy keyed on its -``request_id_headers``) pins all of a rollout's turns to the same engine — keeping the -growing cross-turn prefix warm in that engine's KV cache instead of re-prefilling it -cold on a random shard each turn.""" +"""Per-rollout routing header (the trace id, same value every turn), so a session-affinity +router pins a rollout's turns to one engine and its growing prefix stays KV-cached.""" @dataclass @@ -45,14 +39,10 @@ async def get_response( turn: PendingTurn | None = None, headers: Mapping[str, str] | None = None, ) -> Response: - """Run one completion -> a vf `Response`. The eval client forwards the native JSON and - eligible end-to-end headers, then parses a copy via `dialect`; the train client derives - the typed prompt from `body` and tokenizes it. - - `session_id` is the rollout's stable id (the trace id); when set, the client sends it - as the `SESSION_ID_HEADER` so a session-affinity router keeps the rollout's turns on - one engine for cross-turn prefix-cache reuse. `turn` is the graph-resolved prompt - prefix; train clients may use it for renderer bridging, while relay clients ignore it.""" + """Run one completion -> a vf `Response`. The eval client forwards the native JSON + and parses a copy via `dialect`; the train client renders `body` to token ids. + `session_id` is the rollout's trace id (sent as `SESSION_ID_HEADER`); `turn` is the + graph-resolved prompt prefix, used by train clients for renderer bridging.""" async def relay( self, diff --git a/verifiers/v1/clients/eval.py b/verifiers/v1/clients/eval.py index 51ec652a57..862ed80e6f 100644 --- a/verifiers/v1/clients/eval.py +++ b/verifiers/v1/clients/eval.py @@ -57,7 +57,7 @@ class EvalClient(Client): """Relay native JSON to the provider and parse a copy for the trace.""" def __init__(self, config: BaseClientConfig) -> None: - self.base_url = config.base_url.rstrip("/") + self.base_url = config.base_url self.api_key = resolve_api_key(config) # Keep endpoint headers separate so they can override intercepted request headers before # the dialect's provider authentication is applied. @@ -101,12 +101,9 @@ def _headers( incoming: Mapping[str, str] | None, session_id: str | None, ) -> httpx.Headers: - """Build provider headers from the intercepted request. - - Preserve provider feature headers such as `openai-beta` / `anthropic-beta`, - discard localhost auth and transport framing, then apply endpoint-configured headers, - session routing, and real provider auth. - """ + """Provider headers from the intercepted request: keep feature headers + (`openai-beta`, `anthropic-beta`), discard localhost auth and framing, then apply + endpoint headers, session routing, and real provider auth.""" headers = httpx.Headers(incoming) connection = headers.pop("connection", "") for name in _BLOCKED_REQUEST_HEADERS | set( diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index f01405d450..2c8d347cb7 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -1,4 +1,4 @@ -"""Train client: intercepts + renders prompts to tokens for training using `renderers`.""" +"""Train client: renders prompts to token ids and calls a vLLM generate endpoint.""" import asyncio import json @@ -184,10 +184,9 @@ def _has_multimodal_content(messages) -> bool: @dataclass class RendererSlot: - """One renderer, the rollouts currently holding it, and the lock that makes it safe: - encoding mutates a fast tokenizer's truncation/padding state, so `run` serializes the - slot's encode-side work (render, bridge) on a thread. Decode-side work (`generate`'s - response parsing) is a pure read and needs neither the lock nor the hop.""" + """One renderer and the rollouts currently holding it. Encoding mutates a fast + tokenizer's state, so `run` serializes encode-side work (render, bridge) on a thread; + decode-side work is a pure read and needs neither the lock nor the hop.""" renderer: Renderer load: int = 0 @@ -207,17 +206,14 @@ class ElasticRendererPool: them, so every client with the same build inputs works one list.""" _renderers: ClassVar[dict[tuple, list[RendererSlot]]] = {} - """The process's renderers, keyed by build inputs. Where the interception pool has an - owner (the env constructs it and tears it down, injecting it into agents), renderers - have none — clients are built per rollout with no injection channel, and the tokenizers - must outlive event loops and clients alike. So the shared state lives on the type (v0's - `RendererClient._shared_pools`, the same shape) and there is no `start`/`stop`: a - renderer is pure memory, nothing holds a socket or tunnel.""" + """The process's renderers, keyed by build inputs. Unlike the owned interception pool, + renderers have no owner to inject them (clients are built per rollout) and must outlive + loops and clients, so the shared state lives on the type — and needs no `start`/`stop`: + a renderer is pure memory.""" _locks: ClassVar[dict[tuple, tuple[asyncio.AbstractEventLoop, asyncio.Lock]]] = {} - """Per-key single-flight lock for `grow`. An asyncio lock binds to the loop that first - awaits it while the renderers outlive loops, so each key keeps (loop, lock) and a loop - change mints a fresh lock — nothing from a dead loop can still hold it.""" + """Per-key single-flight lock for `grow`, minted fresh on a loop change: an asyncio + lock binds to the loop that first awaits it, while renderers outlive loops.""" def __init__( self, @@ -241,11 +237,9 @@ def __init__( self.renderers = self._renderers.setdefault(self.key, []) def warm(self) -> None: - """Start building the first renderer if none exists — the counterpart of the - interception pool's `start()`: a client is built while its rollout is still - provisioning, so the tokenizer loads now rather than in front of the first - turn. A no-op off the event loop (tests, sync construction) — `acquire` - builds on demand anyway.""" + """Start building the first renderer if none exists, so the tokenizer loads while + the rollout provisions rather than in front of its first turn. A no-op off the + event loop — `acquire` builds on demand anyway.""" if self.renderers: return try: @@ -257,9 +251,8 @@ def warm(self) -> None: async def grow(self) -> RendererSlot: """A renderer with spare capacity — reuse one under `multiplex`, else load one - more tokenizer on a thread (`create_renderer` is seconds of blocking work). - Single-flight per key: every caller serializes on the key's lock, so concurrent - cold acquires (and warms) wait for one build instead of stacking tokenizers.""" + more tokenizer on a thread. Single-flight per key: concurrent cold acquires wait + for one build instead of stacking tokenizers.""" loop = asyncio.get_running_loop() bound = self._locks.get(self.key) if bound is None or bound[0] is not loop: @@ -303,10 +296,8 @@ async def acquire(self) -> AsyncIterator[RendererSlot]: class TrainClient(Client): """Renders prompts to token ids and calls a vLLM `/inference/v1/generate` engine. - One client per rollout: it owns its engine connection and takes a slot on the elastic - renderer pool for each turn. Building the client warms the pool's first tokenizer, so - the load happens while the rollout provisions rather than in front of its first turn. - The pool itself is shared across clients — see `ElasticRendererPool`.""" + One client per rollout: it owns its engine connection and takes a slot on the shared + `ElasticRendererPool` for each turn.""" def __init__(self, config: TrainClientConfig) -> None: self.config = config diff --git a/verifiers/v1/configs/client.py b/verifiers/v1/configs/client.py index f1211d7854..fea86c512c 100644 --- a/verifiers/v1/configs/client.py +++ b/verifiers/v1/configs/client.py @@ -77,11 +77,10 @@ class TrainClientConfig(BaseClientConfig): adapter name (served only for sampling) never drives tokenizer loading. Falls back to the per-request model when None.""" multiplex: int = Field(256, ge=1) - """Rollouts that share one renderer. The pool warms one and grows on demand, so N - concurrent rollouts hold ~N/multiplex tokenizers instead of a fixed set. A renderer is - held only for the render itself (milliseconds) while a turn takes seconds, so one - absorbs many rollouts; the default keeps 2048 concurrent rollouts at 8 tokenizers. - Lower it when rendering is the slow part (very long prompts), at ~75-95 MB each.""" + """Rollouts that share one renderer (~75-95 MB each): the pool warms one and grows on + demand, so N concurrent rollouts hold ~N/multiplex tokenizers. A renderer is only busy + for the render itself (ms against a multi-second turn), so one absorbs many rollouts; + lower this when rendering is the slow part (very long prompts, frequent bridge misses).""" # Discriminated union for a CLI-selectable client (`--client.type eval|train`). From 01f3a3e6943710a004e8d50b22efc5a0f490e0e6 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Mon, 3 Aug 2026 18:15:15 +0000 Subject: [PATCH 24/25] chore: drop the session client docstring Co-Authored-By: Claude Fable 5 --- verifiers/v1/session.py | 1 - 1 file changed, 1 deletion(-) diff --git a/verifiers/v1/session.py b/verifiers/v1/session.py index be1f33872b..3aa65f56ba 100644 --- a/verifiers/v1/session.py +++ b/verifiers/v1/session.py @@ -62,7 +62,6 @@ def reached(self, trace: Trace) -> str | None: class RolloutSession: ctx: ModelContext client: Client - """The rollout's own live client, resolved from `ctx.client` and closed with the rollout.""" trace: Trace stops: list[Callable[[Trace], Awaitable[bool]]] = field(default_factory=list) limits: RolloutLimits = field(default_factory=RolloutLimits) From c7c19d244bef7421a4ddd638ee767f0889de402a Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Mon, 3 Aug 2026 18:27:23 +0000 Subject: [PATCH 25/25] fix: close the legacy server's cached v0 clients on shutdown EnvServer.run() used to close the base class's client cache; the cache moved to LegacyEnvServer, so the close moves with it. Co-Authored-By: Claude Fable 5 --- verifiers/v1/legacy.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/verifiers/v1/legacy.py b/verifiers/v1/legacy.py index 8a792c005f..a7870aab0b 100644 --- a/verifiers/v1/legacy.py +++ b/verifiers/v1/legacy.py @@ -385,6 +385,14 @@ def serving(self): # are no serving resources to enter. return contextlib.nullcontext() + async def run(self) -> None: + try: + await super().run() + finally: + for client in self._clients.values(): + with contextlib.suppress(Exception): + await client.close() + def _v0_client(self, client_config: ClientConfig, model: str): """Translate a v1 ``ClientConfig`` into a v0 client (cached). A renderer config (token-in/out, training) builds a v0 renderer client whose tokenizer is pinned to