feat: run v0 environments on the eval CLI (legacy bridge)

verifiers/v1/cli/eval.py

@@ -28,7 +28,7 @@
 from verifiers.v1.cli.runner import run_eval
 from verifiers.v1.configs.eval import EvalConfig
 
-USAGE = "usage: uv run eval [<taskset-id>] [--harness.id <id>] [options] [@ file.toml]"
+USAGE = "usage: uv run eval [<taskset-id>] [--harness.id <id>] [--id <env-id> (legacy)] [options] [@ file.toml]"
 
 
 def main(argv: list[str] | None = None) -> None:
@@ -45,25 +45,39 @@ def main(argv: list[str] | None = None) -> None:
             narrow_config(EvalConfig, argv)
         )  # full option help, narrowed to the given ids
         return
-    if not extract_id(argv, "taskset") and not references_config_file(argv):
+    legacy_id = any(a == "--id" or a.startswith("--id=") for a in argv)  # v0 env id
+    if (
+        not extract_id(argv, "taskset")
+        and not legacy_id
+        and not references_config_file(argv)
+    ):
         raise SystemExit(
             USAGE
-        )  # need a taskset: a positional / --taskset.id, or one in @ file.toml
+        )  # need a taskset (positional / --taskset.id), a legacy --id, or a @ file.toml
 
     config_type = narrow_config(EvalConfig, argv)
     sys.argv = [sys.argv[0], *argv]  # let prime-pydantic-config render help/errors
     config = cli(config_type)
     if config.dry_run:  # resolved + validated; dump it and skip the run
         print(config.model_dump_json(indent=2, exclude_none=True))
         return
+    # The --rich dashboard reads live v1 Rollout state, so it's off for a legacy run.
+    rich = config.rich and not config.is_legacy
     # --rich owns the screen, so quiet the per-rollout INFO logs it would replace.
-    setup_logging("DEBUG" if config.verbose else "WARNING" if config.rich else "INFO")
+    setup_logging("DEBUG" if config.verbose else "WARNING" if rich else "INFO")
 
-    env = vf.Environment(config)
-    # Make SIGTERM behave like Ctrl-C (SIGINT) so a killed/timed-out eval still runs
-    # each rollout's `finally` — i.e. tears down its docker container / prime sandbox.
-    signal.signal(signal.SIGTERM, lambda *_: (_ for _ in ()).throw(KeyboardInterrupt()))
-    traces = asyncio.run(run_eval(env, config))
-    if not config.rich:  # --rich is the whole output; otherwise dump each trace as JSON
+    if config.is_legacy:  # v0 backwards-compat: run the classic env, bridged to Traces
+        from verifiers.v1.legacy import run_legacy_eval
+
+        traces = asyncio.run(run_legacy_eval(config))
+    else:
+        env = vf.Environment(config)
+        # Make SIGTERM behave like Ctrl-C (SIGINT) so a killed/timed-out eval still runs
+        # each rollout's `finally` — i.e. tears down its docker container / prime sandbox.
+        signal.signal(
+            signal.SIGTERM, lambda *_: (_ for _ in ()).throw(KeyboardInterrupt())
+        )
+        traces = asyncio.run(run_eval(env, config))
+    if not rich:  # --rich is the whole output; otherwise dump each trace as JSON
         for trace in traces:
             print(trace.model_dump_json(indent=2, exclude_none=True))

verifiers/v1/env.py

@@ -21,6 +21,7 @@
 from verifiers.v1.clients import RolloutContext
 from verifiers.v1.decorators import discover_decorated
 from verifiers.v1.episode import Episode
+from verifiers.v1.ids import EnvId
 from verifiers.v1.interception import RolloutLimits
 from verifiers.v1.retries import RetryConfig
 from verifiers.v1.rollout import Rollout
@@ -68,6 +69,33 @@ class EnvConfig(BaseConfig):
     max_total_tokens: int | None = None
     """Max total (prompt + completion) tokens per rollout (None = no limit). Caps the
     trace's `total_tokens`; framework-enforced between turns."""
+    # --- legacy (v0) backwards-compat -----------------------------------------
+    # Run a classic `verifiers.load_environment(id, **args)` env, bridged to v1 Traces (see
+    # `verifiers.v1.legacy`), instead of a v1 taskset/harness. Set `id` (leave `taskset`
+    # unset) to opt in; native v1 envs leave these untouched. Mirrors prime-rl's EnvConfig
+    # so it inherits these (a v0 env is driven the same way in eval and the env server).
+    id: EnvId | None = None
+    """Classic (v0) env id (`name`, `org/name`, or `org/name@version` — installed from the
+    hub on demand), loaded via `verifiers.load_environment` and run through the legacy
+    bridge. Set this *instead of* `taskset` to run a v0 environment."""
+    args: dict = {}
+    """Construction kwargs forwarded to `load_environment(id, **args)`."""
+    extra_env_kwargs: dict = {}
+    """Post-load kwargs applied to the v0 env via `env.set_kwargs(**extra_env_kwargs)` (e.g.
+    `max_total_completion_tokens`, `max_seq_len`, `timeout_seconds`) — typically
+    auto-populated by the orchestrator, distinct from the `args` passed at construction."""
+
+    @property
+    def is_legacy(self) -> bool:
+        """A v0/legacy env (run via the bridge): a legacy `id` is set and no v1 `taskset`."""
+        return self.id is not None and not self.taskset.id
+
+    @property
+    def env_id(self) -> str:
+        """The env identifier — the v1 taskset id, else the legacy v0 env id."""
+        return self.taskset.id or self.id or ""
+
+    # --- end legacy -----------------------------------------------------------
 
     @model_validator(mode="before")
     @classmethod

