diff --git a/docs/v1/agent.md b/docs/v1/agent.md
index b2bcca4a7d..232b75b155 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
 
@@ -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.
diff --git a/tests/v1/conftest.py b/tests/v1/conftest.py
index 049a08c2b5..d24bc155f0 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-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
diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py
index f991f14682..4f6a5f1940 100644
--- a/tests/v1/test_e2e.py
+++ b/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(
diff --git a/verifiers/v1/agent.py b/verifiers/v1/agent.py
index 5151b06a11..71f5adbb14 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.dialects import parse_message
@@ -228,10 +226,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)."""
 
@@ -239,7 +237,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
@@ -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]
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/base.py b/verifiers/v1/clients/base.py
new file mode 100644
index 0000000000..41317f430d
--- /dev/null
+++ b/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
diff --git a/verifiers/v1/clients/client.py b/verifiers/v1/clients/client.py
index 5940f49f26..283490ac2b 100644
--- a/verifiers/v1/clients/client.py
+++ b/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)
diff --git a/verifiers/v1/clients/eval.py b/verifiers/v1/clients/eval.py
index 2951bb23a8..862ed80e6f 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
@@ -19,7 +7,9 @@
 from pydantic import ValidationError
 from pydantic_core import from_json, to_json
 
+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
 from verifiers.v1.graph import PendingTurn
@@ -57,6 +47,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}")
 
@@ -64,22 +56,13 @@
 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
+        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 {})
-        # 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.
-        self.http = httpx.AsyncClient(
-            timeout=None,
-            limits=httpx.Limits(max_connections=128, max_keepalive_connections=20),
-        )
+        self.headers = dict(config.headers or {})
+        self.client = httpx.AsyncClient(timeout=DEFAULT_TIMEOUT, limits=DEFAULT_LIMITS)
 
     async def get_response(
         self,
@@ -92,7 +75,7 @@ async def get_response(
         headers: Mapping[str, str] | None = None,
     ) -> Response:
         resp = await self._request(
-            self.base_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),
         )
@@ -118,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(
@@ -145,14 +125,14 @@ async def _request(
         stream: bool = False,
     ) -> httpx.Response:
         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:
@@ -163,9 +143,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,
@@ -193,7 +170,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,
+            join_url(self.base_url, dialect.upstream_path),
             dialect.apply_overrides(body, model, sampling_args),
             self._headers(dialect, headers, session_id),
             stream=True,
@@ -228,11 +205,11 @@ 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,
+            join_url(self.base_url, route),
             body,
             self._headers(dialect, headers, None),
         )
         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 785f05a923..2c8d347cb7 100644
--- a/verifiers/v1/clients/train.py
+++ b/verifiers/v1/clients/train.py
@@ -1,22 +1,21 @@
-"""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: renders prompts to token ids and calls a vLLM generate endpoint."""
 
+import asyncio
 import json
-from collections.abc import Mapping
-from typing import Any
+import logging
+import threading
+from collections.abc import AsyncIterator, Callable, Mapping
+from contextlib import asynccontextmanager
+from dataclasses import dataclass, field
+from typing import Any, ClassVar, TypeVar
 
-from openai import AsyncOpenAI, OpenAIError
+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
+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
@@ -33,6 +32,10 @@
     Usage,
 )
 
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T")
+
 
 def tool_to_wire(tool: Tool) -> dict:
     function: dict = {
@@ -179,41 +182,134 @@ def _has_multimodal_content(messages) -> bool:
     return False
 
 
-class TrainClient(Client):
-    """Renders prompts to token ids and calls a vLLM `/inference/v1/generate` engine."""
+@dataclass
+class RendererSlot:
+    """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
+    lock: threading.Lock = field(default_factory=threading.Lock)
+
+    async def run(self, fn: Callable[[], T]) -> T:
+        def locked() -> T:
+            with self.lock:
+                return fn()
+
+        return await asyncio.to_thread(locked)
+
+
+class ElasticRendererPool:
+    """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. 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`, minted fresh on a loop change: an asyncio
+    lock binds to the loop that first awaits it, while renderers outlive loops."""
 
     def __init__(
         self,
-        openai: AsyncOpenAI,
-        pool_size: int = 1,
-        config: RendererConfig | None = None,
-        renderer_model_name: str | None = None,
+        renderer_model: str,
+        config: RendererConfig | None,
+        *,
+        chat_template_kwargs: Mapping[str, Any] | None = None,
+        multiplex: int,
     ) -> None:
