From 103a439c3aaba36fdc80beb66513798c5b897d34 Mon Sep 17 00:00:00 2001 From: Cooper Miller Date: Thu, 20 Aug 2026 13:29:11 -0700 Subject: [PATCH 01/11] feat: prime-runs --- pyproject.toml | 80 ++--- uv.lock | 28 +- verifiers/v1/cli/dashboard/eval.py | 21 +- verifiers/v1/cli/eval/main.py | 8 +- verifiers/v1/cli/eval/runner.py | 73 +++-- verifiers/v1/cli/resolve.py | 16 + verifiers/v1/configs/cli/eval.py | 25 +- verifiers/v1/utils/platform.py | 478 +++++++++++------------------ 8 files changed, 352 insertions(+), 377 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index e3b3d969ec..9543c1730a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,15 +4,28 @@ build-backend = "hatchling.build" [project] name = "verifiers" -authors = [ - {name = "William Brown", email = "williambrown97@gmail.com"}, -] -dynamic = ["version"] # derived from git tags via hatch-vcs +authors = [{ name = "William Brown", email = "williambrown97@gmail.com" }] +dynamic = ["version"] # derived from git tags via hatch-vcs description = "Verifiers: Environments for LLM Reinforcement Learning" readme = "README.md" license = "MIT" requires-python = ">=3.11,<3.14" -keywords = ["reinforcement-learning", "llm", "rl", "grpo", "environments", "multi-turn", "agents", "agentic-rl", "tool-use", "train", "eval", "harness", "verifiers", "rlvr"] +keywords = [ + "reinforcement-learning", + "llm", + "rl", + "grpo", + "environments", + "multi-turn", + "agents", + "agentic-rl", + "tool-use", + "train", + "eval", + "harness", + "verifiers", + "rlvr", +] classifiers = [ "Development Status :: 4 - Beta", "Intended Audience :: Developers", @@ -36,6 +49,7 @@ dependencies = [ "openai-agents>=0.8.2", "prime-tunnel>=0.1.8", "prime-sandboxes>=0.2.37", + "prime-runs>=0.1.0", "pydantic>=2.12.3", "requests", "rich>=11.0.0", @@ -96,30 +110,16 @@ examples = [ ] [project.optional-dependencies] -rg = [ - "reasoning-gym>=0.1.8", -] -ta = [ - "textarena==0.7.4", - "nltk>=3.9.2", -] -browser = [ - "stagehand>=3.4.0", - "python-dotenv>=1.0.0", -] +rg = ["reasoning-gym>=0.1.8"] +ta = ["textarena==0.7.4", "nltk>=3.9.2"] +browser = ["stagehand>=3.4.0", "python-dotenv>=1.0.0"] harbor = [ # Harbor currently publishes Python 3.12+ packages; keep core verifiers installable on 3.11. "harbor==0.21.0 ; python_full_version >= '3.12'", ] -modal = [ - "modal>=1.4.0", -] -notebook = [ - "nest-asyncio>=1.6.0", -] -openenv = [ - "openenv>=0.4.1", -] +modal = ["modal>=1.4.0"] +notebook = ["nest-asyncio>=1.6.0"] +openenv = ["openenv>=0.4.1"] nemo-gym = [ # NeMo Gym currently publishes Python 3.12+ packages; keep core verifiers installable on 3.11. "nemo-gym==0.4.0; python_version >= '3.12'", @@ -163,6 +163,8 @@ url = "https://pypi.org/simple" default = true [tool.uv.sources] +# TEMPORARY: prime-runs is not on PyPI yet +prime-runs = { git = "https://github.com/PrimeIntellect-ai/prime.git", subdirectory = "packages/prime-runs", branch = "feature/prime-runs-sdk" } compact = { path = "environments/compact", editable = true } glossary = { path = "environments/glossary", editable = true } deepwiki = { path = "environments/deepwiki", editable = true } @@ -192,6 +194,7 @@ ty = "2026-07-28T00:00:00Z" # PrimeIntellect-published on PyPI (trusted publisher) prime-tunnel = false prime-sandboxes = false +prime-runs = false prime-pydantic-config = false renderers = false @@ -233,18 +236,8 @@ git_describe_command = [ ] [tool.hatch.build.targets.sdist] -include = [ - "/verifiers", - "/tests", - "/README.md", - "/LICENSE", -] -exclude = [ - "/wandb", - "/scratch", - "/configs", - "/environments", -] +include = ["/verifiers", "/tests", "/README.md", "/LICENSE"] +exclude = ["/wandb", "/scratch", "/configs", "/environments"] [tool.hatch.build.targets.wheel] packages = ["verifiers"] @@ -269,12 +262,7 @@ pythonpath = ["tests/v1/fixtures"] python_files = ["test_*.py"] python_classes = ["Test*"] python_functions = ["test_*"] -addopts = [ - "--strict-markers", - "--tb=short", - "-ra", - "--quiet", -] +addopts = ["--strict-markers", "--tb=short", "-ra", "--quiet"] markers = [ "e2e: marks end-to-end eval-run tests (need a model API key; skipped without one)", "integration: marks tests as integration tests", @@ -334,11 +322,7 @@ not-iterable = "ignore" [tool.coverage.run] source = ["verifiers"] -omit = [ - "*/tests/*", - "*/__pycache__/*", - "*/site-packages/*", -] +omit = ["*/tests/*", "*/__pycache__/*", "*/site-packages/*"] [tool.coverage.report] exclude_lines = [ diff --git a/uv.lock b/uv.lock index 46fa9dd710..f1e62c44a4 100644 --- a/uv.lock +++ b/uv.lock @@ -18,13 +18,14 @@ exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for exclude-newer-span = "P7D" [options.exclude-newer-package] -prime-tunnel = false -prime-sandboxes = false +prime-pydantic-config = false harbor = "2026-08-11T00:00:00Z" ty = "2026-07-28T00:00:00Z" -prime-pydantic-config = false -renderers = false ruff = "2026-07-28T00:00:00Z" +prime-runs = false +prime-tunnel = false +prime-sandboxes = false +renderers = false [manifest] @@ -3331,6 +3332,15 @@ toml = [ { name = "tomli" }, ] +[[package]] +name = "prime-runs" +version = "0.1.0" +source = { git = "https://github.com/PrimeIntellect-ai/prime.git?subdirectory=packages%2Fprime-runs&branch=feature%2Fprime-runs-sdk#4a367dc08f5c56101fc16a458ac4a7895e879bd5" } +dependencies = [ + { name = "httpx" }, + { name = "prime-traces" }, +] + [[package]] name = "prime-sandboxes" version = "0.2.37" @@ -3350,6 +3360,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c3/4c/346abc72f6891267f490abfbe9de867c3ba2ed64ae8a4f964ffc4065492a/prime_sandboxes-0.2.37-py3-none-any.whl", hash = "sha256:9687e1b698c183798138b5e2ca116554773e3e300307a599beb7fbe8ed65e783", size = 49587, upload-time = "2026-08-17T17:07:17.403Z" }, ] +[[package]] +name = "prime-traces" +version = "0.0.2" +source = { git = "https://github.com/PrimeIntellect-ai/prime.git?subdirectory=packages%2Fprime-traces&branch=feature%2Fprime-runs-sdk#4a367dc08f5c56101fc16a458ac4a7895e879bd5" } +dependencies = [ + { name = "httpx" }, + { name = "pydantic" }, +] + [[package]] name = "prime-tunnel" version = "0.1.10" @@ -5100,6 +5119,7 @@ dependencies = [ { name = "openai" }, { name = "openai-agents" }, { name = "prime-pydantic-config", extra = ["toml"] }, + { name = "prime-runs" }, { name = "prime-sandboxes" }, { name = "prime-tunnel" }, { name = "pydantic" }, diff --git a/verifiers/v1/cli/dashboard/eval.py b/verifiers/v1/cli/dashboard/eval.py index 497eacb1c5..9ce65e2ec0 100644 --- a/verifiers/v1/cli/dashboard/eval.py +++ b/verifiers/v1/cli/dashboard/eval.py @@ -279,17 +279,22 @@ def Overview(config: EvalConfig) -> Table: def _push_footer(push: "PushState | None") -> Group | None: - """The `--push` status line under the rollouts, shown once the run finishes and the upload - begins: dim `Pushing traces...` while it runs, then white `Traces pushed ()` or red - `Trace push failed ()`. `None` (no line) until the upload starts and when `--push` is off.""" + """The `--push` status line under the rollouts. The run opens before the first rollout and + its traces stream up as they land, so the line carries the run's URL for the whole eval: + dim `Pushing traces ()` while rollouts are still going, then white `Traces pushed + ()` — with anything that degraded along the way appended in yellow — or red `Trace + push failed ()` when there is no run to show. `None` (no line) when `--push` is off + or the run stayed local.""" if push is None or not push.started: return None - if not push.done: - line = Text("Pushing traces...", style="dim") - elif push.url: - line = Text(f"Traces pushed ({push.url})", style="white", overflow="fold") - else: + if push.error: line = Text(f"Trace push failed ({push.error})", style="red", overflow="fold") + elif not push.done: + line = Text(f"Pushing traces ({push.url})", style="dim", overflow="fold") + else: + line = Text(f"Traces pushed ({push.url})", style="white", overflow="fold") + if push.warning: # pushed, but not all of it - say what went wrong + line.append(f" {push.warning}", style="yellow") return Group(Rule(style="dim"), line) diff --git a/verifiers/v1/cli/eval/main.py b/verifiers/v1/cli/eval/main.py index 68d4bdba36..8ba93e24da 100644 --- a/verifiers/v1/cli/eval/main.py +++ b/verifiers/v1/cli/eval/main.py @@ -18,6 +18,7 @@ write_config, ) from verifiers.v1.cli.resolve import ( + config_file_ref, extract_id, narrow_config, plugin_errors, @@ -68,6 +69,9 @@ def main(argv: list[str] | None = None) -> None: *argv, ] # let prime-pydantic-config render help/errors config = cli(config_type) + # The `@ eval.toml` this run was launched from — uploaded verbatim with the + # run, so its Config tab shows what someone actually wrote. + config.run.record_source(config_file_ref(argv)) # A named run directory is re-entered only by `--resume` or wiped by `--clean`: any # other write into it — the dry-run config.toml included, which would clobber the # config a resume typically re-runs — would overwrite the previous run. @@ -146,10 +150,6 @@ def main(argv: list[str] | None = None) -> None: # Graceful cleanup has already run (each rollout's `finally`); partial results are on # disk. Exit on the conventional Ctrl-C code without a traceback. raise SystemExit(130) - if config.push and not config.rich: - from verifiers.v1.utils.platform import push_traces - - push_traces(episodes, config) if not config.rich: # --rich is the whole output; otherwise dump each trace as JSON for episode in episodes: for trace in episode.traces: diff --git a/verifiers/v1/cli/eval/runner.py b/verifiers/v1/cli/eval/runner.py index 1d9bd4210c..b666cd76bb 100644 --- a/verifiers/v1/cli/eval/runner.py +++ b/verifiers/v1/cli/eval/runner.py @@ -4,6 +4,7 @@ import contextlib import logging import time +from typing import Any from verifiers.v1.cli.dashboard import dashboard from verifiers.v1.cli.eval import resume @@ -18,10 +19,17 @@ from verifiers.v1.env import Env, RunSlot from verifiers.v1.episode import Episode from verifiers.v1.trace import EvalRunInfo +from verifiers.v1.utils.platform import PushState, abort_run, finish_run, open_run logger = logging.getLogger(__name__) +def record_run(episode: "Episode[Any, Any, Any]", config: EvalConfig) -> None: + """Stamp the run onto an episode's traces — the id the platform knows it by.""" + for trace in episode.traces: + trace.record_run(EvalRunInfo(id=config.run.id, name=config.run.name)) + + async def run_eval(env: Env, config: EvalConfig) -> list[Episode]: logger.info("eval config:\n%s", config.model_dump_json(indent=2)) taskset = env.taskset @@ -70,39 +78,49 @@ async def run_eval(env: Env, config: EvalConfig) -> list[Episode]: logger.info("results: %s", out) write_lock = asyncio.Lock() + push_state = PushState() if config.push and config.rich else None + # Opened before the first rollout so the platform's id *is* the run's id: + # every trace is stamped with it once, at rollout time, and nothing is + # re-stamped or rewritten afterwards. + run = open_run(config, push_state) + config.run.adopt_id(run.id) + # A resume's kept rollouts are part of this run too, so they carry its id and + # go up with the rest — otherwise the platform would hold half a run. + for episode in finished: + record_run(episode, config) + run.log_traces(finished) async def on_complete(episode: Episode) -> None: - for trace in episode.traces: - trace.record_run(EvalRunInfo(id=config.run.id, name=config.run.name)) + record_run(episode, config) await append_episode(out, episode, write_lock) + # A queue put, but a bounded one: handing it to a thread keeps a full + # queue from stalling every other rollout (and freezing the dashboard). + await asyncio.to_thread(run.log_traces, [episode]) # 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, 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: - from verifiers.v1.utils.platform import PushState - - push_state = PushState() display = ( dashboard(slots, config, start, push=push_state) if config.rich else contextlib.nullcontext() ) async with display: - results = await asyncio.gather( - *(env.run_slot(slot, ctx, semaphore, on_complete) for slot in planned) - ) + try: + results = await asyncio.gather( + *( + env.run_slot(slot, ctx, semaphore, on_complete) + for slot in planned + ) + ) + except BaseException as e: # a killed eval must not sit at running + await asyncio.to_thread(abort_run, run, e, push_state) + raise episodes = finished + list(results) - if ( - push_state is not None - ): # upload off the event loop so the view keeps refreshing - from verifiers.v1.utils.platform import push_traces - - push_state.started = True - await asyncio.to_thread(push_traces, episodes, config, push_state) + # Drain and close out off the event loop so the view keeps refreshing. + await asyncio.to_thread(finish_run, run, episodes, push_state) return episodes @@ -196,6 +214,13 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]: asyncio.Semaphore(config.max_concurrent) if config.max_concurrent else None ) write_lock = asyncio.Lock() + # Same contract as the in-process runner: the run opens before the first + # rollout, so its id is the one every trace carries. + run = open_run(config) + config.run.adopt_id(run.id) + for episode in finished: + record_run(episode, config) + run.log_traces(finished) async def run_unit(payload: dict) -> list[Episode]: async with semaphore or contextlib.nullcontext(): @@ -205,16 +230,22 @@ async def run_unit(payload: dict) -> list[Episode]: sampling=config.sampling, **payload, ) - for trace in episode.traces: - trace.record_run(EvalRunInfo(id=config.run.id, name=config.run.name)) + record_run(episode, config) await append_episode(out, episode, write_lock) + await asyncio.to_thread(run.log_traces, [episode]) return [episode] # Each rollout is its own `run` request, dispatched least-busy across workers. units = [run_unit(payload) for payload, n in plan for _ in range(n)] - results = await asyncio.gather(*units) + try: + results = await asyncio.gather(*units) + except BaseException as e: # a killed eval must not sit at running + await asyncio.to_thread(abort_run, run, e) + raise await client.close() - return finished + [record for unit in results for record in unit] + episodes = finished + [record for unit in results for record in unit] + await asyncio.to_thread(finish_run, run, episodes) + return episodes finally: proc.terminate() with contextlib.suppress(Exception): diff --git a/verifiers/v1/cli/resolve.py b/verifiers/v1/cli/resolve.py index dc2b671943..eb62705924 100644 --- a/verifiers/v1/cli/resolve.py +++ b/verifiers/v1/cli/resolve.py @@ -46,6 +46,22 @@ def references_config_file(argv: list[str]) -> bool: return any(arg.startswith("@") for arg in argv) +def config_file_ref(argv: list[str]) -> str | None: + """The file a run was launched from — the root-level `@ ` — or None. + + Only the root form counts: `--env @ env.toml` configures one block, whereas + `@ eval.toml` *is* the run's config, which is what gets recorded on the run — + so a `@` right after a flag is somebody else's file. Several root files merge + into one config that no single path describes, so that case records nothing + rather than half of it.""" + paths = [ + argv[i + 1] + for i, arg in enumerate(argv) + if arg == "@" and i + 1 < len(argv) and not (i and argv[i - 1].startswith("--")) + ] + return paths[0] if len(paths) == 1 else None + + def extract_id(argv: list[str], field: str, default: str = "") -> str: """The chosen `.id` from `--.id ` (or `=`) on the CLI, before the typed parse (the positional taskset shorthand is applied upstream). Two diff --git a/verifiers/v1/configs/cli/eval.py b/verifiers/v1/configs/cli/eval.py index 95017fa9fa..039a75101b 100644 --- a/verifiers/v1/configs/cli/eval.py +++ b/verifiers/v1/configs/cli/eval.py @@ -40,13 +40,36 @@ class RunConfig(BaseConfig): """Run directory name — the run writes to `output_dir / dir`. Defaults to `run.name`; set it only when the directory should differ from the display name.""" - # TODO: fetch the id from the Prime SDK once runs are registered there. + # The platform's run id once `prime_runs.init()` has opened the run (see + # `adopt_id`); the local uuid until then, and for a run that never reaches + # the platform. Private so it stays out of the saved config and its digest — + # two runs of the same config differ by id, and resume compares configs. _id: str = PrivateAttr(default_factory=lambda: str(uuid4())) + _source: str | None = PrivateAttr(default=None) + """The `@ file.toml` this run was launched from, recorded by the CLI.""" + @property def id(self) -> str: return self._id + @property + def source(self) -> str | None: + return self._source + + def adopt_id(self, run_id: str) -> None: + """Take the platform's run id as this run's id. + + Called once, before the first rollout, so that every trace is stamped + with the id the platform knows the run by — one id, minted in one place, + never re-stamped afterwards.""" + self._id = run_id + + def record_source(self, path: str | None) -> None: + """Remember the config file the run was launched from, so it can be + uploaded verbatim with the run.""" + self._source = path + class EvalConfig(BaseConfig): env: SerializeAsAny[EnvConfig] = SingleAgentEnvConfig() diff --git a/verifiers/v1/utils/platform.py b/verifiers/v1/utils/platform.py index d81ff7fd92..c57d00280e 100644 --- a/verifiers/v1/utils/platform.py +++ b/verifiers/v1/utils/platform.py @@ -1,317 +1,213 @@ -"""Push a finished eval run to the Prime Intellect platform (`--no-push` to skip). - -Uploads one sample per v1 `Episode` over the `/evaluations/` API (create -> push -samples -> finalize). Each sample keeps the complete native Episode as its source -of truth and includes a flat summary for older Platform consumers. Auth + base URL -come from `$PRIME_API_KEY` / `~/.prime/config.json`. +"""The eval's run on the Prime Intellect platform (`--no-push` to keep it local). + +`prime-runs` owns the run lifecycle now. `open_run` opens the run *before* the +first rollout, so the platform's id is `config.run.id` and every trace carries +it; episodes stream out through `run.log_traces` as they land, instead of being +held in memory and posted in one blast at the end; `finish_run` writes the +terminal status. Ctrl-C, a crash or an exit that never reaches `finish_run` are +reported too, so a killed eval stops showing as running forever. + +What lives here is only the verifiers half of that wiring: which config fields +are the run's identity, what counts as the run's config, and the dashboard's +view of the upload. The batching, retries and the v0 sample projection are the +SDK's — `trace_to_sample`/`build_samples` moved to `prime_runs.projection`, +which is the package that owns that wire format, and are re-exported here for +callers that still import them from verifiers. """ -import json +import asyncio import logging -import os from dataclasses import dataclass from typing import Any -import httpx +import prime_runs as pr +from prime_runs.projection import build_samples, trace_to_sample from verifiers.v1.configs.cli.eval import EvalConfig from verifiers.v1.episode import Episode -from verifiers.v1.trace import Trace -from verifiers.v1.utils.prime import load_prime_config logger = logging.getLogger(__name__) -DEFAULT_API_URL = "https://api.primeintellect.ai" -DEFAULT_FRONTEND_URL = "https://app.primeintellect.ai" -# Repeated /samples posts append; match the Prime Evals client's request ceiling. -_MAX_SAMPLES_PAYLOAD_BYTES = 25 * 1024 * 1024 - +FRAMEWORK = "verifiers" -def json_bytes(value: Any) -> int: - return len( - json.dumps( - value, - ensure_ascii=False, - separators=(",", ":"), - allow_nan=False, - ).encode("utf-8") - ) +__all__ = [ + "PushState", + "abort_run", + "build_samples", + "finish_run", + "open_run", + "trace_to_sample", +] @dataclass class PushState: - """Mutable upload status shared with the dashboard.""" - - started: bool = False - done: bool = False - url: str | None = None - error: str | None = None + """The dashboard's view of the upload. Owns no I/O — the run does. + Reads through to the live run, so the footer can show the run's URL from the + moment it opens rather than only once everything has been uploaded.""" -def trace_to_sample( - trace: Trace, rollout_number: int = 1, episode_id: str | None = None -) -> dict[str, Any]: - """One trace -> the platform's sample dict (the v0 eval-sample format). - - The hub table stays flat — one row per trace; its episode is denormalized onto - the row (`episode_id` from the envelope, plus the trace's own `agent`/`trainable`), - so a multi-trace rollout's grouping travels with each row without a nested - schema. No prompt/completion split (meaningless mid-branch): `completion` is the - final branch's messages, `trajectory` one message list per branch.""" - - def dump(messages): - return [m.model_dump(mode="json", exclude_none=True) for m in messages] - - task = trace.task.data.model_dump(mode="json", exclude_none=True) - branches = trace.branches - sample = { - "sample_id": trace.id, - "example_id": trace.task.data.idx, - "rollout_number": rollout_number, - "episode_id": episode_id, - "agent": trace.agent.name, - "trainable": trace.agent.trainable, - "task": task, - "prompt": [], - "completion": dump(branches[-1].messages) if branches else [], - "answer": task.get("answer"), - # Keyed `tool_defs` because the v0 sample format already carries it there. - "tool_defs": [t.model_dump(mode="json", exclude_none=True) for t in trace.tools] - if trace.tools - else None, - "reward": trace.reward, - "timing": trace.timing.model_dump(mode="json", exclude_none=True), - "is_completed": trace.is_completed, - "is_truncated": trace.is_truncated, - "metrics": trace.metrics, - "error": trace.last_error.model_dump(mode="json", exclude_none=True) - if trace.last_error - else None, - "stop_condition": trace.stop_condition, - "trajectory": [ - { - "messages": dump(branch.messages), - "num_input_tokens": branch.num_input_tokens, - "num_output_tokens": branch.num_output_tokens, - } - for branch in branches - ], - "token_usage": trace.usage.model_dump(mode="json", exclude_none=True) - if trace.usage - else None, - "info": dict(trace.info) or None, - } - # Flatten sub-rewards to top-level keys the way v0 does (raw scores, as v0's - # per-function outputs were); env metrics stay nested. - for name, reward in trace.rewards.items(): - if reward is not None: - sample.setdefault(name, reward.score) - return sample - - -def credentials() -> tuple[str | None, str, str, str | None]: - """(api_key, api_base, frontend_url, team_id) from env vars / `~/.prime/config.json`.""" - cfg = load_prime_config() - api_key = os.getenv("PRIME_API_KEY") or cfg.get("api_key") - base = ( - os.getenv("PRIME_API_BASE_URL") - or os.getenv("PRIME_BASE_URL") - or cfg.get("base_url") - or DEFAULT_API_URL - ) - base = base.rstrip("/").removesuffix("/api/v1") - frontend = ( - os.getenv("PRIME_FRONTEND_URL") - or cfg.get("frontend_url") - or DEFAULT_FRONTEND_URL - ) - team_id = os.getenv("PRIME_TEAM_ID") or cfg.get("team_id") - return api_key, base, frontend, team_id - - -def run_metrics(episodes: list[Episode], traces: list[Trace]) -> dict[str, Any]: - """Run-level aggregates as v0's `GenerateMetadata`. Rewards/metrics aggregate - over the trainable traces only — fixed agents (a judge, a modeled user) often - carry no rewards and would dilute every mean with structural zeros — falling - back to all traces when none are trainable (same rule as the dashboard). - `avg_error` is the share of EPISODES that aren't ok: a hook failure counts - even when its traces are clean or it left none.""" - scored = [t for t in traces if t.agent.trainable] or traces - sums: dict[str, float] = {} - counts: dict[str, int] = {} - for trace in scored: - scores = { - name: reward.score - for name, reward in trace.rewards.items() - if reward is not None - } - metrics = { - name: value for name, value in trace.metrics.items() if value is not None - } - for name, value in {**scores, **metrics}.items(): - sums[name] = sums.get(name, 0.0) + value - counts[name] = counts.get(name, 0) + 1 - n = len(scored) - avg_error = sum(not e.ok for e in episodes) / len(episodes) if episodes else 0.0 - return { - "avg_reward": sum(t.reward for t in scored) / n if n else 0.0, - "avg_metrics": {name: sums[name] / counts[name] for name in sums}, - "avg_error": avg_error, - } - - -def build_samples(episodes: list[Episode]) -> list[dict[str, Any]]: - """One Platform sample per Episode, with a legacy-compatible trace summary. + run: "pr.Run | None" = None + error: str | None = None + """Why there is no run — it could not be opened, or could not be closed out. + Not set for a run that pushed with something degraded along the way; that is + a `warning`, since the records mostly landed.""" + done: bool = False - The native Episode in `info.native_wrapper` is authoritative and contains every - trace. One trainable trace (or the first trace) supplies only the flat summary - used by older consumers. `native_trace_index` identifies that summary trace. - """ - counts: dict[int, int] = {} - samples = [] - for episode in episodes: - if not episode.traces: - continue - summary_trace_index = next( - ( - index - for index, candidate in enumerate(episode.traces) - if candidate.agent.trainable - ), - 0, + @property + def url(self) -> str | None: + """Where to watch the run. `None` for a run that stays local.""" + return self.run.url if self.run is not None else None + + @property + def started(self) -> bool: + """Whether there is anything to report: a live run, or why there isn't one.""" + return self.error is not None or self.url is not None + + @property + def warning(self) -> str | None: + """What went wrong without sinking the upload, or `None` if nothing did. + + Records that were lost first, and how: dropped records reached no sink at + all (the rollouts outran the uploader), while a sink failing says nothing + about the others — with traces and samples both on, those records are + usually still safe in the one that worked. Failing that, the first thing + the SDK contained (`on_error="warn"`), which would otherwise be visible + only in the run's log file.""" + if self.run is None: + return None + parts = ( + [f"{self.run.dropped_records} dropped"] if self.run.dropped_records else [] ) - summary_trace = episode.traces[summary_trace_index] - idx = summary_trace.task.data.idx - counts[idx] = number = counts.get(idx, 0) + 1 - sample = trace_to_sample(summary_trace, number, episode.id) - sample["sample_id"] = episode.id - sample["info"] = { - **(sample["info"] or {}), - "native_wrapper": episode.to_record(), - "native_trace_index": summary_trace_index, - } - if len(b'{"samples":[]}') + json_bytes(sample) <= _MAX_SAMPLES_PAYLOAD_BYTES: - samples.append(sample) - continue + parts += [ + f"{count} failed via {sink}" + for sink, count in sorted(self.run.failed_records.items()) + if count + ] + if parts: + return ", ".join(parts) + return self.run.errors[0] if self.run.errors else None + + +def open_run(config: EvalConfig, state: PushState | None = None) -> "pr.Run": + """Open the run this eval streams into, before the first rollout. + + Never raises: the eval is the point and the upload is not, so a platform + that won't answer (or a missing `PRIME_API_KEY`) degrades to a disabled run + — a real handle whose calls are no-ops — and the reason lands on `state` for + the dashboard. `--no-push` takes the same path, so there is one set of call + sites either way. The run's id is the caller's to adopt (`config.run.adopt_id`).""" + identity: dict[str, Any] = { + "name": config.run.name, + # The environment is resolved by name through the hub's get-or-create, so + # a local env uploads without a prior `prime env push`. A run with no + # taskset has nothing to attach to and can only be a local run — say so + # by passing none, rather than asking the hub to resolve an empty name. + "environments": [config.env.taskset.id] if config.env.taskset.id else [], + "model": config.model, + "framework": FRAMEWORK, + "config": run_config(config), + # verifiers installs its own SIGINT/SIGTERM handler (`install_interrupt`) + # so a killed eval still tears down its sandboxes; the runner reports the + # terminal status from the unwind rather than letting the SDK take the + # signal. The SDK's atexit hook still catches an exit that gets neither. + "handle_signals": False, + } + if config.push: + try: + run = pr.init(mode="online", **identity) + if state is not None: + state.run = run + return run + except Exception as e: # noqa: BLE001 - a failed upload must not fail the eval + logger.warning( + "--push: could not open the run (%s: %s); running without it", + type(e).__name__, + e, + ) + if state is not None: + state.error = f"{type(e).__name__}: {e}" + run = pr.init(mode="disabled", **identity) + if state is not None: + state.run = run + return run + + +def run_config(config: EvalConfig) -> dict[str, Any]: + """What the run was configured with — the fields somebody actually set, plus + the file it was launched from, kept byte for byte. + + `exclude_unset` is the point: a resolved `EvalConfig` dump is hundreds of + defaults nobody chose, and the two or three values that were the experiment + are invisible in it. A `@ eval.toml` is stored verbatim under the SDK's + reserved `config_source` key — comments, key order and section grouping + included — because that file *is* the run's configuration and nothing + reconstructed from a dump can reproduce it.""" + values: dict[str, Any] = config.model_dump(mode="json", exclude_unset=True) + source = config.run.source + if source is not None: + try: + values[pr.CONFIG_SOURCE_KEY] = pr.ConfigSource.from_file(source).to_dict() + except pr.ConfigurationError as e: + logger.warning("--push: not recording the run's config file (%s)", e) + return values + + +def finish_run( + run: "pr.Run", episodes: list[Episode], state: PushState | None = None +) -> None: + """Drain the queued episodes, write the run's aggregates and close it out. + + Blocking — call it off the event loop (`asyncio.to_thread`) so the dashboard + keeps refreshing while the last uploads land.""" + try: + summary = pr.metrics.from_episodes(episodes) + except Exception as e: # noqa: BLE001 - close the run even without its headline logger.warning( - "Episode %s exceeds the Platform sample limit; uploading projected traces", - episode.id, + "--push: could not aggregate the run's metrics (%s: %s)", + type(e).__name__, + e, ) - samples.extend( - trace_to_sample(candidate, number, episode.id) - for candidate in episode.traces + summary = None + _close(run, state, summary=summary) + + +def abort_run( + run: "pr.Run", error: BaseException, state: PushState | None = None +) -> None: + """Close the run out after the eval broke, so it doesn't sit at running. + + Ctrl-C is a decision, not a fault: it lands as `crashed` (the process stopped + without the run ever saying), which is what tells whoever looks whether to + read the run's own error or go look at the machine it ran on. A cancellation + is the same thing arriving from the other direction.""" + if isinstance(error, (KeyboardInterrupt, asyncio.CancelledError)): + status, message = pr.RunStatus.CRASHED, "interrupted" + else: + status, message = pr.RunStatus.FAILED, f"{type(error).__name__}: {error}" + _close(run, state, status=status, error=message) + + +def _close( + run: "pr.Run", + state: PushState | None, + summary: dict[str, Any] | None = None, + status: "pr.RunStatus" = pr.RunStatus.COMPLETED, + error: str | None = None, +) -> None: + """`run.finish()` with the same best-effort contract as the rest of this + module: the eval's results are already on disk, so nothing here may raise.""" + try: + run.finish(summary, status=status, error=error) + except Exception as e: # noqa: BLE001 - the run is over; report, don't raise + logger.warning( + "--push: could not close out the run (%s: %s)", type(e).__name__, e ) - return samples - - -def push_traces( - episodes: list[Episode], - config: EvalConfig, - state: "PushState | None" = None, -) -> str | None: - """Upload a finished run to the platform; return the viewer URL (None if - skipped/failed). Resolves the env by name (get-or-create, so a local run - uploads without a prior `prime env push`); when `state` is given, records the - outcome on it so the dashboard's status line resolves.""" - - def finish(url: str | None = None, error: str | None = None) -> str | None: + if state is not None and state.error is None: + state.error = f"{type(e).__name__}: {e}" + else: + if run.url: + logger.info("--push: %s -> %s", status.value, run.url) + finally: if state is not None: - state.url = url - state.error = error state.done = True - return url - - api_key, base, frontend, team_id = credentials() - if not api_key: - logger.warning( - "--push: no PRIME_API_KEY (set it or run `prime login`); skipping upload" - ) - return finish(error="no PRIME_API_KEY (run `prime login`)") - - traces = [trace for episode in episodes for trace in episode.traces] - env_name = config.env.taskset.id - metrics = run_metrics(episodes, traces) - num_examples = len({t.task.data.idx for t in traces}) - metadata = { - "framework": "verifiers", - "run_id": config.run.id, - "model": config.model, - "num_examples": num_examples, - "rollouts_per_example": config.num_rollouts, - **metrics, - } - - team = {"team_id": team_id} if team_id else {} - api = f"{base}/api/v1" - headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} - # The run is done and its results saved; a network blip here must not crash it - # — log and skip the upload instead. - try: - samples = build_samples(episodes) - batches: list[list[dict[str, Any]]] = [] - batch: list[dict[str, Any]] = [] - payload_bytes = len(b'{"samples":[]}') - for i, sample in enumerate(samples): - sample_bytes = json_bytes(sample) - sample_payload_bytes = len(b'{"samples":[]}') + sample_bytes - if sample_payload_bytes > _MAX_SAMPLES_PAYLOAD_BYTES: - raise ValueError( - f"sample {i} is too large to upload " - f"({sample_payload_bytes} > " - f"{_MAX_SAMPLES_PAYLOAD_BYTES} bytes)" - ) - next_payload_bytes = payload_bytes + (1 if batch else 0) + sample_bytes - if batch and next_payload_bytes > _MAX_SAMPLES_PAYLOAD_BYTES: - batches.append(batch) - batch = [] - payload_bytes = len(b'{"samples":[]}') - next_payload_bytes = payload_bytes + sample_bytes - batch.append(sample) - payload_bytes = next_payload_bytes - if batch or not samples: - batches.append(batch) - - with httpx.Client(headers=headers, timeout=300.0) as client: - - def post(path: str, body: dict) -> dict: - resp = client.post(f"{api}{path}", json=body) - resp.raise_for_status() - return resp.json() - - env_id = post("/environmentshub/resolve", {"name": env_name, **team})[ - "data" - ]["id"] - eval_id = post( - "/evaluations/", - { - "name": config.run.name, - "environments": [{"id": env_id}], - "model_name": config.model, - "dataset": env_name, - "framework": "verifiers", - "metadata": metadata, - "metrics": metrics, - "tags": [], - **team, - }, - )["evaluation_id"] - for batch in batches: - body = json.dumps( - {"samples": batch}, - ensure_ascii=False, - separators=(",", ":"), - allow_nan=False, - ).encode("utf-8") - resp = client.post( - f"{api}/evaluations/{eval_id}/samples", - content=body, - ) - resp.raise_for_status() - post(f"/evaluations/{eval_id}/finalize", {"metrics": metrics}) - except Exception as e: # noqa: BLE001 - push is best-effort across the full upload - logger.warning("--push: upload failed (%s: %s); skipping", type(e).__name__, e) - return finish(error=f"{type(e).__name__}: {e}") - - url = f"{frontend}/dashboard/evaluations/{eval_id}" - logger.info("--push: uploaded %d samples -> %s", len(samples), url) - return finish(url=url) From a7ff2ea31f74e3ba29e84282df3858da2d8e676a Mon Sep 17 00:00:00 2001 From: Cooper Miller Date: Thu, 20 Aug 2026 13:44:25 -0700 Subject: [PATCH 02/11] chore: revert pyproject --- pyproject.toml | 76 +++++++++++++++++++++++++++++++------------------- 1 file changed, 48 insertions(+), 28 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 9543c1730a..7b98bc9bd1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,28 +4,15 @@ build-backend = "hatchling.build" [project] name = "verifiers" -authors = [{ name = "William Brown", email = "williambrown97@gmail.com" }] -dynamic = ["version"] # derived from git tags via hatch-vcs +authors = [ + {name = "William Brown", email = "williambrown97@gmail.com"}, +] +dynamic = ["version"] # derived from git tags via hatch-vcs description = "Verifiers: Environments for LLM Reinforcement Learning" readme = "README.md" license = "MIT" requires-python = ">=3.11,<3.14" -keywords = [ - "reinforcement-learning", - "llm", - "rl", - "grpo", - "environments", - "multi-turn", - "agents", - "agentic-rl", - "tool-use", - "train", - "eval", - "harness", - "verifiers", - "rlvr", -] +keywords = ["reinforcement-learning", "llm", "rl", "grpo", "environments", "multi-turn", "agents", "agentic-rl", "tool-use", "train", "eval", "harness", "verifiers", "rlvr"] classifiers = [ "Development Status :: 4 - Beta", "Intended Audience :: Developers", @@ -110,16 +97,30 @@ examples = [ ] [project.optional-dependencies] -rg = ["reasoning-gym>=0.1.8"] -ta = ["textarena==0.7.4", "nltk>=3.9.2"] -browser = ["stagehand>=3.4.0", "python-dotenv>=1.0.0"] +rg = [ + "reasoning-gym>=0.1.8", +] +ta = [ + "textarena==0.7.4", + "nltk>=3.9.2", +] +browser = [ + "stagehand>=3.4.0", + "python-dotenv>=1.0.0", +] harbor = [ # Harbor currently publishes Python 3.12+ packages; keep core verifiers installable on 3.11. "harbor==0.21.0 ; python_full_version >= '3.12'", ] -modal = ["modal>=1.4.0"] -notebook = ["nest-asyncio>=1.6.0"] -openenv = ["openenv>=0.4.1"] +modal = [ + "modal>=1.4.0", +] +notebook = [ + "nest-asyncio>=1.6.0", +] +openenv = [ + "openenv>=0.4.1", +] nemo-gym = [ # NeMo Gym currently publishes Python 3.12+ packages; keep core verifiers installable on 3.11. "nemo-gym==0.4.0; python_version >= '3.12'", @@ -236,8 +237,18 @@ git_describe_command = [ ] [tool.hatch.build.targets.sdist] -include = ["/verifiers", "/tests", "/README.md", "/LICENSE"] -exclude = ["/wandb", "/scratch", "/configs", "/environments"] +include = [ + "/verifiers", + "/tests", + "/README.md", + "/LICENSE", +] +exclude = [ + "/wandb", + "/scratch", + "/configs", + "/environments", +] [tool.hatch.build.targets.wheel] packages = ["verifiers"] @@ -262,7 +273,12 @@ pythonpath = ["tests/v1/fixtures"] python_files = ["test_*.py"] python_classes = ["Test*"] python_functions = ["test_*"] -addopts = ["--strict-markers", "--tb=short", "-ra", "--quiet"] +addopts = [ + "--strict-markers", + "--tb=short", + "-ra", + "--quiet", +] markers = [ "e2e: marks end-to-end eval-run tests (need a model API key; skipped without one)", "integration: marks tests as integration tests", @@ -322,7 +338,11 @@ not-iterable = "ignore" [tool.coverage.run] source = ["verifiers"] -omit = ["*/tests/*", "*/__pycache__/*", "*/site-packages/*"] +omit = [ + "*/tests/*", + "*/__pycache__/*", + "*/site-packages/*", +] [tool.coverage.report] exclude_lines = [ From 41a104fec3cfdaa20b1ce98a6bad135e6fe8da11 Mon Sep 17 00:00:00 2001 From: Cooper Miller Date: Thu, 20 Aug 2026 14:41:59 -0700 Subject: [PATCH 03/11] fix: close out an eval run whatever breaks --- verifiers/v1/cli/dashboard/eval.py | 8 +++++- verifiers/v1/cli/eval/runner.py | 44 ++++++++++++++++-------------- verifiers/v1/utils/platform.py | 7 ++++- 3 files changed, 36 insertions(+), 23 deletions(-) diff --git a/verifiers/v1/cli/dashboard/eval.py b/verifiers/v1/cli/dashboard/eval.py index 9ce65e2ec0..82df553a4a 100644 --- a/verifiers/v1/cli/dashboard/eval.py +++ b/verifiers/v1/cli/dashboard/eval.py @@ -287,7 +287,13 @@ def _push_footer(push: "PushState | None") -> Group | None: or the run stayed local.""" if push is None or not push.started: return None - if push.error: + if push.error and push.url: + # The run exists and holds everything that streamed up; only closing it out + # failed. Say so and keep the URL, rather than reporting a failed push and + # hiding the run it did produce. + line = Text(f"Traces pushed ({push.url})", style="white", overflow="fold") + line.append(f" not closed out: {push.error}", style="red") + elif push.error: line = Text(f"Trace push failed ({push.error})", style="red", overflow="fold") elif not push.done: line = Text(f"Pushing traces ({push.url})", style="dim", overflow="fold") diff --git a/verifiers/v1/cli/eval/runner.py b/verifiers/v1/cli/eval/runner.py index b666cd76bb..9a65ea2fae 100644 --- a/verifiers/v1/cli/eval/runner.py +++ b/verifiers/v1/cli/eval/runner.py @@ -98,29 +98,31 @@ async def on_complete(episode: Episode) -> None: await asyncio.to_thread(run.log_traces, [episode]) # 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, n in plan for slot in env.slots(task, n=n)] - slots = [RunSlot.finished(episode) for episode in finished] + planned - display = ( - dashboard(slots, config, start, push=push_state) - if config.rich - else contextlib.nullcontext() - ) - async with display: - try: + # run; plan slots inside so the env's agents borrow them. Everything from + # bringing those up to tearing them down is inside the try: a run that was + # opened is closed out whatever breaks, so none of them sits at running. + try: + async with env.serving(): + planned = [slot for task, n in plan for slot in env.slots(task, n=n)] + slots = [RunSlot.finished(episode) for episode in finished] + planned + display = ( + dashboard(slots, config, start, push=push_state) + if config.rich + else contextlib.nullcontext() + ) + async with display: results = await asyncio.gather( *( env.run_slot(slot, ctx, semaphore, on_complete) for slot in planned ) ) - except BaseException as e: # a killed eval must not sit at running - await asyncio.to_thread(abort_run, run, e, push_state) - raise - episodes = finished + list(results) - # Drain and close out off the event loop so the view keeps refreshing. - await asyncio.to_thread(finish_run, run, episodes, push_state) + episodes = finished + list(results) + # Drain and close out off the event loop so the view keeps refreshing. + await asyncio.to_thread(finish_run, run, episodes, push_state) + except BaseException as e: + await asyncio.to_thread(abort_run, run, e, push_state) + raise return episodes @@ -239,12 +241,12 @@ async def run_unit(payload: dict) -> list[Episode]: units = [run_unit(payload) for payload, n in plan for _ in range(n)] try: results = await asyncio.gather(*units) - except BaseException as e: # a killed eval must not sit at running + await client.close() + episodes = finished + [record for unit in results for record in unit] + await asyncio.to_thread(finish_run, run, episodes) + except BaseException as e: await asyncio.to_thread(abort_run, run, e) raise - await client.close() - episodes = finished + [record for unit in results for record in unit] - await asyncio.to_thread(finish_run, run, episodes) return episodes finally: proc.terminate() diff --git a/verifiers/v1/utils/platform.py b/verifiers/v1/utils/platform.py index c57d00280e..5c49bc40ba 100644 --- a/verifiers/v1/utils/platform.py +++ b/verifiers/v1/utils/platform.py @@ -180,7 +180,12 @@ def abort_run( Ctrl-C is a decision, not a fault: it lands as `crashed` (the process stopped without the run ever saying), which is what tells whoever looks whether to read the run's own error or go look at the machine it ran on. A cancellation - is the same thing arriving from the other direction.""" + is the same thing arriving from the other direction. + + A run that already finished keeps the status it reported: teardown failing + after a clean upload does not retroactively make the run a failure.""" + if run.finished: + return if isinstance(error, (KeyboardInterrupt, asyncio.CancelledError)): status, message = pr.RunStatus.CRASHED, "interrupted" else: From 181c97f7b81ffd66238a5ef6e2653b8dc53475ee Mon Sep 17 00:00:00 2001 From: Cooper Miller Date: Thu, 20 Aug 2026 14:57:34 -0700 Subject: [PATCH 04/11] fix: don't close the run while rollouts are in flight --- verifiers/v1/cli/eval/runner.py | 35 ++++++++++++++++++++++++++------- 1 file changed, 28 insertions(+), 7 deletions(-) diff --git a/verifiers/v1/cli/eval/runner.py b/verifiers/v1/cli/eval/runner.py index e7a6be9dff..bbba3a6ab5 100644 --- a/verifiers/v1/cli/eval/runner.py +++ b/verifiers/v1/cli/eval/runner.py @@ -4,7 +4,8 @@ import contextlib import logging import time -from typing import cast +from collections.abc import Awaitable, Iterable +from typing import TypeVar, cast from verifiers.v1.cli.dashboard import dashboard from verifiers.v1.cli.eval import resume @@ -22,6 +23,29 @@ logger = logging.getLogger(__name__) +T = TypeVar("T") + + +async def gather_rollouts(rollouts: Iterable[Awaitable[T]]) -> list[T]: + """`asyncio.gather`, but nothing is left running when it raises. + + `gather` re-raises the first failure without touching its siblings, so a + caller that reacts to the failure — closing the run out, tearing the env + down — does so while rollouts are still in flight, and their `on_complete` + streams into a run that has already been finished. Cancelling and awaiting + them first runs each one's teardown `finally` (the same unwind a Ctrl-C + asks for) and leaves nothing behind to race the caller's cleanup.""" + tasks = [asyncio.ensure_future(rollout) for rollout in rollouts] + try: + return await asyncio.gather(*tasks) + except BaseException: + for task in tasks: + task.cancel() + # return_exceptions: every task is awaited, and the failure that is + # about to propagate is the one worth reporting — not a cancellation. + await asyncio.gather(*tasks, return_exceptions=True) + raise + async def run_eval(env: Env, config: EvalConfig) -> list[Episode]: logger.info("eval config:\n%s", config.model_dump_json(indent=2)) @@ -110,11 +134,8 @@ async def on_complete(episode: Episode) -> None: else contextlib.nullcontext() ) async with display: - results = await asyncio.gather( - *( - env.run_slot(slot, ctx, semaphore, on_complete) - for slot in planned - ) + results = await gather_rollouts( + env.run_slot(slot, ctx, semaphore, on_complete) for slot in planned ) episodes = finished + list(results) # Drain and close out off the event loop so the view keeps refreshing. @@ -240,7 +261,7 @@ async def run_unit(payload: dict) -> list[Episode]: # Each rollout is its own `run` request, dispatched least-busy across workers. units = [run_unit(payload) for payload, n in plan for _ in range(n)] try: - results = await asyncio.gather(*units) + results = await gather_rollouts(units) await client.close() episodes = finished + [record for unit in results for record in unit] await asyncio.to_thread(finish_run, run, episodes) From 6dda5cf2083e9003948e10236540cf277390bb4f Mon Sep 17 00:00:00 2001 From: Cooper Miller Date: Thu, 20 Aug 2026 15:21:41 -0700 Subject: [PATCH 05/11] chore: clean up --- verifiers/v1/cli/dashboard/eval.py | 3 +-- verifiers/v1/cli/eval/main.py | 3 +-- verifiers/v1/cli/eval/runner.py | 7 +++---- verifiers/v1/cli/resolve.py | 8 +------- verifiers/v1/configs/cli/eval.py | 3 +-- verifiers/v1/utils/platform.py | 20 +------------------- 6 files changed, 8 insertions(+), 36 deletions(-) diff --git a/verifiers/v1/cli/dashboard/eval.py b/verifiers/v1/cli/dashboard/eval.py index 82df553a4a..67eeebd61a 100644 --- a/verifiers/v1/cli/dashboard/eval.py +++ b/verifiers/v1/cli/dashboard/eval.py @@ -289,8 +289,7 @@ def _push_footer(push: "PushState | None") -> Group | None: return None if push.error and push.url: # The run exists and holds everything that streamed up; only closing it out - # failed. Say so and keep the URL, rather than reporting a failed push and - # hiding the run it did produce. + # failed. line = Text(f"Traces pushed ({push.url})", style="white", overflow="fold") line.append(f" not closed out: {push.error}", style="red") elif push.error: diff --git a/verifiers/v1/cli/eval/main.py b/verifiers/v1/cli/eval/main.py index 8ba93e24da..a8d8d33ab0 100644 --- a/verifiers/v1/cli/eval/main.py +++ b/verifiers/v1/cli/eval/main.py @@ -69,8 +69,7 @@ def main(argv: list[str] | None = None) -> None: *argv, ] # let prime-pydantic-config render help/errors config = cli(config_type) - # The `@ eval.toml` this run was launched from — uploaded verbatim with the - # run, so its Config tab shows what someone actually wrote. + # The `@ eval.toml` this run was launched from. config.run.record_source(config_file_ref(argv)) # A named run directory is re-entered only by `--resume` or wiped by `--clean`: any # other write into it — the dry-run config.toml included, which would clobber the diff --git a/verifiers/v1/cli/eval/runner.py b/verifiers/v1/cli/eval/runner.py index bbba3a6ab5..54d85fa197 100644 --- a/verifiers/v1/cli/eval/runner.py +++ b/verifiers/v1/cli/eval/runner.py @@ -102,13 +102,12 @@ async def run_eval(env: Env, config: EvalConfig) -> list[Episode]: write_lock = asyncio.Lock() push_state = PushState() if config.push and config.rich else None - # Opened before the first rollout so the platform's id *is* the run's id: - # every trace is stamped with it once, at rollout time, and nothing is - # re-stamped or rewritten afterwards. + + # Opened before the first rollout so the platform's id is the run's id run = open_run(config, push_state) config.run.adopt_id(run.id) # A resume's kept rollouts are part of this run too, so they carry its id and - # go up with the rest — otherwise the platform would hold half a run. + # go up with the rest for episode in finished: episode.record_run(EvalRunInfo(id=config.run.id, name=config.run.name)) run.log_traces(finished) diff --git a/verifiers/v1/cli/resolve.py b/verifiers/v1/cli/resolve.py index eb62705924..5441c2995f 100644 --- a/verifiers/v1/cli/resolve.py +++ b/verifiers/v1/cli/resolve.py @@ -47,13 +47,7 @@ def references_config_file(argv: list[str]) -> bool: def config_file_ref(argv: list[str]) -> str | None: - """The file a run was launched from — the root-level `@ ` — or None. - - Only the root form counts: `--env @ env.toml` configures one block, whereas - `@ eval.toml` *is* the run's config, which is what gets recorded on the run — - so a `@` right after a flag is somebody else's file. Several root files merge - into one config that no single path describes, so that case records nothing - rather than half of it.""" + """The file a run was launched from — the root-level `@ ` or None.""" paths = [ argv[i + 1] for i, arg in enumerate(argv) diff --git a/verifiers/v1/configs/cli/eval.py b/verifiers/v1/configs/cli/eval.py index 039a75101b..6faf9e61ef 100644 --- a/verifiers/v1/configs/cli/eval.py +++ b/verifiers/v1/configs/cli/eval.py @@ -61,8 +61,7 @@ def adopt_id(self, run_id: str) -> None: """Take the platform's run id as this run's id. Called once, before the first rollout, so that every trace is stamped - with the id the platform knows the run by — one id, minted in one place, - never re-stamped afterwards.""" + with the id the platform knows the run by.""" self._id = run_id def record_source(self, path: str | None) -> None: diff --git a/verifiers/v1/utils/platform.py b/verifiers/v1/utils/platform.py index 5c49bc40ba..c7a5c42b6b 100644 --- a/verifiers/v1/utils/platform.py +++ b/verifiers/v1/utils/platform.py @@ -1,19 +1,4 @@ -"""The eval's run on the Prime Intellect platform (`--no-push` to keep it local). - -`prime-runs` owns the run lifecycle now. `open_run` opens the run *before* the -first rollout, so the platform's id is `config.run.id` and every trace carries -it; episodes stream out through `run.log_traces` as they land, instead of being -held in memory and posted in one blast at the end; `finish_run` writes the -terminal status. Ctrl-C, a crash or an exit that never reaches `finish_run` are -reported too, so a killed eval stops showing as running forever. - -What lives here is only the verifiers half of that wiring: which config fields -are the run's identity, what counts as the run's config, and the dashboard's -view of the upload. The batching, retries and the v0 sample projection are the -SDK's — `trace_to_sample`/`build_samples` moved to `prime_runs.projection`, -which is the package that owns that wire format, and are re-exported here for -callers that still import them from verifiers. -""" +"""The eval's run on the Prime Intellect platform (`--no-push` to keep it local).""" import asyncio import logging @@ -49,9 +34,6 @@ class PushState: run: "pr.Run | None" = None error: str | None = None - """Why there is no run — it could not be opened, or could not be closed out. - Not set for a run that pushed with something degraded along the way; that is - a `warning`, since the records mostly landed.""" done: bool = False @property From 9761460077c14527a9678c940e9acfe20b2c2df2 Mon Sep 17 00:00:00 2001 From: Cooper Miller Date: Thu, 20 Aug 2026 15:35:32 -0700 Subject: [PATCH 06/11] fix: rm done and trace_to_sample --- verifiers/v1/cli/dashboard/eval.py | 2 +- verifiers/v1/utils/platform.py | 13 +++++++------ 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/verifiers/v1/cli/dashboard/eval.py b/verifiers/v1/cli/dashboard/eval.py index 67eeebd61a..7c6c8ed947 100644 --- a/verifiers/v1/cli/dashboard/eval.py +++ b/verifiers/v1/cli/dashboard/eval.py @@ -294,7 +294,7 @@ def _push_footer(push: "PushState | None") -> Group | None: line.append(f" not closed out: {push.error}", style="red") elif push.error: line = Text(f"Trace push failed ({push.error})", style="red", overflow="fold") - elif not push.done: + elif not push.finished: line = Text(f"Pushing traces ({push.url})", style="dim", overflow="fold") else: line = Text(f"Traces pushed ({push.url})", style="white", overflow="fold") diff --git a/verifiers/v1/utils/platform.py b/verifiers/v1/utils/platform.py index c7a5c42b6b..7f84af30ff 100644 --- a/verifiers/v1/utils/platform.py +++ b/verifiers/v1/utils/platform.py @@ -6,7 +6,7 @@ from typing import Any import prime_runs as pr -from prime_runs.projection import build_samples, trace_to_sample +from prime_runs.projection import build_samples from verifiers.v1.configs.cli.eval import EvalConfig from verifiers.v1.episode import Episode @@ -21,7 +21,6 @@ "build_samples", "finish_run", "open_run", - "trace_to_sample", ] @@ -34,13 +33,18 @@ class PushState: run: "pr.Run | None" = None error: str | None = None - done: bool = False @property def url(self) -> str | None: """Where to watch the run. `None` for a run that stays local.""" return self.run.url if self.run is not None else None + @property + def finished(self) -> bool: + """Whether the run has been closed out. The run's own flag, not a copy of + it: one that a caller has to remember to set is one that can disagree.""" + return self.run is not None and self.run.finished + @property def started(self) -> bool: """Whether there is anything to report: a live run, or why there isn't one.""" @@ -195,6 +199,3 @@ def _close( else: if run.url: logger.info("--push: %s -> %s", status.value, run.url) - finally: - if state is not None: - state.done = True From 4c465d4f085f807c50ab154e2acb8a28164372df Mon Sep 17 00:00:00 2001 From: Cooper Miller Date: Thu, 20 Aug 2026 16:00:59 -0700 Subject: [PATCH 07/11] chore: improve docstring --- verifiers/v1/cli/eval/runner.py | 26 +++++++++++++------------- verifiers/v1/configs/cli/eval.py | 6 ++---- 2 files changed, 15 insertions(+), 17 deletions(-) diff --git a/verifiers/v1/cli/eval/runner.py b/verifiers/v1/cli/eval/runner.py index 54d85fa197..c5e64ff470 100644 --- a/verifiers/v1/cli/eval/runner.py +++ b/verifiers/v1/cli/eval/runner.py @@ -27,22 +27,25 @@ async def gather_rollouts(rollouts: Iterable[Awaitable[T]]) -> list[T]: - """`asyncio.gather`, but nothing is left running when it raises. + """`asyncio.gather`, but one rollout failing stops the others too. - `gather` re-raises the first failure without touching its siblings, so a - caller that reacts to the failure — closing the run out, tearing the env - down — does so while rollouts are still in flight, and their `on_complete` - streams into a run that has already been finished. Cancelling and awaiting - them first runs each one's teardown `finally` (the same unwind a Ctrl-C - asks for) and leaves nothing behind to race the caller's cleanup.""" + Plain `gather` raises the first error and leaves the rest running. They then + keep going while the caller is already handling that error — still uploading + to a run it has just closed, still using an env it is tearing down. + Cancelling them here, and waiting for each one to finish unwinding, keeps + those two things from overlapping. + + The error is re-raised exactly as it arrived, which is why this is not an + `asyncio.TaskGroup`: a TaskGroup wraps everything in an `ExceptionGroup`, and + `main` would stop recognizing a `KeyboardInterrupt` as Ctrl-C.""" tasks = [asyncio.ensure_future(rollout) for rollout in rollouts] try: return await asyncio.gather(*tasks) except BaseException: for task in tasks: task.cancel() - # return_exceptions: every task is awaited, and the failure that is - # about to propagate is the one worth reporting — not a cancellation. + # return_exceptions so this waits for all of them; without it the first + # cancellation would raise and the rest would be left running again. await asyncio.gather(*tasks, return_exceptions=True) raise @@ -115,8 +118,6 @@ async def run_eval(env: Env, config: EvalConfig) -> list[Episode]: 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) - # A queue put, but a bounded one: handing it to a thread keeps a full - # queue from stalling every other rollout (and freezing the dashboard). await asyncio.to_thread(run.log_traces, [episode]) # Serving resources (shared tool servers, interception) come up once for the @@ -236,8 +237,7 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]: asyncio.Semaphore(config.max_concurrent) if config.max_concurrent else None ) write_lock = asyncio.Lock() - # Same contract as the in-process runner: the run opens before the first - # rollout, so its id is the one every trace carries. + run = open_run(config) config.run.adopt_id(run.id) for episode in finished: diff --git a/verifiers/v1/configs/cli/eval.py b/verifiers/v1/configs/cli/eval.py index 6faf9e61ef..6ff697eedb 100644 --- a/verifiers/v1/configs/cli/eval.py +++ b/verifiers/v1/configs/cli/eval.py @@ -40,11 +40,9 @@ class RunConfig(BaseConfig): """Run directory name — the run writes to `output_dir / dir`. Defaults to `run.name`; set it only when the directory should differ from the display name.""" - # The platform's run id once `prime_runs.init()` has opened the run (see - # `adopt_id`); the local uuid until then, and for a run that never reaches - # the platform. Private so it stays out of the saved config and its digest — - # two runs of the same config differ by id, and resume compares configs. _id: str = PrivateAttr(default_factory=lambda: str(uuid4())) + """The platform's run id once `prime_runs.init()` has opened the run (see + `adopt_id`); the local uuid until then, and for a run that stays local.""" _source: str | None = PrivateAttr(default=None) """The `@ file.toml` this run was launched from, recorded by the CLI.""" From c50ffa3309ba5a3a090c416087eff2c91f89df05 Mon Sep 17 00:00:00 2001 From: Cooper Miller Date: Thu, 20 Aug 2026 16:09:31 -0700 Subject: [PATCH 08/11] chore: clean platform.py --- verifiers/v1/utils/platform.py | 30 ++++-------------------------- 1 file changed, 4 insertions(+), 26 deletions(-) diff --git a/verifiers/v1/utils/platform.py b/verifiers/v1/utils/platform.py index 7f84af30ff..10889b69b2 100644 --- a/verifiers/v1/utils/platform.py +++ b/verifiers/v1/utils/platform.py @@ -41,8 +41,7 @@ def url(self) -> str | None: @property def finished(self) -> bool: - """Whether the run has been closed out. The run's own flag, not a copy of - it: one that a caller has to remember to set is one that can disagree.""" + """Whether the run has been closed out.""" return self.run is not None and self.run.finished @property @@ -76,13 +75,7 @@ def warning(self) -> str | None: def open_run(config: EvalConfig, state: PushState | None = None) -> "pr.Run": - """Open the run this eval streams into, before the first rollout. - - Never raises: the eval is the point and the upload is not, so a platform - that won't answer (or a missing `PRIME_API_KEY`) degrades to a disabled run - — a real handle whose calls are no-ops — and the reason lands on `state` for - the dashboard. `--no-push` takes the same path, so there is one set of call - sites either way. The run's id is the caller's to adopt (`config.run.adopt_id`).""" + """Open the run this eval streams into, before the first rollout.""" identity: dict[str, Any] = { "name": config.run.name, # The environment is resolved by name through the hub's get-or-create, so @@ -121,14 +114,7 @@ def open_run(config: EvalConfig, state: PushState | None = None) -> "pr.Run": def run_config(config: EvalConfig) -> dict[str, Any]: """What the run was configured with — the fields somebody actually set, plus - the file it was launched from, kept byte for byte. - - `exclude_unset` is the point: a resolved `EvalConfig` dump is hundreds of - defaults nobody chose, and the two or three values that were the experiment - are invisible in it. A `@ eval.toml` is stored verbatim under the SDK's - reserved `config_source` key — comments, key order and section grouping - included — because that file *is* the run's configuration and nothing - reconstructed from a dump can reproduce it.""" + the file it was launched from, kept byte for byte.""" values: dict[str, Any] = config.model_dump(mode="json", exclude_unset=True) source = config.run.source if source is not None: @@ -161,15 +147,7 @@ def finish_run( def abort_run( run: "pr.Run", error: BaseException, state: PushState | None = None ) -> None: - """Close the run out after the eval broke, so it doesn't sit at running. - - Ctrl-C is a decision, not a fault: it lands as `crashed` (the process stopped - without the run ever saying), which is what tells whoever looks whether to - read the run's own error or go look at the machine it ran on. A cancellation - is the same thing arriving from the other direction. - - A run that already finished keeps the status it reported: teardown failing - after a clean upload does not retroactively make the run a failure.""" + """Close the run out after the eval broke, so it doesn't sit at running.""" if run.finished: return if isinstance(error, (KeyboardInterrupt, asyncio.CancelledError)): From b006fecf91634b98e43d7bc5b176adc21068e6e0 Mon Sep 17 00:00:00 2001 From: Cooper Miller Date: Thu, 20 Aug 2026 21:37:30 -0700 Subject: [PATCH 09/11] fix: rm handle_signals --- verifiers/v1/utils/platform.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/verifiers/v1/utils/platform.py b/verifiers/v1/utils/platform.py index 10889b69b2..5427b95009 100644 --- a/verifiers/v1/utils/platform.py +++ b/verifiers/v1/utils/platform.py @@ -86,11 +86,6 @@ def open_run(config: EvalConfig, state: PushState | None = None) -> "pr.Run": "model": config.model, "framework": FRAMEWORK, "config": run_config(config), - # verifiers installs its own SIGINT/SIGTERM handler (`install_interrupt`) - # so a killed eval still tears down its sandboxes; the runner reports the - # terminal status from the unwind rather than letting the SDK take the - # signal. The SDK's atexit hook still catches an exit that gets neither. - "handle_signals": False, } if config.push: try: From e3631005531474dfdea84b2b779fa05f7eb50e8b Mon Sep 17 00:00:00 2001 From: Cooper Miller Date: Fri, 21 Aug 2026 15:34:00 -0700 Subject: [PATCH 10/11] bump --- uv.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/uv.lock b/uv.lock index f1e62c44a4..09c0135bfc 100644 --- a/uv.lock +++ b/uv.lock @@ -3335,7 +3335,7 @@ toml = [ [[package]] name = "prime-runs" version = "0.1.0" -source = { git = "https://github.com/PrimeIntellect-ai/prime.git?subdirectory=packages%2Fprime-runs&branch=feature%2Fprime-runs-sdk#4a367dc08f5c56101fc16a458ac4a7895e879bd5" } +source = { git = "https://github.com/PrimeIntellect-ai/prime.git?subdirectory=packages%2Fprime-runs&branch=feature%2Fprime-runs-sdk#97d6482a21b31fcdb6bfdbcf7ccdb962e5659c56" } dependencies = [ { name = "httpx" }, { name = "prime-traces" }, @@ -3363,7 +3363,7 @@ wheels = [ [[package]] name = "prime-traces" version = "0.0.2" -source = { git = "https://github.com/PrimeIntellect-ai/prime.git?subdirectory=packages%2Fprime-traces&branch=feature%2Fprime-runs-sdk#4a367dc08f5c56101fc16a458ac4a7895e879bd5" } +source = { git = "https://github.com/PrimeIntellect-ai/prime.git?subdirectory=packages%2Fprime-traces&branch=feature%2Fprime-runs-sdk#97d6482a21b31fcdb6bfdbcf7ccdb962e5659c56" } dependencies = [ { name = "httpx" }, { name = "pydantic" }, From 261bb93535e0bc89b92f611897cbe734993622f9 Mon Sep 17 00:00:00 2001 From: Cooper Miller Date: Fri, 21 Aug 2026 18:20:55 -0700 Subject: [PATCH 11/11] log_episodes --- uv.lock | 4 ++-- verifiers/v1/cli/eval/runner.py | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/uv.lock b/uv.lock index 09c0135bfc..a54a06acf9 100644 --- a/uv.lock +++ b/uv.lock @@ -3335,7 +3335,7 @@ toml = [ [[package]] name = "prime-runs" version = "0.1.0" -source = { git = "https://github.com/PrimeIntellect-ai/prime.git?subdirectory=packages%2Fprime-runs&branch=feature%2Fprime-runs-sdk#97d6482a21b31fcdb6bfdbcf7ccdb962e5659c56" } +source = { git = "https://github.com/PrimeIntellect-ai/prime.git?subdirectory=packages%2Fprime-runs&branch=feature%2Fprime-runs-sdk#34faa4aaf4e45ed4b0fe387089585ddb698423fd" } dependencies = [ { name = "httpx" }, { name = "prime-traces" }, @@ -3363,7 +3363,7 @@ wheels = [ [[package]] name = "prime-traces" version = "0.0.2" -source = { git = "https://github.com/PrimeIntellect-ai/prime.git?subdirectory=packages%2Fprime-traces&branch=feature%2Fprime-runs-sdk#97d6482a21b31fcdb6bfdbcf7ccdb962e5659c56" } +source = { git = "https://github.com/PrimeIntellect-ai/prime.git?subdirectory=packages%2Fprime-traces&branch=feature%2Fprime-runs-sdk#34faa4aaf4e45ed4b0fe387089585ddb698423fd" } dependencies = [ { name = "httpx" }, { name = "pydantic" }, diff --git a/verifiers/v1/cli/eval/runner.py b/verifiers/v1/cli/eval/runner.py index c5e64ff470..39ac4fc3f0 100644 --- a/verifiers/v1/cli/eval/runner.py +++ b/verifiers/v1/cli/eval/runner.py @@ -113,12 +113,12 @@ async def run_eval(env: Env, config: EvalConfig) -> list[Episode]: # go up with the rest for episode in finished: episode.record_run(EvalRunInfo(id=config.run.id, name=config.run.name)) - run.log_traces(finished) + run.log_episodes(finished) 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) - await asyncio.to_thread(run.log_traces, [episode]) + await asyncio.to_thread(run.log_episodes, [episode]) # Serving resources (shared tool servers, interception) come up once for the # run; plan slots inside so the env's agents borrow them. Everything from @@ -242,7 +242,7 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]: config.run.adopt_id(run.id) for episode in finished: episode.record_run(EvalRunInfo(id=config.run.id, name=config.run.name)) - run.log_traces(finished) + run.log_episodes(finished) async def run_unit(payload: dict) -> list[Episode]: async with semaphore or contextlib.nullcontext(): @@ -254,7 +254,7 @@ async def run_unit(payload: dict) -> list[Episode]: ) episode.record_run(EvalRunInfo(id=config.run.id, name=config.run.name)) await append_episode(out, episode, write_lock) - await asyncio.to_thread(run.log_traces, [episode]) + await asyncio.to_thread(run.log_episodes, [episode]) return [cast(Episode, episode)] # Each rollout is its own `run` request, dispatched least-busy across workers.