Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 27 additions & 29 deletions verifiers/v1/cli/dashboard/eval.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,8 @@
from typing import TYPE_CHECKING

from pydantic import BaseModel
from rich.console import Console, Group
from rich import get_console
from rich.console import Group
from rich.markup import escape
from rich.progress_bar import ProgressBar
from rich.rule import Rule
Expand All @@ -33,7 +34,7 @@

# For sizing pages to the terminal: detects the real terminal height/width each access (the live
# view writes to the same terminal). Reused so we don't rebuild it every refresh tick.
_CONSOLE = Console()
_CONSOLE = get_console()
_PAGE_SECONDS = 5.0 # rotate to the next page of rollouts this often when they overflow
# The under-bar breakdown pads its label column to the Overview's widest label, so the two
# `label value` grids (above and below the progress bar) line their values up.
Expand Down Expand Up @@ -188,10 +189,9 @@ def overrides(
skip: frozenset[str] = frozenset(),
) -> list[str]:
"""`field=value` segments for the fields the *user* customized, sorted, diffed against each
field's declared default. Not `model_fields_set`: a `--resume` run reloads its config via
`model_validate(config.toml)`, and that toml is dumped with `exclude_none` (every field), so
`model_fields_set` would flag them all. `default` is the reference instance, threaded through
recursion so a pinned nested default (`taskset.task.tools`) reads as
field's declared default. Not `model_fields_set`: a `--resume` run reloads the full saved
JSON config, so `model_fields_set` would flag every field. `default` is the reference instance,
threaded through recursion so a pinned nested default (`taskset.task.tools`) reads as
unchanged. `skip` holds dotted paths (`runtime.type`)."""
segments: list[str] = []
fields = type(config).model_fields
Expand Down Expand Up @@ -782,56 +782,43 @@ def _paginate(

def _render(
slots: list[RunSlot],
config: EvalConfig,
start: float,
pager: Pager,
header: Group | Table,
runtime_type: str,
push: "PushState | None" = None,
tail: LogTail | None = None,
) -> Group:
now = time.time()
warning = _warning(config)
header = Group(warning, Text(""), Overview(config)) if warning else Overview(config)
# The --push status line (and, on Ctrl-C, the cleanup notice) appear under the rollouts. Measure
# the fixed top (header + progress + rule) and the footer so the rollout rows fill what's left;
# page through them (timer / arrows) when they'd overflow (else rich truncates).
footers = [f for f in (_push_footer(push), _interrupt_footer()) if f is not None]
footer = Group(*footers) if footers else None
top = Group(header, Progress(slots, start), Rule(style="dim"))
progress = Progress(slots, start)
top = Group(header, progress, Rule(style="dim"))
reserved = len(_CONSOLE.render_lines(top))
if footer is not None:
reserved += len(_CONSOLE.render_lines(footer))
rows_per_page = max(1, _CONSOLE.size.height - reserved - 1)
if tail is not None: # --show-logs: the run's log stream in place of rollout rows
parts = [
header,
Progress(slots, start),
progress,
Rule(style="dim"),
tail.view(rows_per_page),
]
if footer is not None:
parts.append(footer)
return Group(*parts)
page_groups, index, count = _paginate(_groups(slots), rows_per_page, pager, now)
progress = Progress(
slots,
start,
page=(index + 1, count) if count > 1 else None,
)
if count > 1:
progress = Progress(slots, start, page=(index + 1, count))
parts = [
header,
progress,
Rule(style="dim"),
Rows(
page_groups,
now,
"/".join(
dict.fromkeys(
getattr(config.env, role).runtime.type
for role in config.env.agent_harnesses()
)
)
or "subprocess",
),
Rows(page_groups, now, runtime_type),
]
if footer is not None:
parts.append(footer)
Expand All @@ -846,13 +833,24 @@ async def dashboard(
push: "PushState | None" = None,
):
pager = Pager()
warning = _warning(config)
header = Group(warning, Text(""), Overview(config)) if warning else Overview(config)
runtime_type = (
"/".join(
dict.fromkeys(
getattr(config.env, role).runtime.type
for role in config.env.agent_harnesses()
)
)
or "subprocess"
)
tail = (
LogTail(attempt_log_file(output_path(config)))
if config.rich is not None and config.rich.show_logs
else None
)
async with live_view(
lambda: _render(slots, config, start, pager, push, tail),
on_key=pager.on_key,
lambda: _render(slots, start, pager, header, runtime_type, push, tail),
on_key=None if tail is not None else pager.on_key,
):
yield
10 changes: 3 additions & 7 deletions verifiers/v1/cli/dashboard/replay.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ class ReplayProgress(TaskProgress):


def _render(
states: list[TaskProgress], taskset_name: str, source: str, out: str, start: float
states: list[ReplayProgress], taskset_name: str, source: str, out: str, start: float
) -> Group:
overview = Table.grid(padding=(0, 2))
overview.add_column(style="dim")
Expand All @@ -57,11 +57,7 @@ def _render(
if s.state == "pending": # show only in-flight/done rows (like eval/validate)
continue
label = f"name={s.name[:40]}" if s.name else f"idx={s.idx}"
parts = [
p
for p in (s.state if s.state in _DONE else "", getattr(s, "detail", ""))
if p
]
parts = [p for p in (s.state if s.state in _DONE else "", s.detail) if p]
result = " ".join(parts) + " ·" if parts else ""
elapsed = format_time((s.end or now) - s.start) if s.start else ""
rows.add_row(
Expand All @@ -75,7 +71,7 @@ def _render(

@contextlib.asynccontextmanager
async def replay_dashboard(
states: list[TaskProgress], taskset_name: str, source: str, out: str, start: float
states: list[ReplayProgress], taskset_name: str, source: str, out: str, start: float
):
"""Refresh the live replay view until the `with` block exits, then a final frame."""
async with live_view(lambda: _render(states, taskset_name, source, out, start)):
Expand Down
5 changes: 3 additions & 2 deletions verifiers/v1/cli/debug.py
Original file line number Diff line number Diff line change
Expand Up @@ -240,7 +240,8 @@ async def debug_task(task: Task, config: DebugConfig) -> tuple[Trace, bool]:
trace.timing.agent.start = time.time()
debug.update(await run_action(runtime, config))
trace.timing.agent.end = time.time()
if not debug.get("ok"):
trace.ok = bool(debug["ok"])
if not trace.ok:
record_action_failure(trace, debug)
trace.stop(str(debug["reason"]))
except asyncio.CancelledError as e:
Expand Down Expand Up @@ -282,7 +283,7 @@ async def run_debug(config: DebugConfig) -> list[Trace]:
)

out = output_path(config)
save_config(config, out)
save_config(config, out, "debug.json")
logger.info(
"debugging %d task(s) from %s on the %s runtime",
len(tasks),
Expand Down
17 changes: 9 additions & 8 deletions verifiers/v1/cli/eval/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@
from verifiers.v1.cli.eval.runner import run_eval
from verifiers.v1.cli.output import (
TRACES_FILE,
config_digest,
create_attempt_log_dir,
output_path,
saved_config_path,
Expand Down Expand Up @@ -69,7 +68,7 @@ def main(argv: list[str] | None = None) -> None:
] # let prime-pydantic-config render help/errors
config = cli(config_type)
# 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
# other write into it — the dry-run config included, which would clobber the
# config a resume typically re-runs — would overwrite the previous run.
run_path = output_path(config)
if config.clean and not config.resume and run_path.exists():
Expand Down Expand Up @@ -97,10 +96,12 @@ def main(argv: list[str] | None = None) -> None:
f"--resume: no saved config under {run_path} - not a run dir"
)
with plugin_errors():
saved = config_type.model_validate(json.loads(saved_path.read_text()))
if config_digest(saved) != config_digest(config):
saved_dump = saved.model_dump(mode="json")
current_dump = config.model_dump(mode="json")
saved = config_type.model_validate_json(saved_path.read_text())
saved_dump = saved.model_dump(mode="json")
current_dump = config.model_dump(mode="json")
saved_json = json.dumps(saved_dump, sort_keys=True, separators=(",", ":"))
current_json = json.dumps(current_dump, sort_keys=True, separators=(",", ":"))
if saved_json != current_json:
changed = sorted(
key
for key in set(saved_dump) | set(current_dump)
Expand All @@ -114,12 +115,12 @@ def main(argv: list[str] | None = None) -> None:
)
if config.dry_run: # resolved + validated; write it to the output dir and exit
setup_logging("DEBUG" if config.verbose else "INFO")
logger.info("wrote config to %s", write_config(config, output_path(config)))
logger.info("wrote config to %s", write_config(config, run_path))
return
# Always tee this attempt's logs to `logs/attempt_<n>/eval.log` (`logs/latest`
# points there) — in server mode (the default) the workers write there too, and
# `--rich.show-logs` tails it live.
log_file = str(create_attempt_log_dir(output_path(config)) / "eval.log")
log_file = str(create_attempt_log_dir(run_path) / "eval.log")
level = "DEBUG" if config.verbose else "INFO"
setup_logging(level, log_file=log_file, console=config.rich is None)
# First Ctrl-C / SIGTERM warns and raises KeyboardInterrupt so a killed/timed-out eval still
Expand Down
21 changes: 5 additions & 16 deletions verifiers/v1/cli/eval/resume.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,11 @@
it. `load` keeps the good saved rollouts and re-runs what's owed: missing rollouts
(never written) and errored ones (dropped and redone).

A saved episode is matched to a selected task by hashing `episode.task.data`.
A saved episode is matched to a selected task by its persisted `episode.task.hash`.
Tasks with identical data are interchangeable, a task whose data changed since the
interrupted run re-runs, and nothing depends on `data.idx`.
"""

import json
from collections import Counter, defaultdict
from collections.abc import Callable
from pathlib import Path
Expand All @@ -18,7 +17,6 @@

from verifiers.v1.cli.output import TRACES_FILE
from verifiers.v1.episode import WireEpisode
from verifiers.v1.task import task_key


def load(
Expand Down Expand Up @@ -47,17 +45,16 @@ def load(
if not line.strip():
continue
try:
try:
row = from_json(line)
except ValueError:
row = json.loads(line)
row = from_json(line)
if "traces" not in row:
continue
key = task_key(row["task"]["data"])
key = row["task"]["hash"]
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
Outdated
except (ValueError, KeyError, IndexError, TypeError):
Comment thread
xeophon marked this conversation as resolved.
Outdated
# A torn final line (the run died mid-write) or a foreign shape
# is not a keepable rollout — it's owed again, never a crash.
continue
if not isinstance(key, str):
continue
if key not in targets or len(good[key]) >= targets[key]:
continue
try:
Expand All @@ -83,11 +80,3 @@ def load(
tmp.write_bytes(b"".join(keep))
tmp.replace(path)
return episodes, owed


def nothing_to_resume_msg(resume_dir: Path, num_tasks: int, num_rollouts: int) -> str:
"""Shown (before exit 0) when every selected rollout already completed."""
return (
f"nothing to resume in {resume_dir}: all {num_tasks}x{num_rollouts} rollouts "
f"already completed without error"
)
5 changes: 4 additions & 1 deletion verifiers/v1/cli/eval/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,10 @@ async def run_eval(config: EvalConfig) -> list[Episode]:
loaded, owed = resume.load(out, keys, config.num_rollouts, complete)
finished = [cast(Episode, episode) for episode in loaded]
if not owed: # already complete - report it and exit successfully
print(resume.nothing_to_resume_msg(out, len(tasks), config.num_rollouts))
print(
f"nothing to resume in {out}: all {len(tasks)}x{config.num_rollouts} "
"rollouts already completed without error"
)
raise SystemExit(0)
counts = distribute(keys, owed, config.num_rollouts)
plan = [(task, n) for task, n in zip(tasks, counts) if n]
Expand Down
37 changes: 10 additions & 27 deletions verifiers/v1/cli/init.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,14 +24,10 @@ def _names(name: str) -> tuple[str, str, str, str]:
return dash, pkg, stem, prefix


def _write(path: Path, content: str, force: bool) -> bool:
if path.exists() and not force:
print(f" skip {path} (exists)")
return False
def _write(path: Path, content: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content)
print(f" write {path}")
return True


def _pyproject(dash: str, pkg: str) -> str:
Expand Down Expand Up @@ -67,7 +63,6 @@ def _taskset_py(pkg: str, prefix: str, *, add_tool: bool) -> str:
local_imports: list[str] = []
task_config_fields = ""
task_decls = ""
state = "vf.State"
if add_tool:
local_imports.append(f"from {pkg}.servers.tool import {prefix}Toolset")
task_config_fields += "\n tools: vf.ToolsetConfig = vf.ToolsetConfig()"
Expand All @@ -79,24 +74,18 @@ def _taskset_py(pkg: str, prefix: str, *, add_tool: bool) -> str:
)
if local_imports:
imports += "\n\n" + "\n".join(local_imports)
methods_block = ""
has_task_config = add_tool
task_config = (
f"\n\nclass {prefix}TaskConfig(vf.TaskConfig):\n"
' """Knobs the task reads from ``self.config``; configure them under '
f'``--env.taskset.task.*``."""{task_config_fields}\n'
if has_task_config
if add_tool
else ""
)
task_generic = (
f"{prefix}Data, {state}, {prefix}TaskConfig"
if has_task_config
else f"{prefix}Data"
f"{prefix}Data, vf.State, {prefix}TaskConfig" if add_tool else f"{prefix}Data"
)
config_body = ' num_tasks: int = 5\n """How many tasks to build."""\n' + (
f" task: {prefix}TaskConfig = {prefix}TaskConfig()\n"
if has_task_config
else ""
f" task: {prefix}TaskConfig = {prefix}TaskConfig()\n" if add_tool else ""
)
return f'''\
{imports}
Expand All @@ -107,7 +96,7 @@ class {prefix}Data(vf.TaskData):
{task_config}

class {prefix}Task(vf.Task[{task_generic}]):
"""Rewards, hooks, and servers, with row data available on ``self.data``."""{task_decls}{methods_block}
"""Rewards, hooks, and servers, with row data available on ``self.data``."""{task_decls}
@vf.reward(weight=1.0)
async def reward(self, trace: vf.Trace) -> float:
raise NotImplementedError("Score the rollout and return a float (e.g. in [0, 1]).")
Expand Down Expand Up @@ -223,25 +212,21 @@ def scaffold(config: InitConfig) -> Path:
)
print(f"scaffolding v1 environment {dash!r} in {env_dir}")

_write(env_dir / "pyproject.toml", _pyproject(dash, pkg), config.force)
_write(env_dir / "pyproject.toml", _pyproject(dash, pkg))
_write(
env_dir / "README.md",
_readme(dash, pkg, add_tool=config.add_tool, add_harness=config.add_harness),
config.force,
)
_write(
pkg_dir / "__init__.py", _init_py(pkg, prefix, config.add_harness), config.force
)
_write(pkg_dir / "__init__.py", _init_py(pkg, prefix, config.add_harness))
_write(
pkg_dir / "taskset.py",
_taskset_py(pkg, prefix, add_tool=config.add_tool),
config.force,
)
if config.add_harness:
_write(pkg_dir / "harness.py", _harness_py(prefix), config.force)
_write(pkg_dir / "harness.py", _harness_py(prefix))
if config.add_tool:
_write(pkg_dir / "servers" / "__init__.py", "", config.force)
_write(pkg_dir / "servers" / "tool.py", _tool_py(stem, prefix), config.force)
_write(pkg_dir / "servers" / "__init__.py", "")
_write(pkg_dir / "servers" / "tool.py", _tool_py(stem, prefix))

print(f"\ndone. next:\n uv pip install -e {env_dir}\n uv run eval {dash} -n 3")
return env_dir
Expand All @@ -260,8 +245,6 @@ def main(argv: list[str] | None = None) -> None:

sys.argv = [sys.argv[0], *argv]
config = cli(InitConfig)
if not config.name:
raise SystemExit(USAGE)
scaffold(config)


Expand Down
Loading