feat(v1): grade in an isolated box, with Harbor-native artifacts

docs/v1/harbor.md

@@ -87,10 +87,21 @@ added automatically; evaluator-provided `allow` entries add exceptions and `bloc
 entries can narrow them. Restricted Harbor tasks require Docker or a Prime VM; Prime
 accepts host-level entries and rejects combinations that need both policy modes.
 
+## Artifacts and collect hooks
+
+`artifacts = [...]` and `[[verifier.collect]]` are read from `task.toml` ([Harbor Docs](https://www.harborframework.com/docs/run-jobs/results-and-artifacts)). Collect hooks run in the agent's box from the task's `finalize`, which is Harbor's own ordering — after the agent phase, before collection — and declared paths plus the `/logs/artifacts/` convention dir are then carried into the grading box and restored at their original paths ("no translation", as in Harbor).
+
+Two deliberate differences from `harbor run`:
+
+- **A failing collect hook fails the rollout.** Harbor logs it and carries on, because there the output is observability; here it is a grading input, and a silently absent file makes the verifier score a stale state.
+- **`destination` has no effect.** It positions a file in Harbor's host trial directory; verifiers has no trial directory (the trace is the record), and Harbor never lets `destination` affect verifier-side placement.
+
+Sidecar `service` entries, `[verifier].user`, and an explicit `[verifier.environment]` image are rejected at load. The grading box is built from the task's own image, so only the agent's delta has to travel; a different verifier image would need the whole working tree copied across.
+
 ## Shortcomings
 
 verifiers does not have parity with Harbor yet, so some features are missing and currently being worked on. The most notable missing features right now are:
 
 - Switching to a different verifier-phase network policy ([Harbor Docs](https://www.harborframework.com/docs/tasks/network-policy))
-- Shared & separate verifiers ([Harbor Docs](https://www.harborframework.com/docs/tasks#verifier-environment-shared-vs-separate))
+- Sidecar services, and the sidecar artifacts and collect hooks that go with them ([Harbor Docs](https://www.harborframework.com/docs/tasks#sidecar-artifacts-and-collect-hooks))
 - Multi-step tasks ([Harbor Docs](https://www.harborframework.com/docs/tasks/multi-step))

docs/v1/tasksets.md

@@ -224,6 +224,37 @@ class JudgeTraceTaskset(vf.Taskset[JudgedTask, SetConfig]):
 
 To override the judge model, set `env.taskset.task.judge.model` in your config (it is a string).
 
+## Grading artifacts
+
+An environment can grade in a second box rather than the one the agent worked in, so nothing the agent did to its environment can reach the grader. Only what the task declares crosses over.
+
+Two channels, and they do different jobs:
+
+- **`trace.info` is the record.** `capture_patch` puts the diff in `trace.info["patch"]`, a judge puts its verdict there, and both ride `traces.jsonl`. It never travels to another box.
+- **`/logs/artifacts/` is transport.** Anything written there is collected with no declaration at all, carried to the host, and restored in the grading box at the same path.
+
+Produce artifacts in `finalize`, while the runtime is live and before scoring mutates anything:
+
+```python
+class MySweTask(vf.Task[MyData]):
+    async def finalize(self, trace: vf.Trace, runtime: vf.Runtime) -> None:
+        await vf.capture_patch(
+            trace, runtime, self.data.base_commit,
+            write_path=f"{vf.CONVENTION_DIR}/patch.diff",   # record + transport, one call
+        )
+```
+
+Declare paths outside the convention dir on the task row:
+
+```python
+class MyData(vf.TaskData):
+    artifacts: list[vf.Artifact] = [vf.Artifact(source="/work/report", exclude=[".git"])]
+```
+
+A declared path that is missing at collection time fails the rollout: it was declared because grading needs it, and grading a partial state scores the rollout wrong rather than loudly failing it. The convention dir is exempt — it is collected for every task, and most never write to it.
+
+The grading box boots from the same image as the agent's, so the repo and its dependencies are already present. Only the agent's delta has to travel, which is why the collection cap (`vf.artifacts.MAX_ARTIFACT_BYTES`) is sized for a patch rather than a tree.
+
 ## Beyond one agent
 
 One episode doesn't have to be one agent run: agents, the control flow between agents, and cross-agent rewards are the environment's job — see [The Env](env.md).

verifiers/v1/__init__.py

@@ -25,6 +25,7 @@
 from verifiers.v1.env import Env
 from verifiers.v1.envs.single_agent import SingleAgentEnv, SingleAgentEnvConfig
 from verifiers.v1.errors import (
+    ArtifactError,
     EnvError,
     HarnessError,
     InterceptionError,
@@ -89,6 +90,14 @@
 )
 from verifiers.v1.state import State, StateT
 from verifiers.v1.configs.task import TaskConfig
+from verifiers.v1.artifacts import (
+    CONVENTION_DIR,
+    Artifact,
+    CollectedArtifact,
+    collect,
+    release,
+    restore,
+)
 from verifiers.v1.task import Task, TaskData, TaskResources, TaskTimeout, WireTaskData
 from verifiers.v1.configs.taskset import TasksetConfig
 from verifiers.v1.taskset import Taskset
@@ -206,6 +215,7 @@
     "ToolsetError",
     "SandboxError",
     "TaskError",
+    "ArtifactError",
     "InterceptionError",
     "TunnelError",
     # clients
@@ -279,6 +289,13 @@
     "PATCH_CAP_BYTES",
     "capture_patch",
     "resolve_head",
+    # grading artifacts
+    "CONVENTION_DIR",
+    "Artifact",
+    "CollectedArtifact",
+    "collect",
+    "release",
+    "restore",
     # scoring
     "compare_stdout_results",
     "extract_boxed_answer",

verifiers/v1/artifacts.py

@@ -0,0 +1,198 @@
+"""Carry a task's declared files out of the agent's box and into a grading box.
+
+Grading in the box the agent worked in leaves a seam a policy under RL pressure will
+find. Grading in a second box removes it — only what the task declares crosses over.
+
+Two channels, non-overlapping: `Trace.info` is the durable record (`capture_patch` puts
+the diff there, a judge puts its verdict there) and never travels; `/logs/artifacts/` is
+transport, Harbor's in-sandbox convention, collected with no declaration and restored at
+the same path in the grading box ("no translation", as in Harbor).
+
+`collect` runs while the agent's box is alive, right after `Task.finalize` produced the
+files. It is the barrier: once it returns the box can be torn down in the background.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import shlex
+import uuid
+from dataclasses import dataclass
+from pathlib import PurePosixPath
+from typing import TYPE_CHECKING
+
+from verifiers.v1.errors import ArtifactError
+from verifiers.v1.types import StrictBaseModel
+
+if TYPE_CHECKING:
+    from verifiers.v1.runtimes import Runtime
+
+logger = logging.getLogger(__name__)
+
+CONVENTION_DIR = "/logs/artifacts"
+"""Harbor's in-sandbox publish directory, swept implicitly so a task that writes here
+needs no declaration."""
+
+MAX_ARTIFACT_BYTES = 32 * 1024 * 1024
+"""Ceiling per collection. Sized for a delta, not a tree: the grading box boots from the
+agent's image, so the repo is already there and only its output has to travel."""
+
+
+class Artifact(StrictBaseModel):
+    """One path to carry into the grading box, where it lands at this same path.
+
+    Harbor's `ArtifactConfig` minus `destination` (host trial-directory placement, which
+    verifiers has no equivalent for) and `service` (compose sidecars, unsupported).
+    """
+
+    source: str
+    exclude: list[str] = []
+    """`tar --exclude` patterns, applied when `source` is a directory."""
+
+
+@dataclass(frozen=True)
+class CollectedArtifact:
+    """One source, tarred out of the agent's box and held on the host in between.
+    Transient: only `Trace.info` is durable."""
+
+    root: str
+    archive: bytes
+
+
+async def collect(
+    runtime: Runtime, artifacts: list[Artifact] | None = None
+) -> list[CollectedArtifact]:
+    """Tar the convention dir and every declared path out of `runtime`.
+
+    A declared source that is missing raises: it was declared because grading needs it,
+    and grading a partial state scores the rollout wrong rather than failing it. The
+    implicit convention sweep is exempt — most tasks never write there.
+
+    One archive per source: BusyBox `tar` (every alpine-based image) implements only
+    `c`/`x`/`t` with no `-r` to append, and each source carries its own excludes anyway.
+    """
+    declared = [_checked(a) for a in artifacts or []]
+    convention = PurePosixPath(CONVENTION_DIR)
+    sweep = not any(
+        (p := PurePosixPath(a.source)) == convention
+        or p.is_relative_to(convention)
+        or convention.is_relative_to(p)
+        for a in declared
+    )
+    entries = ([Artifact(source=CONVENTION_DIR)] if sweep else []) + declared
+
+    collected: list[CollectedArtifact] = []
+    budget = MAX_ARTIFACT_BYTES
+    for artifact in entries:
+        source = artifact.source
+        if (await runtime.run(["test", "-e", source], {})).exit_code != 0:
+            if sweep and source == CONVENTION_DIR:
+                continue
+            raise ArtifactError(
+                f"declared artifact {source!r} does not exist in the box; the task must "
+                "produce it in finalize() (or a [[verifier.collect]] hook)"
+            )
+        archive = await _tar_out(runtime, artifact, budget)
+        budget -= len(archive)
+        collected.append(CollectedArtifact(root=source, archive=archive))
+
+    logger.debug("collected artifact roots: %s", [c.root for c in collected])
+    return collected
+
+
+async def restore(runtime: Runtime, collected: list[CollectedArtifact]) -> None:
+    """Extract `collected` in `runtime` at the original absolute paths."""
+    if not collected:
+        return
+    # Extraction writes to absolute paths. In a container that is the point; under the
+    # subprocess runtime it is the developer's own filesystem.
+    if getattr(runtime.config, "type", None) == "subprocess":
+        raise ArtifactError(
+            "refusing to restore artifacts into the subprocess runtime: extraction "
+            "writes to absolute paths on the host. Grade in a container."
+        )
+    # Clear every root up front, not per entry: a later nested root would otherwise
+    # delete content an earlier one just restored. Clearing also drops any file or
+    # symlink the image left at the target.
+    roots = " ".join(shlex.quote(entry.root) for entry in collected)
+    await _run(runtime, f"rm -rf -- {roots}", "clear artifact roots")
+    for entry in collected:
+        path = f"/tmp/vf-artifact-{uuid.uuid4().hex}.tar"
+        await runtime.write(path, entry.archive)
+        await _run(
+            runtime,
+            f"tar -xf {shlex.quote(path)} -C / && rm -f {shlex.quote(path)}",
+            f"restore artifact {entry.root!r}",
+        )
+
+
+def release(runtime: Runtime) -> None:
+    """Begin `runtime`'s teardown and return without waiting for it.
+
+    Once `collect` has returned nothing needs the agent's box, and on a remote runtime
+    its teardown is an API round trip the grading box should not wait behind. Safe
+    inside a `provision()` block only if the caller detached it (`pop_all()`); otherwise
+    the context manager awaits a teardown anyway and nothing is gained.
+    """
+    if runtime.stopped:
+        return
+    # Mark it synchronously, as `Runtime.stop` does: `create_task` only schedules, so
+    # otherwise a second call would start a second teardown.
+    runtime.stopped = True
+    task = asyncio.create_task(runtime.stop())
+    # asyncio holds only a weak reference to a running task, so a fire-and-forget
+    # teardown can be collected mid-flight.
+    _PENDING.add(task)
+    task.add_done_callback(_PENDING.discard)
+
+
+_PENDING: set[asyncio.Task] = set()
+
+
+def _checked(artifact: Artifact) -> Artifact:
+    path = PurePosixPath(artifact.source)
+    if not path.is_absolute() or ".." in path.parts:
+        raise ArtifactError(
+            f"artifact source {artifact.source!r} must be an absolute path with no '..'"
+        )
+    resolved = path.as_posix().rstrip("/")
+    if not resolved:
+        raise ArtifactError("artifact source '/' would sweep the whole image")
+    return artifact.model_copy(update={"source": resolved})
+
+
+async def _tar_out(runtime: Runtime, artifact: Artifact, budget: int) -> bytes:
+    path = f"/tmp/vf-artifact-{uuid.uuid4().hex}.tar"
+    excludes = " ".join(f"--exclude={shlex.quote(p)}" for p in artifact.exclude)
+    try:
+        await _run(
+            runtime,
+            f"tar -cf {shlex.quote(path)} -C / {excludes} -- "
+            f"{shlex.quote(artifact.source.lstrip('/'))}",
+            f"collect artifact {artifact.source!r}",
+        )
+        # Size it in the box: an oversized collection is refused before it reaches host
+        # memory, not after.
+        sized = await runtime.run(["sh", "-c", f"wc -c < {shlex.quote(path)}"], {})
+        if (raw := sized.stdout.strip()).isdigit() and int(raw) > budget:
+            raise ArtifactError(
+                f"artifact {artifact.source!r} takes the collection over the "
+                f"{MAX_ARTIFACT_BYTES} byte limit. The grading box boots from the "
+                "agent's image, so only the delta needs to travel — narrow the source "
+                "or add `exclude` patterns."
+            )
+        return await runtime.read(path)
+    finally:
+        # Best-effort: the box is about to be destroyed and the name is unique per call.
+        try:
+            await runtime.run(["rm", "-f", path], {})
+        except Exception:  # noqa: BLE001 - cleanup must not mask a collection error.
+            logger.debug("failed to remove %s", path, exc_info=True)
+
+
+async def _run(runtime: Runtime, command: str, action: str) -> None:
+    result = await runtime.run(["sh", "-c", command], {})
+    if result.exit_code:
+        detail = (result.stderr or result.stdout).strip()[-500:]
+        raise ArtifactError(f"failed to {action}: {detail}")