From bd71697dd7b1dd11fb9e82a830d871a6bc38e0aa Mon Sep 17 00:00:00 2001
From: Xeophon <46377542+xeophon@users.noreply.github.com>
Date: Thu, 16 Jul 2026 17:38:17 +0200
Subject: [PATCH 1/6] Support separate Harbor verifier runtimes

---
 docs/v1/harbor.md                       |  18 +-
 verifiers/v1/env.py                     |  23 +-
 verifiers/v1/rollout.py                 |  26 +-
 verifiers/v1/runtimes/base.py           |  21 +-
 verifiers/v1/runtimes/docker.py         |  65 ++-
 verifiers/v1/runtimes/modal.py          |  39 +-
 verifiers/v1/runtimes/prime.py          |  74 ++-
 verifiers/v1/runtimes/subprocess.py     |  17 +-
 verifiers/v1/task.py                    |  12 +-
 verifiers/v1/tasksets/harbor/taskset.py | 683 +++++++++++++++++++++++-
 10 files changed, 920 insertions(+), 58 deletions(-)

diff --git a/docs/v1/harbor.md b/docs/v1/harbor.md
index e1207e5311..e8b8e63199 100644
--- a/docs/v1/harbor.md
+++ b/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. With `[verifier.environment]`, its `docker_image` must be a pullable, prebuilt image that already contains `/tests/test.sh`. Verifiers does not build Harbor Dockerfiles: if `tests/Dockerfile` installs verifier-only dependencies, publish that image and configure it explicitly. 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))
diff --git a/verifiers/v1/env.py b/verifiers/v1/env.py
index 29c22ab85a..ddaf736c9d 100644
--- a/verifiers/v1/env.py
+++ b/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,
diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py
index 8a04ef7851..04444d6431 100644
--- a/verifiers/v1/rollout.py
+++ b/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)
diff --git a/verifiers/v1/runtimes/base.py b/verifiers/v1/runtimes/base.py
index b8a31fcddb..a9e75f9182 100644
--- a/verifiers/v1/runtimes/base.py
+++ b/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
diff --git a/verifiers/v1/runtimes/docker.py b/verifiers/v1/runtimes/docker.py
index f99919db87..0c7f873de8 100644
--- a/verifiers/v1/runtimes/docker.py
+++ b/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
diff --git a/verifiers/v1/runtimes/modal.py b/verifiers/v1/runtimes/modal.py
index b2ee7b7a6d..a7c887086d 100644
--- a/verifiers/v1/runtimes/modal.py
+++ b/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.
diff --git a/verifiers/v1/runtimes/prime.py b/verifiers/v1/runtimes/prime.py
index b7afd147a4..7dfa77cb7f 100644
--- a/verifiers/v1/runtimes/prime.py
+++ b/verifiers/v1/runtimes/prime.py
@@ -8,6 +8,7 @@
 
 import asyncio
 import contextlib
+import io
 import logging
 import math
 import shlex
@@ -202,7 +203,8 @@ 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, max_bytes: int | None = None) -> bytes:
+        self._validate_read_limit(max_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 = (
@@ -211,10 +213,40 @@ async def read(self, path: str) -> bytes:
             else f"{self.config.workdir.rstrip('/')}/{path}"
         )
         try:
+            if max_bytes is not None:
+                # The SDK's download_file buffers response.content. Use its gateway
+                # connection directly so an untrusted file is capped while streaming.
+                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"
+                )
+                headers = {"Authorization": f"Bearer {auth['token']}"}
+                params = {"path": target, "sandbox_id": self.info.id}
+                buffer = io.BytesIO()
+                client = self._client._get_gateway_client()
+                async with client.stream(
+                    "GET",
+                    url,
+                    headers=headers,
+                    params=params,
+                    timeout=300,
+                ) as response:
+                    response.raise_for_status()
+                    async for chunk in response.aiter_bytes(1024 * 1024):
+                        if buffer.tell() + len(chunk) > max_bytes:
+                            raise SandboxError(
+                                f"read {path!r}: exceeds the {max_bytes} byte limit"
+                            )
+                        buffer.write(chunk)
+                return buffer.getvalue()
+
             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)
+        except SandboxError:
+            raise
         except Exception as e:
             raise SandboxError(f"read {path!r}: {e}") from e
 
@@ -240,6 +272,46 @@ 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:
+        client = self._client
+        if client is None:
+            if self.info.id is None:
+                return
+            raise SandboxError("prime cannot confirm teardown without its client")
+
+        confirmed = False
+        try:
+            await client.delete(self.info.id)
+        except Exception as e:
+            if "HTTP 404" not in str(e):
+                raise SandboxError(
+                    f"prime could not delete sandbox {self.info.id}: {e}"
+                ) from e
+            confirmed = True
+        for _ in range(60):
+            if confirmed:
+                break
+            try:
+                sandbox = await client.get(self.info.id)
+            except Exception as e:
+                if "HTTP 404" in str(e):
+                    confirmed = True
+                    break
+                raise SandboxError(
+                    f"prime could not confirm teardown of {self.info.id}: {e}"
+                ) from e
+            if sandbox.status in {"TERMINATED", "ERROR", "TIMEOUT"}:
+                confirmed = True
+                break
+            await asyncio.sleep(0.5)
+        if not confirmed:
+            raise SandboxError(
+                f"prime sandbox {self.info.id} remained active after deletion"
+            )
+        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.
diff --git a/verifiers/v1/runtimes/subprocess.py b/verifiers/v1/runtimes/subprocess.py
index daf6483621..d1bb21750a 100644
--- a/verifiers/v1/runtimes/subprocess.py
+++ b/verifiers/v1/runtimes/subprocess.py
@@ -10,6 +10,7 @@
 
 from pydantic_config import BaseConfig
 
+from verifiers.v1.errors import SandboxError
 from verifiers.v1.runtimes.base import BaseRuntimeInfo, ProgramResult, Runtime
 
 # Implicit host inheritance removes every name containing "API_KEY" while keeping
@@ -94,8 +95,20 @@ async def run_background(
             proc
         )  # killed in stop() — a host process won't die on its own
 
-    async def read(self, path: str) -> bytes:
-        return await asyncio.to_thread((self.workdir / path).read_bytes)
+    async def read(self, path: str, max_bytes: int | None = None) -> bytes:
+        self._validate_read_limit(max_bytes)
+        target = self.workdir / path
+        if max_bytes is None:
+            return await asyncio.to_thread(target.read_bytes)
+
+        def read_bounded() -> bytes:
+            with target.open("rb") as file:
+                data = file.read(max_bytes + 1)
+            if len(data) > max_bytes:
+                raise SandboxError(f"read {path!r}: exceeds the {max_bytes} byte limit")
+            return data
+
+        return await asyncio.to_thread(read_bounded)
 
     async def write(self, path: str, data: bytes) -> None:
         target = self.workdir / path
diff --git a/verifiers/v1/task.py b/verifiers/v1/task.py
index db955b7bad..4e8c104a24 100644
--- a/verifiers/v1/task.py
+++ b/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,20 @@ async def finalize(self, trace: Trace, runtime: Runtime) -> None:
     async def validate(self, runtime: Runtime) -> bool:
         return True
 
+    def scoring_runtime_config(self, base: RuntimeConfig) -> RuntimeConfig | None:
+        """Return a separate runtime config when this task must score in isolation.
+
+        ``None`` keeps scoring in the agent runtime. A separate scorer owns its
+        runtime transition in ``score``; the rollout first records harness metrics
+        while the agent runtime is still live, then passes this config to ``score``.
+        """
+        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}
diff --git a/verifiers/v1/tasksets/harbor/taskset.py b/verifiers/v1/tasksets/harbor/taskset.py
index 2a0070fa7e..cb288558bd 100644
--- a/verifiers/v1/tasksets/harbor/taskset.py
+++ b/verifiers/v1/tasksets/harbor/taskset.py
@@ -1,8 +1,8 @@
 """Harbor tasksets backed by Harbor Hub packages.
 
-The Harbor CLI downloads and caches each task directory. Its verifier runs in the
-same runtime the harness edited, then writes the score to
-``/logs/verifier/reward.txt``.
+Shared verifiers run in the agent runtime. Separate verifiers collect only Harbor's
+declared artifacts, stop the agent runtime, start a clean verifier runtime, restore the
+artifacts at their original paths, and run the verifier there.
 
 A pullable ``[environment].docker_image`` becomes ``TaskData.image``. Verifiers does
 not build Dockerfile-only environments, so those are rejected unless ``ignore_dockerfile``
@@ -10,29 +10,170 @@
 image unless ``require_image`` is set.
 """
 
+import contextlib
+import gzip
 import hashlib
 import io
+import json
+import math
+import shlex
 import shutil
 import subprocess
 import sys
 import tarfile
 import tempfile
 import tomllib
+import uuid
 from collections.abc import Iterator
 from functools import lru_cache
-from pathlib import Path
+from pathlib import Path, PurePosixPath
+from typing import BinaryIO
 
 from pydantic import Field
 
 from verifiers.v1.decorators import reward
 from verifiers.v1.errors import SandboxError
-from verifiers.v1.runtimes import Runtime
+from verifiers.v1.runtimes import (
+    Runtime,
+    RuntimeConfig,
+    SubprocessConfig,
+    make_runtime,
+)
 from verifiers.v1.task import Task, TaskData, TaskResources, TaskTimeout
 from verifiers.v1.taskset import Taskset, TasksetConfig
+from verifiers.v1.trace import Trace
 from verifiers.v1.types import StrictBaseModel
 
 CACHE = Path.home() / ".cache" / "harbor"
 HARBOR_INSTALL_HINT = "uv sync --python 3.12 --extra harbor"
+MAX_ARTIFACT_ARCHIVE_BYTES = 256 * 1024 * 1024
+MAX_ARTIFACT_CONTENT_BYTES = 1024 * 1024 * 1024
+MAX_ARTIFACT_MEMBERS = 100_000
+MAX_ARTIFACT_TAR_BYTES = MAX_ARTIFACT_CONTENT_BYTES + MAX_ARTIFACT_ARCHIVE_BYTES
+MAX_ARTIFACT_HEADERS = MAX_ARTIFACT_MEMBERS * 2 + 8
+MAX_ARTIFACT_METADATA_BYTES = 1024 * 1024
+MAX_PAX_RECORDS = 4096
+
+
+class _LimitedTarReader:
+    """Keep tarfile from turning one hostile metadata header into a huge allocation."""
+
+    def __init__(self, file: BinaryIO) -> None:
+        self.file = file
+
+    def read(self, size: int = -1) -> bytes:
+        if size < 0 or size > MAX_ARTIFACT_METADATA_BYTES + tarfile.BLOCKSIZE:
+            raise SandboxError("Harbor artifact archive has oversized metadata")
+        return self.file.read(size)
+
+    def seek(self, offset: int, whence: int = io.SEEK_SET) -> int:
+        return self.file.seek(offset, whence)
+
+    def tell(self) -> int:
+        return self.file.tell()
+
+    def __getattr__(self, name: str) -> object:
+        return getattr(self.file, name)
+
+
+def _validate_pax_payload(payload: bytes) -> None:
+    """Reject structural overrides and bound tarfile's per-record allocations."""
+    position = 0
+    records = 0
+    while position < len(payload) and payload[position]:
+        separator = payload.find(b" ", position)
+        if separator < 0:
+            raise SandboxError("Harbor artifact archive has invalid PAX metadata")
+        try:
+            length = int(payload[position:separator])
+        except ValueError as exc:
+            raise SandboxError(
+                "Harbor artifact archive has invalid PAX metadata"
+            ) from exc
+        end = position + length
+        if length < 5 or end > len(payload) or payload[end - 1] != 0x0A:
+            raise SandboxError("Harbor artifact archive has invalid PAX metadata")
+        key, equals, value = payload[separator + 1 : end - 1].partition(b"=")
+        if not key or not equals:
+            raise SandboxError("Harbor artifact archive has invalid PAX metadata")
+        records += 1
+        if records > MAX_PAX_RECORDS:
+            raise SandboxError("Harbor artifact archive has too many PAX records")
+        if key == b"size":
+            raise SandboxError(
+                "Harbor artifact archive uses an unsupported PAX size override"
+            )
+        if (
+            key.startswith(b"GNU.sparse.")
+            or key == b"SCHILY.realsize"
+            or (key == b"SCHILY.filetype" and value == b"sparse")
+        ):
+            raise SandboxError("Harbor artifact archive uses unsupported sparse files")
+        position = end
+
+
+def _validate_tar_archive(file: BinaryIO) -> None:
+    """Bound raw extensions before tarfile recursively interprets them."""
+    file.seek(0, io.SEEK_END)
+    archive_size = file.tell()
+    file.seek(0)
+    headers = 0
+    consecutive_extensions = 0
+    extensions = {
+        tarfile.XHDTYPE,
+        tarfile.XGLTYPE,
+        tarfile.SOLARIS_XHDTYPE,
+        tarfile.GNUTYPE_LONGNAME,
+        tarfile.GNUTYPE_LONGLINK,
+    }
+    while file.tell() < archive_size:
+        header = file.read(tarfile.BLOCKSIZE)
+        if len(header) != tarfile.BLOCKSIZE:
+            raise SandboxError("Harbor artifact archive has a truncated header")
+        if not any(header):
+            break
+        try:
+            member = tarfile.TarInfo.frombuf(header, "utf-8", "surrogateescape")
+        except tarfile.TarError as exc:
+            raise SandboxError("Harbor artifact archive has an invalid header") from exc
+        if member.size < 0:
+            raise SandboxError("Harbor artifact archive has a negative entry size")
+        headers += 1
+        if headers > MAX_ARTIFACT_HEADERS:
+            raise SandboxError("Harbor artifact archive has too many headers")
+        if member.type == tarfile.GNUTYPE_SPARSE:
+            raise SandboxError("Harbor artifact archive uses unsupported sparse files")
+
+        stored_size = (
+            member.size
+            if member.type in extensions
+            or member.isreg()
+            or member.type not in tarfile.SUPPORTED_TYPES
+            else 0
+        )
+        data_end = file.tell() + stored_size
+        padded_end = data_end + (-stored_size % tarfile.BLOCKSIZE)
+        if padded_end > archive_size:
+            raise SandboxError("Harbor artifact archive has a truncated entry")
+        if member.type in extensions:
+            consecutive_extensions += 1
+            if consecutive_extensions > 8:
+                raise SandboxError(
+                    "Harbor artifact archive has too many metadata headers"
+                )
+            if member.size > MAX_ARTIFACT_METADATA_BYTES:
+                raise SandboxError("Harbor artifact archive has oversized metadata")
+            payload = file.read(member.size)
+            if member.type in {
+                tarfile.XHDTYPE,
+                tarfile.XGLTYPE,
+                tarfile.SOLARIS_XHDTYPE,
+            }:
+                _validate_pax_payload(payload)
+        else:
+            consecutive_extensions = 0
+        file.seek(padded_end)
+    file.seek(0)
 
 
 class HarborConfig(TasksetConfig):
