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

@@ -134,7 +134,7 @@ 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).
+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. 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
 

tests/v1/test_configs.py

@@ -24,4 +24,4 @@
 def test_eval_config_parses(path: Path) -> None:
     config = EvalConfig.model_validate(tomllib.load(path.open("rb")))
     # resolved to a v1 taskset or a v0 env id
-    assert (config.env.taskset is not None and config.env.taskset.id) or config.id
+    assert config.env.taskset.id or config.id

verifiers/v1/cli/dashboard/eval.py

@@ -232,7 +232,7 @@ def Overview(config: EvalConfig) -> Table:
     grid.add_column()
     seats = config.env.agent_harnesses()
     taskset = config.env.taskset
-    env_label = taskset.name if taskset is not None else "no taskset"
+    env_label = taskset.name if taskset.id else "no taskset"
     if config.env.id:
         env_label = f"{env_name(config.env.id)}+{env_label}"
     # One seat story when every seat resolves the same way (the common case); one
@@ -254,9 +254,7 @@ def Overview(config: EvalConfig) -> Table:
     # value (or our `[...]`/`{...}` delimiters) can carry Rich markup that would otherwise be
     # parsed as styling and dropped. `id` is in the `env` row; harness `runtime.type` too (hidden
     # here), but only for the harness — `taskset.task.tools.runtime.type` has no other display.
-    if taskset is not None and (
-        taskset_over := overrides(taskset, skip=frozenset({"id"}))
-    ):
+    if taskset_over := overrides(taskset, skip=frozenset({"id"})):
         grid.add_row("taskset", escape("  ·  ".join(taskset_over)))
     for role, h in seats.items():
         if harness_over := overrides(h, skip=frozenset({"id", "runtime.type"})):

verifiers/v1/cli/eval/resume.py

@@ -4,13 +4,23 @@
 flags) and writes back into the same dir. `load` keeps the good saved rollouts and
 re-runs what's owed: missing rollouts (never written) and errored ones (dropped and
 redone).
