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

docs/v1/env.md

@@ -53,6 +53,25 @@ Just like tasksets and harnesses, an `Env` can be user-defined for full expressi
 | `best-of-n` | `agent` | `n` independent attempts per episode; its metrics mark the argmax-reward sibling (`best`) and whether any reached `--env.threshold` (`pass_at_n`) — rejection sampling and pass@k. |
 | `agentic-judge` | `solver`, `judge` | the solver plays the task; a code-executing judge agent verifies the finished attempt with real execution. |
 
+## Grading artifacts
+
+Use artifacts to carry files between runtimes. Files written to
+`/logs/artifacts/` are collected implicitly; declare other paths on the task data:
+
+```python
+class MyData(vf.TaskData):
+    artifacts: list[vf.Artifact] = [
+        vf.Artifact(source="/work/report", exclude=[".git"])
+    ]
+
+
+class MyTask(vf.Task[MyData]):
+    async def finalize(self, trace: vf.Trace, runtime: vf.Runtime) -> None:
+        trace.state.artifacts = await vf.collect(runtime, self.data.artifacts)
+```
+
+Declared paths must exist when collected. The implicit directory is optional.
+
 ## Concurrency
 
 Write independent agents as independent (`asyncio.gather`, a `TaskGroup`) — how many actually run at once is the run's call, not the env's. Two knobs bound it, and the **episode is the unit** at the outer one:

docs/v1/harbor.md

@@ -101,10 +101,13 @@ routes take precedence over deny rules, while ordinary Prime deny rules are appl
 unchanged and may block a matching route. Restricted Harbor tasks require Docker or a
 Prime VM; Prime accepts host-level entries.
 