@@ -69,6 +210,21 @@ class Author(StrictBaseModel):
     email: str | None = None
 
 
+class HarborArtifact(StrictBaseModel):
+    source: str
+
+
+class HarborVerifier(StrictBaseModel):
+    separate: bool = False
+    image: str | None = None
+    workdir: str | None = None
+    resources: TaskResources = TaskResources()
+    network_access: bool = True
+    env: dict[str, str] = {}
+    artifacts: list[HarborArtifact] = []
+    upload_tests: bool = False
+
+
 class HarborData(TaskData):
     """Parsed ``task.toml`` metadata plus the host-side verifier directory.
 
@@ -83,38 +239,379 @@ class HarborData(TaskData):
     tags: list[str] = []
     task_dir: str = Field("", exclude=True)
     """Host path to the task dir; used to stage tests/ to verify, not serialized."""
-    verifier_env: dict[str, str] = {}
-    """Raw [verifier.env] entries (literals or `${VAR}`/`${VAR:-default}` templates).
-    Resolved against the host environment at scoring time, like `harbor run` — so a
-    verifier that needs judge API keys or configuration actually receives them."""
+    verifier: HarborVerifier = Field(default_factory=HarborVerifier, exclude=True)
+    """Resolved verifier mode and runtime data, kept host-side for scoring."""
 
 
 class HarborTask(Task[HarborData]):
-    """Stage and run Harbor's verifier inside the task's live runtime."""
+    """Run a Harbor verifier in its shared or separate runtime."""
 
-    @reward(weight=1.0)
-    async def solved(self, runtime: Runtime) -> float:
-        await runtime.write(
-            "/tmp/tests.tgz", make_tar(Path(self.data.task_dir) / "tests")
+    def scoring_runtime_config(self, base: RuntimeConfig) -> RuntimeConfig | None:
+        verifier = self.data.verifier
+        if not verifier.separate:
+            return None
+
+        from verifiers.v1.env import resolve_runtime_config
+
+        data = (
+            self.data
+            if verifier.upload_tests
+            else TaskData(
+                idx=self.data.idx,
+                prompt=None,
+                image=verifier.image,
+                workdir=verifier.workdir,
+                resources=verifier.resources,
+            )
         )
-        await runtime.run(
+        config = resolve_runtime_config(base, data)
+        if isinstance(config, SubprocessConfig):
+            raise ValueError(
+                "separate Harbor verification needs a docker, prime, or modal runtime"
+            )
+
+        network = type(config).model_fields.get("network_access")
+        if (
+            network is not None
+            and getattr(config, "network_access") == network.default
+            and verifier.network_access != network.default
+        ):
+            config = config.model_copy(
+                update={"network_access": verifier.network_access}
+            )
+        return config
+
+    async def score(
+        self,
+        trace: Trace,
+        runtime: Runtime | None = None,
+        scoring_runtime_config: RuntimeConfig | None = None,
+    ) -> None:
+        if not self.data.verifier.separate:
+            await super().score(trace, runtime)
+            return
+        if runtime is None:
+            await super().score(trace)
+            return
+        if scoring_runtime_config is None:
+            raise ValueError(
+                "separate Harbor scoring requires a verifier runtime config"
+            )
+
+        artifacts, paths = await self._collect_artifacts(runtime)
+        # The agent runtime must be gone before any verifier process starts.
+        await runtime.stop_confirmed()
+        target = make_runtime(scoring_runtime_config, name=f"{runtime.name}-verifier")
+        try:
+            await target.start()
+            await self._prepare_verifier(
+                target,
+                artifacts,
+                paths,
+                getattr(scoring_runtime_config, "workdir", "/"),
+            )
+            await super().score(trace, target)
+        finally:
+            await target.stop()
+
+    async def _collect_artifacts(self, runtime: Runtime) -> tuple[bytes, list[str]]:
+        buffer = tempfile.SpooledTemporaryFile(max_size=8 * 1024 * 1024)
+        seen: set[str] = set()
+        paths: list[str] = []
+        content_size = 0
+        member_count = 0
+        try:
+            await run_checked(
+                runtime,
+                [
+                    "sh",
+                    "-c",
+                    "rm -rf /logs/verifier && mkdir -p /logs/verifier",
+                ],
+                {},
+                "Harbor artifact staging setup",
+            )
+            with tarfile.open(fileobj=buffer, mode="w:gz") as safe:
+                for artifact in self.data.verifier.artifacts:
+                    exists = await runtime.run(["test", "-e", artifact.source], {})
+                    if exists.exit_code == 1:
+                        continue
+                    if exists.exit_code:
+                        detail = (exists.stderr or exists.stdout).strip()
+                        raise SandboxError(
+                            f"Harbor artifact probe failed for {artifact.source!r}: {detail}"
+                        )
+
+                    archive = (
+                        f"/logs/verifier/.vf-harbor-artifacts-{uuid.uuid4().hex}.tgz"
+                    )
+                    packed = await runtime.run(
+                        [
+                            "tar",
+                            "-chzf",
+                            archive,
+                            "--exclude",
+                            f"./{archive.lstrip('/')}",
+                            "-C",
+                            "/",
+                            f"./{artifact.source.lstrip('/')}",
+                        ],
+                        {},
+                    )
+                    if packed.exit_code:
+                        with contextlib.suppress(Exception):
+                            await runtime.run(["rm", "-f", archive], {})
+                        exists = await runtime.run(["test", "-e", artifact.source], {})
+                        if exists.exit_code == 1:
+                            continue
+                        detail = (packed.stderr or packed.stdout).strip()
+                        raise SandboxError(
+                            f"Harbor artifact collection failed for "
+                            f"{artifact.source!r}: {detail}"
+                        )
+                    try:
+                        unsafe = await runtime.read(
+                            archive, max_bytes=MAX_ARTIFACT_ARCHIVE_BYTES
+                        )
+                    finally:
+                        with contextlib.suppress(Exception):
+                            await runtime.run(["rm", "-f", archive], {})
+
+                    root = PurePosixPath(artifact.source.lstrip("/"))
+                    # The agent controls these bytes. Flatten links and retain only
+                    # regular files/directories rooted under the declared artifact.
+                    with tempfile.SpooledTemporaryFile(
+                        max_size=8 * 1024 * 1024
+                    ) as unpacked:
+                        with gzip.GzipFile(fileobj=io.BytesIO(unsafe)) as compressed:
+                            tar_size = 0
+                            while chunk := compressed.read(1024 * 1024):
+                                tar_size += len(chunk)
+                                if tar_size > MAX_ARTIFACT_TAR_BYTES:
+                                    raise SandboxError(
+                                        "Harbor artifact archive exceeds the "
+                                        "decompression limit"
+                                    )
+                                unpacked.write(chunk)
+                        unpacked.seek(0)
+                        _validate_tar_archive(unpacked)
+                        with tarfile.open(
+                            fileobj=_LimitedTarReader(unpacked), mode="r:"
+                        ) as source:
+                            for member in source:
+                                path = PurePosixPath(member.name)
+                                if (
+                                    member.size < 0
+                                    or path.is_absolute()
+                                    or ".." in path.parts
+                                    or not (path == root or root in path.parents)
+                                    or not (
+                                        member.isfile()
+                                        or member.islnk()
+                                        or member.isdir()
+                                    )
+                                ):
+                                    raise SandboxError(
+                                        "Harbor artifact collection produced unsafe "
+                                        f"entry {member.name!r}"
+                                    )
+                                member_count += 1
+                                content_size += member.size
+                                if member_count > MAX_ARTIFACT_MEMBERS:
+                                    raise SandboxError(
+                                        "Harbor artifacts exceed the "
+                                        f"{MAX_ARTIFACT_MEMBERS:,} entry limit"
+                                    )
+                                if content_size > MAX_ARTIFACT_CONTENT_BYTES:
+                                    raise SandboxError(
+                                        "Harbor artifacts exceed the "
+                                        f"{MAX_ARTIFACT_CONTENT_BYTES // 1024**3} GiB "
+                                        "uncompressed limit"
+                                    )
+                                if path.as_posix() in seen:
+                                    continue
+                                seen.add(path.as_posix())
+                                clean = tarfile.TarInfo(path.as_posix())
+                                clean.mode = member.mode & 0o777
+                                clean.mtime = member.mtime
+                                if member.isdir():
+                                    clean.type = tarfile.DIRTYPE
+                                    safe.addfile(clean)
+                                    continue
+                                file = source.extractfile(member)
+                                if file is None:
+                                    raise SandboxError(
+                                        "Harbor artifact collection could not read "
+                                        f"{member.name!r}"
+                                    )
+                                if member.islnk():
+                                    with tempfile.SpooledTemporaryFile(
+                                        max_size=8 * 1024 * 1024
+                                    ) as linked:
+                                        linked_size = 0
+                                        while chunk := file.read(1024 * 1024):
+                                            linked_size += len(chunk)
+                                            if (
+                                                content_size + linked_size
+                                                > MAX_ARTIFACT_CONTENT_BYTES
+                                            ):
+                                                raise SandboxError(
+                                                    "Harbor artifacts exceed the "
+                                                    f"{MAX_ARTIFACT_CONTENT_BYTES // 1024**3} "
+                                                    "GiB uncompressed limit"
+                                                )
+                                            linked.write(chunk)
+                                        content_size += linked_size
+                                        clean.size = linked_size
+                                        linked.seek(0)
+                                        safe.addfile(clean, linked)
+                                    continue
+                                clean.size = member.size
+                                safe.addfile(clean, file)
+                    paths.append(artifact.source)
+        except (OSError, tarfile.TarError) as exc:
+            buffer.close()
+            raise SandboxError(
+                "Harbor artifact collection produced an invalid archive"
+            ) from exc
+        except BaseException:
+            buffer.close()
+            raise
+        try:
+            buffer.seek(0, io.SEEK_END)
+            if buffer.tell() > MAX_ARTIFACT_ARCHIVE_BYTES:
+                raise SandboxError(
+                    "Harbor artifacts exceed the "
+                    f"{MAX_ARTIFACT_ARCHIVE_BYTES // 1024**2} MiB archive limit"
+                )
+            buffer.seek(0)
+            return buffer.read(), paths
+        finally:
+            buffer.close()
+
+    async def _prepare_verifier(
+        self, runtime: Runtime, artifacts: bytes, paths: list[str], workdir: str
+    ) -> None:
+        await run_checked(
+            runtime,
             [
                 "sh",
                 "-c",
-                "mkdir -p /logs/verifier /tests && tar -xzf /tmp/tests.tgz -C /tests",
+                "rm -rf /logs/verifier && mkdir -p /logs/verifier /logs/artifacts",
             ],
             {},
+            "Harbor verifier setup",
+        )
+        archive = f"/logs/verifier/.vf-harbor-artifacts-{uuid.uuid4().hex}.tgz"
+        if paths:
+            command = "rm -rf " + " ".join(shlex.quote(path) for path in paths)
+            command += f" && mkdir -p {shlex.quote(workdir)}"
+            await run_checked(
+                runtime,
+                ["sh", "-c", command],
+                {},
+                "Harbor artifact target cleanup",
+            )
+        await runtime.write(archive, artifacts)
+        await run_checked(
+            runtime,
+            [
+                "sh",
+                "-c",
+                f"tar -xzf {shlex.quote(archive)} -C / && rm -f {shlex.quote(archive)}",
+            ],
+            {},
+            "Harbor artifact restore",
+        )
+        if self.data.verifier.upload_tests:
+            tests = f"/logs/verifier/.vf-harbor-tests-{uuid.uuid4().hex}.tgz"
+            await runtime.write(tests, make_tar(Path(self.data.task_dir) / "tests"))
+            await run_checked(
+                runtime,
+                [
+                    "sh",
+                    "-c",
+                    "rm -rf /tests && mkdir -p /tests && "
+                    f"tar -xzf {shlex.quote(tests)} -C /tests && "
+                    f"rm -f {shlex.quote(tests)}",
+                ],
+                {},
+                "Harbor test upload",
+            )
+
+    @reward(weight=1.0)
+    async def solved(self, runtime: Runtime) -> float | dict[str, float]:
+        if not self.data.verifier.separate:
+            await runtime.write(
+                "/tmp/tests.tgz", make_tar(Path(self.data.task_dir) / "tests")
+            )
+            await run_checked(
+                runtime,
+                [
+                    "sh",
+                    "-c",
+                    "rm -rf /logs/verifier /tests && "
+                    "mkdir -p /logs/verifier /tests && "
+                    "tar -xzf /tmp/tests.tgz -C /tests",
+                ],
+                {},
+                "Harbor verifier setup",
+            )
+        await run_checked(
+            runtime,
+            ["test", "-f", "/tests/test.sh"],
+            {},
+            "Harbor verifier test discovery",
         )
         await runtime.run(
-            ["sh", "-c", "cd /tests && bash test.sh"], verifier_env(self.data)
+            [
+                "sh",
+                "-c",
+                "bash /tests/test.sh > /logs/verifier/test-stdout.txt 2>&1",
+            ],
+            verifier_env(self.data),
         )
         try:
-            reward = (await runtime.read("/logs/verifier/reward.txt")).decode().strip()
-            return float(reward or 0)
-        except (SandboxError, OSError, ValueError):
+            raw = (await runtime.read("/logs/verifier/reward.json")).decode()
+        except UnicodeDecodeError:
+            return 0.0
+        except (SandboxError, OSError):
+            raw = None
+        if raw is not None:
+            try:
+                values = json.loads(raw)
+                if not isinstance(values, dict):
+                    return 0.0
+                rewards = {
+                    str(key): float(value)
+                    for key, value in values.items()
+                    if isinstance(value, (int, float)) and not isinstance(value, bool)
+                }
+                return (
+                    rewards
+                    if len(rewards) == len(values)
+                    and all(math.isfinite(value) for value in rewards.values())
+                    else 0.0
+                )
+            except (TypeError, ValueError, OverflowError):
+                return 0.0
+        try:
+            raw = (await runtime.read("/logs/verifier/reward.txt")).decode().strip()
+            value = float(raw)
+            return value if math.isfinite(value) else 0.0
+        except (SandboxError, OSError, ValueError, OverflowError):
             return 0.0
 
 
+async def run_checked(
+    runtime: Runtime, argv: list[str], env: dict[str, str], action: str
+) -> None:
+    result = await runtime.run(argv, env)
+    if result.exit_code:
+        detail = (result.stderr or result.stdout).strip()
+        raise SandboxError(f"{action} failed: {detail}")
+
+
 def harbor_cli() -> str:
     scripts_dir = Path(sys.executable).parent
     harbor_bin = shutil.which("harbor", path=str(scripts_dir))
@@ -266,8 +763,147 @@ def parse_resources(env: dict, multiplier: float = 1.0) -> TaskResources:
     )
 
 
+def parse_verifier(
+    task_dir: Path, config: dict, resource_multiplier: float
+) -> HarborVerifier:
+    environment = config.get("environment", {})
+    verifier = config.get("verifier", {})
+    mode = verifier.get("environment_mode")
+    verifier_environment = verifier.get("environment")
+    if mode not in (None, "shared", "separate"):
+        raise ValueError(f"{task_dir.name}: invalid verifier environment_mode {mode!r}")
+    if mode == "shared" and verifier_environment is not None:
+        raise ValueError(
+            f"{task_dir.name}: [verifier.environment] is incompatible with "
+            "environment_mode='shared'"
+        )
+
+    separate = mode == "separate" or verifier_environment is not None
+    if not separate:
+        return HarborVerifier(env=verifier.get("env", {}))
+    if verifier.get("collect"):
+        raise ValueError(
+            f"{task_dir.name}: [[verifier.collect]] needs compose services, which "
+            "Verifiers runtimes do not support"
+        )
+    if verifier.get("user") is not None:
+        raise ValueError(
+            f"{task_dir.name}: [verifier].user is not supported for separate runtimes"
+        )
+
+    explicit_environment = verifier_environment is not None
+    verifier_environment = verifier_environment if explicit_environment else environment
+    if verifier_environment.get("os", "linux").lower() != "linux":
+        raise ValueError(
+            f"{task_dir.name}: separate Harbor verification supports Linux images only"
+        )
+    if explicit_environment and not verifier_environment.get("docker_image"):
+        raise ValueError(
+            f"{task_dir.name}: [verifier.environment] needs a pullable docker_image; "
+            "building tests/Dockerfile is not supported"
+        )
+    unsupported = [
+        field
+        for field in ("healthcheck", "mcp_servers", "skills_dir", "gpu_types", "tpu")
+        if verifier_environment.get(field)
+    ]
+    if unsupported:
+        raise ValueError(
+            f"{task_dir.name}: separate verifier environment fields are not supported: "
+            + ", ".join(unsupported)
+        )
+
+    network_mode = verifier.get("network_mode")
+    if network_mode is None:
+        network_mode = verifier_environment.get("network_mode")
+    if network_mode is None:
+        network_mode = (
+            "public"
+            if verifier_environment.get("allow_internet", True)
+            else "no-network"
+        )
+    if network_mode == "allowlist":
+        raise ValueError(
+            f"{task_dir.name}: verifier network_mode='allowlist' is not supported"
+        )
+    if verifier.get("allowed_hosts") is not None or verifier_environment.get(
+        "allowed_hosts"
+    ):
+        raise ValueError(
+            f"{task_dir.name}: verifier network allowlists are not supported"
+        )
+    if network_mode not in ("public", "no-network"):
+        raise ValueError(
+            f"{task_dir.name}: invalid verifier network_mode {network_mode!r}"
+        )
+
+    raw_artifacts = config.get("artifacts", [])
+    if not isinstance(raw_artifacts, list):
+        raise ValueError(f"{task_dir.name}: artifacts must be a list")
+    raw_artifacts = ["/logs/artifacts", *raw_artifacts]
+    artifacts: list[HarborArtifact] = []
+    for raw in raw_artifacts:
+        artifact = {"source": raw} if isinstance(raw, str) else raw
+        if not isinstance(artifact, dict) or not isinstance(
+            artifact.get("source"), str
+        ):
+            raise ValueError(f"{task_dir.name}: invalid artifact entry {raw!r}")
+        if artifact.get("service") not in (None, "main"):
+            raise ValueError(f"{task_dir.name}: sidecar artifacts are not supported")
+        if artifact.get("destination"):
+            raise ValueError(
+                f"{task_dir.name}: artifact destinations are not supported"
+            )
+        if artifact.get("exclude"):
+            raise ValueError(
+                f"{task_dir.name}: artifact exclude patterns are not supported"
+            )
+        source = artifact["source"]
+        if not source.startswith("/") or ".." in PurePosixPath(source).parts:
+            raise ValueError(
+                f"{task_dir.name}: artifact source must be an absolute non-root path, "
+                f"got {source!r}"
+            )
+        path = PurePosixPath(f"/{source.lstrip('/')}")
+        source = path.as_posix()
+        if path == PurePosixPath("/"):
+            raise ValueError(
+                f"{task_dir.name}: artifact source must be an absolute non-root path, "
+                f"got {source!r}"
+            )
+        protected = (PurePosixPath("/tests"), PurePosixPath("/logs/verifier"))
+        if any(
+            path == target or path in target.parents or target in path.parents
+            for target in protected
+        ):
+            raise ValueError(
+                f"{task_dir.name}: artifact source {source!r} overlaps verifier-owned files"
+            )
+        existing = [PurePosixPath(entry.source) for entry in artifacts]
+        if any(
+            path == other or path in other.parents or other in path.parents
+            for other in existing
+        ):
+            continue
+        artifacts.append(HarborArtifact(source=source))
+
+    env = verifier_environment.get("env", {}) | verifier.get("env", {})
+    return HarborVerifier(
+        separate=True,
+        image=verifier_environment.get("docker_image"),
+        workdir=verifier_environment.get("workdir"),
+        resources=parse_resources(verifier_environment, resource_multiplier),
+        network_access=network_mode == "public",
+        env=env,
+        artifacts=artifacts,
+        upload_tests=not explicit_environment,
+    )
+
+
 def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborData:
     config = tomllib.loads((task_dir / "task.toml").read_text())
+    if config.get("steps"):
+        raise ValueError(f"{task_dir.name}: Harbor multi-step tasks are not supported")
     task, meta = config.get("task", {}), config.get("metadata", {})
     authors = [Author(**a) for a in task.get("authors", [])]
     # Older registry entries stored one author in [metadata].
@@ -286,6 +922,7 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD
             harbor_config.require_image,
             harbor_config.ignore_dockerfile,
         ),
+        workdir=config.get("environment", {}).get("workdir"),
         timeout=TaskTimeout(
             harness=harness_timeout * harbor_config.timeout_multiplier
             if harness_timeout is not None
@@ -303,20 +940,20 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD
         category=meta.get("category"),
         tags=meta.get("tags", []),
         task_dir=str(task_dir),
-        verifier_env=config.get("verifier", {}).get("env", {}),
+        verifier=parse_verifier(task_dir, config, harbor_config.resource_multiplier),
     )
 
 
 def verifier_env(task: HarborData) -> dict[str, str]:
     """Resolve templates at scoring time so host secrets are never serialized."""
-    if not task.verifier_env:
+    if not task.verifier.env:
         return {}
 
     # Harbor is an optional dependency, so importing this module must still work
     # for users who do not install the Harbor extra.
     from harbor.utils.env import resolve_env_vars
 
-    return resolve_env_vars(task.verifier_env)
+    return resolve_env_vars(task.verifier.env)
 
 
 # Downloaded test directories are immutable. Cache only the latest archive to

From eb972e17959c2cfc7fe8101b2506e513dc109c48 Mon Sep 17 00:00:00 2001
From: Xeophon <46377542+xeophon@users.noreply.github.com>
Date: Thu, 16 Jul 2026 18:05:43 +0200
Subject: [PATCH 2/6] Address Harbor verifier runtime feedback

---
 docs/v1/harbor.md                       |  2 +-
 verifiers/v1/tasksets/harbor/taskset.py | 22 ++++++++++------------
 2 files changed, 11 insertions(+), 13 deletions(-)

diff --git a/docs/v1/harbor.md b/docs/v1/harbor.md
index e8b8e63199..56a94f5fbc 100644
--- a/docs/v1/harbor.md
+++ b/docs/v1/harbor.md
@@ -76,7 +76,7 @@ artifacts = ["/workspace/final-output"]
 environment_mode = "separate"
 ```
 
-Without `[verifier.environment]`, the clean runtime reuses the resolved agent image and Verifiers uploads the trusted `tests/` directory. With `[verifier.environment]`, its `docker_image` must be a pullable, prebuilt image that already contains `/tests/test.sh`. Verifiers does not build Harbor Dockerfiles: if `tests/Dockerfile` installs verifier-only dependencies, publish that image and configure it explicitly. The verifier environment can independently declare its image, workdir, resources, environment variables, and `public` or `no-network` network access.
+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.
 
diff --git a/verifiers/v1/tasksets/harbor/taskset.py b/verifiers/v1/tasksets/harbor/taskset.py
index cb288558bd..182ba04f1d 100644
--- a/verifiers/v1/tasksets/harbor/taskset.py
+++ b/verifiers/v1/tasksets/harbor/taskset.py
@@ -269,17 +269,7 @@ def scoring_runtime_config(self, base: RuntimeConfig) -> RuntimeConfig | None:
             raise ValueError(
                 "separate Harbor verification needs a docker, prime, or modal runtime"
             )
-
-        network = type(config).model_fields.get("network_access")
-        if (
-            network is not None
-            and getattr(config, "network_access") == network.default
-            and verifier.network_access != network.default
-        ):
-            config = config.model_copy(
-                update={"network_access": verifier.network_access}
-            )
-        return config
+        return config.model_copy(update={"network_access": verifier.network_access})
 
     async def score(
         self,
@@ -567,10 +557,11 @@ async def solved(self, runtime: Runtime) -> float | dict[str, float]:
             [
                 "sh",
                 "-c",
-                "bash /tests/test.sh > /logs/verifier/test-stdout.txt 2>&1",
+                "cd /tests && bash test.sh > /logs/verifier/test-stdout.txt 2>&1",
             ],
             verifier_env(self.data),
         )
+        # Harbor gives structured rewards precedence when both files exist.
         try:
             raw = (await runtime.read("/logs/verifier/reward.json")).decode()
         except UnicodeDecodeError:
@@ -802,6 +793,12 @@ def parse_verifier(
             f"{task_dir.name}: [verifier.environment] needs a pullable docker_image; "
             "building tests/Dockerfile is not supported"
         )
+    if not explicit_environment and (task_dir / "tests" / "Dockerfile").exists():
+        raise ValueError(
+            f"{task_dir.name}: tests/Dockerfile needs a pullable "
+            "[verifier.environment].docker_image; building verifier Dockerfiles "
+            "is not supported"
+        )
     unsupported = [
         field
         for field in ("healthcheck", "mcp_servers", "skills_dir", "gpu_types", "tpu")
@@ -880,6 +877,7 @@ def parse_verifier(
                 f"{task_dir.name}: artifact source {source!r} overlaps verifier-owned files"
             )
         existing = [PurePosixPath(entry.source) for entry in artifacts]
+        # Harbor keeps the first declaration when artifact sources overlap.
         if any(
             path == other or path in other.parents or other in path.parents
             for other in existing

From 916c4dea41552decc6c988ab7fb2897712ad6f09 Mon Sep 17 00:00:00 2001
From: Xeophon <46377542+xeophon@users.noreply.github.com>
Date: Thu, 16 Jul 2026 18:51:01 +0200
Subject: [PATCH 3/6] Retry bounded Prime downloads

---
 verifiers/v1/runtimes/prime.py | 79 +++++++++++++++++++++++-----------
 1 file changed, 55 insertions(+), 24 deletions(-)

diff --git a/verifiers/v1/runtimes/prime.py b/verifiers/v1/runtimes/prime.py
index 7dfa77cb7f..a1cfda24cf 100644
--- a/verifiers/v1/runtimes/prime.py
+++ b/verifiers/v1/runtimes/prime.py
@@ -16,6 +16,7 @@
 from pathlib import Path, PurePosixPath
 from typing import ClassVar, Literal
 
+import httpx
 from pydantic import model_validator
 from pydantic_config import BaseConfig
 
@@ -33,6 +34,14 @@
 
 MAX_LIFETIME = 24 * 60 * 60
 """Prime's fixed cap (seconds) on any sandbox's total lifetime."""
+DOWNLOAD_ATTEMPTS = 4
+DOWNLOAD_RETRY_STATUSES = {500, 502, 503, 504, 524}
+DOWNLOAD_RETRY_ERRORS = (
+    httpx.RemoteProtocolError,
+    httpx.ConnectError,
+    httpx.PoolTimeout,
+    httpx.ReadError,
+)
 
 
 class PrimeConfig(BaseConfig):
@@ -216,30 +225,52 @@ async def read(self, path: str, max_bytes: int | None = None) -> bytes:
             if max_bytes is not None:
                 # The SDK's download_file buffers response.content. Use its gateway
                 # connection directly so an untrusted file is capped while streaming.
-                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"
-                )
-                headers = {"Authorization": f"Bearer {auth['token']}"}
-                params = {"path": target, "sandbox_id": self.info.id}
-                buffer = io.BytesIO()
-                client = self._client._get_gateway_client()
-                async with client.stream(
-                    "GET",
-                    url,
-                    headers=headers,
-                    params=params,
-                    timeout=300,
-                ) as response:
-                    response.raise_for_status()
-                    async for chunk in response.aiter_bytes(1024 * 1024):
-                        if buffer.tell() + len(chunk) > max_bytes:
-                            raise SandboxError(
-                                f"read {path!r}: exceeds the {max_bytes} byte limit"
-                            )
-                        buffer.write(chunk)
-                return buffer.getvalue()
+                for attempt in range(DOWNLOAD_ATTEMPTS):
+                    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"
+                    )
+                    headers = {"Authorization": f"Bearer {auth['token']}"}
+                    params = {"path": target, "sandbox_id": self.info.id}
+                    buffer = io.BytesIO()
+                    try:
+                        client = self._client._get_gateway_client()
+                        async with client.stream(
+                            "GET",
+                            url,
+                            headers=headers,
+                            params=params,
+                            timeout=300,
+                        ) as response:
+                            response.raise_for_status()
+                            async for chunk in response.aiter_bytes(1024 * 1024):
+                                if buffer.tell() + len(chunk) > max_bytes:
+                                    raise SandboxError(
+                                        f"read {path!r}: exceeds the "
+                                        f"{max_bytes} byte limit"
+                                    )
+                                buffer.write(chunk)
+                        return buffer.getvalue()
+                    except httpx.HTTPStatusError as error:
+                        status = error.response.status_code
+                        if status == 401 and attempt < DOWNLOAD_ATTEMPTS - 1:
+                            await self._client.clear_auth_cache()
+                            continue
+                        if status == 409:
+                            if await self._client._should_retry_409(
+                                self.info.id, error, attempt
+                            ):
+                                continue
+                        if (
+                            status not in DOWNLOAD_RETRY_STATUSES
+                            or attempt == DOWNLOAD_ATTEMPTS - 1
+                        ):
+                            raise
+                    except DOWNLOAD_RETRY_ERRORS:
+                        if attempt == DOWNLOAD_ATTEMPTS - 1:
+                            raise
+                    await asyncio.sleep(2**attempt)
 
             with tempfile.TemporaryDirectory() as directory:
                 download = Path(directory) / "download"

From 915b3924a4adb4e9e442d1125b334b35b9f152a4 Mon Sep 17 00:00:00 2001
From: Xeophon <46377542+xeophon@users.noreply.github.com>
Date: Thu, 16 Jul 2026 20:27:32 +0200
Subject: [PATCH 4/6] Simplify separate Harbor verifier runtimes

---
 docs/v1/harbor.md                       |  10 +-
 verifiers/v1/env.py                     |   4 -
 verifiers/v1/rollout.py                 |   5 +-
 verifiers/v1/runtimes/base.py           |  34 +-
 verifiers/v1/runtimes/docker.py         |  69 +--
 verifiers/v1/runtimes/modal.py          |  39 +-
 verifiers/v1/runtimes/prime.py          | 151 ++---
 verifiers/v1/runtimes/subprocess.py     |  17 +-
 verifiers/v1/tasksets/harbor/taskset.py | 721 +++++++-----------------
 9 files changed, 335 insertions(+), 715 deletions(-)

diff --git a/docs/v1/harbor.md b/docs/v1/harbor.md
index 56a94f5fbc..6f4195a0d2 100644
--- a/docs/v1/harbor.md
+++ b/docs/v1/harbor.md
@@ -74,14 +74,18 @@ artifacts = ["/workspace/final-output"]
 
 [verifier]
 environment_mode = "separate"
+
+[verifier.environment]
+docker_image = "registry.example.com/my-benchmark-verifier:latest"
+network_mode = "no-network"
 ```
 
-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.
+Separate verification requires a prebuilt `[verifier.environment].docker_image` containing `/tests/test.sh`; Verifiers does not build `tests/Dockerfile`. The verifier environment can independently declare its workdir, resources, environment variables, and `public` or `no-network` network access. It is supported by the Docker and Prime runtimes.
 
-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.
+Only filesystem artifacts can cross this boundary. Artifact sources must be absolute, non-root paths and either regular files or directories containing only regular files and directories. Symlinks and special files are rejected; nested empty directories and file mode bits are not retained. Transfer is limited to 256 MiB across 10,000 files and a 4 MiB path manifest. Runtime images must use a default user that can create `/logs` 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: 
+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:
 - 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))
diff --git a/verifiers/v1/env.py b/verifiers/v1/env.py
index ddaf736c9d..e9f150f816 100644
--- a/verifiers/v1/env.py
+++ b/verifiers/v1/env.py
@@ -314,9 +314,6 @@ 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
@@ -357,7 +354,6 @@ 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,
diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py
index 04444d6431..fa64e9b679 100644
--- a/verifiers/v1/rollout.py
+++ b/verifiers/v1/rollout.py
@@ -52,7 +52,6 @@ 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,
@@ -65,7 +64,9 @@ def __init__(
         self.harness = harness
         self.ctx = ctx
         self.runtime_config = runtime_config
-        self.scoring_runtime_config = scoring_runtime_config
+        self.scoring_runtime_config = task.scoring_runtime_config(
+            harness.config.runtime
+        )
         self.setup_timeout = setup_timeout
         self.harness_timeout = harness_timeout
         self.finalize_timeout = finalize_timeout
diff --git a/verifiers/v1/runtimes/base.py b/verifiers/v1/runtimes/base.py
index a9e75f9182..c5cf456fd0 100644
--- a/verifiers/v1/runtimes/base.py
+++ b/verifiers/v1/runtimes/base.py
@@ -15,6 +15,7 @@
 
 from pydantic_config import BaseConfig
 
+from verifiers.v1.errors import SandboxError
 from verifiers.v1.utils.aio import run_shielded
 
 logger = logging.getLogger(__name__)
@@ -128,19 +129,6 @@ 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.
@@ -236,16 +224,26 @@ 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:
+    def _validate_read_limit(self, max_bytes: int) -> None:
+        if max_bytes < 0:
             raise ValueError("max_bytes must be non-negative")
 
     @abstractmethod
-    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."""
+    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 a file.
+
+        Runtimes with streaming file APIs should override this to enforce the limit
+        before buffering the full file.
+        """
+        self._validate_read_limit(max_bytes)
+        data = await self.read(path)
+        if len(data) > max_bytes:
+            raise SandboxError(f"read {path!r}: exceeds the {max_bytes} byte limit")
+        return data
+
     @abstractmethod
     async def write(self, path: str, data: bytes) -> None:
         pass
diff --git a/verifiers/v1/runtimes/docker.py b/verifiers/v1/runtimes/docker.py
index 0c7f873de8..f070774123 100644
--- a/verifiers/v1/runtimes/docker.py
+++ b/verifiers/v1/runtimes/docker.py
@@ -157,7 +157,26 @@ 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, max_bytes: int | None = None) -> bytes:
+    async def read(self, path: str) -> bytes:
+        proc = await asyncio.create_subprocess_exec(
+            "docker",
+            "exec",
+            "--workdir",
+            self.config.workdir,
+            self._container,
+            "cat",
+            path,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await proc.communicate()
+        if proc.returncode != 0:
+            raise SandboxError(
+                f"read {path!r}: {stderr.decode(errors='replace').strip()}"
+            )
+        return stdout
+
+    async def read_bounded(self, path: str, max_bytes: int) -> bytes:
         self._validate_read_limit(max_bytes)
         proc = await asyncio.create_subprocess_exec(
             "docker",
@@ -168,32 +187,21 @@ async def read(self, path: str, max_bytes: int | None = None) -> bytes:
             "cat",
             path,
             stdout=asyncio.subprocess.PIPE,
-            stderr=(
-                asyncio.subprocess.PIPE
-                if max_bytes is None
-                else asyncio.subprocess.DEVNULL
-            ),
+            stderr=asyncio.subprocess.DEVNULL,
         )
         try:
-            if max_bytes is None:
-                stdout, stderr = await proc.communicate()
+            try:
+                stdout = await proc.stdout.readexactly(max_bytes + 1)
+            except asyncio.IncompleteReadError as exc:
+                stdout = exc.partial
             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"
-                    )
+                with contextlib.suppress(ProcessLookupError):
+                    proc.kill()
                 await proc.wait()
-                stderr = b""
+                raise SandboxError(f"read {path!r}: exceeds the {max_bytes} byte limit")
+            await proc.wait()
             if proc.returncode != 0:
-                detail = stderr.decode(errors="replace").strip()
-                raise SandboxError(f"read {path!r}: {detail or 'docker exec failed'}")
+                raise SandboxError(f"read {path!r}: docker exec failed")
             return stdout
         finally:
             if proc.returncode is None:
@@ -223,23 +231,6 @@ 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
diff --git a/verifiers/v1/runtimes/modal.py b/verifiers/v1/runtimes/modal.py
index a7c887086d..b2ee7b7a6d 100644
--- a/verifiers/v1/runtimes/modal.py
+++ b/verifiers/v1/runtimes/modal.py
@@ -9,7 +9,6 @@
 
 import asyncio
 import contextlib
-import io
 import logging
 import shlex
 from pathlib import PurePosixPath
@@ -166,31 +165,9 @@ def _abs(self, path: str) -> str:
             return path
         return f"{self.config.workdir.rstrip('/')}/{path}"
 
-    async def read(self, path: str, max_bytes: int | None = None) -> bytes:
-        self._validate_read_limit(max_bytes)
+    async def read(self, path: str) -> bytes:
         try:
-            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
+            return await self._sandbox.filesystem.read_bytes.aio(self._abs(path))
         except Exception as e:
             raise SandboxError(f"read {path!r}: {e}") from e
 
@@ -205,18 +182,6 @@ 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.
diff --git a/verifiers/v1/runtimes/prime.py b/verifiers/v1/runtimes/prime.py
index a1cfda24cf..0a9668dc6e 100644
--- a/verifiers/v1/runtimes/prime.py
+++ b/verifiers/v1/runtimes/prime.py
@@ -35,12 +35,14 @@
 MAX_LIFETIME = 24 * 60 * 60
 """Prime's fixed cap (seconds) on any sandbox's total lifetime."""
 DOWNLOAD_ATTEMPTS = 4
-DOWNLOAD_RETRY_STATUSES = {500, 502, 503, 504, 524}
+DOWNLOAD_RETRY_STATUSES = {408, 500, 502, 503, 504, 524}
 DOWNLOAD_RETRY_ERRORS = (
     httpx.RemoteProtocolError,
     httpx.ConnectError,
     httpx.PoolTimeout,
     httpx.ReadError,
+    httpx.ConnectTimeout,
+    httpx.ReadTimeout,
 )
 
 
@@ -212,8 +214,7 @@ async def run_background(
                 f"prime background launch failed: {result.stderr.strip()}"
             )
 
-    async def read(self, path: str, max_bytes: int | None = None) -> bytes:
-        self._validate_read_limit(max_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 = (
@@ -222,60 +223,64 @@ async def read(self, path: str, max_bytes: int | None = None) -> bytes:
             else f"{self.config.workdir.rstrip('/')}/{path}"
         )
         try:
-            if max_bytes is not None:
-                # The SDK's download_file buffers response.content. Use its gateway
-                # connection directly so an untrusted file is capped while streaming.
-                for attempt in range(DOWNLOAD_ATTEMPTS):
-                    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"
-                    )
-                    headers = {"Authorization": f"Bearer {auth['token']}"}
-                    params = {"path": target, "sandbox_id": self.info.id}
-                    buffer = io.BytesIO()
-                    try:
-                        client = self._client._get_gateway_client()
-                        async with client.stream(
-                            "GET",
-                            url,
-                            headers=headers,
-                            params=params,
-                            timeout=300,
-                        ) as response:
-                            response.raise_for_status()
-                            async for chunk in response.aiter_bytes(1024 * 1024):
-                                if buffer.tell() + len(chunk) > max_bytes:
-                                    raise SandboxError(
-                                        f"read {path!r}: exceeds the "
-                                        f"{max_bytes} byte limit"
-                                    )
-                                buffer.write(chunk)
-                        return buffer.getvalue()
-                    except httpx.HTTPStatusError as error:
-                        status = error.response.status_code
-                        if status == 401 and attempt < DOWNLOAD_ATTEMPTS - 1:
-                            await self._client.clear_auth_cache()
-                            continue
-                        if status == 409:
-                            if await self._client._should_retry_409(
-                                self.info.id, error, attempt
-                            ):
-                                continue
-                        if (
-                            status not in DOWNLOAD_RETRY_STATUSES
-                            or attempt == DOWNLOAD_ATTEMPTS - 1
-                        ):
-                            raise
-                    except DOWNLOAD_RETRY_ERRORS:
-                        if attempt == DOWNLOAD_ATTEMPTS - 1:
-                            raise
-                    await asyncio.sleep(2**attempt)
-
             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)
+        except Exception as e:
+            raise SandboxError(f"read {path!r}: {e}") from e
+
+    async def read_bounded(self, path: str, max_bytes: int) -> bytes:
+        self._validate_read_limit(max_bytes)
+        target = (
+            path
+            if path.startswith("/")
+            else f"{self.config.workdir.rstrip('/')}/{path}"
+        )
+        try:
+            # The SDK's download_file buffers response.content. Use its gateway
+            # connection directly so an untrusted file is capped while streaming.
+            for attempt in range(DOWNLOAD_ATTEMPTS):
+                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"
+                )
+                headers = {"Authorization": f"Bearer {auth['token']}"}
+                params = {"path": target, "sandbox_id": self.info.id}
+                buffer = io.BytesIO()
+                try:
+                    client = self._client._get_gateway_client()
+                    async with client.stream(
+                        "GET", url, headers=headers, params=params, timeout=300
+                    ) as response:
+                        response.raise_for_status()
+                        async for chunk in response.aiter_bytes(1024 * 1024):
+                            if buffer.tell() + len(chunk) > max_bytes:
+                                raise SandboxError(
+                                    f"read {path!r}: exceeds the {max_bytes} byte limit"
+                                )
+                            buffer.write(chunk)
+                    return buffer.getvalue()
+                except httpx.HTTPStatusError as error:
+                    status = error.response.status_code
+                    if status == 401 and attempt < DOWNLOAD_ATTEMPTS - 1:
+                        await self._client.clear_auth_cache()
+                        continue
+                    if status == 409:
+                        if await self._client._should_retry_409(
+                            self.info.id, error, attempt
+                        ):
+                            continue
+                    if (
+                        status not in DOWNLOAD_RETRY_STATUSES
+                        or attempt == DOWNLOAD_ATTEMPTS - 1
+                    ):
+                        raise
+                except DOWNLOAD_RETRY_ERRORS:
+                    if attempt == DOWNLOAD_ATTEMPTS - 1:
+                        raise
+                await asyncio.sleep(2**attempt)
         except SandboxError:
             raise
         except Exception as e:
@@ -303,46 +308,6 @@ 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:
-        client = self._client
-        if client is None:
-            if self.info.id is None:
-                return
-            raise SandboxError("prime cannot confirm teardown without its client")
-
-        confirmed = False
-        try:
-            await client.delete(self.info.id)
-        except Exception as e:
-            if "HTTP 404" not in str(e):
-                raise SandboxError(
-                    f"prime could not delete sandbox {self.info.id}: {e}"
-                ) from e
-            confirmed = True
-        for _ in range(60):
-            if confirmed:
-                break
-            try:
-                sandbox = await client.get(self.info.id)
-            except Exception as e:
-                if "HTTP 404" in str(e):
-                    confirmed = True
-                    break
-                raise SandboxError(
-                    f"prime could not confirm teardown of {self.info.id}: {e}"
-                ) from e
-            if sandbox.status in {"TERMINATED", "ERROR", "TIMEOUT"}:
-                confirmed = True
-                break
-            await asyncio.sleep(0.5)
-        if not confirmed:
-            raise SandboxError(
-                f"prime sandbox {self.info.id} remained active after deletion"
-            )
-        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.
diff --git a/verifiers/v1/runtimes/subprocess.py b/verifiers/v1/runtimes/subprocess.py
index d1bb21750a..daf6483621 100644
--- a/verifiers/v1/runtimes/subprocess.py
+++ b/verifiers/v1/runtimes/subprocess.py
@@ -10,7 +10,6 @@
 
 from pydantic_config import BaseConfig
 
-from verifiers.v1.errors import SandboxError
 from verifiers.v1.runtimes.base import BaseRuntimeInfo, ProgramResult, Runtime
 
 # Implicit host inheritance removes every name containing "API_KEY" while keeping
@@ -95,20 +94,8 @@ async def run_background(
             proc
         )  # killed in stop() — a host process won't die on its own
 
-    async def read(self, path: str, max_bytes: int | None = None) -> bytes:
-        self._validate_read_limit(max_bytes)
-        target = self.workdir / path
-        if max_bytes is None:
-            return await asyncio.to_thread(target.read_bytes)
-
-        def read_bounded() -> bytes:
-            with target.open("rb") as file:
-                data = file.read(max_bytes + 1)
-            if len(data) > max_bytes:
-                raise SandboxError(f"read {path!r}: exceeds the {max_bytes} byte limit")
-            return data
-
-        return await asyncio.to_thread(read_bounded)
+    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:
         target = self.workdir / path
diff --git a/verifiers/v1/tasksets/harbor/taskset.py b/verifiers/v1/tasksets/harbor/taskset.py
index 182ba04f1d..3289e40638 100644
--- a/verifiers/v1/tasksets/harbor/taskset.py
+++ b/verifiers/v1/tasksets/harbor/taskset.py
@@ -10,13 +10,9 @@
 image unless ``require_image`` is set.
 """
 
