feat(v1): support Harbor's separate verifier environments

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.

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

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

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:

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",

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:

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

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:

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.

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: