feat: add vf-run, a minimal in-process runner

docs/getting_started.md

@@ -22,13 +22,14 @@ To work with other tasksets, including packages you build yourself following [bu
 
 ## Commands
 
-verifiers installs five CLI commands, which you run with `uv run` in the environment where verifiers and your taskset package are installed:
+verifiers installs six CLI commands, which you run with `uv run` in the environment where verifiers and your taskset package are installed:
 
 | Command | Use |
 | --- | --- |
 | [`vf-validate <taskset>`](debugging.md#before-running-an-agent) | Check task setup, a known gold solution, or the untouched task |
 | [`vf-debug <taskset>`](debugging.md#before-running-an-agent) | Run a command or script inside the prepared task runtime |
 | [`vf-replay <run-dir>`](debugging.md#re-score-without-rerunning-the-agent) | Re-score saved `vf-eval` traces without recreating their sandbox |
+| `vf-run <taskset>` | Run a few episodes in-process and save their traces, with a one-line summary |
 | [`vf-gepa <taskset>`](gepa.md) | Optimize a system prompt against task rewards |
 | [`vf-eval <taskset>`](evaluation.md) | Run the bundled evaluator |
 

pyproject.toml

@@ -151,6 +151,7 @@ renderers = false
 
 [project.scripts]
 vf-eval = "verifiers.v1.cli.eval.main:main"
+vf-run = "verifiers.v1.cli.run:main"
 vf-validate = "verifiers.v1.cli.validate:main"
 vf-debug = "verifiers.v1.cli.debug:main"
 vf-replay = "verifiers.v1.cli.replay:main"

tests/integration/conftest.py

@@ -4,24 +4,24 @@
 model-driven tests skip.
 
 Each behavior module fixes one task and fans it across the harness x runtime grid in
-`grid.py`, which also says why a cell is skipped. `run` drives the `Env` API in-process,
-the calls the eval CLI makes. The placement fixtures below are indirect-only: they
-translate a parametrized value, and using one without `indirect=True` fails loudly.
+`grid.py`, which also says why a cell is skipped. `run` drives `vf-run`'s in-process
+runner. The placement fixtures below are indirect-only: they translate a parametrized
+value, and using one without `indirect=True` fails loudly.
 """
 
-import asyncio
 import os
 from pathlib import Path
 
 import pytest
 from grid import agent_seats, runtime_config
 
+from verifiers.v1.cli.run import run_episodes
 from verifiers.v1.clients import EvalClientConfig, ModelContext
-from verifiers.v1.configs.cli.eval import EvalConfig
+from verifiers.v1.configs.cli.run import RunnerConfig
 from verifiers.v1.configs.harness import HarnessConfig
 from verifiers.v1.configs.sampling import SamplingConfig
 from verifiers.v1.episode import Episode
-from verifiers.v1.utils.loaders import harness_config_type, load_environment
+from verifiers.v1.utils.loaders import harness_config_type
 
 CI_MODEL = "internal/deepseek-v4.1-flash"
 CI_REASONING_EFFORT = "low"
@@ -117,8 +117,8 @@ def _eval_config(
     reasoning_effort: str | None = CI_REASONING_EFFORT,
     model: str = CI_MODEL,
     client: dict | None = None,
-) -> EvalConfig:
-    """Build the smallest `EvalConfig` that still exercises the path, used by the `run`
+) -> RunnerConfig:
+    """Build the smallest `RunnerConfig` that still exercises the path, used by the `run`
     fixture. `taskset_overrides` merges onto the
     `{id: ...}` config; `runtime` places the `agent` seat's harness (an agent field, not a
     harness one).
@@ -156,7 +156,7 @@ def _eval_config(
         seat_cfg.setdefault("timeout", {"rollout": rollout_timeout, "scoring": 60})
         # Agent runs retry locally; interactions retry with their whole episode.
         seat_cfg.setdefault("retries", retries)
-    return EvalConfig(
+    return RunnerConfig(
         env={
             "taskset": taskset_cfg,
             **env_cfg,
@@ -173,28 +173,14 @@ def _eval_config(
 
 
 @pytest.fixture
-def run():
-    """Run a taskset's episodes in-process through the `Env` API: the calls `vf-eval`
-    makes (`serving`, `slots`, `run_slot`, retries included), without its run
-    directory or dashboard. The Examples job covers that CLI layer."""
+def run(tmp_path):
+    """Run a taskset's episodes in-process with `vf-run`'s runner, saving them under
+    `tmp_path`."""
 
     async def _episodes(taskset: str, **kwargs) -> list[Episode]:
         config = _eval_config(taskset, **kwargs)
-        env = load_environment(config.env)
-        tasks = list(env.taskset.select(config.select))
-        ctx = ModelContext(
-            client=config.client, model=config.model, sampling=config.sampling
-        )
-        async with env.serving():
-            return list(
-                await asyncio.gather(
-                    *(
-                        env.run_slot(slot, ctx)
-                        for task in tasks
-                        for slot in env.slots(task, config.num_rollouts)
-                    )
-                )
-            )
+        config.output_dir = tmp_path
+        return await run_episodes(config)
 
     return _episodes
 

tests/integration/test_examples.py

@@ -1,4 +1,4 @@
-"""Smoke-eval every example taskset in `environments/` through the `eval` CLI.
+"""Smoke-run every example taskset in `environments/` through the `vf-run` CLI.
 
 Each taskset runs with its required harness for capped rollouts, so a broken
 example taskset fails CI. Interception exercises all four of its behaviors. `compact` is excluded (it's a harness, not a taskset);
@@ -90,10 +90,10 @@ def test_eval(taskset: str, tmp_path: Path):
     ]
     num_tasks = "4" if taskset == "interception" else "1"
     cmd = [
-        "uv", "run", "--no-sync", "vf-eval", taskset,
+        "uv", "run", "--no-sync", "vf-run", taskset,
         *model,
         "-n", num_tasks, "-r", "1", *caps,
-        "--sampling.max-tokens", "4096", "--no-rich",
+        "--sampling.max-tokens", "4096",
         "--output-dir", str(tmp_path), "--run.dir", "run",
         # Recipe envs run their seats one at a time by default; in parallel, a
         # multi-seat episode (proposer + n solvers) fits the timeout.

tests/unit/cli/test_run.py

@@ -0,0 +1,94 @@
+"""`vf-run` resolves a typed config from the CLI without a model."""
+
+import asyncio
+import contextlib
+import json
+
+import pytest
+
+from verifiers.v1.cli import run
+from verifiers.v1.cli.output import CONFIG_DIR, RESOLVED_DIR
+from verifiers.v1.cli.run import USAGE, main
+from verifiers.v1.configs.cli.run import RunnerConfig
+
+
+@pytest.fixture(autouse=True)
+def keep_stdlib_logging(monkeypatch):
+    """`main` routes logging through loguru for the whole process; keep other tests'
+    `caplog` working."""
+    monkeypatch.setattr(run, "setup_logging", lambda *args, **kwargs: None)
+
+
+def test_dry_run_writes_config(tmp_path):
+    main(
+        [
+            "reverse-text",
+            "-n",
+            "2",
+            "-r",
+            "3",
+            "-o",
+            str(tmp_path),
+            "--run.dir",
+            "run",
+            "--dry-run",
+        ]
+    )
+    run_dir = tmp_path / "run"
+    config = json.loads((run_dir / CONFIG_DIR / RESOLVED_DIR / "run.json").read_text())
+    assert config["env"]["taskset"]["id"] == "reverse-text"
+    assert config["select"]["limit"] == 2
+    assert config["num_rollouts"] == 3
+    assert not (run_dir / "traces.jsonl").exists()
+
+
+def test_refuses_a_run_dir_with_results(tmp_path):
+    run_dir = tmp_path / "run"
+    run_dir.mkdir()
+    (run_dir / "traces.jsonl").write_text("{}\n")
+    with pytest.raises(SystemExit, match="already contains results"):
+        main(["reverse-text", "-o", str(tmp_path), "--run.dir", "run", "--dry-run"])
+
+
+def test_needs_a_taskset():
+    with pytest.raises(SystemExit, match="usage") as exc:
+        main(["-n", "1"])
+    assert str(exc.value) == USAGE
+
+
+async def test_failed_rollout_cancels_siblings_before_teardown(tmp_path, monkeypatch):
+    events: list[str] = []
+
+    class Taskset(list):
+        bounded = True
+
+        def select(self, _select):
+            return self
+
+    class FakeEnv:
+        taskset = Taskset(["fail", "slow"])
+
+        def slots(self, task, n):
+            return [task] * n
+
+        async def run_slot(self, slot, ctx, semaphore, on_complete):
+            if slot == "fail":
+                await asyncio.sleep(0)
+                raise RuntimeError("boom")
+            try:
+                await asyncio.sleep(60)
+            finally:
+                events.append("sibling unwound")
+
+        @contextlib.asynccontextmanager
+        async def serving(self):
+            try:
+                yield
+            finally:
+                events.append("teardown")
+
+    monkeypatch.setattr(run, "load_environment", lambda env: FakeEnv())
+    config = RunnerConfig(env={"taskset": {"id": "reverse-text"}}, output_dir=tmp_path)
+    with pytest.raises(RuntimeError, match="boom"):
+        await run.run_episodes(config)
+    assert events == ["sibling unwound", "teardown"]

verifiers/v1/cli/output.py

@@ -3,10 +3,11 @@
 import json
 import os
 from pathlib import Path
+from typing import Protocol
 
 from pydantic import BaseModel
 
-from verifiers.v1.configs.cli.eval import EvalConfig
+from verifiers.v1.configs.cli.run import RunConfig
 from verifiers.v1.utils.trace_store import TRACES_FILE
 
 CONFIG_DIR = "configs"
@@ -60,7 +61,14 @@ def saved_config_path(run_dir: Path) -> Path | None:
     return candidates[0] if candidates else None
 
 
-def output_path(config: EvalConfig) -> Path:
+class RunOutput(Protocol):
+    """A config that owns a run directory: `output_dir / run.dir`."""
+
+    output_dir: Path
+    run: RunConfig
+
+
+def output_path(config: RunOutput) -> Path:
     """Where this run writes: `output_dir / run.dir` — the same grouping convention as
     training. The run directory defaults to the auto-generated run name
     (`<env>--<model>--<harness>--<short-id>`)."""

verifiers/v1/cli/run.py

@@ -0,0 +1,116 @@
+"""The `vf-run` entrypoint: run a taskset's episodes in-process and save their traces.
+
+Prints one summary line and exits non-zero when any episode is not ok, so CI can call
+it directly. For full evaluations, use prime-rl's `uv run eval`.
+"""
+
+import asyncio
+import logging
+import os
+import sys
+
+from pydantic_config import cli
+
+from verifiers.v1.cli.output import output_path, save_config, write_config
+from verifiers.v1.cli.resolve import (
+    extract_id,
+    narrow_config,
+    plugin_errors,
+    references_config_file,
+    with_positional_taskset,
+)
+from verifiers.v1.clients import ModelContext
+from verifiers.v1.configs.cli.run import RunnerConfig
+from verifiers.v1.episode import Episode
+from verifiers.v1.trace import EvalRunInfo
+from verifiers.v1.utils.interrupt import install_interrupt
+from verifiers.v1.utils.loaders import load_environment
+from verifiers.v1.utils.logging import setup_logging
+from verifiers.v1.utils.trace_store import TRACES_FILE, append_episode
+
+logger = logging.getLogger(__name__)
+
+USAGE = "usage: uv run vf-run [<taskset-id>] [--env.id <id>] [-m <model>] [-n <tasks>] [-r <rollouts>] [options] [@ file.toml]"
+
+
+async def run_episodes(config: RunnerConfig) -> list[Episode]:
+    """Run `num_rollouts` episodes per selected task and append each to `traces.jsonl`
+    as it completes."""
+    env = load_environment(config.env)
+    taskset = env.taskset.select(config.select)
+    if not taskset.bounded:
+        raise ValueError(
+            f"{type(taskset).__name__} is infinite - bound the run with -n"
+        )
+    out = output_path(config)
+    save_config(config, out, "run.json")
+    ctx = ModelContext(
+        client=config.client, model=config.model, sampling=config.sampling
+    )
+    semaphore = (
+        asyncio.Semaphore(config.max_concurrent) if config.max_concurrent else None
+    )
+    write_lock = asyncio.Lock()
+
+    async def on_complete(episode: Episode) -> None:
+        episode.record_run(EvalRunInfo(id=config.run.id, name=config.run.name))
+        await append_episode(out, episode, write_lock)
+
+    async with env.serving():
+        rollouts = [
+            asyncio.ensure_future(env.run_slot(slot, ctx, semaphore, on_complete))
+            for task in taskset
+            for slot in env.slots(task, config.num_rollouts)
+        ]
+        try:
+            return list(await asyncio.gather(*rollouts))
+        except BaseException:
+            # Unwind every sibling before `serving()` tears down the resources they use.
+            for rollout in rollouts:
+                rollout.cancel()
+            await asyncio.gather(*rollouts, return_exceptions=True)
+            raise
+
+
+def main(argv: list[str] | None = None) -> None:
+    argv = with_positional_taskset(list(sys.argv[1:]) if argv is None else list(argv))
+    if not argv or any(arg in ("-h", "--help") for arg in argv):
+        print(USAGE)
+        sys.argv = [sys.argv[0], "--help"]
+        with plugin_errors():
+            cli(narrow_config(RunnerConfig, argv))
+        return
+    if (
+        not extract_id(argv, "env.taskset")
+        and not references_config_file(argv)
+        and not any(a.startswith("--env.") for a in argv)
+    ):
+        raise SystemExit(USAGE)
+
+    with plugin_errors():
+        config_type = narrow_config(RunnerConfig, argv)
+        sys.argv = [sys.argv[0], *argv]
+        config = cli(config_type)
+    setup_logging("DEBUG" if config.verbose else "INFO")
+    out = output_path(config)
+    traces_file = out / TRACES_FILE
+    if traces_file.exists() and traces_file.stat().st_size > 0:
+        raise SystemExit(f"{out} already contains results - pick another --run.dir")
+    if config.dry_run:
+        logger.info("wrote config to %s", write_config(config, out, "run.json"))
+        return
+
+    # The run identity: every process this run spawns inherits it.
+    os.environ.setdefault("VF_RUN_ID", config.run.id)
+    # A first Ctrl-C raises KeyboardInterrupt so each rollout's teardown still runs.
+    install_interrupt()
+    try:
+        episodes = asyncio.run(run_episodes(config))
+    except KeyboardInterrupt:
+        raise SystemExit(130)
+    traces = [trace for episode in episodes for trace in episode.traces]
+    ok = sum(episode.ok for episode in episodes)
+    reward = sum(trace.reward for trace in traces) / len(traces) if traces else 0.0
+    print(f"{ok}/{len(episodes)} episodes ok, mean reward {reward:.3f}: {out}")
+    if ok < len(episodes):
+        raise SystemExit(1)