-import contextlib
-import gzip
+import asyncio
 import hashlib
 import io
-import json
-import math
-import shlex
 import shutil
 import subprocess
 import sys
@@ -27,16 +23,17 @@
 from collections.abc import Iterator
 from functools import lru_cache
 from pathlib import Path, PurePosixPath
-from typing import BinaryIO
+from typing import TYPE_CHECKING
 
 from pydantic import Field
 
 from verifiers.v1.decorators import reward
 from verifiers.v1.errors import SandboxError
 from verifiers.v1.runtimes import (
+    DockerConfig,
+    PrimeConfig,
     Runtime,
     RuntimeConfig,
-    SubprocessConfig,
     make_runtime,
 )
 from verifiers.v1.task import Task, TaskData, TaskResources, TaskTimeout
@@ -46,134 +43,15 @@
 
 CACHE = Path.home() / ".cache" / "harbor"
 HARBOR_INSTALL_HINT = "uv sync --python 3.12 --extra harbor"
-MAX_ARTIFACT_ARCHIVE_BYTES = 256 * 1024 * 1024
-MAX_ARTIFACT_CONTENT_BYTES = 1024 * 1024 * 1024
-MAX_ARTIFACT_MEMBERS = 100_000
-MAX_ARTIFACT_TAR_BYTES = MAX_ARTIFACT_CONTENT_BYTES + MAX_ARTIFACT_ARCHIVE_BYTES
-MAX_ARTIFACT_HEADERS = MAX_ARTIFACT_MEMBERS * 2 + 8
-MAX_ARTIFACT_METADATA_BYTES = 1024 * 1024
-MAX_PAX_RECORDS = 4096
-
-
-class _LimitedTarReader:
-    """Keep tarfile from turning one hostile metadata header into a huge allocation."""
-
-    def __init__(self, file: BinaryIO) -> None:
-        self.file = file
-
-    def read(self, size: int = -1) -> bytes:
-        if size < 0 or size > MAX_ARTIFACT_METADATA_BYTES + tarfile.BLOCKSIZE:
-            raise SandboxError("Harbor artifact archive has oversized metadata")
-        return self.file.read(size)
-
-    def seek(self, offset: int, whence: int = io.SEEK_SET) -> int:
-        return self.file.seek(offset, whence)
-
-    def tell(self) -> int:
-        return self.file.tell()
-
-    def __getattr__(self, name: str) -> object:
-        return getattr(self.file, name)
-
-
-def _validate_pax_payload(payload: bytes) -> None:
-    """Reject structural overrides and bound tarfile's per-record allocations."""
-    position = 0
-    records = 0
-    while position < len(payload) and payload[position]:
-        separator = payload.find(b" ", position)
-        if separator < 0:
-            raise SandboxError("Harbor artifact archive has invalid PAX metadata")
-        try:
-            length = int(payload[position:separator])
-        except ValueError as exc:
-            raise SandboxError(
-                "Harbor artifact archive has invalid PAX metadata"
-            ) from exc
-        end = position + length
-        if length < 5 or end > len(payload) or payload[end - 1] != 0x0A:
-            raise SandboxError("Harbor artifact archive has invalid PAX metadata")
-        key, equals, value = payload[separator + 1 : end - 1].partition(b"=")
-        if not key or not equals:
-            raise SandboxError("Harbor artifact archive has invalid PAX metadata")
-        records += 1
-        if records > MAX_PAX_RECORDS:
-            raise SandboxError("Harbor artifact archive has too many PAX records")
-        if key == b"size":
-            raise SandboxError(
-                "Harbor artifact archive uses an unsupported PAX size override"
-            )
-        if (
-            key.startswith(b"GNU.sparse.")
-            or key == b"SCHILY.realsize"
-            or (key == b"SCHILY.filetype" and value == b"sparse")
-        ):
-            raise SandboxError("Harbor artifact archive uses unsupported sparse files")
-        position = end
-
-
-def _validate_tar_archive(file: BinaryIO) -> None:
-    """Bound raw extensions before tarfile recursively interprets them."""
-    file.seek(0, io.SEEK_END)
-    archive_size = file.tell()
-    file.seek(0)
-    headers = 0
-    consecutive_extensions = 0
-    extensions = {
-        tarfile.XHDTYPE,
-        tarfile.XGLTYPE,
-        tarfile.SOLARIS_XHDTYPE,
-        tarfile.GNUTYPE_LONGNAME,
-        tarfile.GNUTYPE_LONGLINK,
-    }
-    while file.tell() < archive_size:
-        header = file.read(tarfile.BLOCKSIZE)
-        if len(header) != tarfile.BLOCKSIZE:
-            raise SandboxError("Harbor artifact archive has a truncated header")
-        if not any(header):
-            break
-        try:
-            member = tarfile.TarInfo.frombuf(header, "utf-8", "surrogateescape")
-        except tarfile.TarError as exc:
-            raise SandboxError("Harbor artifact archive has an invalid header") from exc
-        if member.size < 0:
-            raise SandboxError("Harbor artifact archive has a negative entry size")
-        headers += 1
-        if headers > MAX_ARTIFACT_HEADERS:
-            raise SandboxError("Harbor artifact archive has too many headers")
-        if member.type == tarfile.GNUTYPE_SPARSE:
-            raise SandboxError("Harbor artifact archive uses unsupported sparse files")
-
-        stored_size = (
-            member.size
-            if member.type in extensions
-            or member.isreg()
-            or member.type not in tarfile.SUPPORTED_TYPES
-            else 0
-        )
-        data_end = file.tell() + stored_size
-        padded_end = data_end + (-stored_size % tarfile.BLOCKSIZE)
-        if padded_end > archive_size:
-            raise SandboxError("Harbor artifact archive has a truncated entry")
-        if member.type in extensions:
-            consecutive_extensions += 1
-            if consecutive_extensions > 8:
-                raise SandboxError(
-                    "Harbor artifact archive has too many metadata headers"
-                )
-            if member.size > MAX_ARTIFACT_METADATA_BYTES:
-                raise SandboxError("Harbor artifact archive has oversized metadata")
-            payload = file.read(member.size)
-            if member.type in {
-                tarfile.XHDTYPE,
-                tarfile.XGLTYPE,
-                tarfile.SOLARIS_XHDTYPE,
-            }:
-                _validate_pax_payload(payload)
-        else:
-            consecutive_extensions = 0
-        file.seek(padded_end)
-    file.seek(0)
+MAX_ARTIFACT_BYTES = 256 * 1024 * 1024
+MAX_ARTIFACT_FILES = 10_000
+MAX_ARTIFACT_MANIFEST_BYTES = 4 * 1024 * 1024
+
+if TYPE_CHECKING:
+    from harbor.models.task.config import (
+        EnvironmentConfig,
+        TaskConfig as HarborTaskConfig,
+    )
 
 
 class HarborConfig(TasksetConfig):
