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)). The harbor env — this taskset's default — grades such tasks in `finalize`: the solver plays the task as usual, its declared artifacts and the `/logs/artifacts/` convention directory are collected while its box is alive, the box is torn down, and the env then provisions a fresh box, restores those artifacts, stages `tests/` fresh, and grades there, recording the verifier's rewards and metrics onto the solver's trace. The grading box derives from the solver's runtime policy unless `--env.verifier-runtime.*` names its own (a network-restricted verifier on Prime needs `vm true`); infrastructure failures around it retry per `--env.verifier-retries` before the episode fails — a grading box that can't be reached never reads as reward 0. The score is read from `/logs/verifier/reward.json` — a finite number, or an object of finite numbers: with a `reward` key that key is the score and the rest are recorded as metrics; without one every key is recorded as a separate reward. Missing or invalid, it falls 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.
+
+Under any other env, a separate-verifier task refuses to grade in the agent's box rather than silently losing its isolation. `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`, which Harbor does when a declared `[verifier.environment]` names no `docker_image`. A separate verifier image itself is supported — it just has to be pre-built and pullable (see above), because verifiers never builds images
+- 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
@@ -561,14 +561,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/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
@@ -285,9 +286,42 @@ async def run_uv_script(
         argv = await self.prepare_uv_script(script, env)
         return await self.run([*argv, *(args or [])], env or {})
 
+    async def read(self, path: str, max_bytes: int | None = None) -> bytes:
+        """Read `path` into host memory. `max_bytes` caps the transfer, raising past
+        the cap — for a file written by something we don't control, whose size we
+        can't assume. The cap is enforced inside the box rather than after the
+        transfer, and base64 because `run` returns decoded text. Framework method —
+        override `_read`, not this."""
+        if max_bytes is None:
+            return await self._read(path)
+        # Through a temp file, not a pipe: `head | base64` exits with base64's 0
+        # even when the path is missing, and a missing file must raise here just
+        # as it does from `_read`.
+        result = await self.run(
+            [
+                "sh",
+                "-c",
+                (
+                    "t=$(mktemp) || exit 1; "
+                    'head -c "$1" -- "$2" > "$t" || { rm -f "$t"; exit 1; }; '
+                    'base64 < "$t"; rc=$?; rm -f "$t"; exit $rc'
+                ),
+                "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 read(self, path: str) -> bytes:
-        pass
+    async def _read(self, path: str) -> bytes:
+        """Read the whole file at `path`; `read` adds the optional transfer cap."""
 
     @abstractmethod
     async def write(self, path: str, data: bytes) -> None:

verifiers/v1/runtimes/docker/__init__.py

@@ -334,7 +334,7 @@ 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) -> bytes:
         proc = await asyncio.create_subprocess_exec(
             "docker",
             "exec",

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

@@ -270,7 +270,7 @@ async def run_background(
                 f"prime background launch failed: {result.stderr.strip()}"
             )
 
-    async def read(self, path: str) -> bytes:
+    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.
         target = (

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:

verifiers/v1/tasksets/harbor/__init__.py

@@ -1,8 +1,16 @@
+from verifiers.v1.tasksets.harbor.env import HarborEnv, HarborEnvConfig
 from verifiers.v1.tasksets.harbor.taskset import (
     HarborConfig,
     HarborData,
     HarborTask,
     HarborTaskset,
 )
 
-__all__ = ["HarborConfig", "HarborData", "HarborTask", "HarborTaskset"]
+__all__ = [
+    "HarborConfig",
+    "HarborData",
+    "HarborEnv",
+    "HarborEnvConfig",
+    "HarborTask",
+    "HarborTaskset",
+]

verifiers/v1/tasksets/harbor/env.py

@@ -0,0 +1,124 @@
+"""The harbor taskset's own env: the single solver seat, plus separate-verifier
+grading for tasks that declare ``[verifier].environment_mode = "separate"``.
+
+The default env for harbor runs (the taskset package exports it). A shared-verifier
+task runs exactly as under the single-agent env: one `agent` trace, graded in the
+box it worked in. A separate-verifier task is graded by `finalize` instead: the
+solver's declared artifacts travel (collected by its task `finalize` while its box
+is alive), a fresh box is provisioned from the task's verifier declaration,
+`tests/` is staged there, and the verifier's rewards land on the solver's trace.
+No second agent is involved — the verifier is the task's own `tests/test.sh`.
+"""
+
+import asyncio
+import logging
+from contextlib import AsyncExitStack
+
+from pydantic import Field
+
+import verifiers.v1 as vf
+from verifiers.v1.runtimes import RuntimeConfig, provision_runtime
+from verifiers.v1.tasksets.harbor.taskset import (
+    HarborTask,
+    verifier_box_data,
+)
+from verifiers.v1.utils.artifacts import restore
+from verifiers.v1.utils.compile import resolve_runtime_config
+from verifiers.v1.utils.retries import backoff
+
+logger = logging.getLogger(__name__)
+
+
+class HarborEnvConfig(vf.EnvConfig):
+    agent: vf.AgentConfig = vf.AgentConfig()
+    """The one seat — the policy under evaluation/training; pin
+    `--env.agent.harness.*` to choose its program or runtime."""
+    verifier_runtime: RuntimeConfig | None = None
+    """Where a separate-verifier task grades. None derives the grading box from
+    the solver's runtime policy; set it (e.g. `--env.verifier-runtime.type prime
+    --env.verifier-runtime.vm true`) when the verifier needs different placement
+    than the agent."""
+    verifier_retries: int = Field(2, ge=0)
+    """Extra attempts at provisioning-and-grading the separate box before the
+    episode fails. Grading is deterministic; what these retry is the
+    infrastructure around it (image pulls, provisioning)."""
+
+
+class HarborEnv(vf.Env[HarborEnvConfig]):
+    async def run(self, task: vf.Task, agents: vf.Agents) -> None:
+        if not isinstance(task, HarborTask):
+            raise TypeError(
+                f"the harbor env runs harbor tasks; got {type(task).__name__}"
+            )
+        if task.data.verifier is None:
+            await agents.agent.run(task)
+            return
+        # Resolve the verifier's box before the solve, so an impossible pairing
+        # (e.g. a restricted Prime verifier without vm=true) costs nothing
+        # rather than a full agent run.
+        self._verifier_config(task)
+        await agents.agent.run(task.graded_elsewhere())
+
+    def _verifier_config(self, task: HarborTask) -> RuntimeConfig:
+        base = (
+            self.config.verifier_runtime
+            if self.config.verifier_runtime is not None
+            else self.config.agent.runtime
+        )
+        return resolve_runtime_config(base, HarborTask(verifier_box_data(task.data)))
+
+    async def finalize(self, task: vf.Task, episode: vf.Episode) -> None:
+        """Grade a separate-verifier task in its own box, onto the solver's trace.
+
+        Provision a fresh box from the task's verifier declaration, restore the
+        solver's collected artifacts, stage `tests/`, run the verifier, and record
+        its rewards (and any extra reward.json keys as metrics) on the solver's
+        trace. Infrastructure failures retry per `verifier_retries`; the last one
+        fails the episode — a grading box that can't be reached must never read
+        as reward 0."""
+        if not isinstance(task, HarborTask) or task.data.verifier is None:
+            return
+        solution = episode.traces[0]
+        if not solution.ok:
+            return
+        grader = HarborTask(verifier_box_data(task.data))
+        scores = await self._grade(self._verifier_config(task), grader, solution)
+        items = scores.items() if isinstance(scores, dict) else [("solved", scores)]
+        for name, value in items:
+            solution.record_reward(name, value)
+
+    async def _grade(
+        self, config: RuntimeConfig, grader: HarborTask, solution: vf.Trace
+    ) -> float | dict[str, float]:
+        last: Exception | None = None
+        for attempt in range(self.config.verifier_retries + 1):
+            if attempt:
+                delay = backoff(attempt - 1)
+                logger.warning(
+                    "harbor verifier attempt %d/%d failed (%s); retrying in %.1fs",
+                    attempt,
+                    self.config.verifier_retries + 1,
+                    last,
+                    delay,
+                )
+                await asyncio.sleep(delay)
+            try:
+                # The scoring deadline covers provisioning and grading, but not the
+                # box's teardown: a score already in hand must not be discarded
+                # because the teardown ran out the clock.
+                async with AsyncExitStack() as boxes:
+                    async with asyncio.timeout(grader.data.timeout.scoring):
+                        box = await boxes.enter_async_context(provision_runtime(config))
+                        await box.prepare_setup()
+                        # Artifacts first, tests second: an artifact entry pointing
+                        # into /tests must not survive staging, which wipes and
+                        # rebuilds that directory.
+                        await restore(box, solution.state.artifacts)
+                        await grader._stage_tests(box, wipe=True)
+                        await box.prepare_execution([])
+                        scores = await grader._graded(box, solution)
+                    return scores
+            except Exception as e:  # noqa: BLE001 - each attempt's failure is retried
+                last = e
+        assert last is not None
+        raise last