diff --git a/docs/v1/harbor.md b/docs/v1/harbor.md index 85717b7e6f..f2a40d850c 100644 --- a/docs/v1/harbor.md +++ b/docs/v1/harbor.md @@ -69,6 +69,10 @@ 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 from `task.toml`. Docker and Prime VM runtimes collect declared artifacts, confirm that the agent runtime is gone, then grade in a fresh runtime on the same provider. Explicit verifier environments need a pullable image unless `ignore_dockerfile = true`. + ## Network policies Harbor's effective agent network policy is applied to Docker or Prime VM harness @@ -89,8 +93,4 @@ accepts host-level entries and rejects combinations that need both policy modes. ## 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)) +verifiers does not have complete Harbor parity yet. The main missing features are verifier network allowlists, shared-runtime phase switching, compose sidecars, and multi-step tasks. diff --git a/verifiers/v1/agent.py b/verifiers/v1/agent.py index d7ffda028f..62c506538d 100644 --- a/verifiers/v1/agent.py +++ b/verifiers/v1/agent.py @@ -32,7 +32,7 @@ Runtime, RuntimeConfig, SubprocessConfig, - make_runtime, + provision_runtime, runtime_is_local, ) from verifiers.v1.session import RolloutLimits @@ -566,14 +566,8 @@ async def provision(self, task: Task | None = None) -> AsyncIterator[Runtime]: if task is not None else self.runtime_config ) - runtime = make_runtime(config) - try: - # start() inside the try: a failed start may already hold a remote - # sandbox, so it must reach stop() (safe on a partially-started runtime). - await runtime.start() + async with provision_runtime(config) as runtime: yield runtime - finally: - await runtime.stop() class _EpisodeAgent(Agent): diff --git a/verifiers/v1/mcp/launch.py b/verifiers/v1/mcp/launch.py index 7d92a31f2c..10684da5e8 100644 --- a/verifiers/v1/mcp/launch.py +++ b/verifiers/v1/mcp/launch.py @@ -22,7 +22,7 @@ from verifiers.v1.runtimes import ( NetworkPolicyConfig, Runtime, - make_runtime, + provision_runtime, runtime_is_local, ) from verifiers.v1.runtimes.base import _ENSURE_UV @@ -268,9 +268,7 @@ async def serve( if colocated and harness_runtime is not None: runtime = harness_runtime else: - runtime = make_runtime(cfg.runtime) - await runtime.start() - stack.push_async_callback(runtime.stop) + runtime = await stack.enter_async_context(provision_runtime(cfg.runtime)) # Only consumers outside the server runtime need its fixed published port. Colocated tools # use independent OS-assigned ports, avoiding clashes on the runtime's service port. exposed = runtime is not harness_runtime diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index 3df8632349..99216520e5 100644 --- a/verifiers/v1/rollout.py +++ b/verifiers/v1/rollout.py @@ -22,7 +22,7 @@ import logging import time from collections.abc import AsyncIterator, Callable -from contextlib import AsyncExitStack, asynccontextmanager +from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager from verifiers import __version__ from verifiers.v1.configs.agent import AgentConfig @@ -141,6 +141,7 @@ def __init__( self._interception = interception self.runtime = runtime self._owns_runtime = runtime is None + self._scoring_runtime: AbstractAsyncContextManager[Runtime] | None = None self.trace: Trace = Trace( task=TraceTask( type=type(task).__name__, @@ -238,6 +239,15 @@ async def open(self) -> bool: "task.prompt, or drive the run through agent.interaction() and open " "it with the first turn(message)" ) + self._scoring_runtime = self.task.scoring_runtime( + runtime, self.trace.agent.config.runtime + ) + if self._scoring_runtime is not None and ( + not self._owns_runtime or self._shared_tools + ): + raise TaskError( + "separate scoring needs an owned runtime without shared tool servers" + ) if self._owns_runtime: await runtime.start() await runtime.prepare_setup() @@ -411,13 +421,21 @@ async def close(self) -> Trace: trace.timing.scoring.start = now async with boundary(TaskError, "scoring"): # Cross-trace judgement runs 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, - ) + assert runtime is not None + if self._scoring_runtime is None: + async with asyncio.timeout(self._scoring_timeout): + await asyncio.gather( + self.task.score(trace, runtime), + self.harness.score(trace, runtime), + ) + else: + async with AsyncExitStack() as scoring: + async with asyncio.timeout(self._scoring_timeout): + await self.harness.score(trace, runtime) + target = await scoring.enter_async_context( + self._scoring_runtime + ) + await self.task.score(trace, target) trace.timing.scoring.end = time.time() except Exception as e: self.fail(e) @@ -435,7 +453,7 @@ async def close(self) -> Trace: if span.start and not span.end: span.end = now trace.split_generation() - if runtime is not None: + if runtime is not None and (not runtime.stopped or runtime.stop_failed): try: await self.harness.cleanup(trace, runtime) except Exception: diff --git a/verifiers/v1/runtimes/__init__.py b/verifiers/v1/runtimes/__init__.py index f84e2b273b..7fab1c7a1c 100644 --- a/verifiers/v1/runtimes/__init__.py +++ b/verifiers/v1/runtimes/__init__.py @@ -1,3 +1,5 @@ +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager from typing import Annotated from pydantic import Field @@ -45,6 +47,18 @@ def make_runtime(config: RuntimeConfig, name: str | None = None) -> Runtime: return runtime +@asynccontextmanager +async def provision_runtime( + config: RuntimeConfig, name: str | None = None +) -> AsyncIterator[Runtime]: + runtime = make_runtime(config, name) + try: + await runtime.start() + yield runtime + finally: + await runtime.stop() + + def runtime_is_local(config: RuntimeConfig) -> bool: """Whether a runtime of this config exchanges host-local URLs without a public tunnel, read off the runtime class without provisioning one.""" @@ -59,6 +73,7 @@ def runtime_is_local(config: RuntimeConfig) -> bool: "BaseRuntimeInfo", "NetworkPolicyConfig", "make_runtime", + "provision_runtime", "runtime_is_local", "SubprocessConfig", "SubprocessRuntime", diff --git a/verifiers/v1/runtimes/base.py b/verifiers/v1/runtimes/base.py index e8d80cb3b9..00576810d4 100644 --- a/verifiers/v1/runtimes/base.py +++ b/verifiers/v1/runtimes/base.py @@ -2,6 +2,7 @@ import asyncio import atexit +import base64 import contextlib import hashlib import logging @@ -146,6 +147,7 @@ def __init__(self, name: str | None = None) -> None: """Whether teardown has begun (set by `stop`). A stopped runtime is dead: a rollout refuses to borrow one — the owner tore it down, so any use is a lifetime bug in the borrowing program, caught up front instead of failing opaquely mid-harness.""" + self.stop_failed = False @property def type(self) -> str: @@ -164,6 +166,22 @@ async def stop(self) -> None: self.stopped = True # before the await: no new borrows once teardown begins await run_shielded(self.teardown()) + async def stop_confirmed(self) -> None: + """Free the resource or raise unless deletion is confirmed.""" + self.stopped = True + task = asyncio.create_task(self.teardown_confirmed()) + try: + await run_shielded(task) + finally: + self.stop_failed = ( + not task.done() or task.cancelled() or task.exception() is not None + ) + + async def teardown_confirmed(self) -> None: + 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. @@ -259,9 +277,33 @@ async def run_uv_script( argv = await self.prepare_uv_script(script, env) return await self.run([*argv, *(args or [])], env or {}) - @abstractmethod - async def read(self, path: str) -> bytes: - pass + async def read(self, path: str, max_bytes: int | None = None) -> bytes: + """Read a file, optionally bounding its output inside the runtime.""" + if max_bytes is None: + return await self._read(path) + if max_bytes < 0: + raise ValueError("max_bytes must be non-negative") + result = await self.run( + [ + "sh", + "-c", + 'head -c "$1" -- "$2" | base64', + "sh", + str(max_bytes + 1), + path, + ], + {}, + ) + if result.exit_code or result.stderr: + detail = (result.stderr or result.stdout).strip() + raise SandboxError(f"read {path!r}: {detail}") + data = base64.b64decode(result.stdout) + if len(data) > max_bytes: + raise SandboxError(f"read {path!r}: exceeds the {max_bytes} byte limit") + return data + + async def _read(self, path: str) -> bytes: + raise NotImplementedError(f"{type(self).__name__} does not support read") @abstractmethod async def write(self, path: str, data: bytes) -> None: diff --git a/verifiers/v1/runtimes/docker/__init__.py b/verifiers/v1/runtimes/docker/__init__.py index bbeee85409..defcc65461 100644 --- a/verifiers/v1/runtimes/docker/__init__.py +++ b/verifiers/v1/runtimes/docker/__init__.py @@ -334,15 +334,21 @@ async def run_background( if run.exit_code != 0: raise SandboxError(f"docker exec -d failed: {run.stderr.strip()}") - async def read(self, path: str) -> bytes: + async def read(self, path: str, max_bytes: int | None = None) -> bytes: + if max_bytes is not None and max_bytes < 0: + raise ValueError("max_bytes must be non-negative") + command = ( + ["cat", "--", path] + if max_bytes is None + else ["head", "-c", str(max_bytes + 1), "--", path] + ) proc = await asyncio.create_subprocess_exec( "docker", "exec", "--workdir", self.config.workdir, self._container, - "cat", - path, + *command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) @@ -351,6 +357,8 @@ async def read(self, path: str) -> bytes: raise SandboxError( f"read {path!r}: {stderr.decode(errors='replace').strip()}" ) + if max_bytes is not None and len(stdout) > max_bytes: + raise SandboxError(f"read {path!r}: exceeds the {max_bytes} byte limit") return stdout async def write(self, path: str, data: bytes) -> None: @@ -375,6 +383,18 @@ async def write(self, path: str, data: bytes) -> None: f"write {path!r}: {stderr.decode(errors='replace').strip()}" ) + async def teardown_confirmed(self) -> None: + if self._container is not None: + removed = await docker("rm", "--force", self._container) + if removed.exit_code and "No such container" not in removed.stderr: + raise SandboxError( + f"docker container {self._container!r} deletion failed: " + f"{removed.stderr.strip()}" + ) + self._stopped = True + if self._proxy is not None: + await self._proxy.stop() + def cleanup(self) -> None: if self._container is None or self._stopped: return diff --git a/verifiers/v1/runtimes/modal.py b/verifiers/v1/runtimes/modal.py index 14f5a67b8f..e44d6957c3 100644 --- a/verifiers/v1/runtimes/modal.py +++ b/verifiers/v1/runtimes/modal.py @@ -165,7 +165,7 @@ def _abs(self, path: str) -> str: return path return f"{self.config.workdir.rstrip('/')}/{path}" - async def read(self, path: str) -> bytes: + async def _read(self, path: str) -> bytes: try: return await self._sandbox.filesystem.read_bytes.aio(self._abs(path)) except Exception as e: diff --git a/verifiers/v1/runtimes/prime.py b/verifiers/v1/runtimes/prime.py index b8de3386b7..5e7e7ee4d4 100644 --- a/verifiers/v1/runtimes/prime.py +++ b/verifiers/v1/runtimes/prime.py @@ -274,19 +274,47 @@ async def run_background( f"prime background launch failed: {result.stderr.strip()}" ) - async def read(self, path: str) -> bytes: - # Avoid background-job log limits and base64 overhead by downloading binary data directly. - # The temporary file is removed on every exit, and its byte read stays off the event loop. + async def read(self, path: str, max_bytes: int | None = None) -> bytes: + if max_bytes is not None and max_bytes < 0: + raise ValueError("max_bytes must be non-negative") target = ( path if path.startswith("/") else f"{self.config.workdir.rstrip('/')}/{path}" ) try: - with tempfile.TemporaryDirectory() as directory: - download = Path(directory) / "download" - await self._client.download_file(self.info.id, target, str(download)) - return await asyncio.to_thread(download.read_bytes) + if max_bytes is None: + # Avoid background-job log limits and base64 overhead. + with tempfile.TemporaryDirectory() as directory: + download = Path(directory) / "download" + await self._client.download_file( + self.info.id, target, str(download) + ) + return await asyncio.to_thread(download.read_bytes) + # Stream bounded reads because the SDK buffers the complete response. + 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" + ) + data = bytearray() + async with self._client._get_gateway_client().stream( + "GET", + url, + headers={"Authorization": f"Bearer {auth['token']}"}, + params={"path": target, "sandbox_id": self.info.id}, + timeout=300, + ) as response: + response.raise_for_status() + async for chunk in response.aiter_bytes(1024 * 1024): + data.extend(chunk) + if len(data) > max_bytes: + raise SandboxError( + f"read {path!r}: exceeds the {max_bytes} byte limit" + ) + return bytes(data) + except SandboxError: + raise except Exception as e: raise SandboxError(f"read {path!r}: {e}") from e @@ -312,6 +340,18 @@ async def write(self, path: str, data: bytes) -> None: except Exception as e: raise SandboxError(f"write {path!r}: {e}") from e + async def teardown_confirmed(self) -> None: + runtime_id = self.info.id + client = self._client + if client is None or runtime_id is None: + raise RuntimeError( + "prime sandbox deletion cannot be confirmed without its client and ID" + ) + await client.delete(runtime_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/runtimes/subprocess.py b/verifiers/v1/runtimes/subprocess.py index e4a74deae3..e1a470211c 100644 --- a/verifiers/v1/runtimes/subprocess.py +++ b/verifiers/v1/runtimes/subprocess.py @@ -96,7 +96,7 @@ async def run_background( proc ) # killed in stop() — a host process won't die on its own - async def read(self, path: str) -> bytes: + async def _read(self, path: str) -> bytes: return await asyncio.to_thread((self.workdir / path).read_bytes) async def write(self, path: str, data: bytes) -> None: diff --git a/verifiers/v1/task.py b/verifiers/v1/task.py index f3c819741a..659680b644 100644 --- a/verifiers/v1/task.py +++ b/verifiers/v1/task.py @@ -41,9 +41,11 @@ from verifiers.v1.utils.generic import generic_type if TYPE_CHECKING: + from contextlib import AbstractAsyncContextManager + from verifiers.v1.judge import Judge from verifiers.v1.mcp import Toolset - from verifiers.v1.runtimes import Runtime + from verifiers.v1.runtimes import Runtime, RuntimeConfig from verifiers.v1.trace import Trace logger = logging.getLogger(__name__) @@ -223,6 +225,12 @@ async def finalize(self, trace: Trace, runtime: Runtime) -> None: async def validate(self, runtime: Runtime) -> bool: return True + def scoring_runtime( + self, runtime: Runtime, runtime_policy: RuntimeConfig + ) -> AbstractAsyncContextManager[Runtime] | None: + """Optionally replace the runtime passed to task scoring.""" + return None + async def score( self, trace: Trace, diff --git a/verifiers/v1/tasksets/harbor/taskset.py b/verifiers/v1/tasksets/harbor/taskset.py index e295ed8f64..84f28df644 100644 --- a/verifiers/v1/tasksets/harbor/taskset.py +++ b/verifiers/v1/tasksets/harbor/taskset.py @@ -1,39 +1,68 @@ """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 shutil import subprocess import sys import tarfile import tempfile import tomllib -from collections.abc import Iterator +from collections.abc import AsyncIterator, Iterator +from contextlib import AbstractAsyncContextManager, asynccontextmanager from functools import lru_cache -from pathlib import Path +from pathlib import Path, PurePosixPath +from typing import TYPE_CHECKING, Any 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, + provision_runtime, +) from verifiers.v1.task import Task, TaskData, TaskResources, TaskTimeout from verifiers.v1.configs.taskset import TasksetConfig from verifiers.v1.taskset import Taskset 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" +MAX_ARTIFACT_BYTES = 256 * 1024 * 1024 +MAX_ARTIFACT_ARCHIVE_BYTES = 2 * MAX_ARTIFACT_BYTES +MAX_ARTIFACT_FILES = 10_000 +MAX_ARTIFACT_MANIFEST_BYTES = 4 * 1024 * 1024 +MAX_REWARD_BYTES = 1024 * 1024 +SYSTEM_ARTIFACT_ROOTS = { + "/", + "/bin", + "/etc", + "/lib", + "/lib64", + "/sbin", + "/tmp", + "/usr", + "/var", +} + +if TYPE_CHECKING: + from harbor.models.task.config import ( + EnvironmentConfig, + TaskConfig as HarborTaskConfig, + ) class HarborConfig(TasksetConfig): @@ -76,6 +105,19 @@ class Author(StrictBaseModel): email: str | None = None +class Artifact(StrictBaseModel): + source: str + exclude: list[str] = [] + + +class VerifierConfig(StrictBaseModel): + image: str | None = None + resources: TaskResources = TaskResources() + workdir: str | None = None + fresh_copy: bool = False + network_access: bool = True + + class HarborData(TaskData): """Parsed ``task.toml`` metadata plus the host-side verifier directory. @@ -89,38 +131,281 @@ class HarborData(TaskData): category: str | None = None tags: list[str] = [] task_dir: str = "" - """Host path to the task dir; used to stage tests/ to verify.""" 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.""" + verifier: VerifierConfig | None = None + artifacts: list[Artifact] = [] class HarborTask(Task[HarborData]): - """Stage and run Harbor's verifier inside the task's live runtime.""" - - @reward(weight=1.0) - async def solved(self, runtime: Runtime) -> float: + def scoring_runtime( + self, runtime: Runtime, runtime_policy: RuntimeConfig + ) -> AbstractAsyncContextManager[Runtime] | None: + verifier_data = self.data.verifier + if verifier_data is None: + return None + config = verifier_runtime_config( + runtime, + runtime_policy, + verifier_data, + ) + return self._separate_scoring_runtime(runtime, config) + + @asynccontextmanager + async def _separate_scoring_runtime( + self, runtime: Runtime, config: RuntimeConfig + ) -> AsyncIterator[Runtime]: + artifacts = await self._collect_artifacts(runtime) + # No verifier process starts until the agent runtime is confirmed gone. + await runtime.stop_confirmed() + async with provision_runtime( + config, name=f"{runtime.name}-verifier" + ) as verifier: + await verifier.prepare_setup() + await self._prepare_verifier(verifier, artifacts) + await verifier.prepare_execution([]) + yield verifier + + async def _prepare_verifier( + self, + runtime: Runtime, + artifacts: tuple[bytes, list[str]] | None = None, + ) -> None: await runtime.write( "/tmp/tests.tgz", make_tar(Path(self.data.task_dir) / "tests") ) - await runtime.run( + await run_checked( + runtime, [ "sh", "-c", - "mkdir -p /logs/verifier /tests && tar -xzf /tmp/tests.tgz -C /tests", + "rm -rf /logs/verifier /tests && " + "mkdir -p /logs/verifier /tests && " + "tar -xzf /tmp/tests.tgz -C /tests", ], {}, + "Harbor verifier setup", + ) + if artifacts is None: + return + archive, roots = artifacts + archive_path = "/logs/verifier/artifacts.tar" + await runtime.write(archive_path, archive) + await run_checked( + runtime, + [ + "sh", + "-c", + "set -e; archive=$1; workdir=$2; shift 2; " + 'for path do rm -rf -- "$path"; done; ' + 'mkdir -p -- "$workdir"; ' + 'tar -xf "$archive" -C /; rm -f -- "$archive"', + "sh", + archive_path, + runtime.config.workdir, + *roots, + ], + {}, + "Harbor artifact restore", + ) + + @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 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) ) + if self.data.verifier is not None: + try: + data = json.loads( + await runtime.read("/logs/verifier/reward.json", MAX_REWARD_BYTES) + ) + if type(data) in (int, float): + return float(data) + if ( + data + and isinstance(data, dict) + and all(type(value) in (int, float) for value in data.values()) + ): + rewards = {key: float(value) for key, value in data.items()} + return rewards.get("reward", rewards) + except (SandboxError, OSError, TypeError, ValueError): + pass try: - reward = (await runtime.read("/logs/verifier/reward.txt")).decode().strip() - return float(reward or 0) + data = await runtime.read("/logs/verifier/reward.txt", MAX_REWARD_BYTES) + return float(data.decode().strip() or 0) except (SandboxError, OSError, ValueError): return 0.0 + async def _collect_artifacts(self, runtime: Runtime) -> tuple[bytes, list[str]]: + """Run the artifact hook, then create and validate one bounded archive.""" + await run_checked( + runtime, + ["sh", "-c", "rm -rf /logs/verifier && mkdir -p /logs/verifier"], + {}, + "Harbor artifact staging setup", + ) + script = Path(self.data.task_dir) / "pre_artifacts.sh" + if script.is_file(): + script_path = "/logs/verifier/pre-artifacts.sh" + await runtime.write(script_path, script.read_bytes()) + result = await runtime.run(["bash", script_path], {}) + 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(), + ) + + workdir = PurePosixPath(runtime.config.workdir) + artifacts: list[Artifact] = [] + protected = (PurePosixPath("/tests"), PurePosixPath("/logs/verifier")) + for artifact in self.data.artifacts: + source = PurePosixPath(artifact.source) + if not source.is_absolute(): + source = workdir / source + source = PurePosixPath(f"/{source.as_posix().lstrip('/')}") + source_text = source.as_posix() + if ( + ".." in source.parts + or source_text in SYSTEM_ARTIFACT_ROOTS + or any( + source.is_relative_to(target) or target.is_relative_to(source) + for target in protected + ) + ): + raise ValueError( + f"task {self.data.name!r} has unsafe artifact source " + f"{source_text!r}" + ) + if any( + source.is_relative_to(other := PurePosixPath(entry.source)) + or other.is_relative_to(source) + for entry in artifacts + ): + continue + artifacts.append(artifact.model_copy(update={"source": source_text})) + + archive_path = "/logs/verifier/artifacts.tar" + await runtime.write(archive_path, b"\0" * tarfile.RECORDSIZE) + for artifact in artifacts: + prefix = artifact.source.lstrip("/").replace("\\", "\\\\") + prefix = prefix.replace("&", "\\&").replace("|", "\\|") + result = await runtime.run( + [ + "sh", + "-c", + "source=$1; archive=$2; path=$3; transform=$4; shift 4; " + 'if [ ! -e "$source" ] && [ ! -L "$source" ]; then exit; fi; ' + 'if [ -d "$source" ]; then ' + 'exec tar -rf "$archive" --format=posix --hard-dereference ' + '"$@" "$transform" -C "$source" -- .; fi; ' + 'exec tar -rf "$archive" --format=posix --hard-dereference ' + '-C / -- "$path"', + "sh", + artifact.source, + archive_path, + artifact.source.lstrip("/"), + f"--transform=s|^\\.|{prefix}|", + *(f"--exclude={pattern}" for pattern in artifact.exclude), + ], + {"LC_ALL": "C"}, + ) + if result.exit_code or result.stderr: + detail = (result.stderr or result.stdout).strip() + raise SandboxError( + f"Harbor artifact collection for {artifact.source!r} " + f"failed: {detail}" + ) + data = await runtime.read(archive_path, MAX_ARTIFACT_ARCHIVE_BYTES) + await runtime.run(["rm", "-f", archive_path], {}) + + roots = [artifact.source for artifact in artifacts] + allowed = [PurePosixPath(root.lstrip("/")) for root in roots] + seen: set[str] = set() + total_bytes = manifest_bytes = 0 + try: + with tarfile.open(fileobj=io.BytesIO(data), mode="r:") as archive: + for member in archive: + name = member.name + path = PurePosixPath(name) + if ( + not name + or path.is_absolute() + or ".." in path.parts + or path.as_posix() != name + or not any(path.is_relative_to(root) for root in allowed) + or name in seen + ): + raise SandboxError( + f"Harbor artifact archive contains unsafe path {name!r}" + ) + if not (member.isfile() or member.isdir()): + raise SandboxError( + f"Harbor artifact {name!r} must be a regular file or directory" + ) + seen.add(name) + manifest_bytes += len(name.encode()) + 1 + if member.isfile(): + total_bytes += member.size + if ( + manifest_bytes > MAX_ARTIFACT_MANIFEST_BYTES + or len(seen) > MAX_ARTIFACT_FILES + or total_bytes > MAX_ARTIFACT_BYTES + ): + raise SandboxError("Harbor artifacts exceed transfer limits") + except (tarfile.TarError, UnicodeError) as exc: + raise SandboxError(f"invalid Harbor artifact archive: {exc}") from exc + return data, roots + + +def verifier_runtime_config( + runtime: Runtime, + runtime_policy: RuntimeConfig, + verifier: VerifierConfig, +) -> RuntimeConfig: + """Derive a separate verifier runtime from the agent runtime.""" + config = runtime.config + if not isinstance(config, (DockerConfig, PrimeConfig)) or type( + runtime_policy + ) is not type(config): + raise ValueError( + "separate verification needs matching Docker or Prime runtime policies" + ) + updates: dict[str, Any] = { + "allow": runtime_policy.allow if verifier.network_access else [], + "block": runtime_policy.block if verifier.network_access else [], + } + if not verifier.fresh_copy: + updates.update( + { + field: getattr(runtime_policy, field) + for field in ("workdir", "cpu", "memory", "gpu", "disk") + } + ) + 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 type(config).model_validate({**config.model_dump(), **updates}) + + +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 @@ -243,34 +528,86 @@ 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", harbor_config: HarborConfig +) -> dict[str, Any]: + """Parse Harbor's shared or separate verifier configuration.""" + from harbor.models.task.artifacts import with_convention_entry + 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 {"verifier_env": dict(config.verifier.env)} + if config.verifier.collect or config.verifier.user is not None: + raise ValueError(f"{task_dir.name}: unsupported separate verifier fields") + + environment = resolve_effective_verifier_env_config(config, None) + if environment is None: + raise ValueError(f"{task_dir.name}: separate verifier environment is missing") + explicit_environment = config.verifier.environment is not None + if ( + explicit_environment + and environment.docker_image is None + and not harbor_config.ignore_dockerfile + ): raise ValueError( - f"invalid Harbor size {size!r}: expected a number of MB or a " - "'[G|M|K]' string" + f"{task_dir.name}: explicit verifier environments require a docker_image" + ) + if environment.os != TaskOS.LINUX or any( + getattr(environment, field) + for field in ("healthcheck", "mcp_servers", "skills_dir", "gpu_types", "tpu") + ): + raise ValueError(f"{task_dir.name}: unsupported separate verifier environment") + + 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 aren't supported" ) - return float(size.strip()[:-1]) * scale - -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, + entries = with_convention_entry( + config.artifacts, convention_source="/logs/artifacts" ) + if any(entry.service not in (None, "main") for entry in entries): + raise ValueError(f"{task_dir.name}: artifact sidecars aren't supported") + artifacts = [ + Artifact(source=entry.source, exclude=entry.exclude or []) for entry in entries + ] + + return { + "artifacts": artifacts, + "verifier": VerifierConfig( + image=environment.docker_image if explicit_environment else None, + resources=( + parse_resources(environment, harbor_config.resource_multiplier) + if explicit_environment + else TaskResources() + ), + workdir=environment.workdir if explicit_environment else None, + fresh_copy=not explicit_environment, + network_access=network_mode == NetworkMode.PUBLIC, + ), + "verifier_env": dict(environment.env) | dict(config.verifier.env), + } def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborData: @@ -279,6 +616,8 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD 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 aren't supported") network = ( parsed.agent.explicit_phase_policy() or parsed.environment.resolve_baseline() ) @@ -290,8 +629,8 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD if harbor_config.ignore_timeouts: harness_timeout = scoring_timeout = None else: - harness_timeout = config.get("agent", {}).get("timeout_sec") - scoring_timeout = config.get("verifier", {}).get("timeout_sec") + harness_timeout = parsed.agent.timeout_sec + scoring_timeout = parsed.verifier.timeout_sec return HarborData( idx=idx, name=task.get("name") or task_dir.name, @@ -303,6 +642,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, network_allow=( ["*"] if network.network_mode == NetworkMode.PUBLIC @@ -317,7 +657,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, @@ -325,7 +665,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, parsed, harbor_config), )