fix(v1): collect isolated judge artifacts in owned rollouts

tests/v1/fixtures/echo_agentic_v1.py

@@ -31,13 +31,6 @@ class EchoAgenticData(vf.TaskData):
 
 
 class EchoAgenticTask(vf.Task[EchoAgenticData]):
-    async def finalize(self, trace: vf.Trace, runtime: Runtime) -> None:
-        # Subprocess uses a runtime-owned temporary cwd, not a restorable absolute
-        # path; isolated agentic judging requires a container anyway.
-        if runtime.type == "subprocess":
-            return
-        trace.state.artifacts = await vf.collect(runtime, self.data.artifacts)
-
     @vf.reward(weight=1.0)
     async def wrote_phrase(self, runtime: Runtime) -> float:
         try:

verifiers/v1/agent.py

@@ -362,21 +362,26 @@ async def run(
         runtime: Runtime | None = None,
         tools: Mapping[str, SharedToolServer] | None = None,
         on_trace: Callable[[Trace], None] | None = None,
+        collect_artifacts: bool = False,
     ) -> Trace:
         """Run this agent on `task` once and return the trace: one segment — the
         program runs on the task's prompt until it exits (a multi-turn exchange
         is `interaction()`). `runtime` places it into a live borrowed box instead of
         provisioning one; `tools` are live servers borrowed from their
         owner, counted in the pairing check; `on_trace` observes the trace the
-        moment it's minted, before any I/O. Retries whole while the trace ends
-        with a retryable error (`config.retries`) — never into a borrowed box;
-        the final trace keeps earlier attempts' errors."""
+        moment it's minted, before any I/O. `collect_artifacts` captures the task's
+        declared artifacts after its finalizer while its container runtime is still
+        alive. Retries whole while the trace ends with a retryable error
+        (`config.retries`) — never into a borrowed box; the final trace keeps earlier
+        attempts' errors."""
         if self._closed:
             raise RuntimeError("Agent is closed; create a new agent")
         retry = self.config.retries
         history: list = []
         for attempt in range(retry.max_retries + 1):
-            trace = await self._run_once(task, runtime, tools, on_trace)
+            trace = await self._run_once(
+                task, runtime, tools, on_trace, collect_artifacts
+            )
             if attempt == retry.max_retries or not trace_should_retry(trace, retry):
                 break
             if runtime is not None:
@@ -407,9 +412,20 @@ async def _run_once(
         runtime: Runtime | None,
         shared_tools: Mapping[str, SharedToolServer] | None,
         on_trace: Callable[[Trace], None] | None,
+        collect_artifacts: bool,
     ) -> Trace:
         params = self._rollout_params(task, runtime, dict(shared_tools or {}))
-        run = Rollout(task=task, on_trace=on_trace, **params)
+        if collect_artifacts and isinstance(params["runtime_config"], SubprocessConfig):
+            raise TypeError(
+                "artifact collection requires a container runtime; subprocess "
+                "artifacts live in a host-only temporary working directory"
+            )
+        run = Rollout(
+            task=task,
+            on_trace=on_trace,
+            collect_artifacts=collect_artifacts,
+            **params,
+        )
         try:
             if await run.open():
                 await run.step()
@@ -626,13 +642,15 @@ async def run(
         runtime: Runtime | None = None,
         tools: Mapping[str, SharedToolServer] | None = None,
         on_trace: Callable[[Trace], None] | None = None,
+        collect_artifacts: bool = False,
     ) -> Trace:
         async with self._gate or nullcontext():
             trace = await super().run(
                 task,
                 runtime=runtime,
                 tools=tools if tools is not None else self._shared_for(task),
                 on_trace=self._watch(on_trace),
+                collect_artifacts=collect_artifacts,
             )
         self._completed.append(trace)
         return trace

verifiers/v1/envs/agentic_judge/env.py

@@ -342,7 +342,7 @@ class IsolatedAgenticJudgeEnv(AgenticJudgeEnv):
     """Judge only collected artifacts in a fresh box with the solver's policy."""
 
     async def run(self, task: vf.Task, agents: vf.Agents) -> None:
-        solution = await agents.solver.run(task)
+        solution = await agents.solver.run(task, collect_artifacts=True)
         if not solution.ok:
             raise RuntimeError("the solver's rollout failed, so the judge never ran")
         await agents.judge.run(

verifiers/v1/rollout.py

@@ -32,6 +32,7 @@
 from verifiers.v1.task import Task
 from verifiers.v1.trace import AgentInfo, Trace, TraceTask
 from verifiers.v1.types import Messages, Request, Response, SystemMessage, UserMessage
+from verifiers.v1.utils.artifacts import collect
 from verifiers.v1.utils.decorators import discover_decorated, invoke
 
 logger = logging.getLogger(__name__)
@@ -69,6 +70,7 @@ def __init__(
         interception: Interception | None = None,
         runtime: Runtime | None = None,
         on_trace: Callable[[Trace], None] | None = None,
+        collect_artifacts: bool = False,
     ) -> None:
         self.task = task
         self.harness = harness
@@ -81,6 +83,7 @@ def __init__(
         self._interception = interception
         self.runtime = runtime
         self._borrowed_runtime = runtime
+        self._collect_artifacts = collect_artifacts
         self.trace: Trace = Trace(
             task=TraceTask(
                 type=type(task).__name__,
@@ -466,12 +469,14 @@ async def close(self) -> Trace:
             if not self._failed and self._opened:
                 trace.timing.finalize.start = time.time()
                 async with boundary(TaskError, "task finalize"):
-                    await asyncio.wait_for(
-                        invoke(
+                    async with asyncio.timeout(self._timeouts.finalize):
+                        await invoke(
                             self.task.finalize, {"trace": trace, "runtime": runtime}
-                        ),
-                        self._timeouts.finalize,
-                    )
+                        )
+                        if self._collect_artifacts and not trace.state.artifacts:
+                            trace.state.artifacts = await collect(
+                                runtime, self.task.data.artifacts
+                            )
                 now = time.time()
                 trace.timing.finalize.end = now
                 trace.timing.scoring.start = now