diff --git a/docs/v1/env.md b/docs/v1/env.md index 4d510065f3..d7dcb9a17d 100644 --- a/docs/v1/env.md +++ b/docs/v1/env.md @@ -53,6 +53,25 @@ Just like tasksets and harnesses, an `Env` can be user-defined for full expressi | `best-of-n` | `agent` | `n` independent attempts per episode; its metrics mark the argmax-reward sibling (`best`) and whether any reached `--env.threshold` (`pass_at_n`) — rejection sampling and pass@k. | | `agentic-judge` | `solver`, `judge` | the solver plays the task; a code-executing judge agent verifies the finished attempt with real execution. | +## Grading artifacts + +Use artifacts to carry files between runtimes. Files written to +`/logs/artifacts/` are collected implicitly; declare other paths on the task data: + +```python +class MyData(vf.TaskData): + artifacts: list[vf.Artifact] = [ + vf.Artifact(source="/work/report", exclude=[".git"]) + ] + + +class MyTask(vf.Task[MyData]): + async def finalize(self, trace: vf.Trace, runtime: vf.Runtime) -> None: + trace.state.artifacts = await vf.collect(runtime, self.data.artifacts) +``` + +Declared paths must exist when collected. The implicit directory is optional. + ## Concurrency Write independent agents as independent (`asyncio.gather`, a `TaskGroup`) — how many actually run at once is the run's call, not the env's. Two knobs bound it, and the **episode is the unit** at the outer one: diff --git a/docs/v1/harbor.md b/docs/v1/harbor.md index 6f7f31bc36..1b1941ab58 100644 --- a/docs/v1/harbor.md +++ b/docs/v1/harbor.md @@ -101,10 +101,13 @@ routes take precedence over deny rules, while ordinary Prime deny rules are appl unchanged and may block a matching route. Restricted Harbor tasks require Docker or a Prime VM; Prime accepts host-level entries. +## Artifacts and collect hooks + +`artifacts = [...]` and `[[verifier.collect]]` are read from `task.toml` ([Harbor Docs](https://www.harborframework.com/docs/run-jobs/results-and-artifacts)). Collect hooks run in the agent's box from the task's `finalize`, which is Harbor's own ordering — after the agent phase, before collection — and declared paths plus the `/logs/artifacts/` convention dir are then carried into the grading box and restored at their original paths ("no translation", as in Harbor). + ## Shortcomings verifiers does not have parity with Harbor yet, so some features are missing and currently being worked on. The most notable missing features right now are: - Switching to a different verifier-phase network policy ([Harbor Docs](https://www.harborframework.com/docs/tasks/network-policy)) -- Shared & separate verifiers ([Harbor Docs](https://www.harborframework.com/docs/tasks#verifier-environment-shared-vs-separate)) - Multi-step tasks ([Harbor Docs](https://www.harborframework.com/docs/tasks/multi-step)) diff --git a/tests/v1/test_e2e.py b/tests/v1/test_e2e.py index d8613fec69..0aff964ff8 100644 --- a/tests/v1/test_e2e.py +++ b/tests/v1/test_e2e.py @@ -335,6 +335,7 @@ async def test_agentic(run_v1, harness, harness_runtime, tmp_path): runtime={"type": harness_runtime}, output_dir=tmp_path, max_turns=10, + max_tokens=8192, ) assert trace.ok assert trace.num_turns >= 1 # ran a command, then finished @@ -405,18 +406,18 @@ async def test_env_id_agentic_judge(run_v1, tmp_path): policy.write_text("Check EMPIRICALLY that the agent echoed the word back.") traces = await run_v1( "echo-v1", - harness=None, # seats pin their own harness; there is no run-level one + harness=None, env={ "id": "agentic-judge", - # The solver owns the shared box, so the container is pinned here. "solver": {"harness": {"id": "bash"}, "runtime": {"type": "docker"}}, - # The judge reads the trace and reasons before it writes the - # verdict file; the shared 2048-token run cap truncates it mid-audit. "judge": { "harness": {"id": "bash"}, "max_output_tokens": 8192, }, - "task": {"prompt": str(policy)}, + "task": { + "prompt": {"path": str(policy)}, + "hint": "Do not rely on README.md", + }, "score": {"task_weight": 0.5}, }, output_dir=tmp_path, @@ -431,7 +432,7 @@ async def test_env_id_agentic_judge(run_v1, tmp_path): # The task's own reward keeps its raw score; the rescale lands on the weight. assert solver.rewards["echoed"].score == 1.0 assert solver.rewards["echoed"].weight == 0.5 - assert isinstance(judge.info.get("verdict"), dict) # scraped off the box + assert isinstance(judge.info.get("verdict"), dict) assert 0.0 <= solver.rewards["judge"].score <= 1.0 diff --git a/uv.lock b/uv.lock index 49aa74e435..38298ea398 100644 --- a/uv.lock +++ b/uv.lock @@ -21,9 +21,9 @@ exclude-newer-span = "P7D" prime-tunnel = false prime-sandboxes = false prime-pydantic-config = false -ty = "2026-07-28T07:00:00Z" +ty = "2026-07-28T00:00:00Z" renderers = false -ruff = "2026-07-28T07:00:00Z" +ruff = "2026-07-28T00:00:00Z" [[package]] name = "aiofile" diff --git a/verifiers/v1/__init__.py b/verifiers/v1/__init__.py index 8af4f87e1d..7af9b7f140 100644 --- a/verifiers/v1/__init__.py +++ b/verifiers/v1/__init__.py @@ -4,6 +4,12 @@ from verifiers.v1.acp import ACP from verifiers.v1.agent import Agent, Agents, Interaction, Segment, make_agent +from verifiers.v1.artifacts import ( + ARTIFACTS_DIR, + Artifact, + collect, + restore, +) from verifiers.v1.clients import ( BaseClientConfig, Client, @@ -297,6 +303,11 @@ "PATCH_CAP_BYTES", "capture_patch", "resolve_head", + # grading artifacts + "ARTIFACTS_DIR", + "Artifact", + "collect", + "restore", # scoring "compare_stdout_results", "extract_boxed_answer", diff --git a/verifiers/v1/artifacts.py b/verifiers/v1/artifacts.py new file mode 100644 index 0000000000..1d60e6ae1d --- /dev/null +++ b/verifiers/v1/artifacts.py @@ -0,0 +1,144 @@ +"""Artifact collection and restoration across runtimes.""" + +from __future__ import annotations + +import logging +import shlex +import uuid +from pathlib import PurePosixPath +from typing import TYPE_CHECKING + +from pydantic import Field + +from verifiers.v1.types import StrictBaseModel + +if TYPE_CHECKING: + from verifiers.v1.runtimes import Runtime + +logger = logging.getLogger(__name__) + +ARTIFACTS_DIR = "/logs/artifacts" +"""Implicit artifact directory; tasks that write here need no declaration.""" + +MAX_ARTIFACT_BYTES = 32 * 1024 * 1024 +"""Ceiling per collection. Sized for a delta, not a tree: the grading box boots from the +agent's image, so the repo is already there and only its output has to travel.""" + + +class Artifact(StrictBaseModel): + """One path to restore at the same location in another runtime.""" + + source: str + exclude: list[str] = Field(default_factory=list) + """`tar --exclude` patterns, applied when `source` is a directory.""" + + +async def collect( + runtime: Runtime, artifacts: list[Artifact] | None = None +) -> dict[str, bytes]: + """Tar the convention dir and every declared path out of `runtime`. + + Keyed by source path; the values are tar archives. Insertion order is the order + they were declared, and a path cannot be collected twice. + + A declared source that is missing raises: it was declared because grading needs it, + and grading a partial state scores the rollout wrong rather than failing it. The + implicit convention sweep is exempt — most tasks never write there. + + Each source is archived separately so its exclude patterns stay local. + """ + # Resolve relative sources against the runtime workdir. Joining also normalises + # `/work/` to `/work`, so one tree cannot key two entries (the source is both the + # dict key and `restore`'s rm -rf target). + workdir = PurePosixPath(getattr(runtime.config, "workdir", "") or "/") + declared = [ + a.model_copy(update={"source": str(workdir / a.source)}) + for a in artifacts or [] + ] + convention = PurePosixPath(ARTIFACTS_DIR) + sweep = not any( + (p := PurePosixPath(a.source)) == convention + or p.is_relative_to(convention) + or convention.is_relative_to(p) + for a in declared + ) + entries = ([Artifact(source=ARTIFACTS_DIR)] if sweep else []) + declared + + collected: dict[str, bytes] = {} + budget = MAX_ARTIFACT_BYTES + for artifact in entries: + source = artifact.source + if (await runtime.run(["test", "-e", source], {})).exit_code != 0: + if sweep and source == ARTIFACTS_DIR: + continue + raise RuntimeError( + f"declared artifact {source!r} does not exist in the runtime" + ) + archive = await _tar_out(runtime, artifact, budget) + budget -= len(archive) + collected[source] = archive + + logger.debug("collected artifact roots: %s", list(collected)) + return collected + + +async def restore(runtime: Runtime, collected: dict[str, bytes]) -> None: + """Extract `collected` in `runtime` at the original absolute paths.""" + if not collected: + return + # Restoring into the subprocess runtime would extract absolute paths onto the + # developer's filesystem, so refuse it before any archive reaches the host. + if getattr(runtime.config, "type", None) == "subprocess": + raise RuntimeError( + "refusing to restore artifacts into the subprocess runtime: extraction " + "writes to absolute paths on the host. Grade in a container." + ) + # Clear every root up front, not per entry: a later nested root would otherwise + # delete content an earlier one just restored. Clearing also drops any file or + # symlink the image left at the target. + roots = " ".join(shlex.quote(root) for root in collected) + await _run(runtime, f"rm -rf -- {roots}", "clear artifact roots") + for root, archive in collected.items(): + path = f"/tmp/vf-artifact-{uuid.uuid4().hex}.tar" + await runtime.write(path, archive) + await _run( + runtime, + f"tar -xf {shlex.quote(path)} -C / && rm -f {shlex.quote(path)}", + f"restore artifact {root!r}", + ) + + +async def _tar_out(runtime: Runtime, artifact: Artifact, budget: int) -> bytes: + path = f"/tmp/vf-artifact-{uuid.uuid4().hex}.tar" + excludes = " ".join(f"--exclude={shlex.quote(p)}" for p in artifact.exclude) + try: + await _run( + runtime, + f"tar -cf {shlex.quote(path)} -C / {excludes} -- " + f"{shlex.quote(artifact.source.lstrip('/'))}", + f"collect artifact {artifact.source!r}", + ) + # Size it in the box: an oversized collection is refused before it reaches host + # memory, not after. + sized = await runtime.run(["sh", "-c", f"wc -c < {shlex.quote(path)}"], {}) + if (raw := sized.stdout.strip()).isdigit() and int(raw) > budget: + raise RuntimeError( + f"artifact {artifact.source!r} takes the collection over the " + f"{MAX_ARTIFACT_BYTES} byte limit. The grading box boots from the " + "agent's image, so only the delta needs to travel — narrow the source " + "or add `exclude` patterns." + ) + return await runtime.read(path) + finally: + # Best-effort: the box is about to be destroyed and the name is unique per call. + try: + await runtime.run(["rm", "-f", path], {}) + except Exception: + logger.debug("failed to remove %s", path, exc_info=True) + + +async def _run(runtime: Runtime, command: str, action: str) -> None: + result = await runtime.run(["sh", "-c", command], {}) + if result.exit_code: + detail = (result.stderr or result.stdout).strip()[-500:] + raise RuntimeError(f"failed to {action}: {detail}") diff --git a/verifiers/v1/envs/agentic_judge/__init__.py b/verifiers/v1/envs/agentic_judge/__init__.py index 0c5a9b86fd..408d318d67 100644 --- a/verifiers/v1/envs/agentic_judge/__init__.py +++ b/verifiers/v1/envs/agentic_judge/__init__.py @@ -4,6 +4,7 @@ Criterion, JudgeTaskConfig, ScoreConfig, + TextFile, ) __all__ = [ @@ -12,4 +13,5 @@ "Criterion", "JudgeTaskConfig", "ScoreConfig", + "TextFile", ] diff --git a/verifiers/v1/envs/agentic_judge/env.py b/verifiers/v1/envs/agentic_judge/env.py index a912d3cbab..a4b5472557 100644 --- a/verifiers/v1/envs/agentic_judge/env.py +++ b/verifiers/v1/envs/agentic_judge/env.py @@ -1,14 +1,16 @@ -"""agentic-judge: a solver plays the task, a judge verifies it in the same box. +"""agentic-judge: a solver plays the task, then a judge verifies the work. -A reusable env (`--env.id agentic-judge` over any taskset): the box is -provisioned from the solver's runtime policy, the solver plays the task in it, -and a code-executing judge then inspects the work as the agent left it, with -the solver's full trace record uploaded at `/tmp/trace.json`. The judge grades +A reusable env (`--env.id agentic-judge` over any taskset). The solver plays the +task in a container provisioned from its runtime policy; the judge then grades rubric criteria (`[env.task]`: policy prompt, criteria file) and writes its -verdicts to `/tmp/verdict.json`; `finalize()` validates them strictly onto the -solver's trace — `judge/` metrics plus a weighted-mean `judge` reward, -composed with the taskset's own rewards via `[env.score]` (judge-only by -default). +verdicts to `/tmp/verdict.json`, with the solver's full trace record uploaded at +`/tmp/trace.json`. `finalize()` validates them strictly onto the solver's trace — +`judge/` metrics plus a weighted-mean `judge` reward, composed with the +taskset's own rewards via `[env.score]` (judge-only by default). + +`--env.share-runtime` controls whether the judge uses the solver's runtime. It is +enabled by default. When disabled, the judge gets a fresh runtime containing the +task's collected artifacts. """ import json @@ -21,6 +23,7 @@ import verifiers.v1 as vf from verifiers.v1.types import StrictBaseModel +from verifiers.v1.utils.compile import validate_pairing VERDICT_FILE = "/tmp/verdict.json" TRACE_FILE = "/tmp/trace.json" @@ -97,19 +100,31 @@ def _render(template: str, **fields: str) -> str: return pattern.sub(lambda m: fields[m.group(1)], template) -SANDBOX_NOTE = f"""\ +_RECORD_NOTE = f"""\ +The agent's raw trace record (JSON: messages, tool calls, and its `info` +artifacts) is written by the harness — not the agent — at `{TRACE_FILE}`. The +record can be very large — never dump it whole; peek selectively (list its +keys, then slice out specific fields with python or jq) and pull only what you +need. It is complete — it may also carry the task's own scores/metrics and +reference material (a gold answer, a reference solution, held-out tests). Those +are context, not your standard: recorded scores can be wrong and references can +be narrower than the task; do not over-index on how a reference solves it. Your +verdict is what YOU verified by execution.""" + +SHARED_WORKSPACE_NOTE = f"""\ +## Your workspace + +The graded agent worked in this sandbox. Its edits and any scoring side effects +are present. {_RECORD_NOTE}""" + +ISOLATED_WORKSPACE_NOTE = f"""\ ## Your workspace -Your sandbox is the SAME box the graded agent worked in, in the state the agent -left it — its edits (and any scoring side effects) are applied. The agent's raw -trace record (JSON: messages, tool calls, and its `info` artifacts) is uploaded -at `{TRACE_FILE}`. The record can be very large — never dump it whole; peek -selectively (list its keys, then slice out specific fields with python or jq) -and pull only what you need. It is complete — it may also carry the task's own -scores/metrics and reference material (a gold answer, a reference solution, -held-out tests). Those are context, not your standard: recorded scores can be -wrong and references can be narrower than the task; do not over-index on how a -reference solves it. Your verdict is what YOU verified by execution.""" +This is a fresh sandbox built from the task's image. The task's published +artifacts were restored at their original paths; other changes made by the +graded agent are not present. The usual artifact location is +`{vf.ARTIFACTS_DIR}/` (for code tasks, typically a patch to read or apply), plus +any task-declared paths. {_RECORD_NOTE}""" HINT_SECTION = """\ ## Hints @@ -125,13 +140,30 @@ class JudgeTask(vf.Task): NEEDS_CONTAINER = True - def __init__(self, data: vf.TaskData, files: dict[str, bytes]) -> None: + def __init__( + self, + data: vf.TaskData, + files: dict[str, bytes], + artifacts: dict[str, bytes], + ) -> None: super().__init__(data) self.files = files + self.artifacts = artifacts @classmethod - def from_trace(cls, solution: vf.Trace, config: "JudgeTaskConfig") -> "JudgeTask": - """Mint the judge's task from the solver's finished trace.""" + def from_trace( + cls, + solution: vf.Trace, + config: "JudgeTaskConfig", + share_runtime: bool = True, + ) -> "JudgeTask": + """Mint the judge's task from the solver's finished trace. + + `share_runtime` selects both the workspace note and artifact transport. In + the solver's box the published artifacts are already on disk, so none + travel; a fresh box gets the collected set, restored by `setup` at the + paths they had. + """ solved = solution.task.data files = {TRACE_FILE: json.dumps(solution.to_record()).encode()} template = config.build_prompt() @@ -139,7 +171,10 @@ def from_trace(cls, solution: vf.Trace, config: "JudgeTaskConfig") -> "JudgeTask if "{prompt}" not in template: # A policy that doesn't place the task statement itself still needs it. body += "\n\n" + _render(TASK_SECTION, prompt=solved.prompt_text) - sections = [body, _verdict_section(config.criteria()), SANDBOX_NOTE] + workspace_note = ( + SHARED_WORKSPACE_NOTE if share_runtime else ISOLATED_WORKSPACE_NOTE + ) + sections = [body, _verdict_section(config.criteria()), workspace_note] if (hint := config.build_hint()) is not None: sections.insert(1, _render(HINT_SECTION, hint=hint)) prompt = "\n\n".join(sections) @@ -152,9 +187,11 @@ def from_trace(cls, solution: vf.Trace, config: "JudgeTaskConfig") -> "JudgeTask resources=solved.resources, ), files=files, + artifacts={} if share_runtime else solution.state.artifacts, ) async def setup(self, trace: vf.Trace, runtime: vf.Runtime) -> None: + await vf.restore(runtime, self.artifacts) # The solver had this box first: a pre-seeded verdict must never read as # the judge's own, and a file (or planted symlink) at an upload path must # never survive it — a symlinked TRACE_FILE would redirect the write onto @@ -177,34 +214,53 @@ async def finalize(self, trace: vf.Trace, runtime: vf.Runtime) -> None: trace.info["verdict"] = json.loads(raw) +class TextFile(vf.BaseConfig): + """An explicit file-backed text value for config formats without `Path` values.""" + + path: Path + + +TextSource = str | Path | TextFile + + class JudgeTaskConfig(vf.BaseConfig): """The judge's minted task: the grading policy and what lands in its box.""" - prompt: Path | None = None - """Grading-policy file. Replaces only the policy body — the verdict contract - and workspace note are always appended. May reference `{prompt}` (the solver - task's prompt); if it doesn't, the task statement is appended after.""" - hint: Path | None = None - """Optional hints file injected as their own section: task-family pointers - into the trace or box — e.g. for math, where the reference answer lives in - the record; for SWE, to diff the repo or read `info.patch`.""" + prompt: TextSource | None = None + """Grading policy: a string is inline text; a `Path` in Python or + `{ path = "policy.md" }` in config reads a file. Replaces only the policy + body — the verdict contract and workspace note are always appended. May + reference `{prompt}` (the solver task's prompt); if it doesn't, the task + statement is appended after.""" + hint: TextSource | None = None + """Optional hints, with the same explicit inline/file forms as `prompt`, + injected as their own section: task-family pointers into the trace or box — + e.g. for math, where the reference answer lives in the record; for SWE, to + diff the repo or read `info.patch`.""" rubric: Path | None = None """Criteria the judge grades against: a `.toml`/`.json` file with a `criteria` list — the plugged rubric judge's format, so the same rubric files work for both. None grades the single built-in `solved` criterion.""" + @staticmethod + def _resolve(value: TextSource) -> str: + if isinstance(value, str): + return value + path = value if isinstance(value, Path) else value.path + return path.read_text(encoding="utf-8") + def build_prompt(self) -> str: if self.prompt is None: return GRADE_PROMPT + "\n\n" + TASK_SECTION - return self.prompt.read_text() + return self._resolve(self.prompt) def build_hint(self) -> str | None: - return self.hint.read_text() if self.hint is not None else None + return self._resolve(self.hint) if self.hint is not None else None def criteria(self) -> list[Criterion]: if self.rubric is None: return [SOLVED] - text = self.rubric.read_text() + text = self.rubric.read_text(encoding="utf-8") data = ( tomllib.loads(text) if self.rubric.suffix.lower() == ".toml" @@ -241,23 +297,23 @@ class ScoreConfig(vf.BaseConfig): class AgenticJudgeEnvConfig(vf.EnvConfig): solver: vf.AgentConfig = vf.AgentConfig() - """The solver agent. It owns the shared box, so its runtime must be a - container: `--env.solver.runtime.type docker|prime`.""" + """The solver agent. Its runtime must be a container: + `--env.solver.runtime.type docker|prime`.""" judge: vf.AgentConfig = vf.AgentConfig() - """The judge agent. It plays in the solver's box; its own runtime policy is - ignored (overwritten with the solver's).""" + """The judge agent. Its runtime is ignored when `share_runtime` is enabled; + otherwise it must be a container.""" + share_runtime: bool = True + """Whether the judge grades in the solver's runtime.""" task: JudgeTaskConfig = JudgeTaskConfig() score: ScoreConfig = ScoreConfig() class AgenticJudgeEnv(vf.Env[AgenticJudgeEnvConfig]): def __init__(self, config: AgenticJudgeEnvConfig) -> None: - # The judge plays in the solver's box, so its effective runtime IS the - # solver's policy — aligning the config keeps the base env's subprocess - # warning and the runtime stamped on the judge's trace truthful. - config.judge = config.judge.model_copy( - update={"runtime": config.solver.runtime} - ) + if config.share_runtime: + config.judge = config.judge.model_copy( + update={"runtime": config.solver.runtime} + ) super().__init__(config) self._check_agents() # A missing policy file or a malformed rubric fails here, not mid-episode. @@ -271,30 +327,42 @@ def _check_agents(self) -> None: judge = self._harnesses["judge"] if not judge.EXECUTES_CODE: raise ValueError( - "agentic-judge plays a code-executing judge in its own sandbox, but " + "agentic-judge requires a judge harness that can execute code, but " f"harness {judge.config.id!r} is a tool-less chat loop — a verdict " "that needs no execution is a plugged judge " "(--env.taskset.task.judges), not an agent." ) if isinstance(self.config.solver.runtime, vf.SubprocessConfig): raise TypeError( - "agentic-judge plays its judge in the solver's box, but the solver " - "(which provisions it) resolves to the subprocess runtime; use " + "agentic-judge requires the solver to run in a container, but it " + "resolves to the subprocess runtime; use " "--env.solver.runtime.type docker or prime" ) + validate_pairing(judge, JudgeTask, self.config.judge.runtime) async def setup(self, agents: vf.Agents) -> None: # The judge grades the policy; its tokens are never training data. agents.judge.trainable = False async def run(self, task: vf.Task, agents: vf.Agents) -> None: - async with agents.solver.provision(task) as box: - solution = await agents.solver.run(task, runtime=box) - judge_task = JudgeTask.from_trace(solution, self.config.task) - await agents.judge.run(judge_task, runtime=box) + if self.config.share_runtime: + async with agents.solver.provision(task) as box: + solution = await agents.solver.run(task, runtime=box) + judge_task = JudgeTask.from_trace(solution, self.config.task) + await agents.judge.run(judge_task, runtime=box) + return + + solution = await agents.solver.run(task) + if not solution.ok: + return + await agents.judge.run( + JudgeTask.from_trace(solution, self.config.task, share_runtime=False) + ) async def finalize(self, task: vf.Task, episode: vf.Episode) -> None: by_agent = {t.agent.name: t for t in episode.traces} + if "judge" not in by_agent: + return solution, verdict = by_agent["solver"], by_agent["judge"] data = verdict.info.get("verdict") if not isinstance(data, dict) or not isinstance(data.get("verdicts"), list): diff --git a/verifiers/v1/state.py b/verifiers/v1/state.py index ef2aa12455..951661b9c7 100644 --- a/verifiers/v1/state.py +++ b/verifiers/v1/state.py @@ -1,10 +1,10 @@ """Mutable state shared within one rollout. Tool servers synchronize it through the interception state channel. It is excluded -from serialized traces; persist artifacts in `Trace.info` instead. +from serialized traces. """ -from pydantic import ConfigDict +from pydantic import ConfigDict, Field from typing_extensions import TypeVar from verifiers.v1.types import StrictBaseModel @@ -13,6 +13,7 @@ class State(StrictBaseModel): model_config = ConfigDict(ser_json_inf_nan="constants") + artifacts: dict[str, bytes] = Field(default_factory=dict) StateT = TypeVar("StateT", bound=State, default=State) diff --git a/verifiers/v1/task.py b/verifiers/v1/task.py index 1f4aaca03d..b21cb9a190 100644 --- a/verifiers/v1/task.py +++ b/verifiers/v1/task.py @@ -34,6 +34,7 @@ from pydantic_config import BaseConfig from typing_extensions import TypeVar +from verifiers.v1.artifacts import Artifact from verifiers.v1.configs.task import TaskConfig from verifiers.v1.decorators import discover_decorated, invoke_all from verifiers.v1.errors import TaskError, boundary @@ -123,6 +124,11 @@ class TaskData(StrictBaseModel): """Execution-time destinations denied by this task and combined with runtime blocks. Non-empty concrete allowlists cannot be combined with blocklists. Docker framework routes take precedence; ordinary Prime deny rules pass through unchanged.""" + artifacts: list[Artifact] = Field(default_factory=list) + """Paths collected from one runtime and restored at the same locations in another, + on top of the implicitly collected `/logs/artifacts/` convention dir. Declare + runtime outputs that must cross that boundary. A declared path that is missing at + collection time fails the rollout.""" timeout: TaskTimeout = TaskTimeout() resources: TaskResources = TaskResources() diff --git a/verifiers/v1/tasksets/harbor/taskset.py b/verifiers/v1/tasksets/harbor/taskset.py index 99620f0c91..a3d00d1af6 100644 --- a/verifiers/v1/tasksets/harbor/taskset.py +++ b/verifiers/v1/tasksets/harbor/taskset.py @@ -10,6 +10,7 @@ image unless ``require_image`` is set. """ +import asyncio import hashlib import io import shutil @@ -23,12 +24,14 @@ from pydantic import Field +from verifiers.v1.artifacts import Artifact, collect from verifiers.v1.configs.taskset import TasksetConfig from verifiers.v1.decorators import reward from verifiers.v1.errors import SandboxError from verifiers.v1.runtimes import Runtime from verifiers.v1.task import Task, TaskData, TaskResources, TaskTimeout from verifiers.v1.taskset import Taskset +from verifiers.v1.trace import Trace from verifiers.v1.types import StrictBaseModel CACHE = Path.home() / ".cache" / "harbor" @@ -75,6 +78,13 @@ class Author(StrictBaseModel): email: str | None = None +class CollectHook(StrictBaseModel): + """One `[[verifier.collect]]` command, run in the agent's box by `finalize`.""" + + command: str + timeout_sec: float = 600.0 + + class HarborData(TaskData): """Parsed ``task.toml`` metadata plus the host-side verifier directory. @@ -93,11 +103,43 @@ class HarborData(TaskData): """Raw [verifier.env] entries (literals or `${VAR}`/`${VAR:-default}` templates). Resolved against the host environment at scoring time, like `harbor run` — so a verifier that needs judge API keys or configuration actually receives them.""" + collect: list[CollectHook] = Field(default_factory=list) + """`[[verifier.collect]]` blocks: commands that snapshot runtime state into files + after the agent stops, so the files can travel to a grading box as artifacts.""" class HarborTask(Task[HarborData]): """Stage and run Harbor's verifier inside the task's live runtime.""" + async def finalize(self, trace: Trace, runtime: Runtime) -> None: + """Run Harbor's collect hooks while the agent's box is still alive. + + Harbor runs these after the agent phase and before artifact collection, which + is exactly what `finalize` means here, so the hook maps onto the existing + lifecycle rather than needing a stage of its own. + + Strict, unlike `harbor run`, which logs a failed hook and carries on: there the + output is observability, here it is a grading input, and a silently absent file + makes the verifier score a stale state instead of failing loudly. + """ + for hook in self.data.collect: + try: + result = await asyncio.wait_for( + runtime.run(["sh", "-c", hook.command], {}), + hook.timeout_sec, + ) + except TimeoutError as exc: + raise RuntimeError( + f"collect hook timed out after {hook.timeout_sec}s: {hook.command}" + ) from exc + if result.exit_code: + detail = (result.stderr or result.stdout).strip()[-500:] + raise RuntimeError( + f"collect hook failed (exit {result.exit_code}): " + f"{hook.command}\n{detail}" + ) + trace.state.artifacts = await collect(runtime, self.data.artifacts) + @reward(weight=1.0) async def solved(self, runtime: Runtime) -> float: await runtime.write( @@ -248,6 +290,7 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD harbor_task = HarborModelTask(task_dir) parsed = harbor_task.config + artifacts, collect = parse_verifier_extras(task_dir, parsed) environment = parsed.environment network = parsed.agent.explicit_phase_policy() or environment.resolve_baseline() task, meta = parsed.task, parsed.metadata @@ -316,9 +359,65 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD tags=meta.get("tags", []), task_dir=str(task_dir), verifier_env=parsed.verifier.env, + artifacts=artifacts, + collect=collect, ) +def parse_verifier_extras( + task_dir: Path, parsed +) -> tuple[list[Artifact], list[CollectHook]]: + """Parse supported artifact and collect-hook settings.""" + from harbor.constants import MAIN_SERVICE_NAME + from harbor.models.task.artifacts import ( + effective_artifact_service, + normalize_artifact_entries, + ) + + verifier = parsed.verifier + if verifier.environment is not None: + raise ValueError( + f"{task_dir.name}: [verifier.environment] declares a separate verifier " + "image. Grading runs in a fresh box built from the task's own image, so " + "only the agent's delta has to travel; a different verifier image needs " + "the full working tree copied over and isn't supported yet." + ) + if verifier.user is not None: + raise ValueError(f"{task_dir.name}: [verifier].user is not supported") + + artifacts: list[Artifact] = [] + for entry in normalize_artifact_entries(parsed.artifacts): + if effective_artifact_service(entry) != MAIN_SERVICE_NAME: + raise ValueError( + f"{task_dir.name}: artifact {entry.source!r} targets additional " + f"service {entry.service!r}; verifiers currently supports artifacts " + "from the main service only" + ) + # `destination` positions a file in Harbor's host trial directory. Verifiers has + # no such directory (the trace is the record) and Harbor never lets destination + # affect verifier-side placement, so it cannot change any grading outcome. + artifacts.append( + Artifact(source=entry.source, exclude=list(entry.exclude or [])) + ) + + hooks: list[CollectHook] = [] + for hook in verifier.collect: + if hook.service != MAIN_SERVICE_NAME: + raise ValueError( + f"{task_dir.name}: collect hook targets additional service " + f"{hook.service!r}; verifiers currently supports collect hooks for " + "the main service only" + ) + if hook.user is not None: + raise ValueError( + f"{task_dir.name}: collect hook `user` is not supported " + "(commands run as the runtime's default user)" + ) + hooks.append(CollectHook(command=hook.command, timeout_sec=hook.timeout_sec)) + + return artifacts, hooks + + def verifier_env(task: HarborData) -> dict[str, str]: """Resolve templates at scoring time so host secrets are never serialized.""" if not task.verifier_env: diff --git a/verifiers/v1/utils/git.py b/verifiers/v1/utils/git.py index 43222f1c1b..0a29b835b3 100644 --- a/verifiers/v1/utils/git.py +++ b/verifiers/v1/utils/git.py @@ -3,7 +3,8 @@ SWE-style tasksets call `capture_patch` from `Task.finalize` — after the harness finishes, while the runtime is live, before scoring mutates the repo (restoring test files, switching commits) — so the diff is exactly what the agent produced, -including edits to test files (intentional: they reveal reward hacking). +including edits to test files (intentional: they reveal reward hacking), and +excluding whatever `snapshot_untracked` recorded before the agent started. The diff is taken against `base_commit` when the caller has one — a dataset row field, or a SHA recorded with `resolve_head` at setup time and kept in host @@ -15,8 +16,11 @@ from __future__ import annotations import uuid +from pathlib import PurePosixPath from typing import TYPE_CHECKING +from verifiers.v1.errors import SandboxError + if TYPE_CHECKING: from verifiers.v1.runtimes import Runtime from verifiers.v1.trace import Trace @@ -41,10 +45,15 @@ # tree staged, so reporting success would hide a state later scoring may trip # on; a failed head leaves an empty {capped} (the redirect truncates it before # head runs), which would read back as a silently empty patch. +# +# The unstage step is deliberately outside that accounting: if it fails the patch is +# merely as wide as it used to be, which is a worse patch, not a broken rollout. The +# `$#` test is load-bearing — `git reset -q --` with no pathspec unstages everything. _DIFF = ( "rm -f {full} {capped}; " "git add -A; " "add_rc=$?; " + '[ "$#" -gt 0 ] && git reset -q -- "$@"; ' 'git -c core.quotepath=off diff --cached --binary "$VF_DIFF_BASE" > {full}; ' "diff_rc=$?; " "git reset -q; " @@ -74,32 +83,84 @@ async def resolve_head(runtime: Runtime, env: dict | None = None) -> str: return (result.stdout or "").strip() +async def snapshot_untracked(runtime: Runtime, env: dict | None = None) -> list[str]: + """The repo's untracked files, to hand `capture_patch` as `ignore`. + + Call at the end of `setup`, before the agent runs, and keep the result in host + memory beside `resolve_head`'s SHA. Whatever it lists came with the image, so the + agent cannot be credited with it, and a patch that carries it fails `git apply` in + a fresh container of that same image — which is what an isolated grading box is. + + Sandbox snapshotting will make this free: once runtimes can snapshot and diff a + filesystem, the pre-agent untracked set falls out of the diff with no setup-side + bookkeeping in any taskset. Drop this then. + """ + result = await runtime.run( + ["sh", "-c", "git ls-files --others --exclude-standard -z"], env or {} + ) + if result.exit_code != 0: + return [] + return [path for path in (result.stdout or "").split("\0") if path] + + async def capture_patch( - trace: Trace, runtime: Runtime, base_commit: str = "", env: dict | None = None + trace: Trace, + runtime: Runtime, + base_commit: str = "", + env: dict | None = None, + write_path: str | None = None, + ignore: list[str] | None = None, ) -> None: """Snapshot the agent's cumulative diff into `trace.info["patch"]`. - Best-effort by design: a rollout whose sandbox died or whose repo state is - broken records `info["patch_error"]` instead of failing the rollout — - scoring still runs and the error stays visible in results. + `ignore` names paths to leave out — pass `snapshot_untracked`'s list from setup, or + `git add -A` credits the agent with untracked files the image shipped. R2E-Gym boxes + ship three (`datasets`, `install.sh`, `run_tests.sh`), and a patch carrying them + fails `git apply` in a fresh container of that very image — which is what an + isolated grading box is. + + Two failure modes, attributed differently, because they deserve different outcomes. + + A non-zero exit means the box answered and git refused: a stale `index.lock` from a + killed agent command, a deleted `.git`, a `base_commit` the agent rewrote out of + existence, a disk it filled. That is the agent's own environment, so it records + `info["patch_error"]` and lets the rollout score — a run with no patch grades as a + run that changed nothing, which is the right reward. + + An exception means the box never answered: the sandbox died, the exec timed out, + the transport dropped. Nothing there is the policy's doing, so it raises. Scoring + the rollout anyway would feed a zero to training that says only that our + infrastructure failed. + + `write_path` additionally writes the patch to that path inside the box, for tasks + graded in a second sandbox. Point it at `vf.ARTIFACTS_DIR` (e.g. + `/logs/artifacts/patch.diff`) and collection picks it up with no declaration. Leave + it on the convention sweep rather than declaring it as an `Artifact`: a declared + path is collected strictly, which would turn an agent-broken repo into a rollout + error instead of the low score it should earn. """ nonce = uuid.uuid4().hex full, capped = f"{_FULL}_{nonce}", f"{_CAPPED}_{nonce}" cmd = _DIFF.format(full=full, capped=capped, cap=PATCH_CAP_BYTES + 1) try: result = await runtime.run( - ["sh", "-c", cmd], + ["sh", "-c", cmd, "vf-capture-patch", *(ignore or [])], {**(env or {}), "VF_DIFF_BASE": base_commit or "HEAD"}, ) if result.exit_code != 0: + # Not every runtime raises when the box is gone — Docker returns `docker + # exec`'s own non-zero result, which is indistinguishable from git failing. + # One probe on the failure path tells the two apart before we blame anyone. + if (await runtime.run(["true"], {})).exit_code != 0: + raise SandboxError( + f"patch capture failed and the box stopped answering: " + f"{(result.stderr or '').strip()[-300:]}" + ) trace.info["patch_error"] = ( f"exit={result.exit_code} {(result.stderr or '').strip()[-500:]}" ) return raw = await runtime.read(capped) - except Exception as exc: # noqa: BLE001 - capture must never fail the rollout - trace.info["patch_error"] = f"{type(exc).__name__}: {exc}" - return finally: # Unique names don't overwrite each other, so leftovers would accumulate # on shared-filesystem runtimes; removal is best-effort by design. @@ -111,3 +172,7 @@ async def capture_patch( raw = raw[:PATCH_CAP_BYTES] trace.info["patch_truncated"] = True trace.info["patch"] = raw.decode("utf-8", errors="replace") + if write_path is not None: + parent = str(PurePosixPath(write_path).parent) + await runtime.run(["mkdir", "-p", parent], env or {}) + await runtime.write(write_path, raw)