Support separate Harbor verifier runtimes

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))

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)

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)

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

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

verifiers/v1/runtimes/prime.py

@@ -8,13 +8,15 @@
 
 import asyncio
 import contextlib
+import io
 import logging
 import math
 import shlex
 import tempfile
 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.