feat(v1): client-side tasksets + stateless env server (v1-only)

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 harnesses (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.
 

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
 

verifiers/v1/cli/eval/runner.py

@@ -123,6 +123,20 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]:
         if legacy
         else {"config_data": env_config_data(config.env)}  # picklable across the spawn
     )
+    tasks = []
+    if not legacy:
+        from verifiers.v1.loaders import load_taskset
+
+        if config.env.taskset is None:
+            raise ValueError(
+                "a served env needs a seed taskset — set --env.taskset.id (or the "
+                "positional `eval <taskset-id>`)"
+            )
+        # 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.env.taskset).select(
+            config.num_tasks, config.shuffle
+        )
     # Spawned processes inherit no logging — hand them the main process's setup so
     # their rollout logs land in the output dir.
     level = "DEBUG" if config.verbose else "INFO"
@@ -151,24 +165,22 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]:
         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()
-        # Only a legacy (v0) env group-scores; a v1 env scores siblings in its own
-        # rollout.
-        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 (its dataset lives server-side,
+        # reported via `info`), where idx and row coincide. Only a legacy env
+        # group-scores; a v1 env scores siblings in its own rollout.
+        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 = False
+            idxs = [task.data.idx for task in tasks]
+            payloads = {
+                task.data.idx: {"task_data": task.data.model_dump(mode="json")}
+                for task in tasks
+            }
         out = output_path(config)
         finished: list[Episode] = []
         if config.resume is not None:
@@ -227,10 +239,10 @@ async def run_group_unit(idx: int) -> list[Episode]:
         async def run_unit(idx: int) -> list[Episode]:
             async with semaphore or contextlib.nullcontext():
                 episode = await client.run(
-                    task_idx=idx,
                     client=config.client,
                     model=config.model,
                     sampling=config.sampling,
+                    **payloads[idx],
                 )
             for trace in episode.traces:
                 trace.stamp(EvalRunInfo(id=config.uuid))

verifiers/v1/env.py

@@ -31,7 +31,7 @@
 from verifiers.v1.runtimes import SubprocessConfig, runtime_is_local
 from verifiers.v1.errors import EnvError, boundary
 from verifiers.v1.task import Task, resolve_server_config
-from verifiers.v1.taskset import Taskset, TasksetConfig
+from verifiers.v1.taskset import TasksetConfig
 from verifiers.v1.episode import Episode
 from verifiers.v1.trace import Error, Trace
 from verifiers.v1.utils.generic import deep_merge, generic_type
@@ -247,7 +247,7 @@ def __init__(self, config: ConfigT) -> None:
             )
         self.taskset = load_taskset(config.taskset)
         self._default_harness = default_agent_harness(config.taskset.id)
-        task_cls = generic_type(type(self.taskset), Task, origin=Taskset) or Task
+        task_cls = type(self.taskset).task_type()
         self._task_cls: type[Task] = task_cls
         self._agent_specs: dict[str, AgentConfig] = _declared_agent_configs(self.config)
         if not self._agent_specs:
@@ -520,7 +520,7 @@ def _requires_tunnel(self, shared: dict[str, SharedToolServer]) -> bool:
         """`requires_tunnel` over the consumers known before any rollout: role
         runtimes, live `shared` servers, and the task class's tool/user servers;
         a class overriding `server_config` conservatively counts as 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

verifiers/v1/legacy.py

@@ -431,12 +431,23 @@ async def _run_v0(
             state_columns=["trajectory"],
         )
 
+    @staticmethod
+    def _row(req: RunRequest) -> 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`)."""
+        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(self, req: RunRequest) -> RunResponse:
-        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)
         # Trust the bridge-minted record; serialize it once (mirrors `EnvServer`).
         return RunResponse.model_construct(
             episode=Episode.of(
-                rollout_output_to_trace(out, req.task_idx), env=self.taskset_id
+                rollout_output_to_trace(out, task_idx), env=self.taskset_id
             )
         )
 

verifiers/v1/loaders.py

@@ -260,4 +260,4 @@ def resolve_env_config(data: dict | EnvConfig | None) -> EnvConfig:
 def task_type(taskset_id: str) -> type[Task]:
     """The taskset's `Task` subclass from its generic parameters — no data is
     loaded, so replay can cheaply recover the task type. Falls back to `Task`."""
-    return generic_type(taskset_class(taskset_id), Task, origin=Taskset) or Task
+    return taskset_class(taskset_id).task_type()

