diff --git a/tests/v1/fixtures/echo_agentic_v1.py b/tests/v1/fixtures/echo_agentic_v1.py index 2f43fd5326..e61f1c80cd 100644 --- a/tests/v1/fixtures/echo_agentic_v1.py +++ b/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: diff --git a/verifiers/v1/agent.py b/verifiers/v1/agent.py index 14c2c54e31..5e6a212ed9 100644 --- a/verifiers/v1/agent.py +++ b/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,6 +642,7 @@ 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( @@ -633,6 +650,7 @@ async def run( 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 diff --git a/verifiers/v1/envs/agentic_judge/env.py b/verifiers/v1/envs/agentic_judge/env.py index e16bfe4f8d..e78db97e70 100644 --- a/verifiers/v1/envs/agentic_judge/env.py +++ b/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( diff --git a/verifiers/v1/rollout.py b/verifiers/v1/rollout.py index 76c0550d2c..9576a84d01 100644 --- a/verifiers/v1/rollout.py +++ b/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