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

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

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(

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/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,18 @@ 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: str
-    client: Client
+    client: ClientConfig
     sampling: Sampling = field(default_factory=Sampling)

verifiers/v1/clients/eval.py

@@ -20,6 +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 (
+    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
@@ -57,29 +63,38 @@
         "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}")
 
 
 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 {})
-        # 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 {})
+        # 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:
+        """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,
@@ -92,7 +107,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),
         )
@@ -145,14 +160,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 +178,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 +205,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,11 +240,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,
+            self._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()