diff --git a/docs/v1/harbor.md b/docs/v1/harbor.md index 260e7d05f4..1b3fd1f449 100644 --- a/docs/v1/harbor.md +++ b/docs/v1/harbor.md @@ -70,9 +70,27 @@ resource_multiplier = 2.0 The `timeout_multiplier` multiplies both the agent and verifier timeout, while the `resource_multiplier` multiplies the task's CPU, memory and disk space. You might want to use these multipliers when the tasks set too tight limits and/or the agent is slow. +## Separate verifier runtimes + +Harbor's separate verifier mode is selected directly from `task.toml`. After the agent finishes, Verifiers collects `/logs/artifacts` plus the task's configured `artifacts`, stops the agent runtime, starts a clean verifier runtime, restores those artifacts at their original absolute paths, and runs the verifier there. + +```toml +artifacts = ["/workspace/final-output"] + +[verifier] +environment_mode = "separate" + +[verifier.environment] +docker_image = "registry.example.com/my-benchmark-verifier:latest" +network_mode = "no-network" +``` + +Separate verification requires a prebuilt `[verifier.environment].docker_image` containing `/tests/test.sh`; Verifiers does not build `tests/Dockerfile`. The verifier environment can independently declare its workdir, resources, environment variables, and `public` or `no-network` network access. It is supported by the Docker and Prime runtimes. + +Only filesystem artifacts can cross this boundary. Artifact sources must be absolute, non-root paths and either regular files or directories containing only regular files and directories. Symlinks and special files are rejected; nested empty directories and file mode bits are not retained. Transfer is limited to 256 MiB across 10,000 files and a 4 MiB path manifest. Runtime images must use a default user that can create `/logs` and restore every declared artifact root. Compose sidecar artifacts, artifact destinations or exclude patterns, verifier healthchecks or MCP servers, accelerator constraints, allowlist networking, Windows verifier images, and multi-step tasks are not supported. A task whose verifier needs a live service, installed system state, or another undeclared agent-runtime side effect must remain shared or first export that state as an artifact. + ## 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: -- `no-network` support for sandbox runtimes ([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)) +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: +- Complete network-policy support, including allowlists and shared-runtime phase switching ([Harbor Docs](https://www.harborframework.com/docs/tasks/network-policy)) - Multi-step tasks ([Harbor Docs](https://www.harborframework.com/docs/tasks/multi-step)) diff --git a/verifiers/v1/env.py b/verifiers/v1/env.py index 29c22ab85a..e9f150f816 100644 --- a/verifiers/v1/env.py +++ b/verifiers/v1/env.py @@ -27,7 +27,7 @@ SubprocessConfig, runtime_is_local, ) -from verifiers.v1.task import Task, resolve_server_config +from verifiers.v1.task import Task, TaskData, resolve_server_config from verifiers.v1.taskset import Taskset, TasksetConfig from verifiers.v1.utils.generic import generic_type from verifiers.v1.mcp import SharedToolServer, serve_shared @@ -177,30 +177,33 @@ class EnvServerConfig(EnvConfig): def resolve_runtime_config( - base: RuntimeConfig, task: Task, warned: set[tuple[str, str]] | None = None + base: RuntimeConfig, + task: Task | TaskData, + warned: set[tuple[str, str]] | None = None, ) -> RuntimeConfig: """Resolve a task's runtime config from a `base`: inject the task's `image` (a task with an image must run in a container — refuse subprocess), and apply its `workdir` and requested `resources` to the fields the runtime supports. Precedence is cli/toml > task > default; a resource the runtime doesn't support warns once (deduped via `warned`). Shared by `Environment.runtime_for` (rollouts) and the `validate` entrypoint.""" + data = task.data if isinstance(task, Task) else task config = base updates: dict = {} - if task.data.image is not None: + if data.image is not None: if isinstance(config, SubprocessConfig): raise ValueError( - f"task {task.data.idx!r} requires image {task.data.image!r}, but the subprocess " + f"task {data.idx!r} requires image {data.image!r}, but the subprocess " "runtime has no container; use the docker or prime runtime" ) - updates["image"] = task.data.image + updates["image"] = data.image workdir_spec = type(config).model_fields.get("workdir") if ( - task.data.workdir is not None + data.workdir is not None and workdir_spec is not None and getattr(config, "workdir") == workdir_spec.default ): - updates["workdir"] = task.data.workdir - for field, value in task.data.resources.model_dump(exclude_none=True).items(): + updates["workdir"] = data.workdir + for field, value in data.resources.model_dump(exclude_none=True).items(): spec = type(config).model_fields.get(field) if spec is None: key = (config.type, field) diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index 8a04ef7851..fa64e9b679 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -64,6 +64,9 @@ def __init__( self.harness = harness self.ctx = ctx self.runtime_config = runtime_config + self.scoring_runtime_config = task.scoring_runtime_config( + harness.config.runtime + ) self.setup_timeout = setup_timeout self.harness_timeout = harness_timeout self.finalize_timeout = finalize_timeout @@ -216,14 +219,22 @@ async def run(self) -> Trace: self.phase = Phase.SCORING 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, - ) + # Group rewards run later, after the runtime is gone. A task with an + # isolated scorer must let the harness inspect the agent runtime first; + # its score method then owns the transition to the separate runtime. + async with asyncio.timeout(self.scoring_timeout): + if self.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=self.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 b8a31fcddb..45d7054e14 100644 --- a/verifiers/v1/runtimes/base.py +++ b/verifiers/v1/runtimes/base.py @@ -15,6 +15,7 @@ from pydantic_config import BaseConfig +from verifiers.v1.errors import SandboxError from verifiers.v1.utils.aio import run_shielded logger = logging.getLogger(__name__) @@ -128,6 +129,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. @@ -223,10 +230,26 @@ async def run_uv_script( argv = await self.prepare_uv_script(script, env) return await self.run([*argv, *(args or [])], env or {}) + def _validate_read_limit(self, max_bytes: int) -> None: + if max_bytes < 0: + raise ValueError("max_bytes must be non-negative") + @abstractmethod async def read(self, path: str) -> bytes: pass + async def read_bounded(self, path: str, max_bytes: int) -> bytes: + """Read at most ``max_bytes`` from a file. + + Runtimes with streaming file APIs should override this to enforce the limit + before buffering the full file. + """ + self._validate_read_limit(max_bytes) + data = await self.read(path) + if len(data) > max_bytes: + raise SandboxError(f"read {path!r}: exceeds the {max_bytes} byte limit") + return data + @abstractmethod async def write(self, path: str, data: bytes) -> None: pass diff --git a/verifiers/v1/runtimes/docker.py b/verifiers/v1/runtimes/docker.py index f99919db87..670f9b5a1c 100644 --- a/verifiers/v1/runtimes/docker.py +++ b/verifiers/v1/runtimes/docker.py @@ -1,4 +1,4 @@ -"""Local Docker runtime using host networking.""" +"""Local Docker runtime.""" import asyncio import contextlib @@ -27,6 +27,9 @@ class DockerConfig(BaseConfig): type: Literal["docker"] = "docker" image: str = "python:3.11-slim" workdir: str = "/app" + network_access: bool = True + """Whether the container can access the network. A harness runtime needs this to + reach model interception; disable it only for isolated runtimes such as verifiers.""" # TaskData.resources uses these units; non-default runtime config values take precedence. cpu: float | None = None """Pin the container to this many CPU cores (docker `--cpus`). None = unlimited.""" @@ -99,7 +102,7 @@ async def start(self) -> None: "run", "--detach", "--network", - "host", + "host" if self.config.network_access else "none", *limits, "--workdir", self.config.workdir, @@ -173,6 +176,39 @@ async def read(self, path: str) -> bytes: ) return stdout + async def read_bounded(self, path: str, max_bytes: int) -> bytes: + self._validate_read_limit(max_bytes) + proc = await asyncio.create_subprocess_exec( + "docker", + "exec", + "--workdir", + self.config.workdir, + self._container, + "cat", + path, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL, + ) + try: + try: + stdout = await proc.stdout.readexactly(max_bytes + 1) + except asyncio.IncompleteReadError as exc: + stdout = exc.partial + else: + with contextlib.suppress(ProcessLookupError): + proc.kill() + await proc.wait() + raise SandboxError(f"read {path!r}: exceeds the {max_bytes} byte limit") + await proc.wait() + if proc.returncode != 0: + raise SandboxError(f"read {path!r}: docker exec failed") + return stdout + finally: + if proc.returncode is None: + with contextlib.suppress(ProcessLookupError): + proc.kill() + await proc.wait() + async def write(self, path: str, data: bytes) -> None: parent = shlex.quote(str(PurePosixPath(path).parent)) proc = await asyncio.create_subprocess_exec( @@ -195,6 +231,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 e110f6b844..9ded585176 100644 --- a/verifiers/v1/runtimes/prime.py +++ b/verifiers/v1/runtimes/prime.py @@ -8,6 +8,7 @@ import asyncio import contextlib +import io import logging import math import shlex @@ -15,6 +16,7 @@ from pathlib import Path, PurePosixPath from typing import ClassVar, Literal +import httpx from pydantic import model_validator from pydantic_config import BaseConfig @@ -32,6 +34,16 @@ MAX_LIFETIME = 24 * 60 * 60 """Prime's fixed cap (seconds) on any sandbox's total lifetime.""" +DOWNLOAD_ATTEMPTS = 4 +DOWNLOAD_RETRY_STATUSES = {408, 500, 502, 503, 504, 524} +DOWNLOAD_RETRY_ERRORS = ( + httpx.RemoteProtocolError, + httpx.ConnectError, + httpx.PoolTimeout, + httpx.ReadError, + httpx.ConnectTimeout, + httpx.ReadTimeout, +) class PrimeConfig(BaseConfig): @@ -237,6 +249,62 @@ async def read(self, path: str) -> bytes: except Exception as e: raise SandboxError(f"read {path!r}: {e}") from e + async def read_bounded(self, path: str, max_bytes: int) -> bytes: + self._validate_read_limit(max_bytes) + target = ( + path + if path.startswith("/") + else f"{self.config.workdir.rstrip('/')}/{path}" + ) + try: + # The SDK's download_file buffers response.content. Use its gateway + # connection directly so an untrusted file is capped while streaming. + for attempt in range(DOWNLOAD_ATTEMPTS): + auth = await self._client._auth_cache.get_or_refresh(self.info.id) + url = ( + f"{auth['gateway_url'].rstrip('/')}/" + f"{auth['user_ns']}/{auth['job_id']}/download" + ) + headers = {"Authorization": f"Bearer {auth['token']}"} + params = {"path": target, "sandbox_id": self.info.id} + buffer = io.BytesIO() + try: + client = self._client._get_gateway_client() + async with client.stream( + "GET", url, headers=headers, params=params, timeout=300 + ) as response: + response.raise_for_status() + async for chunk in response.aiter_bytes(1024 * 1024): + if buffer.tell() + len(chunk) > max_bytes: + raise SandboxError( + f"read {path!r}: exceeds the {max_bytes} byte limit" + ) + buffer.write(chunk) + return buffer.getvalue() + except httpx.HTTPStatusError as error: + status = error.response.status_code + if status == 401 and attempt < DOWNLOAD_ATTEMPTS - 1: + await self._client.clear_auth_cache() + continue + if status == 409: + if await self._client._should_retry_409( + self.info.id, error, attempt + ): + continue + if ( + status not in DOWNLOAD_RETRY_STATUSES + or attempt == DOWNLOAD_ATTEMPTS - 1 + ): + raise + except DOWNLOAD_RETRY_ERRORS: + if attempt == DOWNLOAD_ATTEMPTS - 1: + raise + await asyncio.sleep(2**attempt) + except SandboxError: + raise + except Exception as e: + raise SandboxError(f"read {path!r}: {e}") from e + async def write(self, path: str, data: bytes) -> None: # Upload via the gateway (multipart) — never inline the bytes on the command line # (a large file, e.g. a task tarball, overflows the exec command-length limit and @@ -259,6 +327,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 db955b7bad..4e8c104a24 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,20 @@ async def finalize(self, trace: Trace, runtime: Runtime) -> None: async def validate(self, runtime: Runtime) -> bool: return True + def scoring_runtime_config(self, base: RuntimeConfig) -> RuntimeConfig | None: + """Return a separate runtime config when this task must score in isolation. + + ``None`` keeps scoring in the agent runtime. A separate scorer owns its + runtime transition in ``score``; the rollout first records harness metrics + while the agent runtime is still live, then passes this config to ``score``. + """ + 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/taskset.py b/verifiers/v1/tasksets/harbor/taskset.py index c1a1550918..9b1cd1efcd 100644 --- a/verifiers/v1/tasksets/harbor/taskset.py +++ b/verifiers/v1/tasksets/harbor/taskset.py @@ -1,8 +1,8 @@ """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``. +Shared verifiers run in the agent runtime. Separate verifiers collect only Harbor's +declared artifacts, stop the agent runtime, start a clean verifier runtime, restore the +artifacts at their original paths, and run the verifier there. A pullable ``[environment].docker_image`` becomes ``TaskData.image``. Verifiers does not build Dockerfile-only environments, so those are rejected unless ``ignore_dockerfile`` @@ -10,6 +10,7 @@ image unless ``require_image`` is set. """ +import asyncio import hashlib import io import shutil @@ -18,21 +19,39 @@ import tarfile import tempfile import tomllib +import uuid from collections.abc import Iterator from functools import lru_cache -from pathlib import Path +from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING 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 ( + DockerConfig, + PrimeConfig, + 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 CACHE = Path.home() / ".cache" / "harbor" HARBOR_INSTALL_HINT = "uv sync --python 3.12 --extra harbor" +MAX_ARTIFACT_BYTES = 256 * 1024 * 1024 +MAX_ARTIFACT_FILES = 10_000 +MAX_ARTIFACT_MANIFEST_BYTES = 4 * 1024 * 1024 + +if TYPE_CHECKING: + from harbor.models.task.config import ( + EnvironmentConfig, + TaskConfig as HarborTaskConfig, + ) class HarborConfig(TasksetConfig): @@ -75,6 +94,19 @@ class Author(StrictBaseModel): email: str | None = None +class HarborArtifact(StrictBaseModel): + source: str + + +class HarborVerifier(StrictBaseModel): + separate: bool = False + image: str | None = None + workdir: str | None = None + resources: TaskResources = TaskResources() + network_access: bool = True + artifacts: list[HarborArtifact] = [] + + class HarborData(TaskData): """Parsed ``task.toml`` metadata plus the host-side verifier directory. @@ -90,29 +122,237 @@ class HarborData(TaskData): task_dir: str = Field("", exclude=True) """Host path to the task dir; used to stage tests/ to verify, 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.""" + """Raw [verifier.env] entries, resolved on the host at scoring time.""" + verifier: HarborVerifier = Field(default_factory=HarborVerifier, exclude=True) + """Resolved verifier mode and runtime data, kept host-side for scoring.""" class HarborTask(Task[HarborData]): - """Stage and run Harbor's verifier inside the task's live runtime.""" + """Run a Harbor verifier in its shared or separate runtime.""" + + def scoring_runtime_config(self, base: RuntimeConfig) -> RuntimeConfig | None: + verifier = self.data.verifier + if not verifier.separate: + return None + if not isinstance(base, (DockerConfig, PrimeConfig)): + raise ValueError( + "separate Harbor verification needs a docker or prime runtime" + ) + + from verifiers.v1.env import resolve_runtime_config + + data = TaskData( + idx=self.data.idx, + prompt=None, + image=verifier.image, + workdir=verifier.workdir, + resources=verifier.resources, + ) + config = resolve_runtime_config(base, data) + return config.model_copy(update={"network_access": verifier.network_access}) + + async def score( + self, + trace: Trace, + runtime: Runtime | None = None, + scoring_runtime_config: RuntimeConfig | None = None, + ) -> None: + if not self.data.verifier.separate: + await super().score(trace, runtime) + return + if runtime is None: + await super().score(trace) + return + if scoring_runtime_config is None: + raise ValueError( + "separate Harbor scoring requires a verifier runtime config" + ) + + with tempfile.TemporaryDirectory() as temp: + files, directories = await self._collect_artifacts(runtime, Path(temp)) + # No verifier process starts until the agent runtime is gone. + await runtime.stop_confirmed() + target = make_runtime( + scoring_runtime_config, name=f"{runtime.name}-verifier" + ) + try: + await target.start() + await self._prepare_verifier(target, files, directories) + await super().score(trace, target) + finally: + await target.stop() + + async def _collect_artifacts( + self, runtime: Runtime, staging: Path + ) -> tuple[list[tuple[str, Path]], list[str]]: + await run_checked( + runtime, + ["sh", "-c", "rm -rf /logs/verifier && mkdir -p /logs/verifier"], + {}, + "Harbor artifact staging setup", + ) + files: list[tuple[str, Path]] = [] + directories: list[str] = [] + total_bytes = 0 + for artifact in self.data.verifier.artifacts: + source = artifact.source + result = await runtime.run( + [ + "sh", + "-c", + 'if [ -L "$1" ]; then printf unsupported; ' + 'elif [ ! -e "$1" ]; then printf missing; ' + 'elif [ -f "$1" ]; then printf file; ' + 'elif [ -d "$1" ]; then printf directory; ' + "else printf unsupported; fi", + "sh", + source, + ], + {}, + ) + if result.exit_code: + detail = (result.stderr or result.stdout).strip() + raise SandboxError( + f"Harbor artifact probe failed for {source!r}: {detail}" + ) + kind = result.stdout + if kind == "missing": + continue + if kind not in ("file", "directory"): + raise SandboxError( + f"Harbor artifact {source!r} must be a regular file or directory" + ) + + paths = [source] + if kind == "directory": + directories.append(source) + manifest = f"/logs/verifier/.vf-harbor-{uuid.uuid4().hex}" + listed = await runtime.run( + [ + "sh", + "-c", + 'find "$1" -mindepth 1 ! -type d ! -type f ' + '-print -quit > "$3" && [ ! -s "$3" ] && ' + 'find "$1" -type f -print0 > "$2"', + "sh", + source, + manifest, + f"{manifest}.bad", + ], + {}, + ) + try: + if listed.exit_code: + raise SandboxError( + f"Harbor artifact directory {source!r} must contain only " + "regular files and directories" + ) + raw = await runtime.read_bounded( + manifest, MAX_ARTIFACT_MANIFEST_BYTES + ) + finally: + await runtime.run(["rm", "-f", manifest, f"{manifest}.bad"], {}) + try: + paths = [ + value.decode() + for value in raw.rstrip(b"\0").split(b"\0") + if value + ] + except UnicodeDecodeError as exc: + raise SandboxError( + f"Harbor artifact directory {source!r} has a non-UTF-8 path" + ) from exc + + if len(files) + len(paths) > MAX_ARTIFACT_FILES: + raise SandboxError( + f"Harbor artifacts exceed the {MAX_ARTIFACT_FILES:,} file limit" + ) + root = PurePosixPath(source) + for path in paths: + artifact_path = PurePosixPath(path) + if ( + not artifact_path.is_absolute() + or ".." in artifact_path.parts + or not (artifact_path == root or root in artifact_path.parents) + ): + raise SandboxError( + f"Harbor artifact directory produced unsafe path {path!r}" + ) + data = await runtime.read_bounded( + artifact_path.as_posix(), MAX_ARTIFACT_BYTES - total_bytes + ) + total_bytes += len(data) + host_path = staging / str(len(files)) + await asyncio.to_thread(host_path.write_bytes, data) + files.append((artifact_path.as_posix(), host_path)) + return files, directories + + async def _prepare_verifier( + self, + runtime: Runtime, + files: list[tuple[str, Path]], + directories: list[str], + ) -> None: + await run_checked( + runtime, + ["rm", "-rf", "/logs/verifier"], + {}, + "Harbor verifier setup", + ) + await run_checked( + runtime, + ["mkdir", "-p", "/logs/verifier"], + {}, + "Harbor verifier setup", + ) + sources = [artifact.source for artifact in self.data.verifier.artifacts] + if sources: + await run_checked( + runtime, + ["rm", "-rf", *sources], + {}, + "Harbor artifact target cleanup", + ) + if all(path != "/logs/artifacts" for path, _ in files): + directories.insert(0, "/logs/artifacts") + directories = list(dict.fromkeys(directories)) + if directories: + await run_checked( + runtime, + ["mkdir", "-p", *directories], + {}, + "Harbor artifact directory restore", + ) + for path, host_path in files: + await runtime.write(path, await asyncio.to_thread(host_path.read_bytes)) @reward(weight=1.0) async def solved(self, runtime: Runtime) -> float: - await runtime.write( - "/tmp/tests.tgz", make_tar(Path(self.data.task_dir) / "tests") - ) - await runtime.run( - [ - "sh", - "-c", - "mkdir -p /logs/verifier /tests && tar -xzf /tmp/tests.tgz -C /tests", - ], + if not self.data.verifier.separate: + await runtime.write( + "/tmp/tests.tgz", make_tar(Path(self.data.task_dir) / "tests") + ) + await run_checked( + runtime, + [ + "sh", + "-c", + "rm -rf /logs/verifier /tests && " + "mkdir -p /logs/verifier /tests && " + "tar -xzf /tmp/tests.tgz -C /tests", + ], + {}, + "Harbor verifier setup", + ) + await run_checked( + runtime, + ["test", "-f", "/tests/test.sh"], {}, + "Harbor verifier test discovery", ) await runtime.run( - ["sh", "-c", "cd /tests && bash test.sh"], verifier_env(self.data) + ["sh", "-c", "cd /tests && bash test.sh"], + verifier_env(self.data), ) try: reward = (await runtime.read("/logs/verifier/reward.txt")).decode().strip() @@ -121,6 +361,15 @@ async def solved(self, runtime: Runtime) -> float: return 0.0 +async def run_checked( + runtime: Runtime, argv: list[str], env: dict[str, str], action: str +) -> None: + result = await runtime.run(argv, env) + if result.exit_code: + detail = (result.stderr or result.stdout).strip() + raise SandboxError(f"{action} failed: {detail}") + + def harbor_cli() -> str: scripts_dir = Path(sys.executable).parent harbor_bin = shutil.which("harbor", path=str(scripts_dir)) @@ -242,38 +491,135 @@ def resolve_image( return None -def size_to_mb(size: str | int | float) -> float: - """A Harbor size in MB, from either schema: current integer-MB fields or the - legacy schema-1.0 size strings ("8G", "512M", "64K").""" - if not isinstance(size, str): - return float(size) - scale = {"G": 1024.0, "M": 1.0, "K": 1 / 1024}.get(size.strip().upper()[-1:]) - if scale is None: +def parse_resources(env: "EnvironmentConfig", multiplier: float = 1.0) -> TaskResources: + """Convert Harbor's validated MB resources to Verifiers' GB resources.""" + return TaskResources( + cpu=env.cpus * multiplier if env.cpus else None, + memory=env.memory_mb / 1024 * multiplier if env.memory_mb else None, + gpu=str(env.gpus) if env.gpus else None, + disk=env.storage_mb / 1024 * multiplier if env.storage_mb else None, + ) + + +def parse_verifier( + task_dir: Path, config: "HarborTaskConfig", resource_multiplier: float +) -> tuple[HarborVerifier, dict[str, str]]: + # Harbor is optional, so importing this module still works without the extra. + from harbor.models.task.config import NetworkMode, TaskOS + from harbor.models.task.verifier_mode import ( + VerifierEnvironmentMode, + resolve_effective_verifier_env_config, + resolve_task_verifier_mode, + ) + + mode = resolve_task_verifier_mode(config) + if mode == VerifierEnvironmentMode.SHARED: + return HarborVerifier(), dict(config.verifier.env) + if config.verifier.collect: + raise ValueError( + f"{task_dir.name}: [[verifier.collect]] needs compose services, which " + "Verifiers runtimes do not support" + ) + if config.verifier.user is not None: + raise ValueError( + f"{task_dir.name}: [verifier].user is not supported for separate runtimes" + ) + + environment = resolve_effective_verifier_env_config(config, None) + if environment is None: + raise ValueError(f"{task_dir.name}: separate verifier environment is missing") + if config.verifier.environment is None or not environment.docker_image: + raise ValueError( + f"{task_dir.name}: separate verification needs a pullable " + "[verifier.environment].docker_image" + ) + if environment.os != TaskOS.LINUX: raise ValueError( - f"invalid Harbor size {size!r}: expected a number of MB or a " - "'[G|M|K]' string" + f"{task_dir.name}: separate Harbor verification supports Linux images only" + ) + unsupported = [ + field + for field in ("healthcheck", "mcp_servers", "skills_dir", "gpu_types", "tpu") + if getattr(environment, field) + ] + if unsupported: + raise ValueError( + f"{task_dir.name}: separate verifier environment fields are not supported: " + + ", ".join(unsupported) ) - return float(size.strip()[:-1]) * scale + network_mode = config.verifier.network_mode or environment.network_mode + if ( + network_mode == NetworkMode.ALLOWLIST + or config.verifier.allowed_hosts + or environment.allowed_hosts + ): + raise ValueError( + f"{task_dir.name}: verifier network allowlists are not supported" + ) -def parse_resources(env: dict, multiplier: float = 1.0) -> TaskResources: - # Harbor's current schema reports memory/storage as integer-MB fields; the - # legacy schema 1.0 used size strings under `memory`/`storage`, which Harbor - # still migrates (datasets like senior-swe-bench are authored against it). - # TaskResources stores GB. A zero GPU count means no GPU request rather than - # the string "0". - memory = env.get("memory_mb") or env.get("memory") - disk = env.get("storage_mb") or env.get("storage") - return TaskResources( - cpu=env["cpus"] * multiplier if env.get("cpus") else None, - memory=size_to_mb(memory) / 1024 * multiplier if memory else None, - gpu=str(env["gpus"]) if env.get("gpus") else None, - disk=size_to_mb(disk) / 1024 * multiplier if disk else None, + artifacts: list[HarborArtifact] = [] + for artifact in ["/logs/artifacts", *config.artifacts]: + source = artifact if isinstance(artifact, str) else artifact.source + if not isinstance(artifact, str) and artifact.service not in (None, "main"): + raise ValueError(f"{task_dir.name}: sidecar artifacts are not supported") + if not isinstance(artifact, str) and artifact.destination: + raise ValueError( + f"{task_dir.name}: artifact destinations are not supported" + ) + if not isinstance(artifact, str) and artifact.exclude: + raise ValueError( + f"{task_dir.name}: artifact exclude patterns are not supported" + ) + if not source.startswith("/") or ".." in PurePosixPath(source).parts: + raise ValueError( + f"{task_dir.name}: artifact source must be an absolute non-root path, " + f"got {source!r}" + ) + path = PurePosixPath(f"/{source.lstrip('/')}") + source = path.as_posix() + if path == PurePosixPath("/"): + raise ValueError( + f"{task_dir.name}: artifact source must be an absolute non-root path, " + f"got {source!r}" + ) + protected = (PurePosixPath("/tests"), PurePosixPath("/logs/verifier")) + if any( + path == target or path in target.parents or target in path.parents + for target in protected + ): + raise ValueError( + f"{task_dir.name}: artifact source {source!r} overlaps verifier-owned files" + ) + existing = [PurePosixPath(entry.source) for entry in artifacts] + # Harbor keeps the first declaration when artifact sources overlap. + if any( + path == other or path in other.parents or other in path.parents + for other in existing + ): + continue + artifacts.append(HarborArtifact(source=source)) + + return ( + HarborVerifier( + separate=True, + image=environment.docker_image, + workdir=environment.workdir, + resources=parse_resources(environment, resource_multiplier), + network_access=network_mode == NetworkMode.PUBLIC, + artifacts=artifacts, + ), + dict(environment.env) | dict(config.verifier.env), ) def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborData: + from harbor.models.task.config import TaskConfig as HarborTaskConfig + config = tomllib.loads((task_dir / "task.toml").read_text()) + parsed = HarborTaskConfig.model_validate(config) + if parsed.steps: + raise ValueError(f"{task_dir.name}: Harbor multi-step tasks are not supported") task, meta = config.get("task", {}), config.get("metadata", {}) authors = [Author(**a) for a in task.get("authors", [])] # Older registry entries stored one author in [metadata]. @@ -284,6 +630,9 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD else: harness_timeout = config.get("agent", {}).get("timeout_sec") scoring_timeout = config.get("verifier", {}).get("timeout_sec") + verifier, verifier_environment = parse_verifier( + task_dir, parsed, harbor_config.resource_multiplier + ) return HarborData( idx=idx, name=task.get("name") or task_dir.name, @@ -295,6 +644,7 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD harbor_config.require_image, harbor_config.ignore_dockerfile, ), + workdir=parsed.environment.workdir, timeout=TaskTimeout( harness=harness_timeout * harbor_config.timeout_multiplier if harness_timeout is not None @@ -304,7 +654,7 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD else None, ), resources=parse_resources( - config.get("environment", {}), harbor_config.resource_multiplier + parsed.environment, harbor_config.resource_multiplier ), keywords=task.get("keywords", []), authors=authors, @@ -312,7 +662,8 @@ 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", {}), + verifier_env=verifier_environment, + verifier=verifier, )