Support separate Harbor verifier runtimes

docs/v1/harbor.md

@@ -65,9 +65,23 @@ 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"
+```
+
+Without `[verifier.environment]`, the clean runtime reuses the resolved agent image and Verifiers uploads the trusted `tests/` directory. Tasks with `tests/Dockerfile` must instead publish a verifier image and set `[verifier.environment].docker_image`; Verifiers fails closed because it does not build Harbor Dockerfiles. The prebuilt image must contain `/tests/test.sh`. The verifier environment can independently declare its image, workdir, resources, environment variables, and `public` or `no-network` network access.
+
+Only filesystem artifacts can cross this boundary. Artifact sources must be absolute paths, and their combined transfer is limited to 1 GiB uncompressed, 256 MiB compressed, and 100,000 entries. Runtime images must use a default user that can create the verifier-owned `/logs` and `/tests` paths 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))
+- 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)
@@ -311,6 +314,9 @@ def episode(self, task: Task, ctx: ModelContext, n: int = 1) -> Episode:
                 f"and need >=2; got n={n} (pass -r/--num-rollouts >= 2)"
             )
         runtime_config = self.runtime_for(task)
+        scoring_runtime_config = task.scoring_runtime_config(
+            self.harness.config.runtime
+        )
         setup_timeout = (
             self.setup_timeout
             if self.setup_timeout is not None
@@ -351,6 +357,7 @@ def episode(self, task: Task, ctx: ModelContext, n: int = 1) -> Episode:
                 harness=self.harness,
                 ctx=ctx,
                 runtime_config=runtime_config,
+                scoring_runtime_config=scoring_runtime_config,
                 setup_timeout=setup_timeout,
                 harness_timeout=harness_timeout,
                 finalize_timeout=finalize_timeout,

verifiers/v1/rollout.py

@@ -52,6 +52,7 @@ def __init__(
         harness: Harness,
         ctx: ModelContext,
         runtime_config: RuntimeConfig,
+        scoring_runtime_config: RuntimeConfig | None = None,
         setup_timeout: float | None = None,
         harness_timeout: float | None = None,
         finalize_timeout: float | None = None,
@@ -64,6 +65,7 @@ def __init__(
         self.harness = harness
         self.ctx = ctx
         self.runtime_config = runtime_config
+        self.scoring_runtime_config = scoring_runtime_config
         self.setup_timeout = setup_timeout
         self.harness_timeout = harness_timeout
         self.finalize_timeout = finalize_timeout
@@ -216,14 +218,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

@@ -128,6 +128,19 @@ async def stop(self) -> None:
         `teardown`, not this."""
         await run_shielded(self.teardown())
 
+    async def stop_confirmed(self) -> None:
+        """Stop the resource and fail unless its provider confirms it can no longer run.
+
+        Clean-slate runtime transitions use this stronger boundary. Normal rollout
+        cleanup remains best-effort through ``stop``.
+        """
+        await run_shielded(self.teardown_confirmed())
+
+    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.
@@ -223,8 +236,14 @@ 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) -> None:
+        if max_bytes is not None and max_bytes < 0:
+            raise ValueError("max_bytes must be non-negative")
+
     @abstractmethod
-    async def read(self, path: str) -> bytes:
+    async def read(self, path: str, max_bytes: int | None = None) -> bytes:
+        """Read a file, raising ``SandboxError`` before returning more than a
+        non-negative ``max_bytes`` limit."""
         pass
 
     @abstractmethod

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,
@@ -154,7 +157,8 @@ 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:
+        self._validate_read_limit(max_bytes)
         proc = await asyncio.create_subprocess_exec(
             "docker",
             "exec",
@@ -164,14 +168,38 @@ async def read(self, path: str) -> bytes:
             "cat",
             path,
             stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
+            stderr=(
+                asyncio.subprocess.PIPE
+                if max_bytes is None
+                else asyncio.subprocess.DEVNULL
+            ),
         )
-        stdout, stderr = await proc.communicate()
-        if proc.returncode != 0:
-            raise SandboxError(
-                f"read {path!r}: {stderr.decode(errors='replace').strip()}"
-            )
-        return stdout
+        try:
+            if max_bytes is None:
+                stdout, stderr = await proc.communicate()
+            else:
+                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()
+                stderr = b""
+            if proc.returncode != 0:
+                detail = stderr.decode(errors="replace").strip()
+                raise SandboxError(f"read {path!r}: {detail or '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))
@@ -195,6 +223,23 @@ 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 None:
+            return
+        removed = await docker("rm", "--force", self._container)
+        if removed.exit_code:
+            inspected = await docker("container", "inspect", self._container)
+            missing = any(
+                text in inspected.stderr.lower()
+                for text in ("no such container", "no such object")
+            )
+            if inspected.exit_code == 0 or not missing:
+                detail = (removed.stderr or inspected.stderr).strip()
+                raise SandboxError(
+                    f"docker could not confirm removal of {self._container!r}: {detail}"
+                )
+        self._stopped = True
+
     def cleanup(self) -> None:
         if self._container is None or self._stopped:
             return

verifiers/v1/runtimes/modal.py

@@ -9,6 +9,7 @@
 
 import asyncio
 import contextlib
+import io
 import logging
 import shlex
 from pathlib import PurePosixPath
@@ -165,9 +166,31 @@ 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, max_bytes: int | None = None) -> bytes:
+        self._validate_read_limit(max_bytes)
         try:
-            return await self._sandbox.filesystem.read_bytes.aio(self._abs(path))
+            target = self._abs(path)
+            if max_bytes is None:
+                return await self._sandbox.filesystem.read_bytes.aio(target)
+            from modal.stream_type import StreamType
+
+            proc = await self._sandbox.exec.aio(
+                "cat", "--", target, text=False, stderr=StreamType.DEVNULL
+            )
+            buffer = io.BytesIO()
+            async for chunk in proc.stdout:
+                if buffer.tell() + len(chunk) > max_bytes:
+                    await proc.stdout.aclose()
+                    raise SandboxError(
+                        f"read {path!r}: exceeds the {max_bytes} byte limit"
+                    )
+                buffer.write(chunk)
+            await proc.wait.aio()
+            if proc.returncode:
+                raise SandboxError(f"read {path!r}: modal exec failed")
+            return buffer.getvalue()
+        except SandboxError:
+            raise
         except Exception as e:
             raise SandboxError(f"read {path!r}: {e}") from e
 
@@ -182,6 +205,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:
+        sandbox = self._sandbox
+        if sandbox is None:
+            return
+        try:
+            await sandbox.terminate.aio(wait=True)
+        except Exception as e:
+            raise SandboxError(
+                f"modal could not confirm teardown of {self.info.id}: {e}"
+            ) from e
+        self._sandbox = None
+
     def cleanup(self) -> None:
         # Synchronous atexit backstop (the async API can't run once the loop is gone): terminate
         # the sandbox via Modal's sync API so the costly resource isn't left to its max-lifetime.