From f05bde7ee55504a9b00329875540e31dff7a0647 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Wed, 15 Jul 2026 20:29:26 +0000 Subject: [PATCH 1/5] =?UTF-8?q?feat(v1):=20client-side=20tasksets=20?= =?UTF-8?q?=E2=80=94=20the=20env=20server=20runs=20tasks=20it=20is=20sent?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The env server no longer owns a taskset: the client (eval entrypoint, prime-rl orchestrator) loads the taskset once and ships each dispatched task's data on the run request; the server pydantic-validates it into the taskset's declared TaskData type and rebuilds the Task exactly as load() would. This removes the server-side task cache (and MAX_LAZY_TASKS), the requirement that infinite tasksets generate identical sequences in every pool worker, and the duplicate dataset load per worker. It also fixes resume-after-interrupt through the server path structurally: the client now dispatches and resumes in one coordinate system (task.data.idx), so tasksets whose load() filters rows (gappy idx) no longer rerun the wrong tasks and drop good traces (supersedes #2017). The legacy v0 bridge keeps task_idx addressing — its dataset genuinely lives server-side. Co-Authored-By: Claude Fable 5 --- docs/v1/tasksets.md | 9 +++-- verifiers/v1/__init__.py | 3 +- verifiers/v1/cli/eval/runner.py | 40 ++++++++++--------- verifiers/v1/decorators.py | 13 +++++++ verifiers/v1/env.py | 5 +-- verifiers/v1/legacy.py | 22 ++++++++--- verifiers/v1/loaders.py | 2 +- verifiers/v1/serve/client.py | 32 ++++++++++++---- verifiers/v1/serve/server.py | 68 ++++++++++++--------------------- verifiers/v1/serve/types.py | 30 ++++++++++++--- verifiers/v1/task.py | 16 +++++++- verifiers/v1/taskset.py | 10 ++++- 12 files changed, 160 insertions(+), 90 deletions(-) diff --git a/docs/v1/tasksets.md b/docs/v1/tasksets.md index da35b6176..d285b0e72 100644 --- a/docs/v1/tasksets.md +++ b/docs/v1/tasksets.md @@ -141,10 +141,11 @@ class AdditionTaskset(vf.Taskset[AdditionTask, vf.TasksetConfig]): Two rules follow from infinity: a run over an infinite taskset must be bounded with `num_tasks` (`-n` on the CLI — omitting it is an error), and `shuffle` is a no-op (warned): there is no whole set to sample from, and the first `n` generated tasks are already an -arbitrary sample. Generation must be deterministic — env-server pool workers each run -their own `load()` and rely on every worker producing the same sequence, so seed any -randomness with a constant (see `alphabet_sort_v1`, `color_codeword_v1`, or the built-in -`textarena` taskset). +arbitrary sample. The generator runs once, client-side (the eval entrypoint or the +prime-rl orchestrator pulls tasks off it and ships each task's data to the env server), +so nothing needs to re-produce the same sequence across processes; keep `load()` +deterministic only if you want `--resume` to regenerate the same first `n` tasks (see +`alphabet_sort_v1`, `color_codeword_v1`, or the built-in `textarena` taskset). ## Adding Tools diff --git a/verifiers/v1/__init__.py b/verifiers/v1/__init__.py index 7a693248f..fca5ca410 100644 --- a/verifiers/v1/__init__.py +++ b/verifiers/v1/__init__.py @@ -11,7 +11,7 @@ ModelContext, resolve_client, ) -from verifiers.v1.decorators import group_reward, metric, reward, stop, tool +from verifiers.v1.decorators import group_reward, has_decorated, metric, reward, stop, tool from verifiers.v1.env import ( ElasticPoolConfig, EnvConfig, @@ -173,6 +173,7 @@ "Error", # decorators "stop", + "has_decorated", "tool", "metric", "reward", diff --git a/verifiers/v1/cli/eval/runner.py b/verifiers/v1/cli/eval/runner.py index 9a2fb38df..5fa687ee1 100644 --- a/verifiers/v1/cli/eval/runner.py +++ b/verifiers/v1/cli/eval/runner.py @@ -130,6 +130,13 @@ async def run_eval_server(config: EvalConfig) -> list[Trace]: if legacy else {"config_data": env_config_data(config)} # picklable across the spawn ) + tasks = [] + if not legacy: + from verifiers.v1.loaders import load_taskset + + # The client owns the taskset: load it here, once — the server (and its pool + # workers) never load data, they rebuild each dispatched task from its request. + tasks = load_taskset(config.taskset).select(config.num_tasks, config.shuffle) # The pool broker + workers are spawned (fresh interpreters, no logging) — hand them # the same loguru setup the main process uses (stderr + the run's log file) so their # rollout logs come back and land in the output dir. @@ -160,22 +167,21 @@ async def run_eval_server(config: EvalConfig) -> list[Trace]: address = await asyncio.to_thread(address_queue.get, timeout=600) client = EnvClient(address=address) await client.wait_for_server_startup(timeout=600) - info = await client.info() - group_scored = info.requires_group_scoring - if info.num_tasks is None: # infinite taskset - the run must be bounded - if config.num_tasks is None: - raise ValueError( - f"{config.env_id} is infinite - bound the run with -n/--num-tasks" - ) - if config.shuffle: - logger.warning( - "shuffle is a no-op on an infinite taskset - " - "taking the first %d generated tasks", - config.num_tasks, - ) - idxs = list(range(config.num_tasks)) - else: + # Dispatch (and resume) in the tasks' own coordinate system: `data.idx`. Only the + # legacy bridge is addressed by dataset row, where idx and row coincide. + if legacy: + info = await client.info() + group_scored = info.requires_group_scoring idxs = sample(list(range(info.num_tasks)), config.shuffle, config.num_tasks) + payloads: dict[int, dict] = {idx: {"task_idx": idx} for idx in idxs} + else: + group_scored = bool(tasks) and bool( + discover_decorated(tasks[0], "group_reward") + ) + idxs = [task.data.idx for task in tasks] + payloads = { + task.data.idx: {"task_data": task.data.full_dump()} for task in tasks + } out = output_path(config) finished: list[Trace] = [] if config.resume is not None: @@ -214,11 +220,11 @@ async def run_eval_server(config: EvalConfig) -> list[Trace]: async def run_group_unit(idx: int) -> list[Trace]: async with semaphore or contextlib.nullcontext(): traces = await client.run_group( - task_idx=idx, n=config.num_rollouts, client=config.client, model=config.model, sampling=config.sampling, + **payloads[idx], ) for trace in traces: await append_trace(out, trace, write_lock) @@ -227,10 +233,10 @@ async def run_group_unit(idx: int) -> list[Trace]: async def run_rollout_unit(idx: int) -> list[Trace]: async with semaphore or contextlib.nullcontext(): trace = await client.run_rollout( - task_idx=idx, client=config.client, model=config.model, sampling=config.sampling, + **payloads[idx], ) await append_trace(out, trace, write_lock) return [trace] diff --git a/verifiers/v1/decorators.py b/verifiers/v1/decorators.py index 75797d4ec..68a352627 100644 --- a/verifiers/v1/decorators.py +++ b/verifiers/v1/decorators.py @@ -25,6 +25,19 @@ class MRO for tagged functions (not `inspect.getmembers(obj)`, which evaluates e return methods +def has_decorated(cls: type, attr: str) -> bool: + """Whether `cls` defines any method tagged with `attr` — `discover_decorated` for a + class, no instance needed. An undecorated override still suppresses a decorated base + method.""" + names = { + name + for klass in cls.__mro__ + for name, fn in vars(klass).items() + if callable(fn) and hasattr(fn, attr) + } + return any(hasattr(getattr(cls, name), attr) for name in names) + + def invoke(fn: Callable[..., Any], available: dict[str, Any]) -> Any: params = inspect.signature(fn).parameters return fn(**{name: value for name, value in available.items() if name in params}) diff --git a/verifiers/v1/env.py b/verifiers/v1/env.py index 29c22ab85..02b26728f 100644 --- a/verifiers/v1/env.py +++ b/verifiers/v1/env.py @@ -29,7 +29,6 @@ ) from verifiers.v1.task import Task, resolve_server_config from verifiers.v1.taskset import Taskset, TasksetConfig -from verifiers.v1.utils.generic import generic_type from verifiers.v1.mcp import SharedToolServer, serve_shared @@ -226,7 +225,7 @@ def validate_pairing(harness: Harness, taskset: Taskset) -> None: taskset, read off the `Taskset[TaskT, ...]` generic), so a failure here holds for every row the taskset can produce; on the env server it fails worker startup instead of every request.""" - task_cls = generic_type(type(taskset), Task, origin=Taskset) or Task + task_cls = type(taskset).task_type() if not harness.SUPPORTS_MCP and (task_cls.tools or type(taskset).tools): raise ValueError( f"Harness {harness.config.id!r} does not support MCP tools, but " @@ -390,7 +389,7 @@ def _requires_tunnel(self, shared: dict[str, SharedToolServer]) -> bool: the way `Task.server_config` resolves them. A task that *overrides* that pairing isn't statically knowable, so it conservatively counts as remote (the tunnel then reaches everything; a wrongly-assumed localhost would reach nothing remote).""" - task_cls = generic_type(type(self.taskset), Task, origin=Taskset) or Task + task_cls = type(self.taskset).task_type() server_classes = [*task_cls.tools, *([task_cls.user] if task_cls.user else [])] if server_classes and task_cls.server_config is not Task.server_config: return True diff --git a/verifiers/v1/legacy.py b/verifiers/v1/legacy.py index eafa6a0a0..2a141ce39 100644 --- a/verifiers/v1/legacy.py +++ b/verifiers/v1/legacy.py @@ -383,6 +383,16 @@ def _v0_client(self, client_config: ClientConfig, model: str): self._clients[key] = resolve_client(v0_config) return self._clients[key] + @staticmethod + def _row(req: RunRolloutRequest | RunGroupRequest) -> int: + """The dataset row a request addresses — the bridge's dataset lives server-side, + so requests must carry `task_idx` (v1 servers take `task_data` instead).""" + if req.task_idx is None: + raise ValueError( + "legacy env server requests address the dataset by task_idx" + ) + return req.task_idx + async def _run_v0( self, task_idx: int, @@ -400,24 +410,24 @@ async def _run_v0( ) async def _run_rollout(self, req: RunRolloutRequest) -> RunRolloutResponse: - out = await self._run_v0(req.task_idx, req.client, req.model, req.sampling) + task_idx = self._row(req) + out = await self._run_v0(task_idx, req.client, req.model, req.sampling) return RunRolloutResponse( - trace=rollout_output_to_trace(out, req.task_idx).model_dump() + trace=rollout_output_to_trace(out, task_idx).model_dump() ) async def _run_group(self, req: RunGroupRequest) -> RunGroupResponse: + task_idx = self._row(req) client = self._v0_client(req.client, req.model) # run_group scores the rollouts together so group/preference reward funcs apply. outs = await self.env.run_group( - group_inputs=[dict(self.dataset[req.task_idx]) for _ in range(req.n)], + group_inputs=[dict(self.dataset[task_idx]) for _ in range(req.n)], client=client, model=req.model, sampling_args=req.sampling.model_dump(exclude_none=True), state_columns=["trajectory"], ) - traces = [ - rollout_output_to_trace(out, req.task_idx).model_dump() for out in outs - ] + traces = [rollout_output_to_trace(out, task_idx).model_dump() for out in outs] return RunGroupResponse(traces=traces) diff --git a/verifiers/v1/loaders.py b/verifiers/v1/loaders.py index 3e1e04bf0..b1a1ddaf6 100644 --- a/verifiers/v1/loaders.py +++ b/verifiers/v1/loaders.py @@ -144,4 +144,4 @@ def task_type(taskset_id: str) -> type[Task]: """The taskset's `Task` subclass from its `Taskset[TaskT, ConfigT]` generic — no data is loaded, so replay can cheaply recover the task data type. Falls back to the base `Task` when no subclass is given.""" - return generic_type(taskset_class(taskset_id), Task, origin=Taskset) or Task + return taskset_class(taskset_id).task_type() diff --git a/verifiers/v1/serve/client.py b/verifiers/v1/serve/client.py index db85c38cf..02aa6d172 100644 --- a/verifiers/v1/serve/client.py +++ b/verifiers/v1/serve/client.py @@ -136,16 +136,27 @@ async def wait_for_server_startup( ) async def info(self) -> InfoResponse: - """Return the taskset `num_tasks` + whether its tasks group-score.""" + """Return whether tasks group-score (+ `num_tasks`, legacy bridge only).""" return await self._request(InfoRequest(), InfoResponse) async def run_rollout( - self, task_idx: int, client: ClientConfig, model: str, sampling: SamplingConfig + self, + client: ClientConfig, + model: str, + sampling: SamplingConfig, + task_data: dict | None = None, + task_idx: int | None = None, ) -> Trace[WireTaskData]: - """Run one rollout for `task_idx`; return a typed `Trace[WireTaskData]`.""" + """Run one rollout; return a typed `Trace[WireTaskData]`. A v1 server takes the + task itself (`task_data`, a `TaskData.full_dump()`); the legacy bridge addresses + its server-side dataset by `task_idx`.""" response = await self._request( RunRolloutRequest( - task_idx=task_idx, client=client, model=model, sampling=sampling + task_data=task_data, + task_idx=task_idx, + client=client, + model=model, + sampling=sampling, ), RunRolloutResponse, ) @@ -153,16 +164,23 @@ async def run_rollout( async def run_group( self, - task_idx: int, n: int, client: ClientConfig, model: str, sampling: SamplingConfig, + task_data: dict | None = None, + task_idx: int | None = None, ) -> list[Trace[WireTaskData]]: - """Run `n` rollouts for `task_idx` as a scored group; return typed `Trace[WireTaskData]`s.""" + """Run `n` rollouts of one task as a scored group; return typed + `Trace[WireTaskData]`s. Task addressing as in `run_rollout`.""" response = await self._request( RunGroupRequest( - task_idx=task_idx, n=n, client=client, model=model, sampling=sampling + task_data=task_data, + task_idx=task_idx, + n=n, + client=client, + model=model, + sampling=sampling, ), RunGroupResponse, ) diff --git a/verifiers/v1/serve/server.py b/verifiers/v1/serve/server.py index 1396d9858..86fc9b322 100644 --- a/verifiers/v1/serve/server.py +++ b/verifiers/v1/serve/server.py @@ -11,7 +11,7 @@ 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.decorators import discover_decorated +from verifiers.v1.decorators import has_decorated from verifiers.v1.env import EnvConfig, Environment from verifiers.v1.serve.types import ( BaseResponse, @@ -22,13 +22,11 @@ RunRolloutRequest, RunRolloutResponse, ) +from verifiers.v1.task import Task, task_data_cls from verifiers.v1.types import SamplingConfig logger = logging.getLogger(__name__) -MAX_LAZY_TASKS = 1_000_000 -"""Most tasks an infinite taskset's generator is willing to build (and cache) per worker.""" - class EnvServer: def __init__( @@ -37,21 +35,14 @@ def __init__( self.address = address self.taskset_id = config.taskset.id self.env = Environment(config) - # A finite taskset is materialized up front (its count is served via `info`); an - # infinite one is pulled off its generator on demand (see `_task`), so - # `num_tasks=None` on the wire ⟺ the taskset is infinite. - self._task_iter = iter(self.env.taskset.load()) - self._tasks: list = [] + # The client owns the taskset and ships each request's task data; the server is + # stateless — it rebuilds the task from the wire (the taskset is never `load()`ed + # here, so pool workers don't each pull the dataset). `num_tasks` stays None: only + # the legacy bridge, whose dataset does live server-side, reports a count. + self._task_cls = type(self.env.taskset).task_type() + self._data_cls = task_data_cls(self._task_cls) self.num_tasks: int | None = None - if not type(self.env.taskset).INFINITE: - self._tasks = list(self._task_iter) - self.num_tasks = len(self._tasks) - # One task type per taskset (the authoring contract; its `load()` constructs it), - # so group scoring is a run-wide property. - first = self._task(0) if self.num_tasks != 0 else None - self.requires_group_scoring = first is not None and bool( - discover_decorated(first, "group_reward") - ) + self.requires_group_scoring = has_decorated(self._task_cls, "group_reward") self._clients: dict[ tuple[str, str], Client ] = {} # (client_config, model) -> Client @@ -70,9 +61,9 @@ def __init__( @classmethod def run_server(cls, address_queue=None, **kwargs) -> None: """Run a spawned server and report its concrete address when requested.""" - # This worker loads the taskset (and any HF datasets it pulls in) and is killed at - # teardown; pin tqdm to a threading lock first so it never leaks a multiprocessing - # semaphore (resource_tracker warning at shutdown). + # A worker may still pull datasets (legacy bridge) and is killed at teardown; pin + # tqdm to a threading lock first so it never leaks a multiprocessing semaphore + # (resource_tracker warning at shutdown). use_threading_tqdm_lock() server = cls(**kwargs) if address_queue is not None: @@ -85,25 +76,16 @@ def run_server(cls, address_queue=None, **kwargs) -> None: # of a spurious multiprocessing traceback, matching serve_env's own handling. pass - def _task(self, idx: int): - """The task at `idx`; an infinite taskset is generated (and cached) up to `idx` - on demand. Generation must be deterministic — every pool worker runs its own - `load()`, so idx-addressing relies on all of them producing the same sequence. - Lazy generation is capped at `MAX_LAZY_TASKS`: an idx that far ahead is a - runaway driver, and generating (and caching) toward it would hang the worker - and exhaust memory instead of failing the one request.""" - while len(self._tasks) <= idx: - if idx >= MAX_LAZY_TASKS: - raise IndexError( - f"task_idx {idx} exceeds the lazy-generation cap ({MAX_LAZY_TASKS})" - ) - try: - self._tasks.append(next(self._task_iter)) - except StopIteration: - raise IndexError( - f"task_idx {idx} out of range ({len(self._tasks)} tasks)" - ) from None - return self._tasks[idx] + def _build_task(self, task_data: dict | None) -> Task: + """Rebuild a request's task from its wire data: validate into the taskset's declared + `TaskData` type and wrap it in the declared `Task` with the config's task subtree — + the same construction the taskset's own `load()` performs.""" + if task_data is None: + raise ValueError( + "v1 env server requests carry task_data (task_idx addresses the legacy bridge)" + ) + 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.""" @@ -128,14 +110,14 @@ def serving(self): async def _run_rollout(self, req: RunRolloutRequest) -> RunRolloutResponse: ctx = self._context(req.client, req.model, req.sampling) - episode = self.env.episode(self._task(req.task_idx), ctx, n=1) + episode = self.env.episode(self._build_task(req.task_data), ctx, n=1) traces = await episode.run() # Trust the concrete trace; serialize it once before client-side re-typing. return RunRolloutResponse.model_construct(trace=traces[0]) async def _run_group(self, req: RunGroupRequest) -> RunGroupResponse: ctx = self._context(req.client, req.model, req.sampling) - episode = self.env.episode(self._task(req.task_idx), ctx, n=req.n) + episode = self.env.episode(self._build_task(req.task_data), ctx, n=req.n) traces = await episode.run() # Avoid a dump-and-validate copy for every trusted trace in the group. return RunGroupResponse.model_construct(traces=traces) @@ -186,7 +168,7 @@ async def run(self) -> None: "EnvServer up: taskset=%s address=%s tasks=%s group_scoring=%s", self.taskset_id, self.address, - self.num_tasks if self.num_tasks is not None else "infinite", + self.num_tasks if self.num_tasks is not None else "client-side", self.requires_group_scoring, ) poller = zmq.asyncio.Poller() diff --git a/verifiers/v1/serve/types.py b/verifiers/v1/serve/types.py index 7993ec8ea..26beae058 100644 --- a/verifiers/v1/serve/types.py +++ b/verifiers/v1/serve/types.py @@ -1,6 +1,6 @@ from typing import ClassVar -from pydantic import BaseModel, Field, field_serializer +from pydantic import BaseModel, Field, field_serializer, model_validator from verifiers.v1.clients.config import ClientConfig from verifiers.v1.task import WireTaskData @@ -33,14 +33,33 @@ class InfoRequest(BaseRequest): class InfoResponse(BaseResponse): num_tasks: int | None = None - """Task count; `None` means the taskset is infinite (bound runs with `num_tasks`).""" + """Task count. Only the legacy bridge (whose dataset lives server-side) reports one; + a v1 server is stateless — its tasks live on the client — so this stays `None`.""" requires_group_scoring: bool = False """Whether tasks must be run and resumed as whole groups.""" -class RunRolloutRequest(BaseRequest): +class TaskAddressing(BaseModel): + """How a run request names its task: v1 ships the task itself (`task_data`, a + `TaskData.full_dump()` the server validates into the taskset's declared type); the + legacy bridge addresses its server-side dataset by row (`task_idx`).""" + + task_data: dict | None = None + """The task's wire data (v1). The server rebuilds and pydantic-validates it.""" + task_idx: int | None = Field(None, ge=0) + """Dataset row index (legacy v0 bridge only).""" + + @model_validator(mode="after") + def _exactly_one(self) -> "TaskAddressing": + if (self.task_data is None) == (self.task_idx is None): + raise ValueError( + "exactly one of task_data (v1) or task_idx (legacy) must be set" + ) + return self + + +class RunRolloutRequest(TaskAddressing, BaseRequest): method: ClassVar[str] = "run_rollout" - task_idx: int = Field(ge=0) client: ClientConfig model: str sampling: SamplingConfig @@ -55,9 +74,8 @@ def _ser_trace(self, trace: "Trace[WireTaskData] | None") -> dict | None: return trace.model_dump() if trace is not None else None -class RunGroupRequest(BaseRequest): +class RunGroupRequest(TaskAddressing, BaseRequest): method: ClassVar[str] = "run_group" - task_idx: int = Field(ge=0) n: int client: ClientConfig model: str diff --git a/verifiers/v1/task.py b/verifiers/v1/task.py index db955b7ba..c2e39c763 100644 --- a/verifiers/v1/task.py +++ b/verifiers/v1/task.py @@ -47,7 +47,7 @@ from collections.abc import Mapping from typing import TYPE_CHECKING, ClassVar, Generic -from pydantic import ConfigDict, model_validator +from pydantic import BaseModel, ConfigDict, model_validator from pydantic_config import BaseConfig from typing_extensions import TypeVar @@ -163,6 +163,20 @@ def prompt_text(self) -> str: texts = [content_text(message.content) for message in self.prompt or []] return "\n\n".join(text for text in texts if text) + def full_dump(self) -> dict: + """`model_dump(mode="json")` plus the fields excluded from serialization — + load-time-only values (e.g. a local task directory) that must not persist in saved + traces but that the env server needs to rebuild the task from a dispatch request.""" + dumped = self.model_dump(mode="json") + for name, field in type(self).model_fields.items(): + if not field.exclude: + continue + value = getattr(self, name) + if isinstance(value, BaseModel): + value = value.model_dump(mode="json") + dumped[name] = value + return dumped + class WireTaskData(TaskData): """Wire form that preserves task-specific fields without importing the task class.""" diff --git a/verifiers/v1/taskset.py b/verifiers/v1/taskset.py index 0cf9c2abb..e4a7c9ddd 100644 --- a/verifiers/v1/taskset.py +++ b/verifiers/v1/taskset.py @@ -36,8 +36,9 @@ def load(self) -> Iterable[MyTask]: from pydantic_config import BaseConfig from typing_extensions import TypeVar -from verifiers.v1.task import TaskConfig, TaskT, resolve_server_config +from verifiers.v1.task import Task, TaskConfig, TaskT, resolve_server_config from verifiers.v1.types import ID +from verifiers.v1.utils.generic import generic_type from verifiers.v1.utils.install import env_name from verifiers.v1.utils.sampling import sample @@ -73,6 +74,13 @@ class Taskset(Generic[TaskT, TasksetConfigT]): def __init__(self, config: TasksetConfigT) -> None: self.config = config + @classmethod + def task_type(cls) -> type[Task]: + """The taskset's declared `Task` subclass, read off the `Taskset[TaskT, ...]` + generic — no data is loaded, so consumers (env server, replay) can cheaply rebuild + wire rows as the declared type.""" + return generic_type(cls, Task, origin=Taskset) or Task + def load(self) -> Iterable[TaskT]: raise NotImplementedError From 601cdd41431b12a6b968918996edab9027ddadc4 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Wed, 15 Jul 2026 21:31:28 +0000 Subject: [PATCH 2/5] =?UTF-8?q?docs(v1):=20architecture=20=E2=80=94=20work?= =?UTF-8?q?ers=20no=20longer=20load=20the=20taskset?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- docs/v1/architecture.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/v1/architecture.md b/docs/v1/architecture.md index 6822c702c..514ab2ffe 100644 --- a/docs/v1/architecture.md +++ b/docs/v1/architecture.md @@ -2,7 +2,7 @@ verifiers is built out of the following parts: -A server-backed evaluation or prime-rl **orchestrator** creates worker processes and distributes rollout requests among them. Each worker loads its taskset and harness, owns an **interception pool**, and creates the runtime used by each rollout it handles. +A server-backed evaluation or prime-rl **orchestrator** creates worker processes and distributes rollout requests among them. The client owns the taskset — it loads the tasks once and ships each dispatched task's data on the request; a worker loads only the harness (rebuilding each task from its request), owns an **interception pool**, and creates the runtime used by each rollout it handles. The orchestrator and workers are managed by verifiers and prime-rl themselves and thus offer few configurable knobs. From 5000484e8038bff6960206033cea8f48c88e98f7 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Wed, 15 Jul 2026 21:43:13 +0000 Subject: [PATCH 3/5] style: ruff format __init__ Co-Authored-By: Claude Fable 5 --- verifiers/v1/__init__.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/verifiers/v1/__init__.py b/verifiers/v1/__init__.py index fca5ca410..ce276863e 100644 --- a/verifiers/v1/__init__.py +++ b/verifiers/v1/__init__.py @@ -11,7 +11,14 @@ ModelContext, resolve_client, ) -from verifiers.v1.decorators import group_reward, has_decorated, metric, reward, stop, tool +from verifiers.v1.decorators import ( + group_reward, + has_decorated, + metric, + reward, + stop, + tool, +) from verifiers.v1.env import ( ElasticPoolConfig, EnvConfig, From 898c2c2b875d2ede2448168db6a59fd6fdf43a02 Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Wed, 15 Jul 2026 21:45:28 +0000 Subject: [PATCH 4/5] chore: TODO on legacy task_idx addressing; unprefix server task/data classes Co-Authored-By: Claude Fable 5 --- verifiers/v1/serve/client.py | 2 ++ verifiers/v1/serve/server.py | 14 +++++--------- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/verifiers/v1/serve/client.py b/verifiers/v1/serve/client.py index 02aa6d172..26dc7aed3 100644 --- a/verifiers/v1/serve/client.py +++ b/verifiers/v1/serve/client.py @@ -145,6 +145,7 @@ async def run_rollout( model: str, sampling: SamplingConfig, task_data: dict | None = None, + # TODO: remove task_idx addressing once v0 (the legacy bridge) is deprecated. task_idx: int | None = None, ) -> Trace[WireTaskData]: """Run one rollout; return a typed `Trace[WireTaskData]`. A v1 server takes the @@ -169,6 +170,7 @@ async def run_group( model: str, sampling: SamplingConfig, task_data: dict | None = None, + # TODO: remove task_idx addressing once v0 (the legacy bridge) is deprecated. task_idx: int | None = None, ) -> list[Trace[WireTaskData]]: """Run `n` rollouts of one task as a scored group; return typed diff --git a/verifiers/v1/serve/server.py b/verifiers/v1/serve/server.py index 86fc9b322..6a380f402 100644 --- a/verifiers/v1/serve/server.py +++ b/verifiers/v1/serve/server.py @@ -35,14 +35,10 @@ def __init__( self.address = address self.taskset_id = config.taskset.id self.env = Environment(config) - # The client owns the taskset and ships each request's task data; the server is - # stateless — it rebuilds the task from the wire (the taskset is never `load()`ed - # here, so pool workers don't each pull the dataset). `num_tasks` stays None: only - # the legacy bridge, whose dataset does live server-side, reports a count. - self._task_cls = type(self.env.taskset).task_type() - self._data_cls = task_data_cls(self._task_cls) + self.task_cls = type(self.env.taskset).task_type() + self.data_cls = task_data_cls(self.task_cls) self.num_tasks: int | None = None - self.requires_group_scoring = has_decorated(self._task_cls, "group_reward") + self.requires_group_scoring = has_decorated(self.task_cls, "group_reward") self._clients: dict[ tuple[str, str], Client ] = {} # (client_config, model) -> Client @@ -84,8 +80,8 @@ def _build_task(self, task_data: dict | None) -> Task: raise ValueError( "v1 env server requests carry task_data (task_idx addresses the legacy bridge)" ) - data = self._data_cls.model_validate(task_data) - return self._task_cls(data, self.env.config.taskset.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.""" From 9ba906bb5effea0b05685cd7352174be0d0fd4ff Mon Sep 17 00:00:00 2001 From: Mika Senghaas Date: Wed, 15 Jul 2026 21:49:58 +0000 Subject: [PATCH 5/5] =?UTF-8?q?feat(v1):=20task=20data=20serializes=20whol?= =?UTF-8?q?e=20=E2=80=94=20drop=20full=5Fdump,=20forbid=20excluded=20field?= =?UTF-8?q?s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dispatch payload is a plain model_dump: a TaskData subclass may not exclude fields (enforced at class definition), since an excluded field would vanish on the wire and the server would rebuild the task with silently-defaulted values. Harbor's task_dir — the only excluded field — now serializes (a host path in saved traces is harmless, and rows become rebuildable for replay). Co-Authored-By: Claude Fable 5 --- verifiers/v1/cli/eval/runner.py | 3 ++- verifiers/v1/cli/replay.py | 9 ++++----- verifiers/v1/serve/client.py | 2 +- verifiers/v1/serve/types.py | 4 ++-- verifiers/v1/task.py | 27 +++++++++++-------------- verifiers/v1/tasksets/harbor/taskset.py | 4 ++-- 6 files changed, 23 insertions(+), 26 deletions(-) diff --git a/verifiers/v1/cli/eval/runner.py b/verifiers/v1/cli/eval/runner.py index 5fa687ee1..1af552133 100644 --- a/verifiers/v1/cli/eval/runner.py +++ b/verifiers/v1/cli/eval/runner.py @@ -180,7 +180,8 @@ async def run_eval_server(config: EvalConfig) -> list[Trace]: ) idxs = [task.data.idx for task in tasks] payloads = { - task.data.idx: {"task_data": task.data.full_dump()} for task in tasks + task.data.idx: {"task_data": task.data.model_dump(mode="json")} + for task in tasks } out = output_path(config) finished: list[Trace] = [] diff --git a/verifiers/v1/cli/replay.py b/verifiers/v1/cli/replay.py index 90f602ae5..b6a14a9e3 100644 --- a/verifiers/v1/cli/replay.py +++ b/verifiers/v1/cli/replay.py @@ -75,11 +75,10 @@ async def run_replay(config: ReplayConfig, source: Path, out: Path) -> list[Trac # time — `trace.task` keeps its wire form, because the trace persists through the # `Trace[WireTaskData, ...]` schema it was read as: a sibling `TaskData` assigned onto # it would have its subclass fields silently dropped from the replay's own output - # (they're real fields, not `model_extra`). A row that can't be rebuilt from the wire - # (a load-time-only field excluded from serialization, like harbor's `task_dir`) is - # scored by the base `Task` on the wire row (judges + base signals only; the - # subclass's own `@reward`s don't run — runtime-dependent ones would be skipped - # offline anyway). + # (they're real fields, not `model_extra`). A row that still can't be rebuilt from + # the wire (e.g. saved before its data type grew a required field) is scored by the + # base `Task` on the wire row (judges + base signals only; the subclass's own + # `@reward`s don't run — runtime-dependent ones would be skipped offline anyway). def rebuild(trace: Trace) -> vf.TaskData: if trace.task.type != task_cls.__name__: # The trace records which class produced it — a mismatch means this row is diff --git a/verifiers/v1/serve/client.py b/verifiers/v1/serve/client.py index 26dc7aed3..5a49746ea 100644 --- a/verifiers/v1/serve/client.py +++ b/verifiers/v1/serve/client.py @@ -149,7 +149,7 @@ async def run_rollout( task_idx: int | None = None, ) -> Trace[WireTaskData]: """Run one rollout; return a typed `Trace[WireTaskData]`. A v1 server takes the - task itself (`task_data`, a `TaskData.full_dump()`); the legacy bridge addresses + task itself (`task_data`, its dumped `TaskData`); the legacy bridge addresses its server-side dataset by `task_idx`.""" response = await self._request( RunRolloutRequest( diff --git a/verifiers/v1/serve/types.py b/verifiers/v1/serve/types.py index 26beae058..b2e476d52 100644 --- a/verifiers/v1/serve/types.py +++ b/verifiers/v1/serve/types.py @@ -40,8 +40,8 @@ class InfoResponse(BaseResponse): class TaskAddressing(BaseModel): - """How a run request names its task: v1 ships the task itself (`task_data`, a - `TaskData.full_dump()` the server validates into the taskset's declared type); the + """How a run request names its task: v1 ships the task itself (`task_data`, the + dumped `TaskData` the server validates into the taskset's declared type); the legacy bridge addresses its server-side dataset by row (`task_idx`).""" task_data: dict | None = None diff --git a/verifiers/v1/task.py b/verifiers/v1/task.py index c2e39c763..25ebc0dd6 100644 --- a/verifiers/v1/task.py +++ b/verifiers/v1/task.py @@ -47,7 +47,7 @@ from collections.abc import Mapping from typing import TYPE_CHECKING, ClassVar, Generic -from pydantic import BaseModel, ConfigDict, model_validator +from pydantic import ConfigDict, model_validator from pydantic_config import BaseConfig from typing_extensions import TypeVar @@ -156,6 +156,17 @@ class TaskData(StrictBaseModel): timeout: TaskTimeout = TaskTimeout() resources: TaskResources = TaskResources() + @classmethod + def __pydantic_init_subclass__(cls, **kwargs) -> None: + super().__pydantic_init_subclass__(**kwargs) + excluded = [name for name, field in cls.model_fields.items() if field.exclude] + if excluded: + raise TypeError( + f"{cls.__name__}: task data fields cannot be excluded from serialization " + f"({excluded}) — a task must survive the wire whole, or the env server " + f"rebuilds it with silently-defaulted fields" + ) + @property def prompt_text(self) -> str: if isinstance(self.prompt, str): @@ -163,20 +174,6 @@ def prompt_text(self) -> str: texts = [content_text(message.content) for message in self.prompt or []] return "\n\n".join(text for text in texts if text) - def full_dump(self) -> dict: - """`model_dump(mode="json")` plus the fields excluded from serialization — - load-time-only values (e.g. a local task directory) that must not persist in saved - traces but that the env server needs to rebuild the task from a dispatch request.""" - dumped = self.model_dump(mode="json") - for name, field in type(self).model_fields.items(): - if not field.exclude: - continue - value = getattr(self, name) - if isinstance(value, BaseModel): - value = value.model_dump(mode="json") - dumped[name] = value - return dumped - class WireTaskData(TaskData): """Wire form that preserves task-specific fields without importing the task class.""" diff --git a/verifiers/v1/tasksets/harbor/taskset.py b/verifiers/v1/tasksets/harbor/taskset.py index 2a0070fa7..59e99f03e 100644 --- a/verifiers/v1/tasksets/harbor/taskset.py +++ b/verifiers/v1/tasksets/harbor/taskset.py @@ -81,8 +81,8 @@ class HarborData(TaskData): difficulty: str | None = None category: str | None = None tags: list[str] = [] - task_dir: str = Field("", exclude=True) - """Host path to the task dir; used to stage tests/ to verify, not serialized.""" + task_dir: str = "" + """Host path to the task dir; used to stage tests/ to verify.""" verifier_env: dict[str, str] = {} """Raw [verifier.env] entries (literals or `${VAR}`/`${VAR:-default}` templates). Resolved against the host environment at scoring time, like `harbor run` — so a