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 73f9e96e8..52f8709b2 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 from verifiers.v1.configs.cli.eval import EvalConfig from verifiers.v1.env import Env, RunSlot @@ -48,14 +49,13 @@ async def run_eval(env: Env, config: EvalConfig) -> list[Episode]: finished: list[Episode] = [] if config.resume is not None: keys = [ - resume.task_key(t.data.model_dump(mode="json", exclude_none=True)) - for t in tasks + task_key(t.data.model_dump(mode="json", exclude_none=True)) for t in tasks ] finished, owed = resume.load(out, keys, config.num_rollouts, env.complete) if not owed: # already complete - report it and exit successfully print(resume.nothing_to_resume_msg(out, len(tasks), config.num_rollouts)) raise SystemExit(0) - counts = resume.distribute(keys, owed, config.num_rollouts) + counts = distribute(keys, owed, config.num_rollouts) plan = [(task, n) for task, n in zip(tasks, counts) if n] logger.info( "resuming %s: %d task(s), %d rollout(s) owed", @@ -179,7 +179,7 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]: client = EnvClient(address=address) await client.wait_for_server_startup(timeout=600) # A v1 run dispatches — and resumes — tasks by content: the client owns them, - # and `resume.task_key` is their identity. Only the legacy bridge is addressed + # and `task_key` is their identity. Only the legacy bridge is addressed # by dataset row (its dataset lives server-side, reported via `info`), and # only a legacy env group-scores; a v1 env scores siblings in its own rollout. if legacy: @@ -209,14 +209,14 @@ async def run_eval_server(config: EvalConfig) -> list[Episode]: whole_task=group_scored, key_of=lambda data: data.get("idx"), ) - counts = resume.distribute(idxs, owed, config.num_rollouts) + counts = distribute(idxs, owed, config.num_rollouts) else: keys = [ - resume.task_key(t.data.model_dump(mode="json", exclude_none=True)) + task_key(t.data.model_dump(mode="json", exclude_none=True)) for t in tasks ] finished, owed = resume.load(out, keys, config.num_rollouts) - counts = resume.distribute(keys, owed, config.num_rollouts) + counts = distribute(keys, owed, config.num_rollouts) if not owed: # already complete - report it and exit successfully print(resume.nothing_to_resume_msg(out, len(plan), config.num_rollouts)) raise SystemExit(0) 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 025e732d5..c1fae6e1e 100644 --- a/verifiers/v1/cli/validate.py +++ b/verifiers/v1/cli/validate.py @@ -2,16 +2,23 @@ import asyncio import contextlib +import json import logging import sys import time +import tomllib +from collections import Counter, defaultdict +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.output import CONFIG_FILE, write_config from verifiers.v1.cli.resolve import ( extract_id, narrow_taskset_config, @@ -19,11 +26,13 @@ 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 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 @@ -31,10 +40,19 @@ 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] + 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,7 +63,149 @@ def _narrow(argv: list[str]) -> type[ValidateConfig]: return narrow_taskset_config(ValidateConfig, extract_id(argv, "taskset")) -ResultRow = dict[str, Any] +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 _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, key: str, mode: str) -> bool: + if not isinstance(row, dict): + return False + reason = row.get("reason") + return ( + row.get("task_key") == 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_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 + targets = Counter(selected_keys) + good: dict[str, list[ResultRow]] = defaultdict(list) + 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 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) + 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 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: @@ -217,27 +377,64 @@ 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_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] = [] + counts = [1] * len(tasks) + else: + rows, owed = load_results(out, selected_keys, mode) + counts = distribute(selected_keys, owed, 1) + write_summary(out, summarize(rows, len(tasks), mode)) + plan = [ + (position, task, key) + for position, (task, key, count) in enumerate(zip(tasks, selected_keys, counts)) + if count + ] 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 +454,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 +478,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, "validate") + 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/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: