diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index 765e03085..a5d3029b7 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -225,13 +225,20 @@ async def run(self) -> Trace: trace.timing.scoring.start = now async with boundary(TaskError, "scoring"): # Group rewards run later, after the runtime is gone. - await asyncio.wait_for( - asyncio.gather( - self.task.score(trace, runtime), - self.harness.score(trace, runtime), - ), - self.scoring_timeout, - ) + scoring_runtime_config = self.task.scoring_runtime_config(runtime) + async with asyncio.timeout(self.scoring_timeout): + if scoring_runtime_config is None: + await asyncio.gather( + self.task.score(trace, runtime), + self.harness.score(trace, runtime), + ) + else: + await self.harness.score(trace, runtime) + await self.task.score( + trace, + runtime, + scoring_runtime_config=scoring_runtime_config, + ) trace.timing.scoring.end = time.time() except RolloutError as e: trace.capture_error(e) diff --git a/verifiers/v1/runtimes/base.py b/verifiers/v1/runtimes/base.py index b8a31fcdd..72cab3455 100644 --- a/verifiers/v1/runtimes/base.py +++ b/verifiers/v1/runtimes/base.py @@ -128,6 +128,12 @@ async def stop(self) -> None: `teardown`, not this.""" await run_shielded(self.teardown()) + async def stop_confirmed(self) -> None: + """Free the resource or raise when deletion cannot be confirmed.""" + raise NotImplementedError( + f"{type(self).__name__} does not support confirmed teardown" + ) + async def teardown(self) -> None: """Free the provisioned resource, off the event loop. Override only for teardown that must be async (e.g. a remote API call); `stop` shields it from cancellation. diff --git a/verifiers/v1/runtimes/docker.py b/verifiers/v1/runtimes/docker.py index f99919db8..5a606d10c 100644 --- a/verifiers/v1/runtimes/docker.py +++ b/verifiers/v1/runtimes/docker.py @@ -195,6 +195,25 @@ async def write(self, path: str, data: bytes) -> None: f"write {path!r}: {stderr.decode(errors='replace').strip()}" ) + async def stop_confirmed(self) -> None: + if self._container is None: + return + removed = await docker("rm", "--force", self._container) + remaining = await docker( + "ps", + "--all", + "--filter", + f"name=^/{self._container}$", + "--format", + "{{.Names}}", + ) + if remaining.exit_code or self._container in remaining.stdout.splitlines(): + detail = (removed.stderr or remaining.stderr or removed.stdout).strip() + raise SandboxError( + f"docker container {self._container!r} deletion was not confirmed: {detail}" + ) + self._stopped = True + def cleanup(self) -> None: if self._container is None or self._stopped: return diff --git a/verifiers/v1/runtimes/prime.py b/verifiers/v1/runtimes/prime.py index e110f6b84..a3b620b92 100644 --- a/verifiers/v1/runtimes/prime.py +++ b/verifiers/v1/runtimes/prime.py @@ -259,6 +259,24 @@ async def write(self, path: str, data: bytes) -> None: except Exception as e: raise SandboxError(f"write {path!r}: {e}") from e + async def stop_confirmed(self) -> None: + """Delete the sandbox or preserve cleanup state and raise.""" + client = self._client + if client is None: + if self.info.id is None: + return + raise RuntimeError( + "prime sandbox deletion cannot be confirmed without its live client" + ) + if self.info.id is None: + raise RuntimeError( + "prime sandbox deletion cannot be confirmed without a provider ID" + ) + await client.delete(self.info.id) + self._client = None + with contextlib.suppress(Exception): + await client.aclose() + def cleanup(self) -> None: # Synchronous atexit backstop (the async client can't run once the loop is gone): delete # the sandbox via the sync client, so the costly resource isn't left to its max-lifetime. diff --git a/verifiers/v1/task.py b/verifiers/v1/task.py index db955b7ba..fcf87eae9 100644 --- a/verifiers/v1/task.py +++ b/verifiers/v1/task.py @@ -61,7 +61,7 @@ if TYPE_CHECKING: from verifiers.v1.judge import Judge from verifiers.v1.mcp import Toolset, User - from verifiers.v1.runtimes import Runtime + from verifiers.v1.runtimes import Runtime, RuntimeConfig from verifiers.v1.trace import Trace logger = logging.getLogger(__name__) @@ -266,10 +266,15 @@ async def finalize(self, trace: Trace, runtime: Runtime) -> None: async def validate(self, runtime: Runtime) -> bool: return True + def scoring_runtime_config(self, runtime: Runtime) -> RuntimeConfig | None: + """Return a separate runtime config when scoring must replace the agent runtime.""" + return None + async def score( self, trace: Trace, runtime: Runtime | None = None, + scoring_runtime_config: RuntimeConfig | None = None, ) -> None: judges = self.plugged_judges() available = {"task": self.data, "trace": trace} diff --git a/verifiers/v1/tasksets/harbor/__init__.py b/verifiers/v1/tasksets/harbor/__init__.py index 2c66acb0d..7f357e0d4 100644 --- a/verifiers/v1/tasksets/harbor/__init__.py +++ b/verifiers/v1/tasksets/harbor/__init__.py @@ -3,6 +3,13 @@ HarborData, HarborTask, HarborTaskset, + VerifierConfig, ) -__all__ = ["HarborConfig", "HarborData", "HarborTask", "HarborTaskset"] +__all__ = [ + "HarborConfig", + "HarborData", + "HarborTask", + "HarborTaskset", + "VerifierConfig", +] diff --git a/verifiers/v1/tasksets/harbor/taskset.py b/verifiers/v1/tasksets/harbor/taskset.py index c1a155091..63beb7ad8 100644 --- a/verifiers/v1/tasksets/harbor/taskset.py +++ b/verifiers/v1/tasksets/harbor/taskset.py @@ -1,17 +1,15 @@ """Harbor tasksets backed by Harbor Hub packages. -The Harbor CLI downloads and caches each task directory. Its verifier runs in the -same runtime the harness edited, then writes the score to -``/logs/verifier/reward.txt``. - -A pullable ``[environment].docker_image`` becomes ``TaskData.image``. Verifiers does -not build Dockerfile-only environments, so those are rejected unless ``ignore_dockerfile`` -deliberately uses the harness runtime image. Tasks without an environment also use that -image unless ``require_image`` is set. +Shared verifiers grade in the agent runtime. Separate verifiers get a fresh runtime on +the same provider, with declared artifacts restored at their original paths. Verifiers +only supports pullable Harbor images unless ``ignore_dockerfile`` is set. """ import hashlib import io +import json +import logging +import shlex import shutil import subprocess import sys @@ -20,19 +18,25 @@ import tomllib from collections.abc import Iterator from functools import lru_cache -from pathlib import Path +from pathlib import Path, PurePosixPath +from typing import Any, cast from pydantic import Field from verifiers.v1.decorators import reward from verifiers.v1.errors import SandboxError -from verifiers.v1.runtimes import Runtime +from verifiers.v1.runtimes import Runtime, RuntimeConfig, make_runtime from verifiers.v1.task import Task, TaskData, TaskResources, TaskTimeout from verifiers.v1.taskset import Taskset, TasksetConfig +from verifiers.v1.trace import Trace from verifiers.v1.types import StrictBaseModel +logger = logging.getLogger(__name__) + CACHE = Path.home() / ".cache" / "harbor" HARBOR_INSTALL_HINT = "uv sync --python 3.12 --extra harbor" +ARTIFACTS_TAR = "/tmp/.vf-artifacts.tgz" +PRE_ARTIFACTS = "/tmp/.vf-pre-artifacts.sh" class HarborConfig(TasksetConfig): @@ -75,6 +79,22 @@ class Author(StrictBaseModel): email: str | None = None +class Artifact(StrictBaseModel): + """An artifact path and optional directory-relative exclude patterns.""" + + source: str + exclude: list[str] = [] + + +class VerifierConfig(StrictBaseModel): + """Configuration for a separate verifier runtime.""" + + image: str | None = None + resources: TaskResources = TaskResources() + workdir: str | None = None + fresh_copy: bool = False + + class HarborData(TaskData): """Parsed ``task.toml`` metadata plus the host-side verifier directory. @@ -88,18 +108,53 @@ class HarborData(TaskData): category: str | None = None tags: list[str] = [] task_dir: str = Field("", exclude=True) - """Host path to the task dir; used to stage tests/ to verify, not serialized.""" + """Host path used to stage ``tests/``; not serialized.""" verifier_env: dict[str, str] = {} - """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.""" + """Unresolved Harbor verifier environment variables.""" + verifier: VerifierConfig | None = None + """Separate verifier configuration; ``None`` means shared mode.""" + artifacts: list[Artifact] = [] + """Artifacts transferred to a separate verifier at their original paths.""" class HarborTask(Task[HarborData]): - """Stage and run Harbor's verifier inside the task's live runtime.""" + """Stage and run a Harbor verifier.""" - @reward(weight=1.0) - async def solved(self, runtime: Runtime) -> float: + def scoring_runtime_config(self, runtime: Runtime) -> RuntimeConfig | None: + verifier_data = self.data.verifier + if verifier_data is None: + return None + return verifier_runtime_config( + runtime, verifier_data, self.data.resources, self.data.name + ) + + async def score( + self, + trace: Trace, + runtime: Runtime | None = None, + scoring_runtime_config: RuntimeConfig | None = None, + ) -> None: + if self.data.verifier is None or runtime is None: + await super().score(trace, runtime) + return + config = scoring_runtime_config or self.scoring_runtime_config(runtime) + assert config is not None + await self._pre_artifacts(runtime) + archives = await self._collect_artifacts(runtime) + await runtime.stop_confirmed() + verifier = make_runtime(config, name=f"{runtime.name}-verifier") + try: + await verifier.start() + await self._prepare_verifier(verifier, archives) + await super().score(trace, verifier) + finally: + await verifier.stop() + + async def _prepare_verifier( + self, + runtime: Runtime, + archives: list[tuple[str, bytes, list[str]]], + ) -> None: await runtime.write( "/tmp/tests.tgz", make_tar(Path(self.data.task_dir) / "tests") ) @@ -107,19 +162,191 @@ async def solved(self, runtime: Runtime) -> float: [ "sh", "-c", - "mkdir -p /logs/verifier /tests && tar -xzf /tmp/tests.tgz -C /tests", + "mkdir -p /logs/verifier && rm -rf /tests && mkdir -p /tests " + "&& tar -xzf /tmp/tests.tgz -C /tests", ], {}, ) + for tar_path, data, extract in archives: + await runtime.write(tar_path, data) + extracted = await runtime.run(extract, {}) + if extracted.exit_code != 0: + raise SandboxError( + "artifact extraction failed in the verifier runtime: " + f"{extracted.stderr.strip()}" + ) + + @reward(weight=1.0) + async def solved(self, runtime: Runtime) -> float | dict[str, float]: + if self.data.verifier is None: + await self._prepare_verifier(runtime, []) await runtime.run( ["sh", "-c", "cd /tests && bash test.sh"], verifier_env(self.data) ) + return await self._read_reward(runtime) + + async def _read_reward(self, runtime: Runtime) -> float | dict[str, float]: + """Read Harbor rewards without changing shared-mode ``reward.txt`` semantics.""" + if self.data.verifier is not None: + try: + data = json.loads(await runtime.read("/logs/verifier/reward.json")) + if isinstance(data, dict): + reward_value = data.get("reward") + if type(reward_value) in (int, float): + return float(reward_value) + if data and all(type(v) in (int, float) for v in data.values()): + return {key: float(value) for key, value in data.items()} + except (SandboxError, OSError, TypeError, ValueError): + pass try: reward = (await runtime.read("/logs/verifier/reward.txt")).decode().strip() return float(reward or 0) except (SandboxError, OSError, ValueError): return 0.0 + async def _pre_artifacts(self, agent: Runtime) -> None: + """Run the optional, best-effort artifact capture hook.""" + script = Path(self.data.task_dir) / "pre_artifacts.sh" + if not script.is_file(): + return + await agent.write(PRE_ARTIFACTS, script.read_bytes()) + result = await agent.run(["bash", PRE_ARTIFACTS], {}) + if result.exit_code != 0: + logger.warning( + "%s: pre_artifacts.sh exited %d — grading whatever artifacts exist: %s", + self.data.name, + result.exit_code, + result.stderr.strip(), + ) + + async def _collect_artifacts( + self, agent: Runtime + ) -> list[tuple[str, bytes, list[str]]]: + """Snapshot agent artifacts for restoration after confirmed teardown.""" + workdir = PurePosixPath(getattr(agent.info, "workdir", None) or "/") + artifacts = [] + for artifact in self.data.artifacts: + source = PurePosixPath(artifact.source) + if not source.is_absolute(): + source = workdir / source + artifacts.append(artifact.model_copy(update={"source": str(source)})) + plain = ["/logs/artifacts"] + [ + artifact.source for artifact in artifacts if not artifact.exclude + ] + excluded = [artifact for artifact in artifacts if artifact.exclude] + probe = await agent.run( + [ + "sh", + "-c", + 'for p in "$@"; do if [ -d "$p" ]; then printf "d %s\\n" "$p"; ' + 'elif [ -e "$p" ]; then printf "f %s\\n" "$p"; fi; done; exit 0', + "sh", + *plain, + *[artifact.source for artifact in excluded], + ], + {}, + ) + # The command exits zero unless the runtime could not execute it. + if probe.exit_code != 0: + raise SandboxError( + f"artifact probe failed in the agent runtime: {probe.stderr.strip()}" + ) + kinds = {} + for line in probe.stdout.splitlines(): + kind, _, path = line.partition(" ") + kinds[path] = kind + # Harbor ignores excludes on files, so include them in the combined tar. + combined = [path for path in plain if path in kinds] + [ + artifact.source + for artifact in excluded + if kinds.get(artifact.source) == "f" + ] + archives = [] + if combined: + archives.append( + await self._archive_tar( + agent, + ARTIFACTS_TAR, + ["tar", "-czf", ARTIFACTS_TAR, "-C", "/", "--"] + + [(path.lstrip("/") or ".") for path in combined], + ["tar", "-xzf", ARTIFACTS_TAR, "-C", "/"], + ) + ) + for index, artifact in enumerate( + a for a in excluded if kinds.get(a.source) == "d" + ): + tar_path = f"/tmp/.vf-artifact-{index}.tgz" + archives.append( + await self._archive_tar( + agent, + tar_path, + [ + "tar", + "-czf", + tar_path, + *[f"--exclude={pattern}" for pattern in artifact.exclude], + "-C", + artifact.source, + ".", + ], + [ + "sh", + "-c", + f"mkdir -p {shlex.quote(artifact.source)} " + f"&& tar -xzf {shlex.quote(tar_path)} -C {shlex.quote(artifact.source)}", + ], + ) + ) + return archives + + @staticmethod + async def _archive_tar( + agent: Runtime, + tar_path: str, + create: list[str], + extract: list[str], + ) -> tuple[str, bytes, list[str]]: + created = await agent.run(create, {}) + if created.exit_code != 0: + raise SandboxError( + f"artifact archival failed in the agent runtime: {created.stderr.strip()}" + ) + return tar_path, await agent.read(tar_path), extract + + +def verifier_runtime_config( + runtime: Runtime, + verifier: VerifierConfig, + task_resources: TaskResources, + task_name: str | None, +) -> RuntimeConfig: + """Derive a separate verifier runtime from the agent runtime.""" + info = runtime.info + if info.type not in ("docker", "prime"): + raise ValueError( + f"task {task_name!r} declares a separate verifier environment, which needs a " + f"runtime with confirmed teardown (docker or prime), not {info.type!r}" + ) + updates: dict[str, Any] = {"id": None} + if not verifier.fresh_copy: + # An explicit verifier environment does not inherit task resource requests. + # Preserve runtime overrides unless they match the task's value exactly. + for field in ("cpu", "memory", "gpu", "disk"): + spec = type(info).model_fields.get(field) + task_value = getattr(task_resources, field) + if ( + spec is not None + and task_value is not None + and getattr(info, field) == task_value + ): + updates[field] = spec.default + if verifier.image is not None: + updates["image"] = verifier.image + updates.update(verifier.resources.model_dump(exclude_none=True)) + if verifier.workdir is not None: + updates["workdir"] = verifier.workdir + return cast(RuntimeConfig, info.model_copy(update=updates)) + def harbor_cli() -> str: scripts_dir = Path(sys.executable).parent @@ -272,6 +499,83 @@ def parse_resources(env: dict, multiplier: float = 1.0) -> TaskResources: ) +def parse_verifier( + task_dir: Path, config: dict, harbor_config: HarborConfig +) -> dict[str, Any]: + """Parse Harbor's shared or separate verifier configuration.""" + verifier = config.get("verifier", {}) + env = verifier.get("environment") + mode = verifier.get("environment_mode") + if mode not in (None, "shared", "separate"): + raise ValueError( + f"{task_dir.name}: invalid [verifier].environment_mode {mode!r} " + "(expected 'shared' or 'separate')" + ) + if mode == "shared" and env is not None: + raise ValueError( + f"{task_dir.name}: [verifier.environment] requires a separate verifier " + '(environment_mode = "separate", or no explicit mode, which it implies)' + ) + fields: dict[str, Any] = {"verifier_env": verifier.get("env", {})} + if mode != "separate" and env is None: + return fields + if verifier.get("collect"): + raise ValueError( + f"{task_dir.name}: [[verifier.collect]] hooks aren't supported (they need " + "compose sidecars); a separate-verifier task that declares them can't run" + ) + verifier_environment = env if env is not None else config.get("environment", {}) + network_sections = (verifier, verifier_environment) + if any( + section.get("allowed_hosts") + or section.get("network_mode") not in (None, "public") + or section.get("allow_internet") is False + for section in network_sections + ): + logger.warning( + "%s: the verifier configuration declares a network policy, which " + "isn't enforced — the verifier grades with the runtime's default network", + task_dir.name, + ) + artifacts: list[Artifact] = [] + for entry in config.get("artifacts", []): + if isinstance(entry, str): + artifacts.append(Artifact(source=entry)) + continue + source, service = entry.get("source"), entry.get("service") + if service not in (None, "main"): + raise ValueError( + f"{task_dir.name}: artifact {source!r} targets compose service " + f"{service!r}, which isn't supported (no compose sidecars); this " + "separate-verifier task can't run" + ) + artifacts.append(Artifact(source=source, exclude=entry.get("exclude", []))) + fields["artifacts"] = artifacts + verifier_config = VerifierConfig(fresh_copy=True) + if env is not None: + image = env.get("docker_image") + if image is None and not harbor_config.ignore_dockerfile: + source = ( + " from tests/Dockerfile" + if (task_dir / "tests" / "Dockerfile").exists() + else "" + ) + raise ValueError( + f"{task_dir.name}: [verifier.environment] has no docker_image, so Harbor " + f"would build the verifier image{source} — building Dockerfiles isn't " + "supported, so this task can't run. Pass --taskset.ignore-dockerfile to " + "grade it on the task's image instead." + ) + verifier_config = VerifierConfig( + image=image, + resources=parse_resources(env, harbor_config.resource_multiplier), + workdir=env.get("workdir"), + ) + fields["verifier_env"] = {**env.get("env", {}), **fields["verifier_env"]} + fields["verifier"] = verifier_config + return fields + + def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborData: config = tomllib.loads((task_dir / "task.toml").read_text()) task, meta = config.get("task", {}), config.get("metadata", {}) @@ -312,7 +616,7 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD category=meta.get("category"), tags=meta.get("tags", []), task_dir=str(task_dir), - verifier_env=config.get("verifier", {}).get("env", {}), + **parse_verifier(task_dir, config, harbor_config), )