""" Verifier execution for AWM environments. Two verification modes: - "sql": Verifier extracts DB state, then an LLM judges the trajectory. - "code": Verifier deterministically returns {"result": "complete"|"others"}. """ import json import logging import os import re import subprocess import sys import textwrap from typing import Any from openai import AsyncOpenAI logger = logging.getLogger(__name__) # Wall-clock backstop for the verifier subprocess VERIFIER_WALL_TIMEOUT_S = float(os.environ.get("OPENENV_AWM_VERIFIER_TIMEOUT", "20.0")) # isolate runner script _RUNNER_PATH = os.path.join( os.path.dirname(os.path.abspath(__file__)), "_verifier_runner.py" ) def _run_verifier_subprocess(payload: dict[str, Any]) -> dict[str, Any]: """Invoke the runner. Returns a dict; never raises.""" try: proc = subprocess.run( [sys.executable, _RUNNER_PATH], input=json.dumps(payload), capture_output=True, text=True, timeout=VERIFIER_WALL_TIMEOUT_S, check=False, ) except subprocess.TimeoutExpired: return { "execution_status": "error", "error_message": (f"Verifier timed out after {VERIFIER_WALL_TIMEOUT_S}s"), } except (OSError, ValueError) as e: return { "execution_status": "error", "error_message": f"Failed to launch verifier subprocess: {e}", } if proc.returncode != 0 and not proc.stdout.strip(): stderr = (proc.stderr or "").strip() return { "execution_status": "error", "error_message": ( f"Verifier subprocess exited {proc.returncode}: {stderr[:500]}" ), } if not proc.stdout.strip(): return { "execution_status": "error", "error_message": "Verifier subprocess produced no output", } try: return json.loads(proc.stdout) except json.JSONDecodeError as e: return { "execution_status": "error", "error_message": ( f"Bad JSON from verifier subprocess ({e}): {proc.stdout[:500]}" ), } def execute_sql_verifier( verifier_code: str, function_name: str, initial_db_path: str, final_db_path: str, ) -> dict[str, Any]: """Execute SQL-mode verifier code that compares initial vs final DB state. Returns: Verifier result dict, or error dict with "execution_status": "error". """ if not os.path.exists(initial_db_path): return { "execution_status": "error", "error_message": f"Initial DB not found: {initial_db_path}", } if not os.path.exists(final_db_path): return { "execution_status": "error", "error_message": f"Final DB not found: {final_db_path}", } original_modes = {} for path in [initial_db_path, final_db_path]: try: original_modes[path] = os.stat(path).st_mode os.chmod(path, 0o444) except Exception: pass try: return _run_verifier_subprocess( { "verifier_code": verifier_code, "function_name": function_name, "initial_db_path": initial_db_path, "final_db_path": final_db_path, "final_answer": None, "mode": "sql", } ) finally: for path, mode in original_modes.items(): try: os.chmod(path, mode) except Exception: pass def execute_code_verifier( verifier_code: str, function_name: str, initial_db_path: str, final_db_path: str, final_answer: str | None = None, ) -> dict[str, Any]: """Execute code-mode verifier that returns {"result": "complete"|"others"}. Returns: Dict with "result" key ("complete" or "others") and "execution_status". """ if not os.path.exists(initial_db_path): return { "result": "others", "execution_status": "error", "error_message": f"Initial DB not found: {initial_db_path}", } if not os.path.exists(final_db_path): return { "result": "others", "execution_status": "error", "error_message": f"Final DB not found: {final_db_path}", } original_modes = {} for path in [initial_db_path, final_db_path]: try: original_modes[path] = os.stat(path).st_mode os.chmod(path, 0o444) except Exception: pass try: result = _run_verifier_subprocess( { "verifier_code": verifier_code, "function_name": function_name, "initial_db_path": initial_db_path, "final_db_path": final_db_path, "final_answer": final_answer, "mode": "code", } ) finally: for path, mode in original_modes.items(): try: os.chmod(path, mode) except Exception: pass # Normalize: code-mode callers expect "result" key on every return. if "result" not in result: result.setdefault("execution_status", "error") result["result"] = "others" return result def run_verifier( verifier_entry: dict, verifier_mode: str, initial_db_path: str, final_db_path: str, final_answer: str | None = None, ) -> tuple[str, dict]: """ Run the appropriate verifier and return (reward_type, details). Returns: (reward_type, verify_result_dict) """ verification = verifier_entry.get("verification", {}) code = verification.get("code", "") if not code or not isinstance(code, str) or len(code.strip()) < 10: return "judge_error", {"error": "No valid verifier code found"} default_func = ( "verify_task_completion" if verifier_mode == "code" else "verify_task" ) func_name = default_func for line in code.split("\n"): line = line.strip() if line.startswith("def verify_") and "(" in line: func_name = line.split("(")[0].replace("def ", "").strip() break if verifier_mode == "code": result = execute_code_verifier( code, func_name, initial_db_path, final_db_path, final_answer ) if result.get("execution_status") == "error": return "judge_error", result return result.get("result", "others"), result result = execute_sql_verifier(code, func_name, initial_db_path, final_db_path) if isinstance(result, dict) and result.get("execution_status") == "error": return "judge_error", result return "incomplete", result def _normalize_azure_url(base_url: str) -> str: """Normalize Azure OpenAI base URLs to include /openai/v1 suffix. The Azure OpenAI v1 API uses the standard OpenAI client with base_url set to https://RESOURCE.openai.azure.com/openai/v1/. This helper auto-appends /openai/v1 when it detects an Azure endpoint. """ if "openai.azure.com" in base_url or "services.ai.azure.com" in base_url: stripped = base_url.rstrip("/") if not stripped.endswith("/openai/v1"): return stripped + "/openai/v1" return base_url async def run_llm_judge( task: str, verifier_result: dict, llm_base_url: str | None = None, llm_api_key: str | None = None, llm_model: str | None = None, trajectory: list | None = None, verifier_reasoning: str = "", success_criteria: str = "", failure_criteria: str = "", ) -> tuple[str, dict]: """Run LLM-as-a-judge combining agent trajectory AND SQL verification results. Aligned with llm_judge_with_sql in agent-world-model-rl-rewards/task_reward.py. Works with both OpenAI and Azure OpenAI (v1 API). For Azure, set llm_base_url to https://YOUR-RESOURCE.openai.azure.com and the URL will be auto-normalized to append /openai/v1. Returns: (classification, judge_result_dict) """ if not llm_base_url or not llm_model: return "judge_error", { "error": "LLM endpoint not configured for sql verifier mode" } llm_base_url = _normalize_azure_url(llm_base_url) try: client = AsyncOpenAI(base_url=llm_base_url, api_key=llm_api_key or "EMPTY") system_prompt = textwrap.dedent("""\ You are an impartial evaluator of automated agent task results with access to database verification. Based on the provided JSON trajectory AND the Python verification results from querying the database, decide the task outcome. This trajectory is generated by an MCP agent on a simulated simplified environment. The environment provides a set of MCP tools to help the agent complete the task. Input: task_json: dict containing task fields such as user task and agent execution_history. verification_json: dict containing Python code, reasoning, success_criteria, failure_criteria, and execution results that verified the database state from the Python function Output: You only output UTF-8 encoded string, avoid any emoji or special characters. You only output English text. Classification categories: - complete: all required steps and closure actions were successfully executed, AND the database state confirms the task was completed - incomplete: partial progress or the database state shows the task is not fully completed - server_error: the agent is blocked by MCP server/environment error, e.g., 5xx errors such as "Internal Server Error". Or the MCP server cannot process the valid tool call and return valid results. This can block the agent from completing the task. - agent_error: the agent made mistakes, invalid parameters, or missing required data without recovery, failed to complete the user's instruction. Priority order for classification: 1) complete (trajectory shows success AND database confirms it) 2) server_error (due to the MCP server/environment error) 3) agent_error (agent-side issue, e.g., invalid mcp_tool_call arguments, hallucination, agent mistakes) 4) incomplete (everything else unfinished or database state doesn't match expected outcome) Key considerations: - The verification_json contains checks performed on the database states. You can read the verification code to understand what checks were performed on the database states. - The verification_json contains the execution results of the verification code. You can use the execution results to help you judge the task completion. - The verification_json contains guidance for how to utilize the code execution results to help you judge the task completion, also including success_criteria and failure_criteria. - The verification results may be empty or error, or even the verification code itself is inaccurate. You should not fully rely on the verification results. You need to comprehensively consider the trajectory information to help you judge the task completion. Output format (must be valid JSON, no markdown fences, no additional commentary): { "reasoning": "", "confidence_score": [0-100, 0-100, 0-100, 0-100] for complete, incomplete, server_error, agent_error respectively, "classification": "", "evidence": { "status": "", "iterations": , "error_signals": [""], "last_actions": [""], "last_status_codes": [], "database_verification": "" } }""") task_payload = { "user_task": task, "actual_execution_steps": len(trajectory) if trajectory else 0, "trajectory": trajectory or [], } verification_json = _sanitize_for_json( { "reasoning": verifier_reasoning, "success_criteria": success_criteria, "failure_criteria": failure_criteria, "code_execution_result": verifier_result, } ) try: verification_json_str = json.dumps( verification_json, ensure_ascii=False, indent=2, default=str ) except Exception: verification_json_str = str(verification_json) user_prompt = ( f"task_json:\n" f"{json.dumps(task_payload, ensure_ascii=False, indent=2, default=str)}\n\n" f"verification_json:\n" f"{verification_json_str}" ) response = await client.chat.completions.create( model=llm_model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=1.0, max_completion_tokens=4096, ) content = response.choices[0].message.content or "" try: result = json.loads(content) except json.JSONDecodeError: match = re.search(r"\{[\s\S]*\}", content) if match: try: result = json.loads(match.group()) except json.JSONDecodeError: return "judge_error", { "error": f"Failed to parse LLM response: {content}" } else: return "judge_error", { "error": f"Failed to parse LLM response: {content}" } classification = result.get("classification", "judge_error").lower().strip() valid = {"complete", "incomplete", "server_error", "agent_error", "judge_error"} if classification not in valid: classification = "judge_error" return classification, result except Exception as e: logger.error(f"LLM judge failed: {e}") return "judge_error", {"error": str(e)} def _sanitize_for_json(obj: Any) -> Any: if isinstance(obj, dict): return { str(k) if isinstance(k, tuple) else k: _sanitize_for_json(v) for k, v in obj.items() } elif isinstance(obj, (list, tuple)): return [_sanitize_for_json(item) for item in obj] elif isinstance(obj, bytes): return obj.decode("utf-8", errors="ignore") else: try: json.dumps(obj) return obj except TypeError: return str(obj)