verifiers/v1/legacy.py

@@ -14,6 +14,7 @@
 """
 
 import logging
+from pathlib import Path
 from typing import Any
 
 import zmq
@@ -262,6 +263,7 @@ def __init__(
         env_id: str,
         env_args: dict | None = None,
         address: str = "tcp://127.0.0.1:5000",
+        extra_env_kwargs: dict | None = None,
     ) -> None:
         from verifiers import load_environment
         from verifiers.v1.ids import ensure_installed, env_name
@@ -272,6 +274,8 @@ def __init__(
         module = ensure_installed(env_id)
         self.taskset_id = env_name(env_id)
         self.env = load_environment(module, **(env_args or {}))
+        if extra_env_kwargs:  # post-load knobs applied via the v0 env's setters
+            self.env.set_kwargs(**extra_env_kwargs)
         # The formatted dataset rows are RolloutInputs (prompt + example_id); index by task_idx.
         self.dataset = self.env.get_dataset()
         self.tasks = self.dataset  # `len(self.tasks)` drives the `info` response
@@ -339,3 +343,102 @@ async def _run_group(self, req: RunGroupRequest) -> RunGroupResponse:
             out = await self._run_v0(req.task_idx, req.client, req.model, req.sampling)
             traces.append(rollout_output_to_trace(out, req.task_idx).to_wire())
         return RunGroupResponse(traces=traces)
+
+
+# --- in-process v0 eval (the `eval` CLI's `--legacy.id` path) ------------------
+
+
+def _eval_client(client_config: ClientConfig, model: str):
+    """A v0 chat-completions client built from the v1 eval `ClientConfig` (base url + api
+    key var + headers). Eval needs no token ids, so this skips the renderer pool the
+    training bridge (`_v0_client`) builds."""
+    from verifiers.clients import resolve_client
+    from verifiers.types import ClientConfig as V0ClientConfig
+
+    return resolve_client(
+        V0ClientConfig(
+            client_type="openai_chat_completions",
+            api_base_url=client_config.base_url,
+            api_key_var=client_config.api_key_var,
+            extra_headers=dict(getattr(client_config, "headers", None) or {}),
+        )
+    )
+
+
+def _legacy_output_dir(config) -> Path:
+    """The legacy run's output dir, mirroring the native `output_path` shape but keyed by
+    the v0 env id (`outputs/<id>--<model>--legacy/<uuid>`); honors `--output-dir`."""
+    from verifiers.v1.ids import env_name
+
+    if config.output_dir is not None:
+        return config.output_dir
+    name = f"{env_name(config.id)}--{config.model.replace('/', '--')}--legacy"
+    return Path("outputs") / name / config.uuid
+
+
+async def run_legacy_eval(config) -> list[Trace]:
+    """In-process v0 eval used by the `eval` CLI when `config.is_legacy` (a legacy `id` is
+    set, no v1 `taskset`).
+
+    Loads the v0 env, runs `num_rollouts` per task with bounded concurrency, maps each v0
+    `RolloutOutput` to a v1 `Trace` (`rollout_output_to_trace`), persists results as they
+    land (the same `results.jsonl` / `config.toml` a native run writes), and returns the
+    traces. The v0 env is run directly (`env.run_rollout`, no env server), so this needs no
+    runtime / interception server. All v0 specifics live here; the CLI only branches on
+    `config.is_legacy`."""
+    import asyncio
+    import random
+
+    from verifiers import load_environment
+
+    from verifiers.v1.cli.output import append_trace, save_config
+    from verifiers.v1.ids import ensure_installed
+
+    # Install from the env hub on demand for an `org/name[@version]` id (a local id is
+    # already importable), then load by module name.
+    env = load_environment(ensure_installed(config.id), **(config.args or {}))
+    if config.extra_env_kwargs:  # post-load knobs (max_total_completion_tokens, …)
+        env.set_kwargs(**config.extra_env_kwargs)
+    dataset = env.get_dataset()
+    idxs = list(range(len(dataset)))
+    if config.shuffle:
+        random.Random(0).shuffle(idxs)  # fixed seed: same sample every run
+    if config.num_tasks is not None:
+        idxs = idxs[: config.num_tasks]
+
+    client = _eval_client(config.client, config.model)
+    sampling_args = config.sampling.model_dump(exclude_none=True)
+    out_dir = _legacy_output_dir(config)
+    save_config(config, out_dir)
+    logger.info("results: %s", out_dir)
+    logger.info(
+        "running %dx%d v0 rollouts on %s (legacy: %s)",
+        len(idxs),
+        config.num_rollouts,
+        config.model,
+        config.id,
+    )
+
+    sem = asyncio.Semaphore(config.max_concurrent) if config.max_concurrent else None
+
+    async def run_one(task_idx: int) -> Trace:
+        async def go() -> Trace:
+            out = await env.run_rollout(
+                input=dict(dataset[task_idx]),
+                client=client,
+                model=config.model,
+                sampling_args=sampling_args,
+                state_columns=["trajectory"],
+            )
+            trace = rollout_output_to_trace(out, task_idx)
+            append_trace(out_dir, trace)
+            return trace
+
+        if sem is None:
+            return await go()
+        async with sem:
+            return await go()
+
+    # `num_rollouts` rollouts per selected task, all bounded by the one semaphore.
+    coros = [run_one(i) for i in idxs for _ in range(config.num_rollouts)]
+    return list(await asyncio.gather(*coros))