Persist and resume model-free task validation

verifiers/v1/cli/eval/main.py

@@ -7,7 +7,7 @@
 from pydantic_config import cli
 
 import verifiers.v1 as vf
-from verifiers.v1.cli.eval.resume import load_resume_config, split_resume
+from verifiers.v1.cli.eval.resume import load_resume_config
 from verifiers.v1.cli.eval.runner import run_eval
 from verifiers.v1.cli.output import output_path, write_config
 from verifiers.v1.cli.resolve import (
@@ -17,6 +17,7 @@
     references_config_file,
     with_positional_taskset,
 )
+from verifiers.v1.cli.resume import split_resume
 from verifiers.v1.configs.cli.eval import EvalConfig
 from verifiers.v1.utils.interrupt import install_interrupt
 from verifiers.v1.utils.logging import setup_logging
@@ -40,7 +41,7 @@ def main(argv: list[str] | None = None) -> None:
                 narrow_config(EvalConfig, argv)
             )  # full option help, narrowed to the given ids
         return
-    resume_dir, rest = split_resume(argv)
+    resume_dir, rest = split_resume(argv, "eval")
     # re-run a previous run's missing/errored rollouts, in place
     if resume_dir is not None:
         if rest:

verifiers/v1/cli/eval/resume.py

@@ -11,7 +11,6 @@
 legacy (v0) bridge still matches by row index (`key_of`).
 """
 
-import hashlib
 import json
 import tomllib
 from collections import Counter, defaultdict
@@ -22,50 +21,14 @@
 from pydantic_core import from_json
 
 from verifiers.v1.cli.output import CONFIG_FILE, TRACES_FILE, sniff_episode
+from verifiers.v1.cli.resume import task_key
 from verifiers.v1.configs.cli.eval import EvalConfig
 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).
-    The caller rejects any leftover args, since resume re-runs the saved config verbatim."""
-    for i, arg in enumerate(argv):
-        if arg == "--resume":
-            if i + 1 >= len(argv):
-                raise SystemExit(
-                    "--resume needs an output dir: uv run eval --resume <dir>"
-                )
-            return Path(argv[i + 1]), argv[:i] + argv[i + 2 :]
-        if arg.startswith("--resume="):
-            return Path(arg.split("=", 1)[1]), argv[:i] + argv[i + 1 :]
-    return None, argv
-
-
 def load_resume_config(resume_dir: Path) -> EvalConfig:
     """Rebuild the run's `EvalConfig` from its saved `config.toml`, pointed back at its own
     output dir so the resumed rollouts append to the same `traces.jsonl`."""

verifiers/v1/cli/eval/runner.py

@@ -14,6 +14,7 @@
     output_path,
     save_config,
 )
+from verifiers.v1.cli.resume import distribute, task_key
 from verifiers.v1.clients import ModelContext
 from verifiers.v1.configs.cli.eval import EvalConfig
 from verifiers.v1.env import Env, RunSlot
@@ -48,14 +49,13 @@ async def run_eval(env: Env, config: EvalConfig) -> list[Episode]:
     finished: list[Episode] = []
     if config.resume is not None:
         keys = [
-            resume.task_key(t.data.model_dump(mode="json", exclude_none=True))
-            for t in tasks
+            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)
-        counts = resume.distribute(keys, owed, config.num_rollouts)
+        counts = 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",
@@ -179,7 +179,7 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]:
         client = EnvClient(address=address)
         await client.wait_for_server_startup(timeout=600)
         # 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
+        # and `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:
@@ -209,14 +209,14 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]:
                     whole_task=group_scored,
                     key_of=lambda data: data.get("idx"),
                 )
-                counts = resume.distribute(idxs, owed, config.num_rollouts)
+                counts = distribute(idxs, owed, config.num_rollouts)
             else:
                 keys = [
-                    resume.task_key(t.data.model_dump(mode="json", exclude_none=True))
+                    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)
+                counts = distribute(keys, owed, config.num_rollouts)
             if not owed:  # already complete - report it and exit successfully
                 print(resume.nothing_to_resume_msg(out, len(plan), config.num_rollouts))
                 raise SystemExit(0)

verifiers/v1/cli/resume.py

@@ -0,0 +1,42 @@
+"""Resume primitives shared by eval-like CLIs."""
+
+import hashlib
+import json
+from collections.abc import Hashable, Mapping
+from pathlib import Path
+from typing import TypeVar
+
+K = TypeVar("K", bound=Hashable)
+
+
+def task_key(data: Mapping) -> str:
+    """Content identity for task wire data, independent of field order."""
+    return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
+
+
+def distribute(
+    selected_keys: list[K], owed: dict[K, int], num_results: int
+) -> list[int]:
+    """Spread each key's owed results over its selected instances, in order."""
+    remaining = dict(owed)
+    counts: list[int] = []
+    for key in selected_keys:
+        take = min(num_results, remaining.get(key, 0))
+        if take:
+            remaining[key] -= take
+        counts.append(take)
+    return counts
+
+
+def split_resume(argv: list[str], command: str) -> tuple[Path | None, list[str]]:
+    """Pull ``--resume <dir>`` from argv, returning the dir and other arguments."""
+    for i, arg in enumerate(argv):
+        if arg == "--resume":
+            if i + 1 >= len(argv):
+                raise SystemExit(
+                    f"--resume needs an output dir: uv run {command} --resume <dir>"
+                )
+            return Path(argv[i + 1]), argv[:i] + argv[i + 2 :]
+        if arg.startswith("--resume="):
+            return Path(arg.split("=", 1)[1]), argv[:i] + argv[i + 1 :]
+    return None, argv