+## 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).
+
 ## 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))
 - Multi-step tasks ([Harbor Docs](https://www.harborframework.com/docs/tasks/multi-step))

tests/v1/test_e2e.py

@@ -335,6 +335,7 @@ async def test_agentic(run_v1, harness, harness_runtime, tmp_path):
         runtime={"type": harness_runtime},
         output_dir=tmp_path,
         max_turns=10,
+        max_tokens=8192,
     )
     assert trace.ok
     assert trace.num_turns >= 1  # ran a command, then finished
@@ -405,18 +406,18 @@ async def test_env_id_agentic_judge(run_v1, tmp_path):
     policy.write_text("Check EMPIRICALLY that the agent echoed the word back.")
     traces = await run_v1(
         "echo-v1",
-        harness=None,  # seats pin their own harness; there is no run-level one
+        harness=None,
         env={
             "id": "agentic-judge",
-            # The solver owns the shared box, so the container is pinned here.
             "solver": {"harness": {"id": "bash"}, "runtime": {"type": "docker"}},
-            # The judge reads the trace and reasons before it writes the
-            # verdict file; the shared 2048-token run cap truncates it mid-audit.
             "judge": {
                 "harness": {"id": "bash"},
                 "max_output_tokens": 8192,
             },
-            "task": {"prompt": str(policy)},
+            "task": {
+                "prompt": {"path": str(policy)},
+                "hint": "Do not rely on README.md",
+            },
             "score": {"task_weight": 0.5},
         },
         output_dir=tmp_path,
@@ -431,7 +432,7 @@ async def test_env_id_agentic_judge(run_v1, tmp_path):
     # The task's own reward keeps its raw score; the rescale lands on the weight.
     assert solver.rewards["echoed"].score == 1.0
     assert solver.rewards["echoed"].weight == 0.5
-    assert isinstance(judge.info.get("verdict"), dict)  # scraped off the box
+    assert isinstance(judge.info.get("verdict"), dict)
     assert 0.0 <= solver.rewards["judge"].score <= 1.0
 
 

verifiers/v1/__init__.py

@@ -4,6 +4,12 @@
 
 from verifiers.v1.acp import ACP
 from verifiers.v1.agent import Agent, Agents, Interaction, Segment, make_agent
+from verifiers.v1.artifacts import (
+    ARTIFACTS_DIR,
+    Artifact,
+    collect,
+    restore,
+)
 from verifiers.v1.clients import (
     BaseClientConfig,
     Client,
@@ -297,6 +303,11 @@
     "PATCH_CAP_BYTES",
     "capture_patch",
     "resolve_head",
+    # grading artifacts
+    "ARTIFACTS_DIR",
+    "Artifact",
+    "collect",
+    "restore",
     # scoring
     "compare_stdout_results",
     "extract_boxed_answer",

verifiers/v1/artifacts.py

@@ -0,0 +1,144 @@
+"""Artifact collection and restoration across runtimes."""
+
+from __future__ import annotations
+
+import logging
+import shlex
+import uuid
+from pathlib import PurePosixPath
+from typing import TYPE_CHECKING
+
+from pydantic import Field
+
+from verifiers.v1.types import StrictBaseModel
+
+if TYPE_CHECKING:
+    from verifiers.v1.runtimes import Runtime
+
+logger = logging.getLogger(__name__)
+
+ARTIFACTS_DIR = "/logs/artifacts"
+"""Implicit artifact directory; tasks that write here need 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 restore at the same location in another runtime."""
+
+    source: str
+    exclude: list[str] = Field(default_factory=list)
+    """`tar --exclude` patterns, applied when `source` is a directory."""
+
+
+async def collect(
+    runtime: Runtime, artifacts: list[Artifact] | None = None
+) -> dict[str, bytes]:
+    """Tar the convention dir and every declared path out of `runtime`.
+
+    Keyed by source path; the values are tar archives. Insertion order is the order
+    they were declared, and a path cannot be collected twice.
+
+    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.
+
+    Each source is archived separately so its exclude patterns stay local.
+    """
+    # Resolve relative sources against the runtime workdir. Joining also normalises
+    # `/work/` to `/work`, so one tree cannot key two entries (the source is both the
+    # dict key and `restore`'s rm -rf target).
+    workdir = PurePosixPath(getattr(runtime.config, "workdir", "") or "/")
+    declared = [
+        a.model_copy(update={"source": str(workdir / a.source)})
+        for a in artifacts or []
+    ]
+    convention = PurePosixPath(ARTIFACTS_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=ARTIFACTS_DIR)] if sweep else []) + declared
+
+    collected: dict[str, bytes] = {}
+    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 == ARTIFACTS_DIR:
+                continue
+            raise RuntimeError(
+                f"declared artifact {source!r} does not exist in the runtime"
+            )
+        archive = await _tar_out(runtime, artifact, budget)
+        budget -= len(archive)
+        collected[source] = archive
+
+    logger.debug("collected artifact roots: %s", list(collected))
+    return collected
+
+
+async def restore(runtime: Runtime, collected: dict[str, bytes]) -> None:
+    """Extract `collected` in `runtime` at the original absolute paths."""
+    if not collected:
+        return
+    # Restoring into the subprocess runtime would extract absolute paths onto the
+    # developer's filesystem, so refuse it before any archive reaches the host.
+    if getattr(runtime.config, "type", None) == "subprocess":
+        raise RuntimeError(
+            "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(root) for root in collected)
+    await _run(runtime, f"rm -rf -- {roots}", "clear artifact roots")
+    for root, archive in collected.items():
+        path = f"/tmp/vf-artifact-{uuid.uuid4().hex}.tar"
+        await runtime.write(path, archive)
+        await _run(
+            runtime,
+            f"tar -xf {shlex.quote(path)} -C / && rm -f {shlex.quote(path)}",
+            f"restore artifact {root!r}",
+        )
+
+
+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 RuntimeError(
+                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:
+            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 RuntimeError(f"failed to {action}: {detail}")

verifiers/v1/envs/agentic_judge/__init__.py

@@ -4,6 +4,7 @@
     Criterion,
     JudgeTaskConfig,
     ScoreConfig,
+    TextFile,
 )
 
 __all__ = [
@@ -12,4 +13,5 @@
     "Criterion",
     "JudgeTaskConfig",
     "ScoreConfig",
+    "TextFile",
 ]