@@ -220,9 +98,7 @@ class HarborVerifier(StrictBaseModel):
     workdir: str | None = None
     resources: TaskResources = TaskResources()
     network_access: bool = True
-    env: dict[str, str] = {}
     artifacts: list[HarborArtifact] = []
-    upload_tests: bool = False
 
 
 class HarborData(TaskData):
@@ -239,6 +115,8 @@ class HarborData(TaskData):
     tags: list[str] = []
     task_dir: str = Field("", exclude=True)
     """Host path to the task dir; used to stage tests/ to verify, not serialized."""
+    verifier_env: dict[str, str] = {}
+    """Raw [verifier.env] entries, resolved on the host at scoring time."""
     verifier: HarborVerifier = Field(default_factory=HarborVerifier, exclude=True)
     """Resolved verifier mode and runtime data, kept host-side for scoring."""
 
@@ -250,25 +128,21 @@ def scoring_runtime_config(self, base: RuntimeConfig) -> RuntimeConfig | None:
         verifier = self.data.verifier
         if not verifier.separate:
             return None
+        if not isinstance(base, (DockerConfig, PrimeConfig)):
+            raise ValueError(
+                "separate Harbor verification needs a docker or prime runtime"
+            )
 
         from verifiers.v1.env import resolve_runtime_config
 
