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

verifiers/v1/rollout.py

@@ -225,13 +225,20 @@ async def run(self) -> Trace:
             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,
-                )
+                scoring_runtime_config = self.task.scoring_runtime_config(runtime)
+                async with asyncio.timeout(self.scoring_timeout):
+                    if 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=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,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.

verifiers/v1/runtimes/docker.py

@@ -195,6 +195,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

@@ -259,6 +259,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.

verifiers/v1/task.py

@@ -61,7 +61,7 @@
 if TYPE_CHECKING:
     from verifiers.v1.judge import Judge
     from verifiers.v1.mcp import Toolset, User
-    from verifiers.v1.runtimes import Runtime
+    from verifiers.v1.runtimes import Runtime, RuntimeConfig
     from verifiers.v1.trace import Trace
 
 logger = logging.getLogger(__name__)
@@ -266,10 +266,15 @@ async def finalize(self, trace: Trace, runtime: Runtime) -> None:
     async def validate(self, runtime: Runtime) -> bool:
         return True
 
+    def scoring_runtime_config(self, runtime: Runtime) -> RuntimeConfig | None:
+        """Return a separate runtime config when scoring must replace the agent runtime."""
+        return None
+
     async def score(
         self,
         trace: Trace,
         runtime: Runtime | None = None,
+        scoring_runtime_config: RuntimeConfig | None = None,
     ) -> None:
         judges = self.plugged_judges()
         available = {"task": self.data, "trace": trace}

verifiers/v1/tasksets/harbor/__init__.py

@@ -3,6 +3,13 @@
     HarborData,
     HarborTask,
     HarborTaskset,
+    VerifierConfig,
 )
 
-__all__ = ["HarborConfig", "HarborData", "HarborTask", "HarborTaskset"]
+__all__ = [
+    "HarborConfig",
+    "HarborData",
+    "HarborTask",
+    "HarborTaskset",
+    "VerifierConfig",
+]