feat(v1): Harbor separate verifier environments

docs/v1/harbor.md

@@ -105,9 +105,26 @@ Prime VM; Prime accepts host-level entries.
 
 `artifacts = [...]` and `[[verifier.collect]]` are read from `task.toml` ([Harbor Docs](https://www.harborframework.com/docs/run-jobs/results-and-artifacts)). Collect hooks run in the agent's box from the task's `finalize`, which is Harbor's own ordering — after the agent phase, before collection — and declared paths plus the `/logs/artifacts/` convention dir are then carried into the grading box and restored at their original paths ("no translation", as in Harbor).
 
+Two deliberate differences from `harbor run`:
+
+- **A failing collect hook fails the rollout.** Harbor logs it and carries on, because there the output is observability; here it is a grading input, and a silently absent file makes the verifier score a stale state.
+- **`destination` has no effect.** It positions a file in Harbor's host trial directory; verifiers has no trial directory (the trace is the record), and Harbor never lets `destination` affect verifier-side placement.
+
+## Separate verifier environments
+
+`[verifier].environment_mode = "separate"` grades in a second box the agent never touched, instead of the one it worked in ([Harbor Docs](https://www.harborframework.com/docs/tasks/verifier)). Only the declared artifacts and the `/logs/artifacts/` convention directory cross over, and the agent's box is confirmed deleted before the verifier starts, so nothing the agent left running can reach the grader. The score is read from `/logs/verifier/reward.json` — a finite number, or an object of finite numbers where a `reward` key is the scalar and any others are recorded as separate rewards — falling back to `reward.txt`.
+
+Which image the verifier boots from follows Harbor: a declared `[verifier.environment]` if there is one, otherwise a fresh copy of `[environment]`, which is the task's own image.
+
+A declared `[verifier.environment]` needs a pullable `docker_image`. Without one Harbor would build the verifier image from `tests/Dockerfile`, and verifiers never builds images — so build and push it yourself and name the resulting reference, exactly as for `[environment]`. `ignore_dockerfile` grades in the agent's image instead, which means the verifier runs somewhere the task never declared; it warns when it does.
+
+`ignore_separate_verifier = true` forces every task back into shared grading, trading the isolation for one sandbox per task.
+
 ## 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))
+- Switching to a different verifier-phase network policy for a *shared* verifier ([Harbor Docs](https://www.harborframework.com/docs/tasks/network-policy)); a separate verifier's own policy is applied
+- Building a verifier image from `tests/Dockerfile`
+- Sidecar services, and the sidecar artifacts and collect hooks that go with them ([Harbor Docs](https://www.harborframework.com/docs/tasks#sidecar-artifacts-and-collect-hooks))
 - Multi-step tasks ([Harbor Docs](https://www.harborframework.com/docs/tasks/multi-step))

verifiers/v1/agent.py

@@ -31,7 +31,7 @@
     Runtime,
     RuntimeConfig,
     SubprocessConfig,
-    make_runtime,
+    provision_runtime,
     runtime_is_local,
 )
 from verifiers.v1.session import RolloutLimits
@@ -563,14 +563,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/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.v1.clients import ModelContext
 from verifiers.v1.configs.agent import AgentConfig
@@ -139,6 +139,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__,
@@ -235,6 +236,19 @@ async def open(self) -> bool:
                     "task.prompt, or drive the run through agent.interaction() and open "
                     "it with the first turn(message)"
                 )
+            # Resolved before the box boots so an impossible pairing costs nothing.
+            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(
+                    f"task {type(self.task).__name__} scores in its own box, which "
+                    "needs this rollout to own the agent's (a borrowed one is its "
+                    "creator's to tear down) and to have no shared tool servers "
+                    "(they outlive the box)"
+                )
             if self._owns_runtime:
                 await runtime.start()
             await runtime.prepare_setup()
@@ -413,13 +427,25 @@ 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,
-                    )
+                    if self._scoring_runtime is None:
+                        await asyncio.wait_for(
+                            asyncio.gather(
+                                self.task.score(trace, runtime),
+                                self.harness.score(trace, runtime),
+                            ),
+                            self._scoring_timeout,
+                        )
+                    else:
+                        # Serial, not gathered: the harness's metrics describe the
+                        # agent's own session, so they are read off the agent's box
+                        # before the task's context manager tears it down. The task's
+                        # box is provisioned inside the scoring deadline — a grader
+                        # that can't be reached in time is a scoring timeout, not a
+                        # zero.
+                        async with asyncio.timeout(self._scoring_timeout):
+                            await self.harness.score(trace, runtime)
+                            async with self._scoring_runtime as scoring_box:
+                                await self.task.score(trace, scoring_box)
                 trace.timing.scoring.end = time.time()
         except Exception as e:  # noqa: BLE001 - finalize boundary records every rollout failure
             self.fail(e)
@@ -437,7 +463,9 @@ async def close(self) -> Trace:
                 if span.start and not span.end:
                     span.end = now
             trace.split_agent_time()
-            if runtime is not None:
+            # A task that scores in its own box tears the agent's down to get there, so
+            # there is nothing left to clean up in it.
+            if runtime is not None and not runtime.stopped:
                 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,22 @@ 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]:
+    """Provision a box from `config` and tear it down on exit.
+
+    `start()` sits inside the `try`: a failed start may already hold a paid sandbox, so
+    it has to reach `stop()` (which is safe on a partially-started 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."""
@@ -71,5 +89,6 @@ def runtime_is_local(config: RuntimeConfig) -> bool:
     "SubprocessRuntime",
     "SubprocessRuntimeInfo",
     "make_runtime",
+    "provision_runtime",
     "runtime_is_local",
 ]

verifiers/v1/runtimes/base.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import atexit
+import base64
 import contextlib
 import hashlib
 import logging
@@ -180,6 +181,25 @@ 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 provisioned resource, raising unless its removal is confirmed.
+
+        `teardown` is best-effort: a resource that outlives us is a billing problem, and
+        failing the rollout over one would be worse than leaking it. This is the variant
+        for a caller that needs the box *gone* before it does the next thing — an agent
+        can leave a background process running past its final turn, so a grading box
+        provisioned while the agent's box is still up is not isolated from it."""
+        self.stopped = True  # before the await: no new borrows once teardown begins
+        await run_shielded(self.teardown_confirmed())
+
+    async def teardown_confirmed(self) -> None:
+        """Free the provisioned resource and verify it is gone, or raise. Unlike
+        `teardown` this is neither best-effort nor optional to implement: a runtime that
+        cannot prove removal cannot host the agent side of an isolated grading run."""
+        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.
@@ -289,6 +309,30 @@ async def run_uv_script(
     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 `path`, raising past the cap.
+
+        Truncating inside the box rather than after the transfer: `read` pulls the whole
+        file into host memory, and a scoring input written by something we don't control
+        has no size we can assume. Base64 because `run` returns decoded text."""
+        result = await self.run(
+            [
+                "sh",
+                "-c",
+                'head -c "$1" -- "$2" | base64',
+                "sh",
+                str(max_bytes + 1),
+                path,
+            ],
+            {},
+        )
+        if result.exit_code:
+            raise SandboxError(f"read {path!r}: {result.stderr.strip()[-500:]}")
+        data = base64.b64decode(result.stdout)
+        if len(data) > max_bytes:
+            raise SandboxError(f"read {path!r}: over the {max_bytes} byte limit")
+        return data
+
     @abstractmethod
     async def write(self, path: str, data: bytes) -> None:
         pass

verifiers/v1/runtimes/docker/__init__.py

@@ -306,6 +306,35 @@ async def teardown(self) -> None:
             await self._proxy.stop()
         await super().teardown()
 
+    async def teardown_confirmed(self) -> None:
+        proxy_error: Exception | None = None
+        if self._proxy is not None:
+            try:
+                await self._proxy.stop()
+            except Exception as exc:  # noqa: BLE001 - remove and confirm before surfacing
+                proxy_error = exc
+        # Do not call `self.teardown()`: a proxy failure there would skip container
+        # removal, but confirmed teardown's caller needs proof the agent box is gone.
+        await super().teardown()
+        if self._container is None:
+            if proxy_error is not None:
+                raise proxy_error
+            return
+        # `docker rm --force` is suppressed in `cleanup`, so ask the daemon rather than
+        # trusting it: an empty listing is the only proof the container is really gone.
+        check = await docker(
+            "ps", "--all", "--quiet", "--filter", f"name=^{self._container}$"
+        )
+        if check.exit_code:
+            raise SandboxError(
+                f"docker: could not confirm {self._container} was removed: "
+                f"{check.stderr.strip()[-500:]}"
+            )
+        if check.stdout.strip():
+            raise SandboxError(f"docker: container {self._container} is still present")
+        if proxy_error is not None:
+            raise proxy_error
+
     async def run(self, argv: list[str], env: dict[str, str]) -> ProgramResult:
         env = {**env, **(self._proxy_env() if self._cut else {})}
         env_args = [arg for k, v in env.items() for arg in ("--env", f"{k}={v}")]

verifiers/v1/runtimes/prime.py

@@ -334,3 +334,20 @@ async def teardown(self) -> None:
                 )
         with contextlib.suppress(Exception):
             await client.aclose()
+
+    async def teardown_confirmed(self) -> None:
+        client, self._client = (
+            self._client,
+            None,
+        )  # same idempotency guard as `teardown`
+        if client is None or self.info.id is None:
+            return
+        try:
+            await client.delete(self.info.id)
+        except Exception as e:
+            raise SandboxError(
+                f"prime: failed to delete sandbox {self.info.id}: {e}"
+            ) from e
+        finally:
+            with contextlib.suppress(Exception):
+                await client.aclose()

verifiers/v1/task.py

@@ -28,9 +28,11 @@
 from verifiers.v1.utils.generic import concrete_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__)
@@ -171,6 +173,27 @@ async def finalize(self, trace: Trace, runtime: Runtime) -> None:
     async def validate(self, runtime: Runtime) -> bool:
         return True
 
+    def scoring_runtime(
+        self, runtime: Runtime, base_runtime_config: RuntimeConfig
+    ) -> AbstractAsyncContextManager[Runtime] | None:
+        """A box to score this task in, instead of the one the agent worked in.
+
+        `None` (the default) keeps scoring where it has always happened: the agent's own
+        runtime, alive and exactly as the agent left it. Return a context manager and
+        the rollout scores the task inside it instead — for a grader the agent must not
+        be able to reach, since anything living in a box the agent had root in is
+        reachable, and hiding individual files only moves the target.
+
+        Whatever has to survive the move is the task's problem: carry it with
+        `collect`/`restore` while the agent's box is still up, inside the context
+        manager. `base_runtime_config` is the resolved run-level config, before the
+        current task's overrides, so the new box can distinguish task-declared resources
+        from CLI overrides; `runtime.config` alone cannot.
+
+        Only `score` moves. `finalize` and the harness's own scoring stay on `runtime`,
+        which is what they describe."""
+        return None
+
     async def score(
         self,
         trace: Trace,