+
+A saved rollout is matched to a selected task by CONTENT: `task_key` hashes the
+task's wire data, so identity is the data itself. Tasks with identical data are
+interchangeable (a collision resolves to "either one counts"), a task whose data
+changed since the interrupted run re-runs (the saved episode answered a different
+question), and nothing depends on `data.idx` being unique — or set at all. Only
+the legacy (v0) bridge, whose tasks never leave the server, still matches by row
+index (its `key_of`).
 """
 
+import hashlib
 import json
 import tomllib
-from collections import defaultdict
-from collections.abc import Callable
+from collections import Counter, defaultdict
+from collections.abc import Callable, Hashable, Mapping
 from pathlib import Path
+from typing import TypeVar
 
 from pydantic_core import from_json
 
@@ -19,6 +29,30 @@
 from verifiers.v1.episode import Episode, WireEpisode
 from verifiers.v1.trace import WireTrace
 
+K = TypeVar("K", bound=Hashable)
+
+
+def task_key(data: Mapping) -> str:
+    """Content identity of one task's wire data — an `exclude_none` dump, the shape
+    saved rows already have on disk. `sort_keys` so field order can't split identity."""
+    return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
+
+
+def distribute(
+    selected_keys: list[K], owed: dict[K, int], num_rollouts: int
+) -> list[int]:
+    """Spread each key's owed rollouts over its selection instances, in order —
+    content-identical tasks are interchangeable, so any instance can absorb the
+    debt (capped at `num_rollouts` each). Returns one count per selection."""
+    remaining = dict(owed)
+    counts: list[int] = []
+    for key in selected_keys:
+        take = min(num_rollouts, remaining.get(key, 0))
+        if take:
+            remaining[key] -= take
+        counts.append(take)
+    return counts
+
 
 def split_resume(argv: list[str]) -> tuple[Path | None, list[str]]:
     """Pull `--resume <dir>` / `--resume=<dir>` out of argv, returning (dir, the other args).
@@ -51,33 +85,35 @@ def load_resume_config(resume_dir: Path) -> EvalConfig:
 
 def load(
     resume_dir: Path,
-    selected_idxs: list[int],
+    selected_keys: list[K],
     num_rollouts: int,
     complete: Callable[[Episode], bool] | None = None,
     *,
     whole_task: bool = False,
-) -> tuple[list[Episode], dict[int, int]]:
-    """Load the good saved rollouts as finished episodes and diff them against the
-    run's target: returns (kept episodes, rollouts owed per task idx). A rollout is
-    kept or redone as a unit — the episode — so a multi-trace rollout interrupted
-    mid-write is simply owed again. `complete` is the environment's keep-verdict
-    (`Env.complete`); without it (the server path) the default is
-    `episode.ok`, so an errored rollout is dropped and re-run. `whole_task` redoes
-    a partially-kept task as a unit — the legacy group-scored path, where
-    `run_group` always serves the full n. Rewrites `traces.jsonl` to just the kept
-    rows via a temp file + atomic rename, so an interrupted resume can't corrupt
-    the prior results; the resumed rollouts then append. Pre-episode files (one
-    bare trace per line) load each trace as a single-trace episode."""
+    key_of: Callable[[Mapping], K] | None = None,
+) -> tuple[list[Episode], dict[K, int]]:
+    """Load the good saved rollouts and diff them against the run's target: returns
+    (kept episodes, rollouts owed per task key). `selected_keys` is one key per
+    selected task (duplicates allowed — a key selected k times is owed up to
+    `k * num_rollouts`; spread back over the tasks with `distribute`). `key_of` maps
+    a saved row's task data to its key (default `task_key`; the legacy bridge uses
+    row indices). `complete` is the keep-verdict (default `episode.ok`); `whole_task`
+    redoes a partially-kept task whole (legacy group scoring). Rewrites
+    `traces.jsonl` to the kept rows via a temp file + atomic rename; a torn or
+    malformed row is owed again, never a crash."""
     path = resume_dir / TRACES_FILE
-    selected = set(selected_idxs)
+    targets = {
+        key: count * num_rollouts for key, count in Counter(selected_keys).items()
+    }
+    keyed = key_of if key_of is not None else task_key
 
     def parse(row: dict) -> Episode:
         if sniff_episode(row):
             return WireEpisode.model_validate(row)
         return Episode.of(WireTrace.model_validate(row))
 
     verdict = complete if complete is not None else (lambda episode: episode.ok)
-    good: dict[int, list[tuple[bytes, Episode]]] = defaultdict(list)
+    good: dict[K, list[tuple[bytes, Episode]]] = defaultdict(list)
     if path.exists():
         with path.open("rb") as results:
             for line in results:
@@ -89,37 +125,37 @@ def parse(row: dict) -> Episode:
                     except ValueError:
                         row = json.loads(line)
                     # The task rides each trace; a traceless record (a failure
-                    # before any trace minted) has no idx and is owed again.
+                    # before any trace minted) has no task and is owed again.
                     if sniff_episode(row):
-                        idx = row["traces"][0]["task"]["data"]["idx"]
+                        key = keyed(row["traces"][0]["task"]["data"])
                     else:
-                        idx = row["task"]["data"]["idx"]
+                        key = keyed(row["task"]["data"])
                 except (ValueError, KeyError, IndexError, TypeError):
                     # A torn final line (the run died mid-write) or a foreign shape
                     # is not a keepable rollout — it's owed again, never a crash.
                     continue
-                if idx not in selected or len(good[idx]) >= num_rollouts:
+                if key not in targets or len(good[key]) >= targets[key]:
                     continue
                 try:
                     episode = parse(row)
                     if not verdict(episode):
                         continue
                 except Exception:  # malformed row: redo it
                     continue
-                good[idx].append(
+                good[key].append(
                     (line if line.endswith(b"\n") else line + b"\n", episode)
                 )
     keep: list[bytes] = []
     episodes: list[Episode] = []
-    owed: dict[int, int] = {}
-    for idx in selected_idxs:
-        rows = good.get(idx, [])
-        if whole_task and len(rows) < num_rollouts:
+    owed: dict[K, int] = {}
+    for key, target in targets.items():
+        rows = good.get(key, [])
+        if whole_task and len(rows) < target:
             rows = []  # a partial unit redoes whole — its kept rows are dropped
         keep.extend(line for line, _ in rows)
         episodes.extend(episode for _, episode in rows)
-        if missing := num_rollouts - len(rows):
-            owed[idx] = missing
+        if missing := target - len(rows):
+            owed[key] = missing
     tmp = path.with_suffix(".jsonl.tmp")
     tmp.write_bytes(b"".join(keep))
     tmp.replace(path)

verifiers/v1/cli/eval/runner.py

@@ -32,21 +32,25 @@ async def run_eval(env: Env, config: EvalConfig) -> list[Episode]:
         asyncio.Semaphore(config.max_concurrent) if config.max_concurrent else None
     )
     out = output_path(config)
-    owed: dict[str, int] | None = None
+    # One (task, rollouts-to-run) pair per selected task; resume shrinks the counts.
+    plan = [(task, config.num_rollouts) for task in tasks]
     # Kept on-disk rollouts rejoin the run as finished episodes; only owed ones re-run.
     finished: list[Episode] = []
     if config.resume is not None:
-        finished, owed = resume.load(
-            out, [t.data.idx for t in tasks], config.num_rollouts, env.complete
-        )
+        keys = [
+            resume.task_key(t.data.model_dump(mode="json", exclude_none=True))
+            for t in tasks
+        ]
+        finished, owed = resume.load(out, keys, config.num_rollouts, env.complete)
         if not owed:  # already complete - report it and exit successfully
             print(resume.nothing_to_resume_msg(out, len(tasks), config.num_rollouts))
             raise SystemExit(0)
-        tasks = [task for task in tasks if owed.get(task.data.idx)]
+        counts = resume.distribute(keys, owed, config.num_rollouts)
+        plan = [(task, n) for task, n in zip(tasks, counts) if n]
         logger.info(
             "resuming %s: %d task(s), %d rollout(s) owed",
             out,
-            len(tasks),
+            len(plan),
             sum(owed.values()),
         )
     else:
@@ -70,13 +74,7 @@ async def on_complete(episode: Episode) -> None:
     # Serving resources (shared tool servers, interception) come up once for the
     # run; plan slots inside so the env's agents borrow them.
     async with env.serving():
-        planned = [
-            slot
-            for task in tasks
-            for slot in env.slots(
-                task, n=owed[task.data.idx] if owed else config.num_rollouts
-            )
-        ]
+        planned = [slot for task, n in plan for slot in env.slots(task, n=n)]
         slots = [RunSlot.finished(episode) for episode in finished] + planned
         push_state = None
         if config.push and config.rich:
@@ -123,6 +121,15 @@ 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
+
+        # 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,48 +158,57 @@ 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:
+        # A v1 run dispatches — and resumes — tasks by content: the client owns them,
+        # and `resume.task_key` is their identity. Only the legacy bridge is addressed
+        # by dataset row (its dataset lives server-side, reported via `info`), and
+        # 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)
+            plan = [({"task_idx": idx}, config.num_rollouts) for idx in idxs]
+        else:
+            group_scored = False
+            plan = [
+                ({"task_data": task.data.model_dump(mode="json")}, config.num_rollouts)
+                for task in tasks
+            ]
         out = output_path(config)
         finished: list[Episode] = []
         if config.resume is not None:
-            # (legacy only) a group is served and scored together, so a partially-kept
-            # task redoes as a whole group — whole_task drops its kept rows.
-            finished, owed = resume.load(
-                out, idxs, config.num_rollouts, whole_task=group_scored
-            )
+            if legacy:
+                # A group is served and scored together, so a partially-kept task
+                # redoes as a whole group — whole_task drops its kept rows.
+                finished, owed = resume.load(
+                    out,
+                    idxs,
+                    config.num_rollouts,
+                    whole_task=group_scored,
+                    key_of=lambda data: data.get("idx"),
+                )
+                counts = resume.distribute(idxs, owed, config.num_rollouts)
+            else:
+                keys = [
+                    resume.task_key(t.data.model_dump(mode="json", exclude_none=True))
+                    for t in tasks
+                ]
+                finished, owed = resume.load(out, keys, config.num_rollouts)
+                counts = resume.distribute(keys, owed, config.num_rollouts)
             if not owed:  # already complete - report it and exit successfully
-                print(resume.nothing_to_resume_msg(out, len(idxs), config.num_rollouts))
+                print(resume.nothing_to_resume_msg(out, len(plan), config.num_rollouts))
                 raise SystemExit(0)
-            idxs = [idx for idx in idxs if owed.get(idx)]
+            plan = [(payload, n) for (payload, _), n in zip(plan, counts) if n]
             logger.info(
                 "resuming %s: %d task(s), %d rollout(s) owed",
                 out,
-                len(idxs),
+                len(plan),
                 sum(owed.values()),
             )
         else:
-            owed = {idx: config.num_rollouts for idx in idxs}
             save_config(config, out)
             logger.info(
                 "running %dx%d rollouts via the env-server %s pool on %s",
-                len(idxs),
+                len(plan),
                 config.num_rollouts,
                 config.pool.type,
                 config.model,
@@ -224,26 +240,26 @@ async def run_group_unit(idx: int) -> list[Episode]:
                 records.append(Episode.of(trace))
             return records
 
-        async def run_unit(idx: int) -> list[Episode]:
+        async def run_unit(payload: dict) -> list[Episode]:
             async with semaphore or contextlib.nullcontext():
                 episode = await client.run(
-                    task_idx=idx,
                     client=config.client,
                     model=config.model,
                     sampling=config.sampling,
+                    **payload,
                 )
             for trace in episode.traces:
                 trace.stamp(EvalRunInfo(id=config.uuid))
             await append_episode(out, episode, write_lock)
             return [episode]
 
         # A group-scored legacy task runs its rollouts together (one `run_group`
-        # request, one worker); otherwise each rollout is its own `run_rollout`
-        # request, dispatched least-busy across workers.
+        # request, one worker); otherwise each rollout is its own `run` request,
+        # dispatched least-busy across workers.
         units = (
-            [run_group_unit(i) for i in idxs]
+            [run_group_unit(payload["task_idx"]) for payload, _ in plan]
             if group_scored
-            else [run_unit(i) for i in idxs for _ in range(owed[i])]
+            else [run_unit(payload) for payload, n in plan for _ in range(n)]
         )
         results = await asyncio.gather(*units)
         await client.close()

verifiers/v1/cli/gepa.py

@@ -66,7 +66,7 @@ def main(argv: list[str] | None = None) -> None:
     # Refuse multi-agent before the dry-run return, so --dry-run can't write a
     # config the real invocation would reject.
     env_cls = vf.environment_class(
-        config.env.taskset.id if config.env.taskset is not None else "",
+        config.env.taskset.id,
         config.env.id,
     )
     if not issubclass(env_cls, vf.SingleAgentEnv):