diff --git a/verifiers/envs/experimental/cli_agent_env.py b/verifiers/envs/experimental/cli_agent_env.py index 1db7005204..f9226d0ce1 100644 --- a/verifiers/envs/experimental/cli_agent_env.py +++ b/verifiers/envs/experimental/cli_agent_env.py @@ -3,6 +3,7 @@ import math import time import uuid +from collections import Counter from typing import Any, cast from prime_sandboxes import ( @@ -17,24 +18,31 @@ from verifiers.clients import Client from verifiers.envs.experimental.sandbox_mixin import SandboxMixin, SandboxMonitorRubric from verifiers.types import ( + AssistantMessage, Messages, MessageType, Response, SamplingArgs, State, Tool, + ToolCall, ) from verifiers.utils.interception_utils import ( InterceptionServer, deliver_response, synthesize_stream, ) +from verifiers.utils.logging_utils import print_time, truncate from verifiers.utils.message_utils import normalize_messages from verifiers.utils.worker_utils import get_free_port logger = logging.getLogger(__name__) +class AgentError(vf.InfraError): + """Raised when the agent process fails or exits unexpectedly.""" + + class CliAgentMonitorRubric(vf.Rubric): """Monitor rubric that tracks CLI agent execution state.""" @@ -90,6 +98,7 @@ def __init__( sandbox_client_max_connections: int = 100, sandbox_client_max_keepalive_connections: int = 50, sandbox_wait_for_creation_max_attempts: int = 120, + keep_sandbox_for_scoring: bool = False, **kwargs, ): super().__init__(max_turns=max_turns, message_type="chat", **kwargs) @@ -104,6 +113,7 @@ def __init__( sandbox_client_max_keepalive_connections=sandbox_client_max_keepalive_connections, sandbox_wait_for_creation_max_attempts=sandbox_wait_for_creation_max_attempts, ) + self.keep_sandbox_for_scoring = keep_sandbox_for_scoring self.run_command = run_command self.poll_interval = poll_interval self.timeout_seconds = timeout_seconds @@ -135,7 +145,6 @@ def init_interception( self.interception_url = interception_url self._tunnel: Tunnel | None = None self._tunnel_lock = asyncio.Lock() - self._tunnel_monitor_task: asyncio.Task | None = None self._interception_server = InterceptionServer(port=interception_port) def _require_interception_server(self) -> InterceptionServer: @@ -148,7 +157,7 @@ async def get_tunnel_url(self) -> str: async with self._tunnel_lock: if self._tunnel is not None and not self._tunnel.is_running: frpc_output = "\n".join(self._tunnel.recent_output) - logger.warning( + self.logger.warning( f"Tunnel process died, recreating. frpc output:\n{frpc_output}" ) self._tunnel.sync_stop() @@ -157,7 +166,7 @@ async def get_tunnel_url(self) -> str: if self._tunnel is None: interception_server = self._require_interception_server() port = interception_server.port - if logger.isEnabledFor(logging.DEBUG): + if self.logger.isEnabledFor(logging.DEBUG): self._tunnel = Tunnel( local_port=port, log_level="debug", @@ -165,48 +174,12 @@ async def get_tunnel_url(self) -> str: else: self._tunnel = Tunnel(local_port=port) url = await self._tunnel.start() - logger.debug(f"Prime Tunnel started: {url}") - - # Lazily start health monitor on first tunnel creation - if ( - self._tunnel_monitor_task is None - or self._tunnel_monitor_task.done() - ): - self._tunnel_monitor_task = asyncio.create_task( - self._tunnel_health_monitor() - ) - + self.logger.debug(f"Prime Tunnel started: {url}") return url else: assert self._tunnel.url is not None, "Tunnel started but URL is None" return self._tunnel.url - async def _tunnel_health_monitor(self, interval: float = 30.0) -> None: - """Background task that checks tunnel liveness and restarts a dead tunnel.""" - try: - while True: - await asyncio.sleep(interval) - async with self._tunnel_lock: - if self._tunnel is not None and not self._tunnel.is_running: - frpc_output = "\n".join(self._tunnel.recent_output) - logger.warning( - f"Health monitor: tunnel dead. frpc output:\n{frpc_output}" - ) - self._tunnel.sync_stop() - interception_server = self._require_interception_server() - port = interception_server.port - if logger.isEnabledFor(logging.DEBUG): - self._tunnel = Tunnel( - local_port=port, - log_level="debug", - ) - else: - self._tunnel = Tunnel(local_port=port) - url = await self._tunnel.start() - logger.info(f"Health monitor: restarted tunnel url={url}") - except asyncio.CancelledError: - return - async def setup_state(self, state: State) -> State: """Setup sandbox + interception for this rollout""" state = await super().setup_state(state) @@ -242,7 +215,7 @@ async def setup_state(self, state: State) -> State: advanced_configs=self.advanced_configs, labels=self.labels if self.labels else [], ) - logger.debug( + self.logger.debug( f"Creating sandbox with OPENAI_BASE_URL={env_vars.get('OPENAI_BASE_URL')} " f"docker_image={docker_image}" ) @@ -255,6 +228,12 @@ async def setup_state(self, state: State) -> State: await self.start_agent(state) + parts = [ + f"Started rollout_id={state['rollout_id']}", + f"example_id={state['example_id']}", + ] + self.logger.info(" | ".join(parts)) + return state async def get_docker_image(self, state: State) -> str: @@ -265,9 +244,9 @@ async def build_env_vars(self, state: State) -> dict[str, str]: """Build environment variables for the sandbox. Override to add custom vars.""" env_vars = dict(self.environment_vars) if self.environment_vars else {} env_vars["OPENAI_BASE_URL"] = state["interception_base_url"] - env_vars.setdefault("OPENAI_TIMEOUT", "600") - env_vars.setdefault("OPENAI_REQUEST_TIMEOUT", "600") - env_vars.setdefault("HTTPX_TIMEOUT", "600") + env_vars.setdefault("OPENAI_TIMEOUT", "3600") + env_vars.setdefault("OPENAI_REQUEST_TIMEOUT", "3600") + env_vars.setdefault("HTTPX_TIMEOUT", "3600") model = state.get("model") if model: env_vars["OPENAI_MODEL"] = model @@ -281,10 +260,16 @@ async def start_agent(self, state: State) -> None: """Start the agent command using background job.""" sandbox_id = state["sandbox_id"] - background_job: BackgroundJob = await self.sandbox_client.start_background_job( - sandbox_id, - self.run_command, - ) + self.logger.debug(f"Starting agent in sandbox {sandbox_id}") + try: + background_job: BackgroundJob = ( + await self.sandbox_client.start_background_job( + sandbox_id, + self.run_command, + ) + ) + except Exception as e: + raise vf.SandboxError(f"Failed to start agent: {e}") from e state["background_job"] = background_job state["agent_start_time"] = time.time() @@ -307,13 +292,15 @@ async def wait_for_completion(self, state: State) -> None: timeout=self.timeout_seconds, ) except asyncio.TimeoutError: - logger.warning(f"Agent timed out after {self.timeout_seconds}s") + self.logger.warning(f"Agent timed out after {self.timeout_seconds}s") state["agent_timed_out"] = True except asyncio.CancelledError: - logger.debug("Completion wait task cancelled") + self.logger.debug("Completion wait task cancelled") raise except Exception as e: - logger.debug(f"Completion wait ended: {e}") + error = AgentError(f"Agent polling failed: {e}") + state["error"] = error + self.logger.error(f"Agent polling failed: {e}") finally: state["agent_completed"] = True @@ -330,11 +317,11 @@ async def poll_job_completion( state["agent_stdout"] = status.stdout state["agent_stderr"] = status.stderr if status.exit_code == 0: - logger.debug( + self.logger.debug( f"Agent completed successfully (exit_code={status.exit_code})" ) else: - logger.warning( + self.logger.warning( f"Agent failed (exit_code={status.exit_code}) stdout={status.stdout}, stderr={status.stderr}" ) return @@ -525,24 +512,13 @@ async def add_model_response( @vf.teardown async def teardown_resources(self): """Stop Prime Tunnel and HTTP interception server.""" - if ( - self._tunnel_monitor_task is not None - and not self._tunnel_monitor_task.done() - ): - self._tunnel_monitor_task.cancel() - try: - await self._tunnel_monitor_task - except asyncio.CancelledError: - pass - self._tunnel_monitor_task = None - async with self._tunnel_lock: if self._tunnel is not None: try: self._tunnel.sync_stop() - logger.debug("Prime Tunnel stopped") + self.logger.debug("Prime Tunnel stopped") except Exception as e: - logger.warning(f"Error stopping Prime Tunnel: {e}") + self.logger.warning(f"Error stopping Prime Tunnel: {e}") finally: self._tunnel = None if self._interception_server is not None: @@ -582,15 +558,58 @@ async def post_rollout(self, state: State): Override for custom post-rollout logic. For example, if sandbox state is needed for reward functions, run computation here and cache the result in state before sandbox is destroyed. """ - pass + tool_counts: Counter[str] = Counter() + for step in state.get("trajectory", []): + for msg in step.get("completion", []): + if isinstance(msg, AssistantMessage) and isinstance( + msg.tool_calls, list + ): + for tc in msg.tool_calls: + if isinstance(tc, ToolCall): + tool_counts[tc.name] += 1 + + example_id = state.get("example_id") + num_turns = len(state.get("trajectory", [])) + stop_condition = state.get("stop_condition", "unknown") + error = state.get("error") + error_info = ( + f"{type(error).__name__}: {truncate(str(error), 80)}" if error else None + ) + exit_code = state.get("agent_exit_code") + timed_out = state.get("agent_timed_out", False) + duration_s = state["timing"].get("total_ms", 0) / 1000 + tools_str = ",".join(f"{k}:{v}" for k, v in tool_counts.most_common()) + parts = [ + f"Finished rollout_id={state.get('rollout_id')}", + f"example_id={example_id}", + f"turns={num_turns}", + f"tools=[{tools_str}]", + f"stop={stop_condition}", + f"exit_code={exit_code}", + f"duration={print_time(duration_s)}", + ] + if timed_out: + parts.append("timed_out=True") + if error_info: + parts.append(f"error={error_info}") + self.logger.info(" | ".join(parts)) @vf.cleanup async def destroy_sandbox(self, state: State): - """Cleanup sandbox after rollout.""" + """Cleanup sandbox after rollout. + + When `keep_sandbox_for_scoring` is True, sandbox deletion is deferred + (e.g. when the rubric needs sandbox access during scoring). + The sandbox is still deregistered from active tracking so the + environment teardown does not attempt a redundant bulk-delete. + """ await self.post_rollout(state) sandbox_id = state.get("sandbox_id") if sandbox_id: - await self.delete_sandbox(sandbox_id) + if self.keep_sandbox_for_scoring: + self.deregister_sandbox(sandbox_id) + else: + await self.delete_sandbox(sandbox_id) async def env_response( self, messages: Messages, state: State, **kwargs diff --git a/verifiers/envs/experimental/opencode_env.py b/verifiers/envs/experimental/opencode_env.py index 3fea441e22..014491b52c 100644 --- a/verifiers/envs/experimental/opencode_env.py +++ b/verifiers/envs/experimental/opencode_env.py @@ -10,7 +10,7 @@ import verifiers as vf from verifiers.envs.experimental.cli_agent_env import CliAgentEnv from verifiers.types import AssistantMessage, Messages, ToolCall -from verifiers.utils.interception_utils import _truncate as truncate +from verifiers.utils.logging_utils import truncate logger = logging.getLogger(__name__) @@ -39,25 +39,19 @@ # Tone and style - Only use emojis if the user explicitly requests it. Avoid using emojis in all communication unless asked. - Your output will be displayed on a command line interface. Your responses should be short and concise. You can use Github-flavored markdown for formatting, and will be rendered in a monospace font using the CommonMark specification. -- Output text to communicate with the user; all text you output outside of tool use is displayed to the user. Only use tools to complete tasks. Never use tools like Bash or code comments as means to communicate with the user during the session. +- Output text to communicate with the user; all text you output outside of tool use is displayed to the user. Only use tools to complete tasks. Never use tools like bash or code comments as means to communicate with the user during the session. - NEVER create files unless they're absolutely necessary for achieving your goal. ALWAYS prefer editing an existing file to creating a new one. This includes markdown files. # Professional objectivity Prioritize technical accuracy and truthfulness over validating the user's beliefs. Focus on facts and problem-solving, providing direct, objective technical info without any unnecessary superlatives, praise, or emotional validation. It is best for the user if OpenCode honestly applies the same rigorous standards to all ideas and disagrees when necessary, even if it may not be what the user wants to hear. Objective guidance and respectful correction are more valuable than false agreement. Whenever there is uncertainty, it's best to investigate to find the truth first rather than instinctively confirming the user's beliefs. """ -TASK_MANAGEMENT_SYSTEM_PROMPT = """\ -# Task Management -You have access to the TodoWrite tools to help you manage and plan tasks. Use these tools frequently to ensure that you are tracking your tasks and giving the user visibility into your progress. These tools are also helpful for planning tasks, and for breaking down larger complex tasks into smaller steps. If you do not use this tool when planning, you may forget to do important tasks - and that is unacceptable. It is critical that you mark todos as completed as soon as you are done with a task. Do not batch up multiple tasks before marking them as completed. -""" - - DEFAULT_INSTALL_COMMAND = ( "curl -fsSL https://opencode.ai/install | bash -s -- --version v1.2.15" ) DEFAULT_RUN_COMMAND_TEMPLATE = """\ -set -e +set -eo pipefail apt-get update && apt-get install -y curl @@ -150,9 +144,11 @@ class OpenCodeEnv(CliAgentEnv): DEFAULT_INSTALL_COMMAND = DEFAULT_INSTALL_COMMAND DEFAULT_RUN_COMMAND_TEMPLATE = DEFAULT_RUN_COMMAND_TEMPLATE DEFAULT_SYSTEM_PROMPT = DEFAULT_SYSTEM_PROMPT - DEFAULT_PROVIDER_TIMEOUT_MS = 1_800_000 # 30min + DEFAULT_PROVIDER_TIMEOUT_MS = 3_600_000 # 1h DEFAULT_DISABLE_COMPACTION = True DEFAULT_ENABLE_INTERLEAVED = True + DEFAULT_INCLUDE_TASK_SYSTEM_PROMPT = False + DEFAULT_TASK_SYSTEM_PROMPT = "" def __init__( self, @@ -166,6 +162,8 @@ def __init__( disable_compaction: bool = DEFAULT_DISABLE_COMPACTION, enable_interleaved: bool = DEFAULT_ENABLE_INTERLEAVED, provider_timeout_ms: int = DEFAULT_PROVIDER_TIMEOUT_MS, + task_system_prompt: str = DEFAULT_TASK_SYSTEM_PROMPT, + include_task_system_prompt: bool = DEFAULT_INCLUDE_TASK_SYSTEM_PROMPT, **kwargs, ): self.asset_dir = asset_dir @@ -173,6 +171,9 @@ def __init__( self.disabled_tools = disabled_tools self.provider_timeout_ms = provider_timeout_ms + if system_prompt is not None and include_task_system_prompt: + system_prompt += "\n" + task_system_prompt + run_command = self.build_run_command( run_command_template, agent_workdir, @@ -183,13 +184,6 @@ def __init__( enable_interleaved=enable_interleaved, ) - if ( - disabled_tools is not None - and system_prompt is not None - and "todowrite" not in disabled_tools - ): - system_prompt += "\n" + TASK_MANAGEMENT_SYSTEM_PROMPT - super().__init__( run_command=run_command, dataset=dataset, @@ -308,6 +302,32 @@ def normalize_response(self, response: vf.Response) -> vf.Response: ) return response.model_copy(update={"message": normalized_message}) + async def post_rollout(self, state: vf.State) -> None: + """Collect agent logs from sandbox before teardown.""" + sandbox_id = state.get("sandbox_id") + if sandbox_id: + try: + result = await self.sandbox_client.execute_command( + sandbox_id, + f"cat {self.remote_logs_path} 2>/dev/null || echo ''", + working_dir=None, + ) + agent_logs = (result.stdout or "").strip() + state["agent_logs"] = agent_logs + + # Log agent output on error or empty trajectory for debugging + num_turns = len(state.get("trajectory", [])) + agent_error = state.get("agent_exit_code", 0) != 0 + if (agent_error or num_turns == 0) and agent_logs: + logger.warning( + f"Agent logs (example_id={state.get('example_id')}, " + f"exit_code={state.get('agent_exit_code')}, turns={num_turns}):\n{agent_logs}" + ) + except Exception as e: + logger.warning(f"Failed to collect agent logs: {e}") + + await super().post_rollout(state) + def build_prompt(self, state: vf.State) -> str: """Build the prompt to be uploaded to OpenCode.""" return state["prompt"][-1]["content"] diff --git a/verifiers/envs/experimental/opencode_qa_env.py b/verifiers/envs/experimental/opencode_qa_env.py index bff428d094..976fb287d9 100644 --- a/verifiers/envs/experimental/opencode_qa_env.py +++ b/verifiers/envs/experimental/opencode_qa_env.py @@ -13,6 +13,9 @@ class OpenCodeQAEnv(OpenCodeEnv): DEFAULT_ANSWER_KEY = "answer" DEFAULT_INSTRUCTION_PROMPT = "" DEFAULT_INSTRUCTION_PROMPT_POST = "" + DEFAULT_DIFFICULTY_KEY = None + DEFAULT_MIN_AVG_REWARD = 0.0 + DEFAULT_MAX_AVG_REWARD = 1.0 def __init__( self, @@ -24,6 +27,9 @@ def __init__( answer_key: str = DEFAULT_ANSWER_KEY, instruction_prompt: str = DEFAULT_INSTRUCTION_PROMPT, instruction_prompt_post: str = DEFAULT_INSTRUCTION_PROMPT_POST, + difficulty_key: str | None = DEFAULT_DIFFICULTY_KEY, + min_avg_reward: float = DEFAULT_MIN_AVG_REWARD, + max_avg_reward: float = DEFAULT_MAX_AVG_REWARD, **kwargs, ): dataset = self.construct_dataset( @@ -34,6 +40,9 @@ def __init__( answer_key, instruction_prompt, instruction_prompt_post, + difficulty_key, + min_avg_reward, + max_avg_reward, ) super().__init__(dataset=dataset, rubric=rubric, **kwargs) @@ -47,6 +56,9 @@ def construct_dataset( answer_key: str, instruction_prompt: str, instruction_prompt_post: str, + difficulty_key: str | None = None, + min_avg_reward: float = 0.0, + max_avg_reward: float = 1.0, ) -> Dataset: """Constructs a general QA dataset.""" @@ -57,6 +69,13 @@ def construct_dataset( ) dataset = dataset_obj + if difficulty_key is not None: + column_names = dataset.column_names or [] + if difficulty_key in column_names: + dataset = dataset.filter( + lambda x: min_avg_reward <= x[difficulty_key] <= max_avg_reward + ) + column_names = dataset.column_names if column_names is None: raise ValueError("Dataset has no columns.") diff --git a/verifiers/envs/experimental/sandbox_mixin.py b/verifiers/envs/experimental/sandbox_mixin.py index cf177e2d5f..194cef05c9 100644 --- a/verifiers/envs/experimental/sandbox_mixin.py +++ b/verifiers/envs/experimental/sandbox_mixin.py @@ -1,11 +1,16 @@ import asyncio +import io import logging import os +import tarfile +import tempfile +from pathlib import Path from typing import Any, Callable, cast import httpx import tenacity as tc from prime_sandboxes import ( + APIError, CommandTimeoutError, CreateSandboxRequest, SandboxClient, @@ -60,6 +65,14 @@ class SandboxMixin: sandbox_wait_for_creation_max_attempts: int with_retry: Callable + def register_sandbox(self, sandbox_id: str) -> None: + """Register a sandbox for active tracking and crash teardown.""" + self.active_sandboxes.add(sandbox_id) + + def deregister_sandbox(self, sandbox_id: str) -> None: + """Deregister a sandbox from active tracking.""" + self.active_sandboxes.discard(sandbox_id) + def init_sandbox_client( self, max_retries: int = 5, @@ -73,7 +86,8 @@ def init_sandbox_client( sandbox_wait_for_creation_max_attempts: int = 120, ): """Initialize sandbox client and retry wrapper. Call from subclass __init__.""" - self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}") + if not hasattr(self, "logger"): + self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}") self.active_sandboxes = set() self.sandbox_wait_for_creation_max_attempts = ( sandbox_wait_for_creation_max_attempts @@ -111,7 +125,7 @@ async def create_sandbox(self, state, request: CreateSandboxRequest) -> str: except Exception as e: raise SandboxCreationError(f"Failed to create sandbox: {e}") from e - self.active_sandboxes.add(sandbox.id) + self.register_sandbox(sandbox.id) state["sandbox_id"] = sandbox.id self.logger.debug(f"Created sandbox {sandbox.id}") @@ -143,7 +157,7 @@ async def delete_sandbox(self, sandbox_id: str): async def _delete(sandbox_id: str): await self.sandbox_client.delete(sandbox_id) - self.active_sandboxes.discard(sandbox_id) + self.deregister_sandbox(sandbox_id) self.logger.debug(f"Deleted sandbox {sandbox_id}") try: @@ -156,7 +170,8 @@ async def bulk_delete_sandboxes(self, sandbox_ids: list[str]) -> None: try: await self.with_retry(self.sandbox_client.bulk_delete)(sandbox_ids) self.logger.debug(f"Bulk deleted sandboxes: {sandbox_ids}") - self.active_sandboxes.difference_update(sandbox_ids) + for sandbox_id in sandbox_ids: + self.deregister_sandbox(sandbox_id) except Exception as e: self.logger.error(f"Failed to bulk delete sandboxes {sandbox_ids}: {e}") @@ -211,6 +226,111 @@ async def run_background_job( sandbox_id=sandbox_id, command=command, timeout=timeout ) + async def upload_file( + self, + sandbox_id: str, + remote_path: str, + local_path: str, + ) -> None: + """Upload a local file to the sandbox.""" + try: + await self.sandbox_client.upload_file(sandbox_id, remote_path, local_path) + except SandboxOOMError as e: + raise vf.SandboxError( + f"Sandbox {sandbox_id} OOM during upload to {remote_path}" + ) from e + except SandboxTimeoutError as e: + raise vf.SandboxError( + f"Sandbox {sandbox_id} timeout during upload to {remote_path}" + ) from e + except APIError as e: + raise vf.SandboxError( + f"API error uploading to {remote_path} in {sandbox_id}: {e}" + ) from e + + async def upload_content( + self, + sandbox_id: str, + content: str, + remote_path: str, + ) -> None: + """Upload a string as a file to the sandbox.""" + with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".txt") as f: + f.write(content) + local_path = f.name + try: + await self.upload_file(sandbox_id, remote_path, local_path) + finally: + Path(local_path).unlink(missing_ok=True) + + async def read_file( + self, + sandbox_id: str, + remote_path: str, + timeout: int = 10, + ) -> str | None: + """Read a file from the sandbox, returning its contents or None on failure.""" + try: + result = await self.sandbox_client.execute_command( + sandbox_id, + f"cat {remote_path}", + timeout=timeout, + ) + if result.exit_code == 0: + return result.stdout or "" + return None + except Exception as e: + self.logger.warning( + f"Failed to read {remote_path} from {sandbox_id}: {type(e).__name__}: {e}" + ) + return None + + async def upload_bundle( + self, + sandbox_id: str, + file_map: dict[str, str], + dest_dir: str, + ) -> None: + """Upload a bundle of files to the sandbox. + + Builds a tar.gz archive from ``file_map`` (relative path → UTF-8 + content), uploads it, and extracts into ``dest_dir``. + """ + buf = io.BytesIO() + with tarfile.open(fileobj=buf, mode="w:gz") as tar: + for rel_path, content in file_map.items(): + data = content.encode("utf-8") + info = tarfile.TarInfo(name=rel_path) + info.size = len(data) + tar.addfile(info, io.BytesIO(data)) + bundle_bytes = buf.getvalue() + + archive_remote = f"{dest_dir}/_bundle.tar.gz" + with tempfile.NamedTemporaryFile(delete=False, suffix=".tar.gz") as f: + f.write(bundle_bytes) + tmp_path = f.name + try: + await self.upload_file(sandbox_id, archive_remote, tmp_path) + finally: + Path(tmp_path).unlink(missing_ok=True) + + extract_cmd = ( + f"mkdir -p {dest_dir} && " + f'python3 -c "import tarfile; ' + f"tarfile.open('{archive_remote}', 'r:gz').extractall('{dest_dir}')\" && " + f"rm -f {archive_remote}" + ) + result = await self.sandbox_client.execute_command( + sandbox_id, + extract_cmd, + timeout=60, + ) + if result.exit_code != 0: + raise vf.SandboxError( + f"Bundle extract failed in {sandbox_id} (exit={result.exit_code}): " + f"{(result.stderr or '')[:200]}" + ) + def teardown_sandboxes(self): """Delete all active sandboxes using sync client. @@ -228,7 +348,7 @@ def teardown_sandboxes(self): try: sync_client.bulk_delete(sandbox_ids=batch) for sandbox_id in batch: - self.active_sandboxes.discard(sandbox_id) + self.deregister_sandbox(sandbox_id) self.logger.debug(f"Bulk deleted batch of {len(batch)} sandboxes") except Exception as e: self.logger.warning(f"Bulk delete failed for batch: {e}") diff --git a/verifiers/rubrics/experimental/hybrid_math_rubric.py b/verifiers/rubrics/experimental/hybrid_math_rubric.py index eb5ebc8223..c0e50b815e 100644 --- a/verifiers/rubrics/experimental/hybrid_math_rubric.py +++ b/verifiers/rubrics/experimental/hybrid_math_rubric.py @@ -1,9 +1,15 @@ +from __future__ import annotations + +import asyncio + from math_verify import parse, verify from openai import AsyncOpenAI - -import verifiers as vf +from verifiers.envs.experimental.sandbox_mixin import SandboxMixin from verifiers.parsers.parser import Parser from verifiers.utils.data_utils import extract_boxed_answer +from verifiers.utils.logging_utils import truncate + +import verifiers as vf # https://github.com/open-compass/CompassVerifier/blob/2d7cba6df0b21f9c6121786ac1e5770c68473598/src/prompts.py#L28 DEFAULT_JUDGE_PROMPT = """\ @@ -84,21 +90,31 @@ class HybridMathRubric(vf.JudgeRubric): """Runs rule-based math verification first, with optional LLM judge fallback.""" + DEFAULT_JUDGE_PARSER = None + DEFAULT_JUDGE_MODEL = "gpt-5-nano" + DEFAULT_JUDGE_CLIENT = None + DEFAULT_JUDGE_PROMPT = DEFAULT_JUDGE_PROMPT + DEFAULT_JUDGE_SAMPLING_ARGS = {} + DEFAULT_USE_JUDGE_FALLBACK = False + def __init__( self, - judge_parser: Parser | None = None, - judge_model: str | None = None, - judge_client: AsyncOpenAI | None = None, - judge_sampling_args: dict = {}, + parser: Parser | None = DEFAULT_JUDGE_PARSER, + use_judge_fallback: bool = DEFAULT_USE_JUDGE_FALLBACK, + judge_client: AsyncOpenAI | None = DEFAULT_JUDGE_CLIENT, + judge_model: str = DEFAULT_JUDGE_MODEL, judge_prompt: str = DEFAULT_JUDGE_PROMPT, - timeout_seconds: float = 5, + judge_sampling_args: dict | None = None, **kwargs, ): + judge_sampling_args = judge_sampling_args or self.DEFAULT_JUDGE_SAMPLING_ARGS + if judge_client is None and not use_judge_fallback: + judge_client = AsyncOpenAI(api_key="unused") super().__init__( judge_client=judge_client, judge_sampling_args=judge_sampling_args, judge_prompt=judge_prompt, - parser=judge_parser, + parser=parser, **kwargs, ) # Reward functions @@ -106,9 +122,8 @@ def __init__( self.add_reward_func(self.judge_score, weight=0) self.add_reward_func(self.correct_answer, weight=1) - self.timeout_seconds = timeout_seconds - self.judge_model = judge_model - self.class_objects["judge_model"] = judge_model + self.judge_model = judge_model if use_judge_fallback else None + self.class_objects["judge_model"] = self.judge_model async def math_verify_score( self, completion: vf.Messages, answer: str, state: vf.State, **kwargs @@ -116,30 +131,28 @@ async def math_verify_score( """Basic rule-based math verification.""" response = self.parser.parse_answer(completion) or "" if response == "": - math_verify_score = 0.0 self.logger.debug("Parsed response is empty.") - else: - try: - math_verify_score = float( - verify( - parse( - f"\\boxed{{{answer}}}", - parsing_timeout=int(self.timeout_seconds), - ), - parse( - f"\\boxed{{{response}}}", - parsing_timeout=int(self.timeout_seconds), - ), - timeout_seconds=int(self.timeout_seconds), - ) - ) - except BaseException as e: - self.logger.warning( - f"Math verification failed with {type(e).__name__}: {e!r}" + state["math_verify_score"] = 0.0 + return 0.0 + + try: + score = float( + verify( + parse(f"\\boxed{{{answer}}}", parsing_timeout=5), + parse(f"\\boxed{{{response}}}", parsing_timeout=5), + timeout_seconds=5, ) - math_verify_score = 0.0 - state["math_verify_score"] = math_verify_score - return math_verify_score + ) + answer_str = truncate(answer.strip(), 20) + response_str = truncate(response.strip(), 20) + self.logger.debug( + f"Local math_verify {score} (answer={answer_str}, response={response_str})" + ) + except BaseException as e: + self.logger.warning(f"Math verification failed: {e!r}") + score = 0.0 + state["math_verify_score"] = score + return score async def judge_score( self, @@ -170,3 +183,154 @@ async def correct_answer(self, state: vf.State, **kwargs) -> float: return float( state.get("math_verify_score", 0.0) or state.get("judge_score", 0.0) ) + + +MATH_VERIFY_SCORER_SCRIPT_TEMPLATE = """\ +from pathlib import Path +from math_verify import parse, verify + +solution = Path("{solution_path}").read_text() +answer_file = Path("{answer_path}") +answer = answer_file.read_text() if answer_file.exists() else "" + +if not answer: + print(0.0) +else: + try: + score = float( + verify( + parse("\\\\boxed{{" + solution + "}}", parsing_timeout=5), + parse("\\\\boxed{{" + answer + "}}", parsing_timeout=5), + timeout_seconds=5, + ) + ) + print(score) + except BaseException: + print(0.0) +""" + + +class RemoteHybridMathRubric(SandboxMixin, HybridMathRubric): + """HybridMathRubric that scores inside the sandbox. + + Expects the environment to keep the sandbox alive for scoring + (``keep_sandbox_for_scoring=True``) and the agent's answer written to + ``answer_path``. Ground-truth is uploaded to ``solution_path`` and a + scorer script is uploaded and executed. The sandbox is deleted in the + ``@vf.cleanup`` handler after scoring. + """ + + DEFAULT_ANSWER_PATH = "/app/answer.txt" + DEFAULT_SOLUTION_PATH = "/app/solution.txt" + DEFAULT_SCORER_PATH = "/app/score.py" + DEFAULT_SCORER_TIMEOUT = 30 + + def __init__( + self, + answer_path: str = DEFAULT_ANSWER_PATH, + solution_path: str = DEFAULT_SOLUTION_PATH, + scorer_path: str = DEFAULT_SCORER_PATH, + scorer_timeout: int = DEFAULT_SCORER_TIMEOUT, + sandbox_client_max_workers: int = 10, + sandbox_client_max_connections: int = 100, + sandbox_client_max_keepalive_connections: int = 50, + **kwargs, + ): + super().__init__(**kwargs) + self.answer_path = answer_path + self.solution_path = solution_path + self.scorer_path = scorer_path + self.scorer_timeout = scorer_timeout + self.score_script = MATH_VERIFY_SCORER_SCRIPT_TEMPLATE.format( + answer_path=answer_path, + solution_path=solution_path, + ) + + self.logger.warning( + "RemoteHybridMathRubric expects a sandbox kept alive for scoring " + f"(keep_sandbox_for_scoring=True) and the agent's answer written to {answer_path}" + ) + self.init_sandbox_client( + sandbox_client_max_workers=sandbox_client_max_workers, + sandbox_client_max_connections=sandbox_client_max_connections, + sandbox_client_max_keepalive_connections=sandbox_client_max_keepalive_connections, + ) + + async def math_verify_score( + self, completion: vf.Messages, answer: str, state: vf.State, **kwargs + ) -> float: + """Run math_verify inside the sandbox.""" + sandbox_id = state.get("sandbox_id") + if not sandbox_id: + state["math_verify_score"] = 0.0 + return 0.0 + # Track the sandbox so it is torn down on crash + self.register_sandbox(sandbox_id) + if state.get("error") or state.get("sandbox_error"): + state["math_verify_score"] = 0.0 + return 0.0 + + try: + await asyncio.gather( + self.upload_content(sandbox_id, answer, self.solution_path), + self.upload_content(sandbox_id, self.score_script, self.scorer_path), + ) + result = await self.sandbox_client.execute_command( + sandbox_id, + f"python3 {self.scorer_path}", + timeout=self.scorer_timeout, + ) + if result.exit_code == 0 and result.stdout.strip(): + score = float(result.stdout.strip().splitlines()[-1]) + self.logger.debug(f"Remote math_verify scored {score=}") + else: + stderr = (result.stderr or "")[:200] + self.logger.warning( + f"Remote math_verify failed (exit={result.exit_code}): {stderr}" + ) + score = 0.0 + except Exception as e: + self.logger.warning(f"Remote math_verify error: {type(e).__name__}: {e}") + score = 0.0 + + state["math_verify_score"] = score + return score + + async def judge_score( + self, + prompt: vf.Messages, + completion: vf.Messages, + answer: str, + state: vf.State, + **kwargs, + ) -> float: + """Judge with response read from the sandbox file.""" + if state.get("math_verify_score", 0) == 1 or self.judge_model is None: + return state.get("math_verify_score", 0) + + sandbox_id = state.get("sandbox_id") + if not sandbox_id: + return 0.0 + response = await self.read_file(sandbox_id, self.answer_path) + if not response: + return 0.0 + completion = [vf.AssistantMessage(content=response)] + + judge_response = await self.judge(prompt, completion, answer, state) + judge_result = ( + extract_boxed_answer(judge_response) + if len(judge_response) != 1 + else judge_response + ) + judge_score = 1.0 if judge_result == "A" else 0.0 + self.logger.debug(f"{judge_score=} ({judge_result=})") + state["judge_result"] = judge_result + state["judge_score"] = judge_score + return judge_score + + @vf.cleanup + async def cleanup_sandbox(self, state: vf.State) -> None: + """Delete the sandbox after scoring is complete.""" + sandbox_id = state.get("sandbox_id") + if sandbox_id: + await self.delete_sandbox(sandbox_id) diff --git a/verifiers/utils/interception_utils.py b/verifiers/utils/interception_utils.py index 88e4829e6d..26595a22e9 100644 --- a/verifiers/utils/interception_utils.py +++ b/verifiers/utils/interception_utils.py @@ -24,6 +24,7 @@ ) from verifiers.types import Response +from verifiers.utils.logging_utils import truncate logger = logging.getLogger(__name__) @@ -171,7 +172,9 @@ async def _handle_request(self, request: Any) -> Any: except asyncio.CancelledError: return web.json_response({"error": "Rollout cancelled"}, status=499) except Exception as e: - logger.error(f"Error processing intercepted request: {e}") + logger.debug( + f"[{rollout_id}] Rollout error surfaced in non-streaming request: {type(e).__name__}: {e}" + ) return web.json_response({"error": str(e)}, status=500) response_dict = serialize_intercept_response(response) @@ -206,12 +209,18 @@ async def _handle_streaming_response( chunk_json = json.dumps(chunk_dict) await response.write(f"data: {chunk_json}\n\n".encode()) - await response_future - except asyncio.CancelledError: logger.debug(f"[{rollout_id}] Streaming cancelled") except Exception as e: logger.error(f"[{rollout_id}] Streaming error: {e}") + return response + + try: + await response_future + except BaseException as e: + logger.debug( + f"[{rollout_id}] Rollout error surfaced in stream: {type(e).__name__}: {e}" + ) try: await response.write_eof() @@ -434,36 +443,36 @@ def serialize_intercept_response(response: Any) -> dict[str, Any]: return dict(response) -def _truncate(s: str, limit: int = 200) -> str: - return (s[:limit] + "...") if len(s) > limit else s - - def _log_request(rollout_id: str, body: dict) -> None: """Log an intercepted request.""" + if not logger.isEnabledFor(logging.DEBUG): + return log_msg = f"[{rollout_id}] <- INTERCEPTED REQUEST" tools = body.get("tools", []) log_msg += f" ({len(tools)} tool(s))" if tools: - log_msg += f"\n[tools]\n{', '.join([tool.get('function', {}).get('name', '?') for tool in tools])}" + log_msg += f"\n[tools] {', '.join([tool.get('function', {}).get('name', '?') for tool in tools])}" for msg in body.get("messages", []): content = msg.get("content", "") if isinstance(content, str): - log_msg += f"\n[{msg.get('role', '?')}] {_truncate(content)}" + log_msg += f"\n[{msg.get('role', '?')}] {truncate(content)}" else: log_msg += f"\n[{msg.get('role', '?')}] " for tc in msg.get("tool_calls") or []: func = tc.get("function", {}) - log_msg += f"\n[tool_call]\n{func.get('name')}({_truncate(func.get('arguments', ''), 100)})" + log_msg += f"\n[tool_call]\n{func.get('name')}({truncate(func.get('arguments', ''), 100)})" logger.debug(log_msg) def _log_response(rollout_id: str, response: dict) -> None: """Log the response from the model.""" + if not logger.isEnabledFor(logging.DEBUG): + return log_msg = f"[{rollout_id}] -> RESPONSE" msg = response.get("choices", [{}])[0].get("message", {}) if msg.get("content"): - log_msg += f"\n[assistant]\n{_truncate(msg['content'])}" + log_msg += f"\n[assistant]\n{truncate(msg['content'])}" for tc in msg.get("tool_calls") or []: func = tc.get("function", {}) - log_msg += f"\n[tool_call]\n{func.get('name')}({_truncate(func.get('arguments', ''), 100)})" + log_msg += f"\n[tool_call]\n{func.get('name')}({truncate(func.get('arguments', ''), 100)})" logger.debug(log_msg) diff --git a/verifiers/utils/logging_utils.py b/verifiers/utils/logging_utils.py index 6d0084f644..38591c9f93 100644 --- a/verifiers/utils/logging_utils.py +++ b/verifiers/utils/logging_utils.py @@ -209,3 +209,8 @@ def print_time(time_s: float) -> str: return f"{ms:.0f}ms" else: return f"{time_s:.0f}s" + + +def truncate(s: str, limit: int = 200) -> str: + """Truncate a string to a given length.""" + return (s[:limit] + "...") if len(s) > limit else s