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

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/__init__.py

@@ -11,7 +11,14 @@
     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 +180,7 @@
     "Error",
     # decorators
     "stop",
+    "has_decorated",
     "tool",
     "metric",
     "reward",

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,22 @@ 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.model_dump(mode="json")}
+                for task in tasks
+            }
         out = output_path(config)
         finished: list[Trace] = []
         if config.resume is not None:
@@ -214,11 +221,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 +234,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]

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

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})

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

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)
 
 

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()

verifiers/v1/serve/client.py

@@ -136,33 +136,53 @@ 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,
+        # TODO: remove task_idx addressing once v0 (the legacy bridge) is deprecated.
+        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`, its dumped `TaskData`); 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,
         )
         return response.trace
 
     async def run_group(
         self,
-        task_idx: int,
         n: int,
         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,
     ) -> 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,
         )