verifiers/v1/serve/client.py

@@ -141,13 +141,25 @@ async def info(self) -> InfoResponse:
         return await self._request(InfoRequest(), InfoResponse)
 
     async def run(
-        self, task_idx: int, client: ClientConfig, model: str, sampling: SamplingConfig
+        self,
+        client: ClientConfig,
+        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,
     ) -> WireEpisode:
-        """Run one rollout for `task_idx`; return its episode record — flat traces
-        (typed `Trace[WireTaskData]`) plus the shared stamp."""
+        """Run one rollout; return its episode record — flat traces (typed
+        `Trace[WireTaskData]`) plus the shared stamp. A v1 server takes the task
+        itself (`task_data`, its dumped `TaskData`); the legacy bridge addresses
+        its server-side dataset by `task_idx`."""
         response = await self._request(
             RunRequest(
-                task_idx=task_idx, client=client, model=model, sampling=sampling
+                task_data=task_data,
+                task_idx=task_idx,
+                client=client,
+                model=model,
+                sampling=sampling,
             ),
             RunResponse,
         )

verifiers/v1/serve/server.py

@@ -22,13 +22,11 @@
     RunRequest,
     RunResponse,
 )
+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,14 +35,9 @@ def __init__(
         self.address = address
         self.taskset_id = config.taskset.id if config.taskset is not None else ""
         self.env = load_environment(config)
-        # A finite taskset materializes up front; an infinite one is pulled off its
-        # generator on demand, so `num_tasks=None` on the wire means infinite.
-        self._task_iter = iter(self.env.taskset.load())
-        self._tasks: list = []
+        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)
         # v1 envs never group-score (siblings score inside the env's own rollout);
         # only the legacy (v0) bridge sets this.
         self.requires_group_scoring = False
@@ -69,8 +62,8 @@ def __init__(
     @classmethod
     def run_server(cls, address_queue=None, **kwargs) -> None:
         """Run a spawned server and report its concrete address when requested."""
-        # Pin tqdm to a threading lock first, so the taskset load never leaks a
-        # multiprocessing semaphore (resource_tracker warning at shutdown).
+        # Pin tqdm to a threading lock first, so a dataset pull (legacy bridge) never
+        # leaks a multiprocessing semaphore (resource_tracker warning at shutdown).
         use_threading_tqdm_lock()
         server = cls(**kwargs)
         if address_queue is not None:
@@ -83,24 +76,19 @@ 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 generates (and caches) up to `idx`
-        on demand. Generation must be deterministic — every pool worker runs its own
-        `load()`, so idx-addressing relies on all producing the same sequence. The
-        `MAX_LAZY_TASKS` cap fails a runaway driver's request instead of hanging the
-        worker generating toward it."""
-        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. The
+        client owns the taskset; this server never `load()`s data, so pool workers
+        don't each pull the dataset."""
+        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)
+        assert self.env.config.taskset is not None  # load_environment refused None
+        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."""
@@ -123,7 +111,7 @@ def serving(self):
 
     async def _run(self, req: RunRequest) -> RunResponse:
         ctx = self._context(req.client, req.model, req.sampling)
-        (slot,) = self.env.slots(self._task(req.task_idx))
+        (slot,) = self.env.slots(self._build_task(req.task_data))
         # The gate spans requests: `--env.max-concurrent` bounds this worker's
         # agent runs the same way the in-process eval's semaphore does.
         episode = await self.env.run_slot(slot, ctx, self._gate)
@@ -182,7 +170,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()

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
@@ -34,20 +34,35 @@ 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 as whole groups — legacy (v0) envs only; a v1
     server always reports False (sibling-dependent signals run inside the env's
     own rollout)."""
 
 
 class RunRequest(BaseRequest):
+    """One env-rollout. 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`)."""
+
     method: ClassVar[str] = "run"
-    task_idx: int = Field(ge=0)
+    task_data: dict | None = None
+    task_idx: int | None = Field(None, ge=0)
     client: ClientConfig
     model: str
     sampling: SamplingConfig
 
+    @model_validator(mode="after")
+    def _exactly_one(self) -> "RunRequest":
+        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 RunResponse(BaseResponse):
     episode: WireEpisode | None = None

verifiers/v1/task.py

@@ -140,6 +140,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):

verifiers/v1/taskset.py

@@ -29,8 +29,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
 
@@ -66,6 +67,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
 

verifiers/v1/tasksets/harbor/taskset.py

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