-        data = (
-            self.data
-            if verifier.upload_tests
-            else TaskData(
-                idx=self.data.idx,
-                prompt=None,
-                image=verifier.image,
-                workdir=verifier.workdir,
-                resources=verifier.resources,
-            )
+        data = TaskData(
+            idx=self.data.idx,
+            prompt=None,
+            image=verifier.image,
+            workdir=verifier.workdir,
+            resources=verifier.resources,
         )
         config = resolve_runtime_config(base, data)
-        if isinstance(config, SubprocessConfig):
-            raise ValueError(
-                "separate Harbor verification needs a docker, prime, or modal runtime"
-            )
         return config.model_copy(update={"network_access": verifier.network_access})
 
     async def score(
@@ -288,249 +162,163 @@ async def score(
                 "separate Harbor scoring requires a verifier runtime config"
             )
 
-        artifacts, paths = await self._collect_artifacts(runtime)
-        # The agent runtime must be gone before any verifier process starts.
-        await runtime.stop_confirmed()
-        target = make_runtime(scoring_runtime_config, name=f"{runtime.name}-verifier")
-        try:
-            await target.start()
-            await self._prepare_verifier(
-                target,
-                artifacts,
-                paths,
-                getattr(scoring_runtime_config, "workdir", "/"),
+        with tempfile.TemporaryDirectory() as temp:
+            files, directories = await self._collect_artifacts(runtime, Path(temp))
+            # No verifier process starts until the agent runtime is gone.
+            await runtime.stop()
+            target = make_runtime(
+                scoring_runtime_config, name=f"{runtime.name}-verifier"
             )
-            await super().score(trace, target)
-        finally:
-            await target.stop()
-
-    async def _collect_artifacts(self, runtime: Runtime) -> tuple[bytes, list[str]]:
-        buffer = tempfile.SpooledTemporaryFile(max_size=8 * 1024 * 1024)
-        seen: set[str] = set()
-        paths: list[str] = []
-        content_size = 0
-        member_count = 0
-        try:
-            await run_checked(
-                runtime,
+            try:
+                await target.start()
+                await self._prepare_verifier(target, files, directories)
+                await super().score(trace, target)
+            finally:
+                await target.stop()
+
+    async def _collect_artifacts(
+        self, runtime: Runtime, staging: Path
+    ) -> tuple[list[tuple[str, Path]], list[str]]:
+        await run_checked(
+            runtime,
+            ["sh", "-c", "rm -rf /logs/verifier && mkdir -p /logs/verifier"],
+            {},
+            "Harbor artifact staging setup",
+        )
+        files: list[tuple[str, Path]] = []
+        directories: list[str] = []
+        total_bytes = 0
+        for artifact in self.data.verifier.artifacts:
+            source = artifact.source
+            result = await runtime.run(
                 [
                     "sh",
                     "-c",
-                    "rm -rf /logs/verifier && mkdir -p /logs/verifier",
+                    'if [ -L "$1" ]; then printf unsupported; '
+                    'elif [ ! -e "$1" ]; then printf missing; '
+                    'elif [ -f "$1" ]; then printf file; '
+                    'elif [ -d "$1" ]; then printf directory; '
+                    "else printf unsupported; fi",
+                    "sh",
+                    source,
                 ],
                 {},
-                "Harbor artifact staging setup",
             )
-            with tarfile.open(fileobj=buffer, mode="w:gz") as safe:
-                for artifact in self.data.verifier.artifacts:
-                    exists = await runtime.run(["test", "-e", artifact.source], {})
-                    if exists.exit_code == 1:
-                        continue
-                    if exists.exit_code:
-                        detail = (exists.stderr or exists.stdout).strip()
-                        raise SandboxError(
-                            f"Harbor artifact probe failed for {artifact.source!r}: {detail}"
-                        )
+            if result.exit_code:
+                detail = (result.stderr or result.stdout).strip()
+                raise SandboxError(
+                    f"Harbor artifact probe failed for {source!r}: {detail}"
+                )
+            kind = result.stdout
+            if kind == "missing":
+                continue
+            if kind not in ("file", "directory"):
+                raise SandboxError(
+                    f"Harbor artifact {source!r} must be a regular file or directory"
+                )
 
-                    archive = (
-                        f"/logs/verifier/.vf-harbor-artifacts-{uuid.uuid4().hex}.tgz"
-                    )
-                    packed = await runtime.run(
-                        [
-                            "tar",
-                            "-chzf",
-                            archive,
-                            "--exclude",
-                            f"./{archive.lstrip('/')}",
-                            "-C",
-                            "/",
-                            f"./{artifact.source.lstrip('/')}",
-                        ],
-                        {},
-                    )
-                    if packed.exit_code:
-                        with contextlib.suppress(Exception):
-                            await runtime.run(["rm", "-f", archive], {})
-                        exists = await runtime.run(["test", "-e", artifact.source], {})
-                        if exists.exit_code == 1:
-                            continue
-                        detail = (packed.stderr or packed.stdout).strip()
+            paths = [source]
+            if kind == "directory":
+                directories.append(source)
+                manifest = f"/logs/verifier/.vf-harbor-{uuid.uuid4().hex}"
+                listed = await runtime.run(
+                    [
+                        "sh",
+                        "-c",
+                        'find "$1" -mindepth 1 ! -type d ! -type f '
+                        '-print -quit > "$3" && [ ! -s "$3" ] && '
+                        'find "$1" -type f -print0 > "$2"',
+                        "sh",
+                        source,
+                        manifest,
+                        f"{manifest}.bad",
+                    ],
+                    {},
+                )
+                try:
+                    if listed.exit_code:
                         raise SandboxError(
-                            f"Harbor artifact collection failed for "
-                            f"{artifact.source!r}: {detail}"
+                            f"Harbor artifact directory {source!r} must contain only "
+                            "regular files and directories"
                         )
-                    try:
-                        unsafe = await runtime.read(
-                            archive, max_bytes=MAX_ARTIFACT_ARCHIVE_BYTES
-                        )
-                    finally:
-                        with contextlib.suppress(Exception):
-                            await runtime.run(["rm", "-f", archive], {})
-
-                    root = PurePosixPath(artifact.source.lstrip("/"))
-                    # The agent controls these bytes. Flatten links and retain only
-                    # regular files/directories rooted under the declared artifact.
-                    with tempfile.SpooledTemporaryFile(
-                        max_size=8 * 1024 * 1024
-                    ) as unpacked:
-                        with gzip.GzipFile(fileobj=io.BytesIO(unsafe)) as compressed:
-                            tar_size = 0
-                            while chunk := compressed.read(1024 * 1024):
-                                tar_size += len(chunk)
-                                if tar_size > MAX_ARTIFACT_TAR_BYTES:
-                                    raise SandboxError(
-                                        "Harbor artifact archive exceeds the "
-                                        "decompression limit"
-                                    )
-                                unpacked.write(chunk)
-                        unpacked.seek(0)
-                        _validate_tar_archive(unpacked)
-                        with tarfile.open(
-                            fileobj=_LimitedTarReader(unpacked), mode="r:"
-                        ) as source:
-                            for member in source:
-                                path = PurePosixPath(member.name)
-                                if (
-                                    member.size < 0
-                                    or path.is_absolute()
-                                    or ".." in path.parts
-                                    or not (path == root or root in path.parents)
-                                    or not (
-                                        member.isfile()
-                                        or member.islnk()
-                                        or member.isdir()
-                                    )
-                                ):
-                                    raise SandboxError(
-                                        "Harbor artifact collection produced unsafe "
-                                        f"entry {member.name!r}"
-                                    )
-                                member_count += 1
-                                content_size += member.size
-                                if member_count > MAX_ARTIFACT_MEMBERS:
-                                    raise SandboxError(
-                                        "Harbor artifacts exceed the "
-                                        f"{MAX_ARTIFACT_MEMBERS:,} entry limit"
-                                    )
-                                if content_size > MAX_ARTIFACT_CONTENT_BYTES:
-                                    raise SandboxError(
-                                        "Harbor artifacts exceed the "
-                                        f"{MAX_ARTIFACT_CONTENT_BYTES // 1024**3} GiB "
-                                        "uncompressed limit"
-                                    )
-                                if path.as_posix() in seen:
-                                    continue
-                                seen.add(path.as_posix())
-                                clean = tarfile.TarInfo(path.as_posix())
-                                clean.mode = member.mode & 0o777
-                                clean.mtime = member.mtime
-                                if member.isdir():
-                                    clean.type = tarfile.DIRTYPE
-                                    safe.addfile(clean)
-                                    continue
-                                file = source.extractfile(member)
-                                if file is None:
-                                    raise SandboxError(
-                                        "Harbor artifact collection could not read "
-                                        f"{member.name!r}"
-                                    )
-                                if member.islnk():
-                                    with tempfile.SpooledTemporaryFile(
-                                        max_size=8 * 1024 * 1024
-                                    ) as linked:
-                                        linked_size = 0
-                                        while chunk := file.read(1024 * 1024):
-                                            linked_size += len(chunk)
-                                            if (
-                                                content_size + linked_size
-                                                > MAX_ARTIFACT_CONTENT_BYTES
-                                            ):
-                                                raise SandboxError(
-                                                    "Harbor artifacts exceed the "
-                                                    f"{MAX_ARTIFACT_CONTENT_BYTES // 1024**3} "
-                                                    "GiB uncompressed limit"
-                                                )
-                                            linked.write(chunk)
-                                        content_size += linked_size
-                                        clean.size = linked_size
-                                        linked.seek(0)
-                                        safe.addfile(clean, linked)
-                                    continue
-                                clean.size = member.size
-                                safe.addfile(clean, file)
-                    paths.append(artifact.source)
-        except (OSError, tarfile.TarError) as exc:
-            buffer.close()
-            raise SandboxError(
-                "Harbor artifact collection produced an invalid archive"
-            ) from exc
-        except BaseException:
-            buffer.close()
-            raise
-        try:
-            buffer.seek(0, io.SEEK_END)
-            if buffer.tell() > MAX_ARTIFACT_ARCHIVE_BYTES:
+                    raw = await runtime.read_bounded(
+                        manifest, MAX_ARTIFACT_MANIFEST_BYTES
+                    )
+                finally:
+                    await runtime.run(["rm", "-f", manifest, f"{manifest}.bad"], {})
+                try:
+                    paths = [
+                        value.decode()
+                        for value in raw.rstrip(b"\0").split(b"\0")
+                        if value
+                    ]
+                except UnicodeDecodeError as exc:
+                    raise SandboxError(
+                        f"Harbor artifact directory {source!r} has a non-UTF-8 path"
+                    ) from exc
+
+            if len(files) + len(paths) > MAX_ARTIFACT_FILES:
                 raise SandboxError(
-                    "Harbor artifacts exceed the "
-                    f"{MAX_ARTIFACT_ARCHIVE_BYTES // 1024**2} MiB archive limit"
+                    f"Harbor artifacts exceed the {MAX_ARTIFACT_FILES:,} file limit"
+                )
+            root = PurePosixPath(source)
+            for path in paths:
+                artifact_path = PurePosixPath(path)
+                if (
+                    not artifact_path.is_absolute()
+                    or ".." in artifact_path.parts
+                    or not (artifact_path == root or root in artifact_path.parents)
+                ):
+                    raise SandboxError(
+                        f"Harbor artifact directory produced unsafe path {path!r}"
+                    )
+                data = await runtime.read_bounded(
+                    artifact_path.as_posix(), MAX_ARTIFACT_BYTES - total_bytes
                 )
-            buffer.seek(0)
-            return buffer.read(), paths
-        finally:
-            buffer.close()
+                total_bytes += len(data)
+                host_path = staging / str(len(files))
+                await asyncio.to_thread(host_path.write_bytes, data)
+                files.append((artifact_path.as_posix(), host_path))
+        return files, directories
 
     async def _prepare_verifier(
-        self, runtime: Runtime, artifacts: bytes, paths: list[str], workdir: str
+        self,
+        runtime: Runtime,
+        files: list[tuple[str, Path]],
+        directories: list[str],
     ) -> None:
         await run_checked(
             runtime,
-            [
-                "sh",
-                "-c",
-                "rm -rf /logs/verifier && mkdir -p /logs/verifier /logs/artifacts",
-            ],
+            ["rm", "-rf", "/logs/verifier"],
             {},
             "Harbor verifier setup",
         )
-        archive = f"/logs/verifier/.vf-harbor-artifacts-{uuid.uuid4().hex}.tgz"
-        if paths:
-            command = "rm -rf " + " ".join(shlex.quote(path) for path in paths)
-            command += f" && mkdir -p {shlex.quote(workdir)}"
+        await run_checked(
+            runtime,
+            ["mkdir", "-p", "/logs/verifier"],
+            {},
+            "Harbor verifier setup",
+        )
+        sources = [artifact.source for artifact in self.data.verifier.artifacts]
+        if sources:
             await run_checked(
                 runtime,
-                ["sh", "-c", command],
+                ["rm", "-rf", *sources],
                 {},
                 "Harbor artifact target cleanup",
             )
-        await runtime.write(archive, artifacts)
+        directories = list(dict.fromkeys(["/logs/artifacts", *directories]))
         await run_checked(
             runtime,
-            [
-                "sh",
-                "-c",
-                f"tar -xzf {shlex.quote(archive)} -C / && rm -f {shlex.quote(archive)}",
-            ],
+            ["mkdir", "-p", *directories],
             {},
-            "Harbor artifact restore",
+            "Harbor artifact directory restore",
         )
-        if self.data.verifier.upload_tests:
-            tests = f"/logs/verifier/.vf-harbor-tests-{uuid.uuid4().hex}.tgz"
-            await runtime.write(tests, make_tar(Path(self.data.task_dir) / "tests"))
-            await run_checked(
-                runtime,
-                [
-                    "sh",
-                    "-c",
-                    "rm -rf /tests && mkdir -p /tests && "
-                    f"tar -xzf {shlex.quote(tests)} -C /tests && "
-                    f"rm -f {shlex.quote(tests)}",
-                ],
-                {},
-                "Harbor test upload",
-            )
+        for path, host_path in files:
+            await runtime.write(path, await asyncio.to_thread(host_path.read_bytes))
 
     @reward(weight=1.0)
-    async def solved(self, runtime: Runtime) -> float | dict[str, float]:
+    async def solved(self, runtime: Runtime) -> float:
         if not self.data.verifier.separate:
             await runtime.write(
                 "/tmp/tests.tgz", make_tar(Path(self.data.task_dir) / "tests")
@@ -554,43 +342,13 @@ async def solved(self, runtime: Runtime) -> float | dict[str, float]:
             "Harbor verifier test discovery",
         )
         await runtime.run(
-            [
-                "sh",
-                "-c",
-                "cd /tests && bash test.sh > /logs/verifier/test-stdout.txt 2>&1",
-            ],
+            ["sh", "-c", "cd /tests && bash test.sh"],
             verifier_env(self.data),
         )
-        # Harbor gives structured rewards precedence when both files exist.
-        try:
-            raw = (await runtime.read("/logs/verifier/reward.json")).decode()
-        except UnicodeDecodeError:
-            return 0.0
-        except (SandboxError, OSError):
-            raw = None
-        if raw is not None:
-            try:
-                values = json.loads(raw)
-                if not isinstance(values, dict):
-                    return 0.0
-                rewards = {
-                    str(key): float(value)
-                    for key, value in values.items()
-                    if isinstance(value, (int, float)) and not isinstance(value, bool)
-                }
-                return (
-                    rewards
-                    if len(rewards) == len(values)
-                    and all(math.isfinite(value) for value in rewards.values())
-                    else 0.0
-                )
-            except (TypeError, ValueError, OverflowError):
-                return 0.0
         try:
-            raw = (await runtime.read("/logs/verifier/reward.txt")).decode().strip()
-            value = float(raw)
-            return value if math.isfinite(value) else 0.0
-        except (SandboxError, OSError, ValueError, OverflowError):
+            reward = (await runtime.read("/logs/verifier/reward.txt")).decode().strip()
+            return float(reward or 0)
+        except (SandboxError, OSError, ValueError):
             return 0.0
 
 
@@ -724,85 +482,56 @@ def resolve_image(
     return None
 
 
-def size_to_mb(size: str | int | float) -> float:
-    """A Harbor size in MB, from either schema: current integer-MB fields or the
-    legacy schema-1.0 size strings ("8G", "512M", "64K")."""
-    if not isinstance(size, str):
-        return float(size)
-    scale = {"G": 1024.0, "M": 1.0, "K": 1 / 1024}.get(size.strip().upper()[-1:])
-    if scale is None:
-        raise ValueError(
-            f"invalid Harbor size {size!r}: expected a number of MB or a "
-            "'<number>[G|M|K]' string"
-        )
-    return float(size.strip()[:-1]) * scale
-
-
-def parse_resources(env: dict, multiplier: float = 1.0) -> TaskResources:
-    # Harbor's current schema reports memory/storage as integer-MB fields; the
-    # legacy schema 1.0 used size strings under `memory`/`storage`, which Harbor
-    # still migrates (datasets like senior-swe-bench are authored against it).
-    # TaskResources stores GB. A zero GPU count means no GPU request rather than
-    # the string "0".
-    memory = env.get("memory_mb") or env.get("memory")
-    disk = env.get("storage_mb") or env.get("storage")
+def parse_resources(env: "EnvironmentConfig", multiplier: float = 1.0) -> TaskResources:
+    """Convert Harbor's validated MB resources to Verifiers' GB resources."""
     return TaskResources(
-        cpu=env["cpus"] * multiplier if env.get("cpus") else None,
-        memory=size_to_mb(memory) / 1024 * multiplier if memory else None,
-        gpu=str(env["gpus"]) if env.get("gpus") else None,
-        disk=size_to_mb(disk) / 1024 * multiplier if disk else None,
+        cpu=env.cpus * multiplier if env.cpus else None,
+        memory=env.memory_mb / 1024 * multiplier if env.memory_mb else None,
+        gpu=str(env.gpus) if env.gpus else None,
+        disk=env.storage_mb / 1024 * multiplier if env.storage_mb else None,
     )
 
 
 def parse_verifier(
-    task_dir: Path, config: dict, resource_multiplier: float
-) -> HarborVerifier:
-    environment = config.get("environment", {})
-    verifier = config.get("verifier", {})
-    mode = verifier.get("environment_mode")
-    verifier_environment = verifier.get("environment")
-    if mode not in (None, "shared", "separate"):
-        raise ValueError(f"{task_dir.name}: invalid verifier environment_mode {mode!r}")
-    if mode == "shared" and verifier_environment is not None:
-        raise ValueError(
-            f"{task_dir.name}: [verifier.environment] is incompatible with "
-            "environment_mode='shared'"
-        )
+    task_dir: Path, config: "HarborTaskConfig", resource_multiplier: float
+) -> tuple[HarborVerifier, dict[str, str]]:
+    # Harbor is optional, so importing this module still works without the extra.
+    from harbor.models.task.config import NetworkMode, TaskOS
+    from harbor.models.task.verifier_mode import (
+        VerifierEnvironmentMode,
+        resolve_effective_verifier_env_config,
+        resolve_task_verifier_mode,
+    )
 
-    separate = mode == "separate" or verifier_environment is not None
-    if not separate:
-        return HarborVerifier(env=verifier.get("env", {}))
-    if verifier.get("collect"):
+    mode = resolve_task_verifier_mode(config)
+    if mode == VerifierEnvironmentMode.SHARED:
+        return HarborVerifier(), dict(config.verifier.env)
+    if config.verifier.collect:
         raise ValueError(
             f"{task_dir.name}: [[verifier.collect]] needs compose services, which "
             "Verifiers runtimes do not support"
         )
-    if verifier.get("user") is not None:
+    if config.verifier.user is not None:
         raise ValueError(
             f"{task_dir.name}: [verifier].user is not supported for separate runtimes"
         )
 
-    explicit_environment = verifier_environment is not None
-    verifier_environment = verifier_environment if explicit_environment else environment
-    if verifier_environment.get("os", "linux").lower() != "linux":
-        raise ValueError(
-            f"{task_dir.name}: separate Harbor verification supports Linux images only"
-        )
-    if explicit_environment and not verifier_environment.get("docker_image"):
+    environment = resolve_effective_verifier_env_config(config, None)
+    if environment is None:
+        raise ValueError(f"{task_dir.name}: separate verifier environment is missing")
+    if config.verifier.environment is None or not environment.docker_image:
         raise ValueError(
-            f"{task_dir.name}: [verifier.environment] needs a pullable docker_image; "
-            "building tests/Dockerfile is not supported"
+            f"{task_dir.name}: separate verification needs a pullable "
+            "[verifier.environment].docker_image"
         )
-    if not explicit_environment and (task_dir / "tests" / "Dockerfile").exists():
+    if environment.os != TaskOS.LINUX:
         raise ValueError(
-            f"{task_dir.name}: tests/Dockerfile needs a pullable "
-            "[verifier.environment].docker_image; building verifier Dockerfiles "
-            "is not supported"
+            f"{task_dir.name}: separate Harbor verification supports Linux images only"
         )
     unsupported = [
         field
         for field in ("healthcheck", "mcp_servers", "skills_dir", "gpu_types", "tpu")
-        if verifier_environment.get(field)
+        if getattr(environment, field)
     ]
     if unsupported:
         raise ValueError(
@@ -810,52 +539,29 @@ def parse_verifier(
             + ", ".join(unsupported)
         )
 
-    network_mode = verifier.get("network_mode")
-    if network_mode is None:
-        network_mode = verifier_environment.get("network_mode")
-    if network_mode is None:
-        network_mode = (
-            "public"
-            if verifier_environment.get("allow_internet", True)
-            else "no-network"
-        )
-    if network_mode == "allowlist":
-        raise ValueError(
-            f"{task_dir.name}: verifier network_mode='allowlist' is not supported"
-        )
-    if verifier.get("allowed_hosts") is not None or verifier_environment.get(
-        "allowed_hosts"
+    network_mode = config.verifier.network_mode or environment.network_mode
+    if (
+        network_mode == NetworkMode.ALLOWLIST
+        or config.verifier.allowed_hosts
+        or environment.allowed_hosts
     ):
         raise ValueError(
             f"{task_dir.name}: verifier network allowlists are not supported"
         )
-    if network_mode not in ("public", "no-network"):
-        raise ValueError(
-            f"{task_dir.name}: invalid verifier network_mode {network_mode!r}"
-        )
 
-    raw_artifacts = config.get("artifacts", [])
-    if not isinstance(raw_artifacts, list):
-        raise ValueError(f"{task_dir.name}: artifacts must be a list")
-    raw_artifacts = ["/logs/artifacts", *raw_artifacts]
     artifacts: list[HarborArtifact] = []
-    for raw in raw_artifacts:
-        artifact = {"source": raw} if isinstance(raw, str) else raw
-        if not isinstance(artifact, dict) or not isinstance(
-            artifact.get("source"), str
-        ):
-            raise ValueError(f"{task_dir.name}: invalid artifact entry {raw!r}")
-        if artifact.get("service") not in (None, "main"):
+    for artifact in ["/logs/artifacts", *config.artifacts]:
+        source = artifact if isinstance(artifact, str) else artifact.source
+        if not isinstance(artifact, str) and artifact.service not in (None, "main"):
             raise ValueError(f"{task_dir.name}: sidecar artifacts are not supported")
-        if artifact.get("destination"):
+        if not isinstance(artifact, str) and artifact.destination:
             raise ValueError(
                 f"{task_dir.name}: artifact destinations are not supported"
             )
-        if artifact.get("exclude"):
+        if not isinstance(artifact, str) and artifact.exclude:
             raise ValueError(
                 f"{task_dir.name}: artifact exclude patterns are not supported"
             )
-        source = artifact["source"]
         if not source.startswith("/") or ".." in PurePosixPath(source).parts:
             raise ValueError(
                 f"{task_dir.name}: artifact source must be an absolute non-root path, "
@@ -885,22 +591,25 @@ def parse_verifier(
             continue
         artifacts.append(HarborArtifact(source=source))
 
-    env = verifier_environment.get("env", {}) | verifier.get("env", {})
-    return HarborVerifier(
-        separate=True,
-        image=verifier_environment.get("docker_image"),
-        workdir=verifier_environment.get("workdir"),
-        resources=parse_resources(verifier_environment, resource_multiplier),
-        network_access=network_mode == "public",
-        env=env,
-        artifacts=artifacts,
-        upload_tests=not explicit_environment,
+    return (
+        HarborVerifier(
+            separate=True,
+            image=environment.docker_image,
+            workdir=environment.workdir,
+            resources=parse_resources(environment, resource_multiplier),
+            network_access=network_mode == NetworkMode.PUBLIC,
+            artifacts=artifacts,
+        ),
+        dict(environment.env) | dict(config.verifier.env),
     )
 
 
 def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborData:
+    from harbor.models.task.config import TaskConfig as HarborTaskConfig
+
     config = tomllib.loads((task_dir / "task.toml").read_text())
-    if config.get("steps"):
+    parsed = HarborTaskConfig.model_validate(config)
+    if parsed.steps:
         raise ValueError(f"{task_dir.name}: Harbor multi-step tasks are not supported")
     task, meta = config.get("task", {}), config.get("metadata", {})
     authors = [Author(**a) for a in task.get("authors", [])]
@@ -909,6 +618,9 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD
         authors = [Author(name=meta["author_name"], email=meta.get("author_email"))]
     harness_timeout = config.get("agent", {}).get("timeout_sec")
     scoring_timeout = config.get("verifier", {}).get("timeout_sec")
+    verifier, verifier_environment = parse_verifier(
+        task_dir, parsed, harbor_config.resource_multiplier
+    )
     return HarborData(
         idx=idx,
         name=task.get("name") or task_dir.name,
@@ -920,7 +632,7 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD
             harbor_config.require_image,
             harbor_config.ignore_dockerfile,
         ),
-        workdir=config.get("environment", {}).get("workdir"),
+        workdir=parsed.environment.workdir,
         timeout=TaskTimeout(
             harness=harness_timeout * harbor_config.timeout_multiplier
             if harness_timeout is not None
@@ -930,7 +642,7 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD
             else None,
         ),
         resources=parse_resources(
-            config.get("environment", {}), harbor_config.resource_multiplier
+            parsed.environment, harbor_config.resource_multiplier
         ),
         keywords=task.get("keywords", []),
         authors=authors,
@@ -938,20 +650,21 @@ def parse_task(task_dir: Path, idx: int, harbor_config: HarborConfig) -> HarborD
         category=meta.get("category"),
         tags=meta.get("tags", []),
         task_dir=str(task_dir),
-        verifier=parse_verifier(task_dir, config, harbor_config.resource_multiplier),
+        verifier_env=verifier_environment,
+        verifier=verifier,
     )
 
 
 def verifier_env(task: HarborData) -> dict[str, str]:
     """Resolve templates at scoring time so host secrets are never serialized."""
-    if not task.verifier.env:
+    if not task.verifier_env:
         return {}
 
     # Harbor is an optional dependency, so importing this module must still work
     # for users who do not install the Harbor extra.
     from harbor.utils.env import resolve_env_vars
 
-    return resolve_env_vars(task.verifier.env)
+    return resolve_env_vars(task.verifier_env)
 
 
 # Downloaded test directories are immutable. Cache only the latest archive to

From f0a88abbd88ac4cf853aa123850780acba39e368 Mon Sep 17 00:00:00 2001
From: Xeophon <46377542+xeophon@users.noreply.github.com>
Date: Fri, 17 Jul 2026 15:25:22 +0200
Subject: [PATCH 5/6] Confirm teardown before separate Harbor grading

---
 verifiers/v1/runtimes/base.py           |  6 ++++++
 verifiers/v1/runtimes/docker.py         | 19 +++++++++++++++++++
 verifiers/v1/runtimes/prime.py          | 18 ++++++++++++++++++
 verifiers/v1/tasksets/harbor/taskset.py |  2 +-
 4 files changed, 44 insertions(+), 1 deletion(-)

diff --git a/verifiers/v1/runtimes/base.py b/verifiers/v1/runtimes/base.py
index c5cf456fd0..45d7054e14 100644
--- a/verifiers/v1/runtimes/base.py
+++ b/verifiers/v1/runtimes/base.py
@@ -129,6 +129,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.
diff --git a/verifiers/v1/runtimes/docker.py b/verifiers/v1/runtimes/docker.py
index f070774123..670f9b5a1c 100644
--- a/verifiers/v1/runtimes/docker.py
+++ b/verifiers/v1/runtimes/docker.py
@@ -231,6 +231,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
diff --git a/verifiers/v1/runtimes/prime.py b/verifiers/v1/runtimes/prime.py
index 298cfecdf2..9ded585176 100644
--- a/verifiers/v1/runtimes/prime.py
+++ b/verifiers/v1/runtimes/prime.py
@@ -327,6 +327,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.
diff --git a/verifiers/v1/tasksets/harbor/taskset.py b/verifiers/v1/tasksets/harbor/taskset.py
index d6649dc002..e2d727bc74 100644
--- a/verifiers/v1/tasksets/harbor/taskset.py
+++ b/verifiers/v1/tasksets/harbor/taskset.py
@@ -171,7 +171,7 @@ async def score(
         with tempfile.TemporaryDirectory() as temp:
             files, directories = await self._collect_artifacts(runtime, Path(temp))
             # No verifier process starts until the agent runtime is gone.
-            await runtime.stop()
+            await runtime.stop_confirmed()
             target = make_runtime(
                 scoring_runtime_config, name=f"{runtime.name}-verifier"
             )

From 40580d2ab5b7b2f61b91747855ea8dd60f1d91a2 Mon Sep 17 00:00:00 2001
From: Xeophon <46377542+xeophon@users.noreply.github.com>
Date: Fri, 17 Jul 2026 15:35:48 +0200
Subject: [PATCH 6/6] Preserve Harbor file artifact shape

---
 verifiers/v1/tasksets/harbor/taskset.py | 17 ++++++++++-------
 1 file changed, 10 insertions(+), 7 deletions(-)

diff --git a/verifiers/v1/tasksets/harbor/taskset.py b/verifiers/v1/tasksets/harbor/taskset.py
index e2d727bc74..9b1cd1efcd 100644
--- a/verifiers/v1/tasksets/harbor/taskset.py
+++ b/verifiers/v1/tasksets/harbor/taskset.py
@@ -313,13 +313,16 @@ async def _prepare_verifier(
                 {},
                 "Harbor artifact target cleanup",
             )
-        directories = list(dict.fromkeys(["/logs/artifacts", *directories]))
-        await run_checked(
-            runtime,
-            ["mkdir", "-p", *directories],
-            {},
-            "Harbor artifact directory restore",
-        )
+        if all(path != "/logs/artifacts" for path, _ in files):
+            directories.insert(0, "/logs/artifacts")
+        directories = list(dict.fromkeys(directories))
+        if directories:
+            await run_checked(
+                runtime,
+                ["mkdir", "-p", *directories],
+                {},
+                "Harbor artifact directory restore",
+            )
         for path, host_path in files:
             await runtime.write(path, await asyncio.to_thread(host_path.read_bytes))