feat: build one client per rollout, share renderers process-wide

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
 
@@ -56,19 +54,18 @@ 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).
+## Client
 
-```python
-client = vf.resolve_client(vf.EvalClientConfig())
+The model endpoint is not a borrowed resource — it is config. Set `AgentConfig.client`; each rollout builds and closes its own `Client` from it.
 
-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)
+```python
+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.

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-0731",
-            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-0731",
+        client=EvalClientConfig(),
+        sampling=SamplingConfig(max_tokens=2048),
+    )
 
 
 @pytest.fixture

tests/v1/test_e2e.py

@@ -180,8 +180,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(

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.dialects import parse_message
@@ -228,18 +226,17 @@ 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)."""
 
     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
@@ -259,12 +256,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
@@ -306,22 +300,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
@@ -584,7 +570,6 @@ def __init__(
         self,
         config: AgentConfig,
         *,
-        client: Client,
         interception: Interception | None,
         name: str,
         shared_tools: Mapping[str, SharedToolServer],
@@ -595,7 +580,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
@@ -690,12 +675,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]

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
 
 

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",

verifiers/v1/clients/base.py

@@ -0,0 +1,40 @@
+"""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
+
+# 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
+"""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+")
+
+
+def build_async_openai(config: BaseClientConfig) -> AsyncOpenAI:
+    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

verifiers/v1/clients/client.py

@@ -1,22 +1,21 @@
 """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
 
+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
 
-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
@@ -40,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,
@@ -78,10 +73,20 @@ async def close(self) -> None:
         pass
 
 
+def resolve_client(config: BaseClientConfig) -> Client:
+    if isinstance(config, TrainClientConfig):
+        from verifiers.v1.clients.train import TrainClient
+
+        return TrainClient(config)
+    from verifiers.v1.clients.eval import EvalClient
+
+    return EvalClient(config)
+
+
 @dataclass(frozen=True)
 class ModelContext:
-    """Client, model, and sampling settings for one rollout."""
+    """Model, endpoint config, and sampling for one rollout."""
 
     model: str
-    client: Client
+    client: ClientConfig
     sampling: Sampling = field(default_factory=Sampling)