-        self.openai = openai
-        self.pool_size = pool_size
+        self.renderer_model = renderer_model
         self.config = config
-        self.renderer_model_name = renderer_model_name
-        self._pool = None
+        self.chat_template_kwargs = chat_template_kwargs
+        self.multiplex = multiplex
+        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,
+        )
+        self.renderers = self._renderers.setdefault(self.key, [])
 
-    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,
+    def warm(self) -> None:
+        """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:
+            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. 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:
+            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 self._pool
+            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."""
+        slot = await self.grow()
+        slot.load += 1
+        try:
+            yield slot
+        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 takes a slot on the shared
+    `ElasticRendererPool` for each turn."""
+
+    def __init__(self, config: TrainClientConfig) -> None:
+        self.config = 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:
+            ElasticRendererPool(
+                config.renderer_model_name,
+                config.renderer,
+                multiplex=config.multiplex,
+            ).warm()
 
     async def get_response(
         self,
@@ -244,7 +340,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 = (
@@ -259,63 +355,79 @@ async def get_response(
         )
         chat_template_kwargs = sampling_params.pop("chat_template_kwargs", None)
         sampling_params.update(raw_sampling)
-        renderer = self._renderer_pool(
-            model,
+        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
 
-        # 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,
-                )
+        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 = (
+                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
 
-            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,
-                )
+                def bridge():
+                    return renderer.bridge_to_next_turn(
+                        previous_prompt_ids,
+                        previous_completion_ids,
+                        wire_messages,
+                        tools=wire_tools,
+                    )
 
-        # 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]
+                bridged = await slot.run(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,
+                    )
 
-        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
+            # 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(
+                    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,
+                )
+            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.
@@ -323,4 +435,4 @@ def bridge():
         return response
 
     async def close(self) -> None:
-        await self.openai.close()
+        await self.client.close()
diff --git a/verifiers/v1/clients/config.py b/verifiers/v1/configs/client.py
similarity index 66%
rename from verifiers/v1/clients/config.py
rename to verifiers/v1/configs/client.py
index d8783ee656..fea86c512c 100644
--- a/verifiers/v1/clients/config.py
+++ b/verifiers/v1/configs/client.py
@@ -1,28 +1,26 @@
-"""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
 from typing import Annotated, Literal
 from urllib.parse import urlparse
 
-from openai import AsyncOpenAI
 from pydantic import Field, model_validator
 from pydantic_config import BaseConfig
 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"
 PRIME_TEAM_ID_HEADER = "X-Prime-Team-ID"
 
@@ -74,12 +72,15 @@ 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
     the per-request model when None."""
+    multiplex: int = Field(256, ge=1)
+    """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`).
@@ -100,28 +101,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,
-    )
-
-
-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 c578960e64..c7aa81db96 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()
 
@@ -219,13 +217,11 @@ def make(name: str, spec: AgentConfig) -> Agent:
                     else self._default_harness,
                     "model": spec.model if spec.model is not None else ctx.model,
                     "sampling": 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,
@@ -240,13 +236,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,
@@ -377,10 +366,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..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.config 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/legacy.py b/verifiers/v1/legacy.py
index fbdeb835ef..a7870aab0b 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 (
@@ -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
@@ -406,7 +414,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=client_config.pool_size,
                     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 1cfb261b2c..98050f46c4 100644
--- a/verifiers/v1/rollout.py
+++ b/verifiers/v1/rollout.py
@@ -8,7 +8,7 @@
 from contextlib import AsyncExitStack
 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.errors import (
     HarnessError,
@@ -91,8 +91,9 @@ def __init__(
         )
         if on_trace is not None:
             on_trace(self.trace)
+        self.client = resolve_client(ctx.client)
         self._session = RolloutSession(
-            ctx, self.trace, discover_decorated(task, "stop"), limits
+            ctx, self.client, self.trace, discover_decorated(task, "stop"), limits
         )
         self._stack = AsyncExitStack()
         self._failed = False
@@ -314,13 +315,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)
@@ -399,6 +402,12 @@ async def close(self) -> Trace:
                     logger.warning(
                         "runtime teardown failed (rollout %s)", trace.id, exc_info=True
                     )
+            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..3aa65f56ba 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,7 @@ def reached(self, trace: Trace) -> str | None:
 @dataclass
 class RolloutSession:
     ctx: ModelContext
+    client: Client
     trace: Trace
     stops: list[Callable[[Trace], Awaitable[bool]]] = field(default_factory=list)
     limits: RolloutLimits = field(default_factory=RolloutLimits)