From 01fa2c6580f4f3d16b385eee0183837775d2b49c Mon Sep 17 00:00:00 2001 From: rasdani <73563550+rasdani@users.noreply.github.com> Date: Sun, 2 Aug 2026 00:07:39 +0000 Subject: [PATCH 1/4] feat(v1): persist and resume task validation --- docs/v1/evaluation.md | 35 ++++ tests/v1/test_validate_cli.py | 228 +++++++++++++++++++++++++++ verifiers/v1/cli/validate.py | 144 +++++++++++++---- verifiers/v1/cli/validate_output.py | 176 +++++++++++++++++++++ verifiers/v1/configs/cli/validate.py | 14 ++ 5 files changed, 567 insertions(+), 30 deletions(-) create mode 100644 tests/v1/test_validate_cli.py create mode 100644 verifiers/v1/cli/validate_output.py diff --git a/docs/v1/evaluation.md b/docs/v1/evaluation.md index dc29f38c4..244aa9304 100644 --- a/docs/v1/evaluation.md +++ b/docs/v1/evaluation.md @@ -52,6 +52,41 @@ The output from evaluations are written into `outputs/----/ `--resume ` re-runs only the rollouts a previous run left missing or errored, appending to that run's own `traces.jsonl`. It reloads the run's saved `config.toml` verbatim, so it takes no other arguments. Good rollouts are kept, while errored ones are dropped and redone. +## Model-free task validation + +Validate task setup and gold patches without running a model: + +```bash +uv run validate primeintellect/terminal-bench-2 --runtime.type prime +``` + +The default runs both checks, each in a fresh runtime. `--only-gold` runs task setup +followed by `Task.validate`; `--only-setup` runs setup alone. Both narrow modes use the +same output, summary, and resume behavior. + +Every run gets a fresh `outputs/--validate//` directory by default. Use +`-o` / `--output-dir` for an exact directory. A run contains: + +```text +config.toml # resolved, replayable config +results.jsonl # one result appended as each task finishes +summary.json # aggregate outcomes and remaining resume debt +validate.log +``` + +The JSONL records preserve the task's dataset `index`, selected-run position, and a +content identity. A completed `valid` or `invalid` result is final. `error`, `timeout`, +missing, malformed, or torn records are work still owed. Resume reloads the saved config +verbatim, removes retryable and duplicate rows, and schedules only that owed work: + +```bash +uv run validate --resume outputs/--validate/ +``` + +`--resume` takes no other arguments. `summary.json` is refreshed after every completed +task, reports all four outcomes plus missing work, and includes separate gold/setup +counts when both checks run. + ## Disabling tools Almost every harness comes with a `disabled_tools` list, which can be used to disable one or multiple tools: diff --git a/tests/v1/test_validate_cli.py b/tests/v1/test_validate_cli.py new file mode 100644 index 000000000..1469c5e83 --- /dev/null +++ b/tests/v1/test_validate_cli.py @@ -0,0 +1,228 @@ +import asyncio +import json +import tomllib +from types import SimpleNamespace + +import pytest + +from verifiers.v1.cli import validate +from verifiers.v1.cli.validate_output import ( + CONFIG_FILE, + RESULTS_FILE, + SUMMARY_FILE, + append_result, + identity, + load_results, + load_resume_config, + output_path, + save_run, + summarize, + validation_mode, +) +from verifiers.v1.configs.cli.validate import ValidateConfig + + +def result_row( + position: int, + key: str, + reason: str, + *, + mode: str = "gold", +) -> dict: + return { + "task_position": position, + "task_key": key, + "index": 100 + position, + "name": f"task-{position}", + "mode": mode, + "valid": reason == "valid", + "reason": reason, + "elapsed": 1.25, + "error": "failed" if reason in {"error", "timeout"} else None, + "error_type": "TimeoutError" if reason == "timeout" else None, + } + + +def test_validate_output_is_fresh_and_replayable(tmp_path): + data = { + "taskset": {"id": "alphabet-sort-v1"}, + "only_setup": True, + "num_tasks": 7, + "rich": False, + } + first = ValidateConfig.model_validate(data) + second = ValidateConfig.model_validate(data) + assert output_path(first) != output_path(second) + + run_dir = tmp_path / "run" + first.output_dir = run_dir + save_run(first, run_dir, total=7) + + saved = tomllib.loads((run_dir / CONFIG_FILE).read_text()) + assert saved["only_setup"] is True + assert saved["num_tasks"] == 7 + assert "uuid" not in saved + assert (run_dir / RESULTS_FILE).read_text() == "" + assert json.loads((run_dir / SUMMARY_FILE).read_text())["outcomes"]["missing"] == 7 + + resumed = load_resume_config(run_dir) + assert resumed.taskset.id == "alphabet-sort-v1" + assert resumed.only_setup is True + assert resumed.num_tasks == 7 + assert resumed.resume == run_dir + assert resumed.output_dir == run_dir + + +def test_resume_keeps_valid_and_invalid_but_retries_the_rest(tmp_path): + selected = [identity(i, {"idx": i, "prompt": f"p{i}"}) for i in range(5)] + rows = [ + result_row(0, selected[0][1], "valid"), + result_row(1, selected[1][1], "invalid"), + result_row(2, selected[2][1], "error"), + result_row(3, selected[3][1], "timeout"), + # A duplicate final result must not survive canonicalization. + result_row(0, selected[0][1], "valid"), + ] + tmp_path.mkdir(exist_ok=True) + (tmp_path / RESULTS_FILE).write_text( + "".join(json.dumps(row) + "\n" for row in rows) + '{"task_position":4' + ) + + kept, owed = load_results(tmp_path, selected, "gold") + + assert [row["task_position"] for row in kept] == [0, 1] + assert owed == [2, 3, 4] + canonical = [ + json.loads(line) for line in (tmp_path / RESULTS_FILE).read_text().splitlines() + ] + assert canonical == kept + + +def test_summary_reports_all_checks_and_resume_debt(): + rows = [ + { + **result_row(0, "a", "valid", mode="all"), + "gold": result_row(0, "a", "valid"), + "setup": result_row(0, "a", "valid", mode="setup"), + }, + { + **result_row(1, "b", "error", mode="all"), + "gold": result_row(1, "b", "invalid"), + "setup": result_row(1, "b", "error", mode="setup"), + }, + ] + + summary = summarize(rows, total=3, mode="all") + + assert summary["outcomes"] == { + "valid": 1, + "invalid": 0, + "error": 1, + "timeout": 0, + "missing": 1, + } + assert summary["terminal"] == 1 + assert summary["owed"] == 2 + assert summary["checks"]["gold"]["invalid"] == 1 + assert summary["checks"]["setup"]["error"] == 1 + + +@pytest.mark.parametrize( + ("config", "expected"), + [ + (ValidateConfig(only_gold=True), "gold"), + (ValidateConfig(only_setup=True), "setup"), + ], +) +def test_only_modes_dispatch_symmetrically(monkeypatch, config, expected): + calls = [] + + async def gold(task, config): + calls.append("gold") + return {"mode": "gold"} + + async def setup(task, config): + calls.append("setup") + return {"mode": "setup"} + + monkeypatch.setattr(validate, "_run_gold", gold) + monkeypatch.setattr(validate, "_run_setup", setup) + + row = asyncio.run(validate._validate_task(object(), config)) + + assert validation_mode(config) == expected + assert row["mode"] == expected + assert calls == [expected] + + +class FakeTask: + NEEDS_CONTAINER = False + + def __init__(self, idx: int): + self.data = SimpleNamespace( + idx=idx, + name=f"task-{idx}", + image=None, + model_dump=lambda **_: {"idx": idx, "name": f"task-{idx}"}, + ) + + +class FakeTaskset: + INFINITE = False + + def __init__(self, tasks): + self._tasks = tasks + + def __iter__(self): + return iter(self._tasks) + + def head(self, n): + return FakeTaskset(self._tasks[:n]) + + def shuffle(self): + return self + + +@pytest.mark.parametrize("mode", ["gold", "setup"]) +def test_run_resume_schedules_only_owed_tasks(monkeypatch, tmp_path, mode): + tasks = [FakeTask(i) for i in range(5)] + selected = [identity(i, task.data.model_dump()) for i, task in enumerate(tasks)] + config = ValidateConfig( + only_gold=mode == "gold", + only_setup=mode == "setup", + output_dir=tmp_path, + rich=False, + ) + save_run(config, tmp_path, total=len(tasks)) + for position, reason in enumerate(("valid", "invalid", "error", "timeout")): + append_result( + tmp_path, + result_row(position, selected[position][1], reason, mode=mode), + ) + config.resume = tmp_path + called = [] + + async def run_task(task, config): + called.append(task.data.idx) + return result_row( + task.data.idx, + selected[task.data.idx][1], + "valid", + mode=mode, + ) + + monkeypatch.setattr(validate.vf, "load_taskset", lambda _: FakeTaskset(tasks)) + monkeypatch.setattr(validate, "_validate_task", run_task) + + rows = asyncio.run(validate.run_validate(config)) + + assert sorted(called) == [2, 3, 4] + assert [row["task_position"] for row in rows] == list(range(5)) + persisted = [ + json.loads(line) for line in (tmp_path / RESULTS_FILE).read_text().splitlines() + ] + assert len(persisted) == 5 + assert {row["task_position"] for row in persisted} == set(range(5)) + summary = json.loads((tmp_path / SUMMARY_FILE).read_text()) + assert summary["mode"] == mode + assert summary["owed"] == 0 diff --git a/verifiers/v1/cli/validate.py b/verifiers/v1/cli/validate.py index 025e732d5..45f79fc27 100644 --- a/verifiers/v1/cli/validate.py +++ b/verifiers/v1/cli/validate.py @@ -2,10 +2,10 @@ import asyncio import contextlib +import json import logging import sys import time -from typing import Any from uuid import uuid4 from pydantic_config import cli @@ -19,11 +19,27 @@ references_config_file, with_positional_taskset, ) +from verifiers.v1.cli.validate_output import ( + LOG_FILE, + SUMMARY_FILE, + ResultRow, + append_result, + identity, + load_results, + load_resume_config, + output_path, + save_run, + split_resume, + summarize, + validation_mode, + write_summary, +) from verifiers.v1.configs.cli.validate import ValidateConfig from verifiers.v1.runtimes import make_runtime from verifiers.v1.state import state_cls from verifiers.v1.task import Task from verifiers.v1.trace import Trace, TraceTask +from verifiers.v1.utils.aio import run_shielded from verifiers.v1.utils.compile import resolve_runtime_config from verifiers.v1.utils.decorators import invoke from verifiers.v1.utils.interrupt import install_interrupt @@ -33,8 +49,9 @@ USAGE = ( "usage: uv run validate [] [--only-setup | --only-gold] " - "[--runtime.type subprocess] [options] [@ file.toml]\n" - " runs the gold and setup-only checks per task (no model)" + "[-o ] [--runtime.type subprocess] [options] [@ file.toml]\n" + " uv run validate --resume \n" + " runs persisted gold and setup-only checks per task (no model)" ) @@ -45,9 +62,6 @@ def _narrow(argv: list[str]) -> type[ValidateConfig]: return narrow_taskset_config(ValidateConfig, extract_id(argv, "taskset")) -ResultRow = dict[str, Any] - - def _classify(valid: bool, exc: BaseException | None) -> str: if valid: return "valid" @@ -217,27 +231,68 @@ async def run_validate(config: ValidateConfig) -> list[dict]: raise SystemExit( "taskset needs a container runtime to validate - pass --runtime.type docker (or prime)" ) - checks = ( - "gold" if config.only_gold else "setup" if config.only_setup else "gold+setup" - ) + mode = validation_mode(config) + checks = "gold+setup" if mode == "all" else mode + out = output_path(config) + selected_identities = [ + identity( + position, + task.data.model_dump(mode="json", exclude_none=True), + ) + for position, task in enumerate(tasks) + ] + if config.resume is None: + save_run(config, out, len(tasks)) + rows: list[ResultRow] = [] + owed = list(range(len(tasks))) + else: + rows, owed = load_results(out, selected_identities, mode) + write_summary(out, summarize(rows, len(tasks), mode)) + owed_set = set(owed) + plan = [ + (position, task, key) + for (position, task), (_, key) in zip(enumerate(tasks), selected_identities) + if position in owed_set + ] logger.info( - "validating %d task(s) from %s on the %s runtime (%s)", + "%s %d/%d task(s) from %s on the %s runtime (%s)", + "resuming" if config.resume is not None else "validating", + len(plan), len(tasks), config.name, config.runtime.type, checks, ) + logger.info("results: %s", out) sem = asyncio.Semaphore(config.max_concurrent) if config.max_concurrent else None states = [TaskProgress(idx=t.data.idx, name=t.data.name) for t in tasks] - state_by_idx = {s.idx: s for s in states} + for row in rows: + state = states[row["task_position"]] + state.state = row["reason"] + + write_lock = asyncio.Lock() - async def _one(task) -> dict: - st = state_by_idx[task.data.idx] + async def _one(position: int, task: Task, key: str) -> ResultRow: + st = states[position] async with sem or contextlib.nullcontext(): st.start = time.time() st.state = "running" row = await _validate_task(task, config) + row["task_position"] = position + row["task_key"] = key + + async def persist() -> None: + async with write_lock: + await asyncio.to_thread(append_result, out, row) + rows.append(row) + await asyncio.to_thread( + write_summary, + out, + summarize(rows, len(tasks), mode), + ) + + await run_shielded(persist()) st.end, st.state = time.time(), row["reason"] if not config.rich: # the dashboard shows this live; otherwise log each task detail = f" - {row['error']}" if row["error"] else "" @@ -257,7 +312,17 @@ async def _one(task) -> dict: else contextlib.nullcontext() ) async with display: - return await asyncio.gather(*(_one(t) for t in tasks)) + if not plan: + logger.info( + "nothing to resume: all %d task(s) are valid or invalid", len(tasks) + ) + return rows + await asyncio.gather( + *(_one(position, task, key) for position, task, key in plan) + ) + rows.sort(key=lambda row: row["task_position"]) + write_summary(out, summarize(rows, len(tasks), mode)) + return rows def main(argv: list[str] | None = None) -> None: @@ -271,27 +336,46 @@ def main(argv: list[str] | None = None) -> None: with plugin_errors(): cli(_narrow(argv)) # full option help, narrowed to the given taskset return - if not extract_id(argv, "taskset") and not references_config_file(argv): - raise SystemExit( - USAGE - ) # need a taskset (positional / --taskset.id) or a @ file.toml - - with plugin_errors(): - config_type = _narrow(argv) - sys.argv = [ - sys.argv[0], - *argv, - ] # let prime-pydantic-config render help/errors - config = cli(config_type) - # Nothing is persisted, so logs are the whole output. Under `--rich` the dashboard owns the - # screen, so keep logs off the console (else stray records print over the UI). - setup_logging("DEBUG" if config.verbose else "INFO", console=not config.rich) + resume_dir, rest = split_resume(argv) + if resume_dir is not None: + if rest: + raise SystemExit( + f"{USAGE}\n--resume replays the saved config and takes no other arguments" + ) + with plugin_errors(): + config = load_resume_config(resume_dir) + else: + if not extract_id(argv, "taskset") and not references_config_file(argv): + raise SystemExit( + USAGE + ) # need a taskset (positional / --taskset.id) or a @ file.toml + + with plugin_errors(): + config_type = _narrow(argv) + sys.argv = [ + sys.argv[0], + *argv, + ] # let prime-pydantic-config render help/errors + config = cli(config_type) + out = output_path(config) + setup_logging( + "DEBUG" if config.verbose else "INFO", + log_file=str(out / LOG_FILE), + console=not config.rich, + ) if config.rich: logging.lastResort = None # drop stdlib records that bypass loguru # Graceful shutdown: first Ctrl-C/SIGTERM unwinds each task's teardown `finally` # (containers/sandboxes); a second is swallowed so it can't orphan them mid-cleanup. install_interrupt() - asyncio.run(run_validate(config)) + try: + asyncio.run(run_validate(config)) + except KeyboardInterrupt: + print(f"interrupted; partial results: {out}", file=sys.stderr) + raise SystemExit(130) + summary = json.loads((out / SUMMARY_FILE).read_text()) + print(f"results: {out}") + print(json.dumps(summary, indent=2, sort_keys=True)) if __name__ == "__main__": diff --git a/verifiers/v1/cli/validate_output.py b/verifiers/v1/cli/validate_output.py new file mode 100644 index 000000000..c5dcf5e3f --- /dev/null +++ b/verifiers/v1/cli/validate_output.py @@ -0,0 +1,176 @@ +"""Durable output and resume primitives for model-free validation runs.""" + +import json +import tomllib +from collections import Counter +from collections.abc import Mapping, Sequence +from pathlib import Path +from typing import Any + +from pydantic_core import from_json + +from verifiers.v1.cli.eval.resume import task_key +from verifiers.v1.cli.output import CONFIG_FILE, write_config +from verifiers.v1.configs.cli.validate import ValidateConfig + +RESULTS_FILE = "results.jsonl" +SUMMARY_FILE = "summary.json" +LOG_FILE = "validate.log" +FINAL_REASONS = frozenset({"valid", "invalid"}) +REASONS = ("valid", "invalid", "error", "timeout") + +ResultRow = dict[str, Any] +TaskIdentity = tuple[int, str] + + +def validation_mode(config: ValidateConfig) -> str: + if config.only_gold: + return "gold" + if config.only_setup: + return "setup" + return "all" + + +def output_path(config: ValidateConfig) -> Path: + """Return an exact explicit output dir, else a fresh eval-shaped run dir.""" + if config.output_dir is not None: + return config.output_dir + return Path("outputs") / f"{config.name}--validate" / config.uuid + + +def identity(position: int, data: Mapping) -> TaskIdentity: + """Identify a selected task by stable selection position and eval's content key.""" + return position, task_key(data) + + +def _write_rows(path: Path, rows: Sequence[ResultRow]) -> None: + tmp = path.with_suffix(f"{path.suffix}.tmp") + with tmp.open("w", encoding="utf-8") as f: + for row in rows: + f.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n") + tmp.replace(path) + + +def append_result(results_dir: Path, row: ResultRow) -> None: + """Append one whole task result. A torn final line is ignored on resume.""" + data = json.dumps(row, sort_keys=True, separators=(",", ":")).encode() + with (results_dir / RESULTS_FILE).open("ab") as f: + f.write(data + b"\n") + + +def _is_final(row: object, target: dict[int, str], mode: str) -> bool: + if not isinstance(row, dict): + return False + position = row.get("task_position") + reason = row.get("reason") + return ( + isinstance(position, int) + and target.get(position) == row.get("task_key") + and row.get("mode") == mode + and reason in FINAL_REASONS + and row.get("valid") is (reason == "valid") + ) + + +def load_results( + results_dir: Path, selected: Sequence[TaskIdentity], mode: str +) -> tuple[list[ResultRow], list[int]]: + """Keep one final valid/invalid row per selected task and return owed positions. + + Missing, malformed, error, and timeout rows are owed. The JSONL is atomically + canonicalized to its kept rows before new work starts, so resumed runs never + accumulate duplicate final records. + """ + path = results_dir / RESULTS_FILE + target = dict(selected) + kept: dict[int, ResultRow] = {} + if path.exists(): + with path.open("rb") as f: + for line in f: + if not line.strip(): + continue + try: + row = from_json(line) + except ValueError: + try: + row = json.loads(line) + except (json.JSONDecodeError, UnicodeDecodeError): + continue + if _is_final(row, target, mode): + position = row["task_position"] + if position not in kept: + kept[position] = row + rows = [kept[position] for position, _ in selected if position in kept] + _write_rows(path, rows) + owed = [position for position, _ in selected if position not in kept] + return rows, owed + + +def summarize(rows: Sequence[ResultRow], total: int, mode: str) -> dict[str, Any]: + """Build the partial or final full-run report written to summary.json.""" + counts = Counter(row.get("reason") for row in rows) + missing = max(0, total - len(rows)) + outcomes = {reason: counts[reason] for reason in REASONS} + outcomes["missing"] = missing + terminal = outcomes["valid"] + outcomes["invalid"] + summary: dict[str, Any] = { + "mode": mode, + "total": total, + "recorded": len(rows), + "terminal": terminal, + "owed": missing + outcomes["error"] + outcomes["timeout"], + "outcomes": outcomes, + "valid_rate": round(outcomes["valid"] / total, 6) if total else None, + } + if mode == "all": + checks: dict[str, dict[str, int]] = {} + for check in ("gold", "setup"): + check_counts = Counter( + row.get(check, {}).get("reason") + for row in rows + if isinstance(row.get(check), dict) + ) + checks[check] = {reason: check_counts[reason] for reason in REASONS} + checks[check]["missing"] = missing + summary["checks"] = checks + return summary + + +def write_summary(results_dir: Path, summary: Mapping[str, Any]) -> None: + path = results_dir / SUMMARY_FILE + tmp = path.with_suffix(f"{path.suffix}.tmp") + tmp.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n") + tmp.replace(path) + + +def save_run(config: ValidateConfig, results_dir: Path, total: int) -> None: + """Create a fresh run before dispatching any validation task.""" + write_config(config, results_dir) + (results_dir / RESULTS_FILE).write_text("") + write_summary(results_dir, summarize([], total, validation_mode(config))) + + +def split_resume(argv: list[str]) -> tuple[Path | None, list[str]]: + for i, arg in enumerate(argv): + if arg == "--resume": + if i + 1 >= len(argv): + raise SystemExit( + "--resume needs an output dir: uv run validate --resume " + ) + return Path(argv[i + 1]), argv[:i] + argv[i + 2 :] + if arg.startswith("--resume="): + return Path(arg.split("=", 1)[1]), argv[:i] + argv[i + 1 :] + return None, argv + + +def load_resume_config(resume_dir: Path) -> ValidateConfig: + """Replay the resolved saved config and point output back at the same run.""" + path = resume_dir / CONFIG_FILE + if not path.exists(): + raise SystemExit( + f"--resume: no config.toml in {resume_dir} - not a validate output dir" + ) + config = ValidateConfig.model_validate(tomllib.loads(path.read_text())) + config.resume = resume_dir + config.output_dir = resume_dir + return config diff --git a/verifiers/v1/configs/cli/validate.py b/verifiers/v1/configs/cli/validate.py index 0ab20e129..ac9c0c9b3 100644 --- a/verifiers/v1/configs/cli/validate.py +++ b/verifiers/v1/configs/cli/validate.py @@ -1,5 +1,8 @@ """Configuration for model-free task validation.""" +from pathlib import Path +from uuid import uuid4 + from pydantic import AliasChoices, Field, SerializeAsAny, model_validator from pydantic_config import BaseConfig @@ -16,6 +19,9 @@ class CheckTimeoutConfig(BaseConfig): class ValidateConfig(BaseConfig): + uuid: str = Field(default_factory=lambda: str(uuid4()), exclude=True) + """Auto-generated run id — the default output directory leaf. Excluded from the + saved config so re-running it starts a fresh run.""" taskset: SerializeAsAny[TasksetConfig] = TasksetConfig() runtime: RuntimeConfig = DockerConfig() """Where each task's validation hooks run.""" @@ -40,6 +46,14 @@ class ValidateConfig(BaseConfig): """Log at debug level instead of the default info.""" rich: bool = True """Show a live dashboard (one row per task) instead of per-task log lines.""" + output_dir: Path | None = Field( + None, validation_alias=AliasChoices("output_dir", "o") + ) + """Where to write config.toml, results.jsonl, summary.json, and validate.log. None + creates a fresh run under outputs/--validate/.""" + resume: Path | None = Field(None, exclude=True) + """Set by --resume: re-run missing, errored, and timed-out tasks in this directory. + The saved config is replayed verbatim, so resume takes no other arguments.""" @property def name(self) -> str: From f113e4dfdd74e8edb86cc12042d9a27b8de0ef0e Mon Sep 17 00:00:00 2001 From: rasdani <73563550+rasdani@users.noreply.github.com> Date: Thu, 6 Aug 2026 03:14:30 +0000 Subject: [PATCH 2/4] refactor(v1): simplify validate persistence layout --- docs/v1/evaluation.md | 35 ----- tests/v1/test_validate_cli.py | 228 ---------------------------- verifiers/v1/cli/validate.py | 174 +++++++++++++++++++-- verifiers/v1/cli/validate_output.py | 176 --------------------- 4 files changed, 159 insertions(+), 454 deletions(-) delete mode 100644 tests/v1/test_validate_cli.py delete mode 100644 verifiers/v1/cli/validate_output.py diff --git a/docs/v1/evaluation.md b/docs/v1/evaluation.md index 244aa9304..dc29f38c4 100644 --- a/docs/v1/evaluation.md +++ b/docs/v1/evaluation.md @@ -52,41 +52,6 @@ The output from evaluations are written into `outputs/----/ `--resume ` re-runs only the rollouts a previous run left missing or errored, appending to that run's own `traces.jsonl`. It reloads the run's saved `config.toml` verbatim, so it takes no other arguments. Good rollouts are kept, while errored ones are dropped and redone. -## Model-free task validation - -Validate task setup and gold patches without running a model: - -```bash -uv run validate primeintellect/terminal-bench-2 --runtime.type prime -``` - -The default runs both checks, each in a fresh runtime. `--only-gold` runs task setup -followed by `Task.validate`; `--only-setup` runs setup alone. Both narrow modes use the -same output, summary, and resume behavior. - -Every run gets a fresh `outputs/--validate//` directory by default. Use -`-o` / `--output-dir` for an exact directory. A run contains: - -```text -config.toml # resolved, replayable config -results.jsonl # one result appended as each task finishes -summary.json # aggregate outcomes and remaining resume debt -validate.log -``` - -The JSONL records preserve the task's dataset `index`, selected-run position, and a -content identity. A completed `valid` or `invalid` result is final. `error`, `timeout`, -missing, malformed, or torn records are work still owed. Resume reloads the saved config -verbatim, removes retryable and duplicate rows, and schedules only that owed work: - -```bash -uv run validate --resume outputs/--validate/ -``` - -`--resume` takes no other arguments. `summary.json` is refreshed after every completed -task, reports all four outcomes plus missing work, and includes separate gold/setup -counts when both checks run. - ## Disabling tools Almost every harness comes with a `disabled_tools` list, which can be used to disable one or multiple tools: diff --git a/tests/v1/test_validate_cli.py b/tests/v1/test_validate_cli.py deleted file mode 100644 index 1469c5e83..000000000 --- a/tests/v1/test_validate_cli.py +++ /dev/null @@ -1,228 +0,0 @@ -import asyncio -import json -import tomllib -from types import SimpleNamespace - -import pytest - -from verifiers.v1.cli import validate -from verifiers.v1.cli.validate_output import ( - CONFIG_FILE, - RESULTS_FILE, - SUMMARY_FILE, - append_result, - identity, - load_results, - load_resume_config, - output_path, - save_run, - summarize, - validation_mode, -) -from verifiers.v1.configs.cli.validate import ValidateConfig - - -def result_row( - position: int, - key: str, - reason: str, - *, - mode: str = "gold", -) -> dict: - return { - "task_position": position, - "task_key": key, - "index": 100 + position, - "name": f"task-{position}", - "mode": mode, - "valid": reason == "valid", - "reason": reason, - "elapsed": 1.25, - "error": "failed" if reason in {"error", "timeout"} else None, - "error_type": "TimeoutError" if reason == "timeout" else None, - } - - -def test_validate_output_is_fresh_and_replayable(tmp_path): - data = { - "taskset": {"id": "alphabet-sort-v1"}, - "only_setup": True, - "num_tasks": 7, - "rich": False, - } - first = ValidateConfig.model_validate(data) - second = ValidateConfig.model_validate(data) - assert output_path(first) != output_path(second) - - run_dir = tmp_path / "run" - first.output_dir = run_dir - save_run(first, run_dir, total=7) - - saved = tomllib.loads((run_dir / CONFIG_FILE).read_text()) - assert saved["only_setup"] is True - assert saved["num_tasks"] == 7 - assert "uuid" not in saved - assert (run_dir / RESULTS_FILE).read_text() == "" - assert json.loads((run_dir / SUMMARY_FILE).read_text())["outcomes"]["missing"] == 7 - - resumed = load_resume_config(run_dir) - assert resumed.taskset.id == "alphabet-sort-v1" - assert resumed.only_setup is True - assert resumed.num_tasks == 7 - assert resumed.resume == run_dir - assert resumed.output_dir == run_dir - - -def test_resume_keeps_valid_and_invalid_but_retries_the_rest(tmp_path): - selected = [identity(i, {"idx": i, "prompt": f"p{i}"}) for i in range(5)] - rows = [ - result_row(0, selected[0][1], "valid"), - result_row(1, selected[1][1], "invalid"), - result_row(2, selected[2][1], "error"), - result_row(3, selected[3][1], "timeout"), - # A duplicate final result must not survive canonicalization. - result_row(0, selected[0][1], "valid"), - ] - tmp_path.mkdir(exist_ok=True) - (tmp_path / RESULTS_FILE).write_text( - "".join(json.dumps(row) + "\n" for row in rows) + '{"task_position":4' - ) - - kept, owed = load_results(tmp_path, selected, "gold") - - assert [row["task_position"] for row in kept] == [0, 1] - assert owed == [2, 3, 4] - canonical = [ - json.loads(line) for line in (tmp_path / RESULTS_FILE).read_text().splitlines() - ] - assert canonical == kept - - -def test_summary_reports_all_checks_and_resume_debt(): - rows = [ - { - **result_row(0, "a", "valid", mode="all"), - "gold": result_row(0, "a", "valid"), - "setup": result_row(0, "a", "valid", mode="setup"), - }, - { - **result_row(1, "b", "error", mode="all"), - "gold": result_row(1, "b", "invalid"), - "setup": result_row(1, "b", "error", mode="setup"), - }, - ] - - summary = summarize(rows, total=3, mode="all") - - assert summary["outcomes"] == { - "valid": 1, - "invalid": 0, - "error": 1, - "timeout": 0, - "missing": 1, - } - assert summary["terminal"] == 1 - assert summary["owed"] == 2 - assert summary["checks"]["gold"]["invalid"] == 1 - assert summary["checks"]["setup"]["error"] == 1 - - -@pytest.mark.parametrize( - ("config", "expected"), - [ - (ValidateConfig(only_gold=True), "gold"), - (ValidateConfig(only_setup=True), "setup"), - ], -) -def test_only_modes_dispatch_symmetrically(monkeypatch, config, expected): - calls = [] - - async def gold(task, config): - calls.append("gold") - return {"mode": "gold"} - - async def setup(task, config): - calls.append("setup") - return {"mode": "setup"} - - monkeypatch.setattr(validate, "_run_gold", gold) - monkeypatch.setattr(validate, "_run_setup", setup) - - row = asyncio.run(validate._validate_task(object(), config)) - - assert validation_mode(config) == expected - assert row["mode"] == expected - assert calls == [expected] - - -class FakeTask: - NEEDS_CONTAINER = False - - def __init__(self, idx: int): - self.data = SimpleNamespace( - idx=idx, - name=f"task-{idx}", - image=None, - model_dump=lambda **_: {"idx": idx, "name": f"task-{idx}"}, - ) - - -class FakeTaskset: - INFINITE = False - - def __init__(self, tasks): - self._tasks = tasks - - def __iter__(self): - return iter(self._tasks) - - def head(self, n): - return FakeTaskset(self._tasks[:n]) - - def shuffle(self): - return self - - -@pytest.mark.parametrize("mode", ["gold", "setup"]) -def test_run_resume_schedules_only_owed_tasks(monkeypatch, tmp_path, mode): - tasks = [FakeTask(i) for i in range(5)] - selected = [identity(i, task.data.model_dump()) for i, task in enumerate(tasks)] - config = ValidateConfig( - only_gold=mode == "gold", - only_setup=mode == "setup", - output_dir=tmp_path, - rich=False, - ) - save_run(config, tmp_path, total=len(tasks)) - for position, reason in enumerate(("valid", "invalid", "error", "timeout")): - append_result( - tmp_path, - result_row(position, selected[position][1], reason, mode=mode), - ) - config.resume = tmp_path - called = [] - - async def run_task(task, config): - called.append(task.data.idx) - return result_row( - task.data.idx, - selected[task.data.idx][1], - "valid", - mode=mode, - ) - - monkeypatch.setattr(validate.vf, "load_taskset", lambda _: FakeTaskset(tasks)) - monkeypatch.setattr(validate, "_validate_task", run_task) - - rows = asyncio.run(validate.run_validate(config)) - - assert sorted(called) == [2, 3, 4] - assert [row["task_position"] for row in rows] == list(range(5)) - persisted = [ - json.loads(line) for line in (tmp_path / RESULTS_FILE).read_text().splitlines() - ] - assert len(persisted) == 5 - assert {row["task_position"] for row in persisted} == set(range(5)) - summary = json.loads((tmp_path / SUMMARY_FILE).read_text()) - assert summary["mode"] == mode - assert summary["owed"] == 0 diff --git a/verifiers/v1/cli/validate.py b/verifiers/v1/cli/validate.py index 45f79fc27..6627e54e4 100644 --- a/verifiers/v1/cli/validate.py +++ b/verifiers/v1/cli/validate.py @@ -6,12 +6,20 @@ import logging import sys import time +import tomllib +from collections import Counter +from collections.abc import Mapping, Sequence +from pathlib import Path +from typing import Any from uuid import uuid4 from pydantic_config import cli +from pydantic_core import from_json import verifiers.v1 as vf from verifiers.v1.cli.dashboard import TaskProgress, validate_dashboard +from verifiers.v1.cli.eval.resume import task_key +from verifiers.v1.cli.output import CONFIG_FILE, write_config from verifiers.v1.cli.resolve import ( extract_id, narrow_taskset_config, @@ -19,21 +27,6 @@ references_config_file, with_positional_taskset, ) -from verifiers.v1.cli.validate_output import ( - LOG_FILE, - SUMMARY_FILE, - ResultRow, - append_result, - identity, - load_results, - load_resume_config, - output_path, - save_run, - split_resume, - summarize, - validation_mode, - write_summary, -) from verifiers.v1.configs.cli.validate import ValidateConfig from verifiers.v1.runtimes import make_runtime from verifiers.v1.state import state_cls @@ -47,6 +40,15 @@ logger = logging.getLogger(__name__) +RESULTS_FILE = "results.jsonl" +SUMMARY_FILE = "summary.json" +LOG_FILE = "validate.log" +FINAL_REASONS = frozenset({"valid", "invalid"}) +REASONS = ("valid", "invalid", "error", "timeout") + +ResultRow = dict[str, Any] +TaskIdentity = tuple[int, str] + USAGE = ( "usage: uv run validate [] [--only-setup | --only-gold] " "[-o ] [--runtime.type subprocess] [options] [@ file.toml]\n" @@ -62,6 +64,148 @@ def _narrow(argv: list[str]) -> type[ValidateConfig]: return narrow_taskset_config(ValidateConfig, extract_id(argv, "taskset")) +def validation_mode(config: ValidateConfig) -> str: + if config.only_gold: + return "gold" + if config.only_setup: + return "setup" + return "all" + + +def output_path(config: ValidateConfig) -> Path: + if config.output_dir is not None: + return config.output_dir + return Path("outputs") / f"{config.name}--validate" / config.uuid + + +def identity(position: int, data: Mapping) -> TaskIdentity: + return position, task_key(data) + + +def _write_rows(path: Path, rows: Sequence[ResultRow]) -> None: + tmp = path.with_suffix(f"{path.suffix}.tmp") + with tmp.open("w", encoding="utf-8") as f: + for row in rows: + f.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n") + tmp.replace(path) + + +def append_result(results_dir: Path, row: ResultRow) -> None: + data = json.dumps(row, sort_keys=True, separators=(",", ":")).encode() + with (results_dir / RESULTS_FILE).open("ab") as f: + f.write(data + b"\n") + + +def _is_final(row: object, target: dict[int, str], mode: str) -> bool: + if not isinstance(row, dict): + return False + position = row.get("task_position") + reason = row.get("reason") + return ( + isinstance(position, int) + and target.get(position) == row.get("task_key") + and row.get("mode") == mode + and reason in FINAL_REASONS + and row.get("valid") is (reason == "valid") + ) + + +def load_results( + results_dir: Path, selected: Sequence[TaskIdentity], mode: str +) -> tuple[list[ResultRow], list[int]]: + """Keep one final valid/invalid row per selected task; return owed positions.""" + path = results_dir / RESULTS_FILE + target = dict(selected) + kept: dict[int, ResultRow] = {} + if path.exists(): + with path.open("rb") as f: + for line in f: + if not line.strip(): + continue + try: + row = from_json(line) + except ValueError: + try: + row = json.loads(line) + except (json.JSONDecodeError, UnicodeDecodeError): + continue + if _is_final(row, target, mode): + position = row["task_position"] + if position not in kept: + kept[position] = row + rows = [kept[position] for position, _ in selected if position in kept] + _write_rows(path, rows) + owed = [position for position, _ in selected if position not in kept] + return rows, owed + + +def summarize(rows: Sequence[ResultRow], total: int, mode: str) -> dict[str, Any]: + counts = Counter(row.get("reason") for row in rows) + missing = max(0, total - len(rows)) + outcomes = {reason: counts[reason] for reason in REASONS} + outcomes["missing"] = missing + terminal = outcomes["valid"] + outcomes["invalid"] + summary: dict[str, Any] = { + "mode": mode, + "total": total, + "recorded": len(rows), + "terminal": terminal, + "owed": missing + outcomes["error"] + outcomes["timeout"], + "outcomes": outcomes, + "valid_rate": round(outcomes["valid"] / total, 6) if total else None, + } + if mode == "all": + checks: dict[str, dict[str, int]] = {} + for check in ("gold", "setup"): + check_counts = Counter( + row.get(check, {}).get("reason") + for row in rows + if isinstance(row.get(check), dict) + ) + checks[check] = {reason: check_counts[reason] for reason in REASONS} + checks[check]["missing"] = missing + summary["checks"] = checks + return summary + + +def write_summary(results_dir: Path, summary: Mapping[str, Any]) -> None: + path = results_dir / SUMMARY_FILE + tmp = path.with_suffix(f"{path.suffix}.tmp") + tmp.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n") + tmp.replace(path) + + +def save_run(config: ValidateConfig, results_dir: Path, total: int) -> None: + write_config(config, results_dir) + (results_dir / RESULTS_FILE).write_text("") + write_summary(results_dir, summarize([], total, validation_mode(config))) + + +def split_resume(argv: list[str]) -> tuple[Path | None, list[str]]: + for i, arg in enumerate(argv): + if arg == "--resume": + if i + 1 >= len(argv): + raise SystemExit( + "--resume needs an output dir: uv run validate --resume " + ) + return Path(argv[i + 1]), argv[:i] + argv[i + 2 :] + if arg.startswith("--resume="): + return Path(arg.split("=", 1)[1]), argv[:i] + argv[i + 1 :] + return None, argv + + +def load_resume_config(resume_dir: Path) -> ValidateConfig: + path = resume_dir / CONFIG_FILE + if not path.exists(): + raise SystemExit( + f"--resume: no config.toml in {resume_dir} - not a validate output dir" + ) + config = ValidateConfig.model_validate(tomllib.loads(path.read_text())) + config.resume = resume_dir + config.output_dir = resume_dir + return config + + def _classify(valid: bool, exc: BaseException | None) -> str: if valid: return "valid" diff --git a/verifiers/v1/cli/validate_output.py b/verifiers/v1/cli/validate_output.py deleted file mode 100644 index c5dcf5e3f..000000000 --- a/verifiers/v1/cli/validate_output.py +++ /dev/null @@ -1,176 +0,0 @@ -"""Durable output and resume primitives for model-free validation runs.""" - -import json -import tomllib -from collections import Counter -from collections.abc import Mapping, Sequence -from pathlib import Path -from typing import Any - -from pydantic_core import from_json - -from verifiers.v1.cli.eval.resume import task_key -from verifiers.v1.cli.output import CONFIG_FILE, write_config -from verifiers.v1.configs.cli.validate import ValidateConfig - -RESULTS_FILE = "results.jsonl" -SUMMARY_FILE = "summary.json" -LOG_FILE = "validate.log" -FINAL_REASONS = frozenset({"valid", "invalid"}) -REASONS = ("valid", "invalid", "error", "timeout") - -ResultRow = dict[str, Any] -TaskIdentity = tuple[int, str] - - -def validation_mode(config: ValidateConfig) -> str: - if config.only_gold: - return "gold" - if config.only_setup: - return "setup" - return "all" - - -def output_path(config: ValidateConfig) -> Path: - """Return an exact explicit output dir, else a fresh eval-shaped run dir.""" - if config.output_dir is not None: - return config.output_dir - return Path("outputs") / f"{config.name}--validate" / config.uuid - - -def identity(position: int, data: Mapping) -> TaskIdentity: - """Identify a selected task by stable selection position and eval's content key.""" - return position, task_key(data) - - -def _write_rows(path: Path, rows: Sequence[ResultRow]) -> None: - tmp = path.with_suffix(f"{path.suffix}.tmp") - with tmp.open("w", encoding="utf-8") as f: - for row in rows: - f.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n") - tmp.replace(path) - - -def append_result(results_dir: Path, row: ResultRow) -> None: - """Append one whole task result. A torn final line is ignored on resume.""" - data = json.dumps(row, sort_keys=True, separators=(",", ":")).encode() - with (results_dir / RESULTS_FILE).open("ab") as f: - f.write(data + b"\n") - - -def _is_final(row: object, target: dict[int, str], mode: str) -> bool: - if not isinstance(row, dict): - return False - position = row.get("task_position") - reason = row.get("reason") - return ( - isinstance(position, int) - and target.get(position) == row.get("task_key") - and row.get("mode") == mode - and reason in FINAL_REASONS - and row.get("valid") is (reason == "valid") - ) - - -def load_results( - results_dir: Path, selected: Sequence[TaskIdentity], mode: str -) -> tuple[list[ResultRow], list[int]]: - """Keep one final valid/invalid row per selected task and return owed positions. - - Missing, malformed, error, and timeout rows are owed. The JSONL is atomically - canonicalized to its kept rows before new work starts, so resumed runs never - accumulate duplicate final records. - """ - path = results_dir / RESULTS_FILE - target = dict(selected) - kept: dict[int, ResultRow] = {} - if path.exists(): - with path.open("rb") as f: - for line in f: - if not line.strip(): - continue - try: - row = from_json(line) - except ValueError: - try: - row = json.loads(line) - except (json.JSONDecodeError, UnicodeDecodeError): - continue - if _is_final(row, target, mode): - position = row["task_position"] - if position not in kept: - kept[position] = row - rows = [kept[position] for position, _ in selected if position in kept] - _write_rows(path, rows) - owed = [position for position, _ in selected if position not in kept] - return rows, owed - - -def summarize(rows: Sequence[ResultRow], total: int, mode: str) -> dict[str, Any]: - """Build the partial or final full-run report written to summary.json.""" - counts = Counter(row.get("reason") for row in rows) - missing = max(0, total - len(rows)) - outcomes = {reason: counts[reason] for reason in REASONS} - outcomes["missing"] = missing - terminal = outcomes["valid"] + outcomes["invalid"] - summary: dict[str, Any] = { - "mode": mode, - "total": total, - "recorded": len(rows), - "terminal": terminal, - "owed": missing + outcomes["error"] + outcomes["timeout"], - "outcomes": outcomes, - "valid_rate": round(outcomes["valid"] / total, 6) if total else None, - } - if mode == "all": - checks: dict[str, dict[str, int]] = {} - for check in ("gold", "setup"): - check_counts = Counter( - row.get(check, {}).get("reason") - for row in rows - if isinstance(row.get(check), dict) - ) - checks[check] = {reason: check_counts[reason] for reason in REASONS} - checks[check]["missing"] = missing - summary["checks"] = checks - return summary - - -def write_summary(results_dir: Path, summary: Mapping[str, Any]) -> None: - path = results_dir / SUMMARY_FILE - tmp = path.with_suffix(f"{path.suffix}.tmp") - tmp.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n") - tmp.replace(path) - - -def save_run(config: ValidateConfig, results_dir: Path, total: int) -> None: - """Create a fresh run before dispatching any validation task.""" - write_config(config, results_dir) - (results_dir / RESULTS_FILE).write_text("") - write_summary(results_dir, summarize([], total, validation_mode(config))) - - -def split_resume(argv: list[str]) -> tuple[Path | None, list[str]]: - for i, arg in enumerate(argv): - if arg == "--resume": - if i + 1 >= len(argv): - raise SystemExit( - "--resume needs an output dir: uv run validate --resume " - ) - return Path(argv[i + 1]), argv[:i] + argv[i + 2 :] - if arg.startswith("--resume="): - return Path(arg.split("=", 1)[1]), argv[:i] + argv[i + 1 :] - return None, argv - - -def load_resume_config(resume_dir: Path) -> ValidateConfig: - """Replay the resolved saved config and point output back at the same run.""" - path = resume_dir / CONFIG_FILE - if not path.exists(): - raise SystemExit( - f"--resume: no config.toml in {resume_dir} - not a validate output dir" - ) - config = ValidateConfig.model_validate(tomllib.loads(path.read_text())) - config.resume = resume_dir - config.output_dir = resume_dir - return config From 2f47c90596c265fc03318d146b7520350f58e5a6 Mon Sep 17 00:00:00 2001 From: rasdani <73563550+rasdani@users.noreply.github.com> Date: Thu, 6 Aug 2026 03:46:56 +0000 Subject: [PATCH 3/4] fix(v1): match eval resume identity --- verifiers/v1/cli/validate.py | 77 ++++++++++++++++++++---------------- 1 file changed, 44 insertions(+), 33 deletions(-) diff --git a/verifiers/v1/cli/validate.py b/verifiers/v1/cli/validate.py index 6627e54e4..6b77187a0 100644 --- a/verifiers/v1/cli/validate.py +++ b/verifiers/v1/cli/validate.py @@ -7,7 +7,7 @@ import sys import time import tomllib -from collections import Counter +from collections import Counter, defaultdict from collections.abc import Mapping, Sequence from pathlib import Path from typing import Any @@ -18,7 +18,7 @@ import verifiers.v1 as vf from verifiers.v1.cli.dashboard import TaskProgress, validate_dashboard -from verifiers.v1.cli.eval.resume import task_key +from verifiers.v1.cli.eval.resume import distribute, task_key from verifiers.v1.cli.output import CONFIG_FILE, write_config from verifiers.v1.cli.resolve import ( extract_id, @@ -47,7 +47,6 @@ REASONS = ("valid", "invalid", "error", "timeout") ResultRow = dict[str, Any] -TaskIdentity = tuple[int, str] USAGE = ( "usage: uv run validate [] [--only-setup | --only-gold] " @@ -78,10 +77,6 @@ def output_path(config: ValidateConfig) -> Path: return Path("outputs") / f"{config.name}--validate" / config.uuid -def identity(position: int, data: Mapping) -> TaskIdentity: - return position, task_key(data) - - def _write_rows(path: Path, rows: Sequence[ResultRow]) -> None: tmp = path.with_suffix(f"{path.suffix}.tmp") with tmp.open("w", encoding="utf-8") as f: @@ -96,14 +91,12 @@ def append_result(results_dir: Path, row: ResultRow) -> None: f.write(data + b"\n") -def _is_final(row: object, target: dict[int, str], mode: str) -> bool: +def _is_final(row: object, key: str, mode: str) -> bool: if not isinstance(row, dict): return False - position = row.get("task_position") reason = row.get("reason") return ( - isinstance(position, int) - and target.get(position) == row.get("task_key") + row.get("task_key") == key and row.get("mode") == mode and reason in FINAL_REASONS and row.get("valid") is (reason == "valid") @@ -111,12 +104,12 @@ def _is_final(row: object, target: dict[int, str], mode: str) -> bool: def load_results( - results_dir: Path, selected: Sequence[TaskIdentity], mode: str -) -> tuple[list[ResultRow], list[int]]: - """Keep one final valid/invalid row per selected task; return owed positions.""" + results_dir: Path, selected_keys: list[str], mode: str +) -> tuple[list[ResultRow], dict[str, int]]: + """Keep final rows by eval task-content key; return counts owed per key.""" path = results_dir / RESULTS_FILE - target = dict(selected) - kept: dict[int, ResultRow] = {} + targets = Counter(selected_keys) + good: dict[str, list[ResultRow]] = defaultdict(list) if path.exists(): with path.open("rb") as f: for line in f: @@ -129,13 +122,35 @@ def load_results( row = json.loads(line) except (json.JSONDecodeError, UnicodeDecodeError): continue - if _is_final(row, target, mode): - position = row["task_position"] - if position not in kept: - kept[position] = row - rows = [kept[position] for position, _ in selected if position in kept] + if not isinstance(row, dict): + continue + key = row.get("task_key") + if ( + isinstance(key, str) + and key in targets + and len(good[key]) < targets[key] + and _is_final(row, key, mode) + ): + good[key].append(row) + + owed = { + key: target - len(good.get(key, [])) + for key, target in targets.items() + if len(good.get(key, [])) < target + } + # Eval spreads debt over content-identical selected tasks in order. Assign the + # interchangeable kept rows to the remaining positions so reporting positions + # stay unique even when duplicate task content exists. + counts = distribute(selected_keys, owed, 1) + rows = [] + used = Counter() + for position, (key, count) in enumerate(zip(selected_keys, counts)): + if count: + continue + row = good[key][used[key]] + used[key] += 1 + rows.append({**row, "task_position": position}) _write_rows(path, rows) - owed = [position for position, _ in selected if position not in kept] return rows, owed @@ -378,25 +393,21 @@ async def run_validate(config: ValidateConfig) -> list[dict]: mode = validation_mode(config) checks = "gold+setup" if mode == "all" else mode out = output_path(config) - selected_identities = [ - identity( - position, - task.data.model_dump(mode="json", exclude_none=True), - ) - for position, task in enumerate(tasks) + selected_keys = [ + task_key(task.data.model_dump(mode="json", exclude_none=True)) for task in tasks ] if config.resume is None: save_run(config, out, len(tasks)) rows: list[ResultRow] = [] - owed = list(range(len(tasks))) + counts = [1] * len(tasks) else: - rows, owed = load_results(out, selected_identities, mode) + rows, owed = load_results(out, selected_keys, mode) + counts = distribute(selected_keys, owed, 1) write_summary(out, summarize(rows, len(tasks), mode)) - owed_set = set(owed) plan = [ (position, task, key) - for (position, task), (_, key) in zip(enumerate(tasks), selected_identities) - if position in owed_set + for position, (task, key, count) in enumerate(zip(tasks, selected_keys, counts)) + if count ] logger.info( "%s %d/%d task(s) from %s on the %s runtime (%s)", From e495be150ddda60044ce17ec9f9667d87515caac Mon Sep 17 00:00:00 2001 From: rasdani <73563550+rasdani@users.noreply.github.com> Date: Thu, 6 Aug 2026 03:52:44 +0000 Subject: [PATCH 4/4] refactor(v1): share CLI resume primitives --- verifiers/v1/cli/eval/main.py | 5 ++-- verifiers/v1/cli/eval/resume.py | 39 +----------------------------- verifiers/v1/cli/eval/runner.py | 14 +++++------ verifiers/v1/cli/resume.py | 42 +++++++++++++++++++++++++++++++++ verifiers/v1/cli/validate.py | 17 ++----------- 5 files changed, 55 insertions(+), 62 deletions(-) create mode 100644 verifiers/v1/cli/resume.py diff --git a/verifiers/v1/cli/eval/main.py b/verifiers/v1/cli/eval/main.py index 9cf392413..a4bd87440 100644 --- a/verifiers/v1/cli/eval/main.py +++ b/verifiers/v1/cli/eval/main.py @@ -7,7 +7,7 @@ from pydantic_config import cli import verifiers.v1 as vf -from verifiers.v1.cli.eval.resume import load_resume_config, split_resume +from verifiers.v1.cli.eval.resume import load_resume_config from verifiers.v1.cli.eval.runner import run_eval from verifiers.v1.cli.output import output_path, write_config from verifiers.v1.cli.resolve import ( @@ -17,6 +17,7 @@ references_config_file, with_positional_taskset, ) +from verifiers.v1.cli.resume import split_resume from verifiers.v1.configs.cli.eval import EvalConfig from verifiers.v1.utils.interrupt import install_interrupt from verifiers.v1.utils.logging import setup_logging @@ -40,7 +41,7 @@ def main(argv: list[str] | None = None) -> None: narrow_config(EvalConfig, argv) ) # full option help, narrowed to the given ids return - resume_dir, rest = split_resume(argv) + resume_dir, rest = split_resume(argv, "eval") # re-run a previous run's missing/errored rollouts, in place if resume_dir is not None: if rest: diff --git a/verifiers/v1/cli/eval/resume.py b/verifiers/v1/cli/eval/resume.py index 18bda6897..98e326a14 100644 --- a/verifiers/v1/cli/eval/resume.py +++ b/verifiers/v1/cli/eval/resume.py @@ -11,7 +11,6 @@ legacy (v0) bridge still matches by row index (`key_of`). """ -import hashlib import json import tomllib from collections import Counter, defaultdict @@ -22,6 +21,7 @@ from pydantic_core import from_json from verifiers.v1.cli.output import CONFIG_FILE, TRACES_FILE, sniff_episode +from verifiers.v1.cli.resume import task_key from verifiers.v1.configs.cli.eval import EvalConfig from verifiers.v1.episode import Episode, WireEpisode from verifiers.v1.trace import WireTrace @@ -29,43 +29,6 @@ K = TypeVar("K", bound=Hashable) -def task_key(data: Mapping) -> str: - """Content identity of one task's wire data — an `exclude_none` dump, the shape - saved rows already have on disk. `sort_keys` so field order can't split identity.""" - return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest() - - -def distribute( - selected_keys: list[K], owed: dict[K, int], num_rollouts: int -) -> list[int]: - """Spread each key's owed rollouts over its selection instances, in order — - content-identical tasks are interchangeable, so any instance can absorb the - debt (capped at `num_rollouts` each). Returns one count per selection.""" - remaining = dict(owed) - counts: list[int] = [] - for key in selected_keys: - take = min(num_rollouts, remaining.get(key, 0)) - if take: - remaining[key] -= take - counts.append(take) - return counts - - -def split_resume(argv: list[str]) -> tuple[Path | None, list[str]]: - """Pull `--resume ` / `--resume=` out of argv, returning (dir, the other args). - The caller rejects any leftover args, since resume re-runs the saved config verbatim.""" - for i, arg in enumerate(argv): - if arg == "--resume": - if i + 1 >= len(argv): - raise SystemExit( - "--resume needs an output dir: uv run eval --resume " - ) - return Path(argv[i + 1]), argv[:i] + argv[i + 2 :] - if arg.startswith("--resume="): - return Path(arg.split("=", 1)[1]), argv[:i] + argv[i + 1 :] - return None, argv - - def load_resume_config(resume_dir: Path) -> EvalConfig: """Rebuild the run's `EvalConfig` from its saved `config.toml`, pointed back at its own output dir so the resumed rollouts append to the same `traces.jsonl`.""" diff --git a/verifiers/v1/cli/eval/runner.py b/verifiers/v1/cli/eval/runner.py index c7b98ed69..4f3c173b9 100644 --- a/verifiers/v1/cli/eval/runner.py +++ b/verifiers/v1/cli/eval/runner.py @@ -14,6 +14,7 @@ output_path, save_config, ) +from verifiers.v1.cli.resume import distribute, task_key from verifiers.v1.clients import ModelContext, resolve_client from verifiers.v1.configs.cli.eval import EvalConfig from verifiers.v1.env import Env, RunSlot @@ -47,14 +48,13 @@ async def run_eval(env: Env, config: EvalConfig) -> list[Episode]: finished: list[Episode] = [] if config.resume is not None: keys = [ - resume.task_key(t.data.model_dump(mode="json", exclude_none=True)) - for t in tasks + task_key(t.data.model_dump(mode="json", exclude_none=True)) for t in tasks ] finished, owed = resume.load(out, keys, config.num_rollouts, env.complete) if not owed: # already complete - report it and exit successfully print(resume.nothing_to_resume_msg(out, len(tasks), config.num_rollouts)) raise SystemExit(0) - counts = resume.distribute(keys, owed, config.num_rollouts) + counts = distribute(keys, owed, config.num_rollouts) plan = [(task, n) for task, n in zip(tasks, counts) if n] logger.info( "resuming %s: %d task(s), %d rollout(s) owed", @@ -179,7 +179,7 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]: client = EnvClient(address=address) await client.wait_for_server_startup(timeout=600) # A v1 run dispatches — and resumes — tasks by content: the client owns them, - # and `resume.task_key` is their identity. Only the legacy bridge is addressed + # and `task_key` is their identity. Only the legacy bridge is addressed # by dataset row (its dataset lives server-side, reported via `info`), and # only a legacy env group-scores; a v1 env scores siblings in its own rollout. if legacy: @@ -209,14 +209,14 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]: whole_task=group_scored, key_of=lambda data: data.get("idx"), ) - counts = resume.distribute(idxs, owed, config.num_rollouts) + counts = distribute(idxs, owed, config.num_rollouts) else: keys = [ - resume.task_key(t.data.model_dump(mode="json", exclude_none=True)) + task_key(t.data.model_dump(mode="json", exclude_none=True)) for t in tasks ] finished, owed = resume.load(out, keys, config.num_rollouts) - counts = resume.distribute(keys, owed, config.num_rollouts) + counts = distribute(keys, owed, config.num_rollouts) if not owed: # already complete - report it and exit successfully print(resume.nothing_to_resume_msg(out, len(plan), config.num_rollouts)) raise SystemExit(0) diff --git a/verifiers/v1/cli/resume.py b/verifiers/v1/cli/resume.py new file mode 100644 index 000000000..3089516f8 --- /dev/null +++ b/verifiers/v1/cli/resume.py @@ -0,0 +1,42 @@ +"""Resume primitives shared by eval-like CLIs.""" + +import hashlib +import json +from collections.abc import Hashable, Mapping +from pathlib import Path +from typing import TypeVar + +K = TypeVar("K", bound=Hashable) + + +def task_key(data: Mapping) -> str: + """Content identity for task wire data, independent of field order.""" + return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest() + + +def distribute( + selected_keys: list[K], owed: dict[K, int], num_results: int +) -> list[int]: + """Spread each key's owed results over its selected instances, in order.""" + remaining = dict(owed) + counts: list[int] = [] + for key in selected_keys: + take = min(num_results, remaining.get(key, 0)) + if take: + remaining[key] -= take + counts.append(take) + return counts + + +def split_resume(argv: list[str], command: str) -> tuple[Path | None, list[str]]: + """Pull ``--resume `` from argv, returning the dir and other arguments.""" + for i, arg in enumerate(argv): + if arg == "--resume": + if i + 1 >= len(argv): + raise SystemExit( + f"--resume needs an output dir: uv run {command} --resume " + ) + return Path(argv[i + 1]), argv[:i] + argv[i + 2 :] + if arg.startswith("--resume="): + return Path(arg.split("=", 1)[1]), argv[:i] + argv[i + 1 :] + return None, argv diff --git a/verifiers/v1/cli/validate.py b/verifiers/v1/cli/validate.py index 6b77187a0..c1fae6e1e 100644 --- a/verifiers/v1/cli/validate.py +++ b/verifiers/v1/cli/validate.py @@ -18,7 +18,6 @@ import verifiers.v1 as vf from verifiers.v1.cli.dashboard import TaskProgress, validate_dashboard -from verifiers.v1.cli.eval.resume import distribute, task_key from verifiers.v1.cli.output import CONFIG_FILE, write_config from verifiers.v1.cli.resolve import ( extract_id, @@ -27,6 +26,7 @@ references_config_file, with_positional_taskset, ) +from verifiers.v1.cli.resume import distribute, split_resume, task_key from verifiers.v1.configs.cli.validate import ValidateConfig from verifiers.v1.runtimes import make_runtime from verifiers.v1.state import state_cls @@ -196,19 +196,6 @@ def save_run(config: ValidateConfig, results_dir: Path, total: int) -> None: write_summary(results_dir, summarize([], total, validation_mode(config))) -def split_resume(argv: list[str]) -> tuple[Path | None, list[str]]: - for i, arg in enumerate(argv): - if arg == "--resume": - if i + 1 >= len(argv): - raise SystemExit( - "--resume needs an output dir: uv run validate --resume " - ) - return Path(argv[i + 1]), argv[:i] + argv[i + 2 :] - if arg.startswith("--resume="): - return Path(arg.split("=", 1)[1]), argv[:i] + argv[i + 1 :] - return None, argv - - def load_resume_config(resume_dir: Path) -> ValidateConfig: path = resume_dir / CONFIG_FILE if not path.exists(): @@ -491,7 +478,7 @@ def main(argv: list[str] | None = None) -> None: with plugin_errors(): cli(_narrow(argv)) # full option help, narrowed to the given taskset return - resume_dir, rest = split_resume(argv) + resume_dir, rest = split_resume(argv, "validate") if resume_dir is not None: if rest: raise SystemExit(