""" 让原生 Codex CLI 操作同一假世界，模型循环与独立评分器不共用状态。 """ import asyncio import copy import hashlib import json import os import shutil import signal import subprocess import tempfile import time from contextlib import contextmanager, nullcontext from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from threading import Thread from typing import Any from scripts.evaluation.live import _parse_final, _validate_report, _worker_environment from scripts.evaluation.models import ModelSettings from scripts.evaluation.proxy import NATIVE_BROWSER_TOOLS, NATIVE_SHELL_TOOLS, EvaluationModelProxy from scripts.evaluation.score import evaluate from scripts.evaluation.server import EvaluationMcpServer from scripts.evaluation.world import EvaluationWorld SUPPORTED_CLI_VERSION = "codex-cli 0.153.4" MAX_PROCESS_OUTPUT_BYTES = 8 * 1024 * 1024 NATIVE_TERMINAL_SCENARIOS = frozenset( {"terminal_session", "terminal_pty_session", "subagent_terminal_share"} ) NATIVE_BROWSER_SCENARIOS = frozenset({"browser_navigation"}) NATIVE_STEERING_SCENARIOS = frozenset({"steering_long_context", "steering_multi_message"}) NATIVE_STEERING_MESSAGE_ID = "moviepilot-evaluation-steering" def _native_steering_message_id(index: int, count: int) -> str: """ 为原生多条 steering 事件生成稳定且可审计的消息标识。 """ return NATIVE_STEERING_MESSAGE_ID if count == 1 else f"{NATIVE_STEERING_MESSAGE_ID}-{index + 1}" def _browser_runtime_configuration(executable: str, instruction_dir: Path) -> dict[str, Any] | None: """ 发现本机已安装的浏览器插件运行时，缺少任一受信文件就保持原生边界关闭。 """ codex_home = Path(os.environ.get("CODEX_HOME") or Path.home() / ".codex") plugin_root = Path(os.environ.get( "MOVIEPILOT_EVAL_BROWSER_PLUGIN_ROOT", str(codex_home / ".tmp" / "bundled-marketplaces" / "openai-bundled" / "plugins" / "browser"), )) node_repl = Path(os.environ.get( "MOVIEPILOT_EVAL_NODE_REPL", "/Applications/ChatGPT.app/Contents/Resources/cua_node/bin/node_repl", )) node_path = Path(os.environ.get("MOVIEPILOT_EVAL_NODE_PATH", str(node_repl.parent / "node"))) node_modules = Path(os.environ.get( "MOVIEPILOT_EVAL_NODE_MODULES", str(node_repl.parent.parent / "lib" / "node_modules"), )) browser_service = os.environ.get("MOVIEPILOT_EVAL_BROWSER_SERVICE") service_candidates = ([Path(browser_service)] if browser_service else []) + sorted( (codex_home / "plugins" / "cache" / "openai-bundled" / "browser").glob( "*/scripts/browser-service.mjs" ), reverse=True, ) browser_service_path = next((path for path in service_candidates if path.is_file()), None) browser_client = plugin_root / "scripts" / "browser-client.mjs" if (not node_repl.is_file() or not node_path.is_file() or not node_modules.is_dir() or not browser_client.is_file() or browser_service_path is None): return None marketplace = plugin_root.parent.parent client_hashes = { hashlib.sha256(browser_client.read_bytes()).hexdigest(), } chrome_client = codex_home / "plugins" / "cache" / "openai-bundled" / "chrome" / "latest" / "scripts" / "browser-client.mjs" if chrome_client.is_file(): client_hashes.add(hashlib.sha256(chrome_client.read_bytes()).hexdigest()) trusted_paths = ":".join(str(path) for path in (codex_home, node_modules, plugin_root)) node_environment = { "NODE_REPL_NATIVE_PIPE_CONNECT_TIMEOUT_MS": "1000", "NODE_REPL_NODE_MODULE_DIRS": str(node_modules), "NODE_REPL_NODE_PATH": str(node_path), "NODE_REPL_TRUSTED_CODE_PATHS": trusted_paths, "NODE_REPL_TRUSTED_BROWSER_CLIENT_SHA256S": ",".join(sorted(client_hashes)), "NODE_REPL_TRUSTED_SERVICES": json.dumps( {"browser": str(browser_service_path), "sky": "@oai/sky/service"}, ensure_ascii=False ), "CODEX_HOME": str(codex_home), "BROWSER_USE_AVAILABLE_BACKENDS": "chrome,iab", "BROWSER_USE_TINYSKY_ENABLED": "1", "BROWSER_USE_CODEX_APP_BUILD_FLAVOR": "prod", "BROWSER_USE_CODEX_APP_VERSION": browser_service_path.parent.parent.name, "BROWSER_USE_BROWSER_CLIENT_BUILD": browser_service_path.parent.parent.name, "CODEX_CLI_PATH": executable, } # 安装后的 Skill 使用可移植的 标记；独立命令行没有桌面插件上下文， # 因此复制一份相同指令，并用已验证的绝对路径展开该标记。 skill = plugin_root / "skills" / "control-in-app-browser" / "SKILL.md" expanded_skill = instruction_dir / "browser-control-skill.md" expanded_skill.write_text(skill.read_text(encoding="utf-8").replace("", str(plugin_root)), encoding="utf-8") return { "node_repl": { "command": str(node_repl), "args": [], "enabled": True, "required": False, "startup_timeout_sec": 120, "tool_timeout_sec": 120, "default_tools_approval_mode": "approve", "env": node_environment, }, "marketplace": {"source_type": "local", "source": str(marketplace)}, "plugins": {"browser@openai-bundled": {"enabled": True}, "chrome@openai-bundled": {"enabled": True}}, "skill": str(expanded_skill), } @contextmanager def _native_browser_fixture() -> Any: """ 启动只绑定回环地址的动态测试页，并记录真实浏览器的点击回调。 """ state = {"clicked": False} class _Handler(BaseHTTPRequestHandler): """ 提供最小页面和点击回调，不记录请求正文或客户端环境。 """ def do_GET(self) -> None: # noqa: N802 - 标准库处理器方法名 """ 返回测试页面或确认按钮回调，其他路径返回 404。 """ if self.path == "/fixture": body = ( "MoviePilot Browser Fixture" "

MoviePilot Browser Fixture

" "

PENDING

" "" "
" ).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) return if self.path == "/clicked": state["clicked"] = True body = b"ok" self.send_response(204) self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(body))) self.end_headers() return self.send_error(404) def log_message(self, _format: str, *_args: Any) -> None: """ 禁止标准库把本地测试请求写入 stderr。 """ server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) thread = Thread(target=server.serve_forever, name="moviepilot-native-browser", daemon=True) thread.start() try: yield f"http://127.0.0.1:{server.server_port}/fixture", state finally: server.shutdown() server.server_close() thread.join(timeout=5) def _catalog(payload: Any, model: str) -> tuple[dict[str, Any], dict[str, Any]]: """ 仅修改实际二进制自带模型的工具模式，原生提示词及其他模型字段逐字保留。 """ if not isinstance(payload, dict) or not isinstance(payload.get("models"), list): raise ValueError("原生模型目录结构无效") matches = [row for row in payload["models"] if isinstance(row, dict) and row.get("slug") == model] if len(matches) != 1: raise ValueError("原生二进制没有唯一的指定模型元数据，不能替换为其他模型") original = matches[0] projected = copy.deepcopy(original) projected["tool_mode"] = "direct" def digest(value: Any) -> str: """ 对相同序列化规则取指纹，便于复验单字段投影。 """ return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True).encode()).hexdigest() metadata = {"source": "codex_debug_models_bundled", "model": model, "original_tool_mode": original.get("tool_mode"), "projected_tool_mode": "direct", "original_model_sha256": digest(original), "projected_model_sha256": digest(projected), "base_instructions_sha256": digest(original.get("base_instructions")), "model_messages_sha256": digest(original.get("model_messages"))} return {"models": [projected]}, metadata def _toml(value: Any) -> str: """ 按 TOML 表达配置值，直接传 argv，不经过 shell 或其变量展开。 """ if isinstance(value, dict): return "{ " + ", ".join(f"{json.dumps(key)} = {_toml(item)}" for key, item in value.items()) + " }" return json.dumps(value, ensure_ascii=False) def _redact(value: Any, tokens: tuple[str, ...]) -> Any: """ 在全部事件和最终 JSON 解码后移除局部令牌，Unicode 转义不能绕过。 """ if isinstance(value, str): for token in tokens: value = value.replace(token, "[local evaluation token]") return value if isinstance(value, dict): return {_redact(key, tokens): _redact(item, tokens) for key, item in value.items()} if isinstance(value, list): return [_redact(item, tokens) for item in value] return value def _configuration(settings: ModelSettings, proxy: EvaluationModelProxy, server: EvaluationMcpServer, control_dir: Path, *, scenario_id: str = "", browser_runtime: dict[str, Any] | None = None) -> dict[str, Any]: """ 显式隔离原生客户端的外部能力，仍保留其默认提示词、计划及子代理循环。 """ disabled = [ "apps", "browser_use", "browser_use_external", "browser_use_full_cdp_access", "computer_use", "in_app_browser", "in_app_chat", "image_generation", "view_image", "hooks", "memories", "plugins", "remote_plugin", "plugin_sharing", "skill_search", "skill_mcp_dependency_install", "workspace_dependencies", "goals", "sleep_tool", "code_mode", "code_mode_host", "enable_request_compression", "unbounded_connection_retries", "tool_suggest", ] if scenario_id not in { "command_execution", "terminal_session", "terminal_pty_session", "subagent_terminal_share", }: disabled.extend(("shell_tool", "unified_exec", "shell_snapshot")) if scenario_id == "browser_navigation": for feature in ("browser_use", "browser_use_external", "browser_use_full_cdp_access", "computer_use", "in_app_browser"): disabled.remove(feature) configuration = { "model": settings.model, "model_provider": "evaluation", "model_reasoning_effort": settings.reasoning_effort, "model_context_window": settings.context_window, "approval_policy": "never", "web_search": "disabled", "tools.update_plan.enabled": True, "project_doc_max_bytes": 0, "skills.bundled.enabled": False, "skills.include_instructions": False, "sqlite_home": str(control_dir / "state"), "log_dir": str(control_dir / "logs"), "model_catalog_json": str(control_dir / "models.json"), "features.skip_host_skill_discovery": True, **{f"features.{name}": False for name in disabled}, "features.code_mode_only": False, "features.code_mode_host": {"enabled": False, "disable_in_process_fallback": False}, "model_providers.evaluation": { "name": "Controlled evaluation proxy", "base_url": proxy.endpoint, "env_key": "MOVIEPILOT_EVAL_MODEL_TOKEN", "wire_api": "responses", "requires_openai_auth": False, "request_max_retries": 0, "stream_max_retries": 0, "stream_idle_timeout_ms": settings.timeout_seconds * 1000, }, "mcp_servers": {"evaluation": { "url": server.endpoint, "bearer_token_env_var": "MOVIEPILOT_EVAL_MCP_TOKEN", "enabled": True, "required": True, "startup_timeout_sec": 10, "tool_timeout_sec": 15, "enabled_tools": ["moviepilot_api", "read_skill", "read_tool_result"], "tools": {name: {"approval_mode": "approve"} for name in ("moviepilot_api", "read_skill", "read_tool_result")}, }}, } if scenario_id in { "command_execution", "terminal_session", "terminal_pty_session", "subagent_terminal_share", }: # 命令场景只开放 CLI 已核对的两个终端动作；仍使用只读沙箱和 never 审批。 configuration.update({f"features.{name}": True for name in ("shell_tool", "unified_exec", "shell_snapshot")}) elif scenario_id == "browser_navigation": # 浏览器插件只能在本机受信运行时完整存在时开启，缺失时保留 blocked 证据。 configuration.update({f"features.{name}": True for name in ( "browser_use", "browser_use_external", "browser_use_full_cdp_access", "computer_use", "in_app_browser", )}) if browser_runtime is not None: configuration.update({ "features.plugins": True, "skills.bundled.enabled": True, "skills.include_instructions": False, "mcp_servers": {**configuration["mcp_servers"], "node_repl": browser_runtime["node_repl"]}, "marketplaces.openai-bundled": browser_runtime["marketplace"], "plugins.browser@openai-bundled": browser_runtime["plugins"]["browser@openai-bundled"], "plugins.chrome@openai-bundled": browser_runtime["plugins"]["chrome@openai-bundled"], # 独立执行不继承桌面端已选插件上下文，需通过原生指令入口加载浏览器 Skill。 "model_instructions_file": browser_runtime["skill"], }) return configuration async def _read_output(stream: asyncio.StreamReader, result: bytearray) -> None: """ 持续排空子进程管道，输出超限时让控制器终止整组进程。 """ while chunk := await stream.read(65536): remaining = MAX_PROCESS_OUTPUT_BYTES - len(result) result.extend(chunk[:remaining]) if len(chunk) > remaining: raise RuntimeError("原生评测进程输出超过 8 MiB") async def _stop_process(process: asyncio.subprocess.Process) -> None: """ 父任务取消或超时时回收原生客户端及继承进程组，不遗留后台评测。 """ if os.name == "posix": try: os.killpg(process.pid, signal.SIGKILL) except ProcessLookupError: pass elif process.returncode is None: process.kill() await asyncio.wait_for(process.wait(), timeout=10) async def _execute(command: list[str], prompt: str, environment: dict[str, str], work_dir: Path, timeout: int) -> dict[str, Any]: """ 只向原生 stdin 发送公开任务，结构化事件和退出状态由控制器采集。 """ process = await asyncio.create_subprocess_exec( *command, cwd=work_dir, env=environment, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, start_new_session=os.name == "posix", limit=MAX_PROCESS_OUTPUT_BYTES, ) readers: list[asyncio.Task[None]] = [] stdout, stderr = bytearray(), bytearray() failure = None try: assert process.stdin is not None and process.stdout is not None and process.stderr is not None readers = [asyncio.create_task(_read_output(process.stdout, stdout)), asyncio.create_task(_read_output(process.stderr, stderr))] process.stdin.write(prompt.encode("utf-8")) await process.stdin.drain() process.stdin.close() async with asyncio.timeout(timeout): await asyncio.gather(*readers) await process.wait() except (RuntimeError, TimeoutError, OSError) as error: failure = type(error).__name__ finally: try: await _stop_process(process) finally: for task in readers: if not task.done(): task.cancel() await asyncio.gather(*readers, return_exceptions=True) return {"returncode": process.returncode, "stdout": stdout.decode("utf-8", errors="replace"), "stderr": stderr.decode("utf-8", errors="replace"), "error_type": failure} def _normalize_app_server_item(item: Any) -> dict[str, Any]: """ 把 app-server 的 camelCase item 转成评测器共用的安全事件字段。 """ if not isinstance(item, dict): return {"type": "unknown"} item_type = item.get("type") normalized_type = { "agentMessage": "agent_message", "commandExecution": "command_execution", "collabAgentToolCall": "collab_tool_call", "fileChange": "file_change", "userMessage": "user_message", }.get(item_type, item_type if isinstance(item_type, str) else "unknown") result: dict[str, Any] = {"type": normalized_type, "id": item.get("id")} if normalized_type == "agent_message": result.update(text=item.get("text", ""), phase=item.get("phase")) elif normalized_type == "mcpToolCall": result.update( server=item.get("server"), tool=item.get("tool"), status=item.get("status"), arguments=item.get("arguments"), result=item.get("result"), ) elif normalized_type == "user_message": result.update(text=item.get("text", ""), content=item.get("content")) elif normalized_type == "command_execution": result.update( command=item.get("command", ""), cwd=item.get("cwd"), process_id=item.get("processId"), status=item.get("status"), aggregated_output=item.get("aggregatedOutput"), exit_code=item.get("exitCode"), command_actions=item.get("commandActions"), ) elif normalized_type == "collab_tool_call": # 保留协作调用的可审计合同；如果只留 type/id，无法区分真实派发、取消和等待。 collab_tools = { "spawnAgent": "spawn_agent", "sendInput": "send_input", "resumeAgent": "resume_agent", "wait": "wait_agent", "closeAgent": "close_agent", "sendMessage": "send_message", "followupTask": "followup_task", "interruptAgent": "interrupt_agent", "listAgents": "list_agents", } raw_states = item.get("agentsStates") agents_states = {} if isinstance(raw_states, dict): for agent_id, state in raw_states.items(): if not isinstance(state, dict): continue agents_states[str(agent_id)] = { "status": state.get("status"), "message": state.get("message"), } result.update( tool=collab_tools.get(item.get("tool"), item.get("tool")), status=item.get("status"), prompt=item.get("prompt"), model=item.get("model"), reasoning_effort=item.get("reasoningEffort"), sender_thread_id=item.get("senderThreadId"), receiver_thread_ids=item.get("receiverThreadIds", []), agents_states=agents_states, ) return result def _normalize_app_server_notification(method: str, params: Any) -> dict[str, Any] | None: """ 保留终端、消息和轮次终态，把其他 app-server 通知降为可审计摘要。 """ payload = params if isinstance(params, dict) else {} if method == "turn/started": turn = payload.get("turn") if isinstance(payload.get("turn"), dict) else {} return {"type": "turn.started", "thread_id": payload.get("threadId"), "turn_id": turn.get("id")} if method == "turn/completed": turn = payload.get("turn") if isinstance(payload.get("turn"), dict) else {} return {"type": "turn.completed", "thread_id": payload.get("threadId"), "turn_id": turn.get("id")} if method == "item/started": return { "type": "item.started", "thread_id": payload.get("threadId"), "turn_id": payload.get("turnId"), "item": _normalize_app_server_item(payload.get("item")), } if method == "item/completed": return { "type": "item.completed", "thread_id": payload.get("threadId"), "turn_id": payload.get("turnId"), "item": _normalize_app_server_item(payload.get("item")), } if method == "item/agentMessage/delta": return { "type": "item.agent_message.delta", "thread_id": payload.get("threadId"), "turn_id": payload.get("turnId"), "item_id": payload.get("itemId"), "delta": payload.get("delta", ""), } if method == "item/commandExecution/outputDelta": return { "type": "item.command_execution.output_delta", "thread_id": payload.get("threadId"), "turn_id": payload.get("turnId"), "item_id": payload.get("itemId"), "delta": payload.get("delta", ""), } if method == "item/commandExecution/terminalInteraction": return { "type": "item.command_execution.terminal_interaction", "thread_id": payload.get("threadId"), "turn_id": payload.get("turnId"), "item_id": payload.get("itemId"), "process_id": payload.get("processId"), "stdin": payload.get("stdin", ""), } if method == "error": return {"type": "error", "error": payload} if method == "warning": return {"type": "warning", "warning": payload} return {"type": "native.notification", "method": method} if method else None def _append_thread_history_events(events: list[dict[str, Any]], thread: Any) -> None: """ 把 thread/read 返回的子代理历史转换为带 scope 的完成事件。 """ if not isinstance(thread, dict): return thread_id = thread.get("id") if not isinstance(thread_id, str) or not thread_id: return turns = thread.get("turns") if not isinstance(turns, list): return for turn in turns: if not isinstance(turn, dict): continue turn_id = turn.get("id") if not isinstance(turn_id, str) or not turn_id: continue items = turn.get("items") if not isinstance(items, list): continue for item in items: normalized = _normalize_app_server_item(item) if normalized.get("type") == "unknown": continue events.append({ "type": "item.completed", "thread_id": thread_id, "turn_id": turn_id, "item": normalized, "thread_history": True, }) async def _execute_app_server( command: list[str], prompt: str, environment: dict[str, str], work_dir: Path, timeout: int, *, model: str, reasoning_effort: str, steering_message: str = "", steering_plan: tuple[tuple[int, str], ...] = (), ) -> dict[str, Any]: """ 通过 Codex app-server 的 JSON-RPC 驱动终端和多条中途补充消息。 """ process = await asyncio.create_subprocess_exec( *command, cwd=work_dir, env=environment, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, start_new_session=os.name == "posix", limit=MAX_PROCESS_OUTPUT_BYTES, ) assert process.stdin is not None and process.stdout is not None and process.stderr is not None stderr = bytearray() stderr_reader = asyncio.create_task(_read_output(process.stderr, stderr)) events: list[dict[str, Any]] = [] stdout_bytes = 0 pending: dict[int, str] = {} next_request_id = 1 thread_id: str | None = None active_turn_id: str | None = None turn_completed = False failure: str | None = None schedule = steering_plan if not schedule and steering_message: schedule = ((1, steering_message),) steering_index = 0 business_tool_count = 0 steering_requests: dict[int, tuple[int, str]] = {} steering_request_pending = False child_thread_ids: set[str] = set() child_thread_read_requests: dict[int, str] = {} child_thread_read_requested: set[str] = set() async def send_request(method: str, params: dict[str, Any]) -> int: """ 发送带序号的 JSON-RPC 请求，并记录响应所属方法。 """ nonlocal next_request_id request_id = next_request_id next_request_id += 1 pending[request_id] = method process.stdin.write( (json.dumps({"jsonrpc": "2.0", "id": request_id, "method": method, "params": params}) + "\n").encode() ) await process.stdin.drain() return request_id async def send_notification(method: str) -> None: """ 发送不等待响应的 JSON-RPC 通知。 """ process.stdin.write((json.dumps({"jsonrpc": "2.0", "method": method}) + "\n").encode()) await process.stdin.drain() async def send_server_error(request_id: Any) -> None: """ 拒绝意外的原生服务端请求，避免评测控制器代替用户授权。 """ process.stdin.write( (json.dumps({ "jsonrpc": "2.0", "id": request_id, "error": {"code": -32000, "message": "评测适配器不接受原生交互授权请求"}, }) + "\n").encode() ) await process.stdin.drain() async def send_child_thread_reads_if_ready() -> None: """ 在父轮次结束后读取已完成子代理历史，补齐可能遗漏的终端 item。 """ if not turn_completed: return for child_id in sorted(child_thread_ids - child_thread_read_requested): request_id = await send_request("thread/read", {"threadId": child_id, "includeTurns": True}) child_thread_read_requests[request_id] = child_id child_thread_read_requested.add(child_id) async def send_next_steering_if_ready() -> None: """ 在达到计划回执边界后逐条发送 steering，并保留请求身份。 """ nonlocal failure, steering_index, steering_request_pending if steering_request_pending or steering_index >= len(schedule): return trigger_count, text = schedule[steering_index] if business_tool_count < trigger_count: return if not thread_id or not active_turn_id: events.append({ "type": "evaluation.steering.boundary_missing", "thread_id_present": bool(thread_id), "turn_id_present": bool(active_turn_id), }) failure = "app_server_steering_turn_id_missing" return message_id = _native_steering_message_id(steering_index, len(schedule)) events.append({"type": "evaluation.steering.queued", "status": "queued", "message_id": message_id}) request_id = await send_request( "turn/steer", { "threadId": thread_id, "expectedTurnId": active_turn_id, "input": [{"type": "text", "text": text}], "clientUserMessageId": message_id, }, ) steering_requests[request_id] = (steering_index, message_id) steering_index += 1 steering_request_pending = True async def handle_response(payload: dict[str, Any]) -> None: """ 按 initialize、thread/start、turn/start 的顺序推进原生轮次。 """ nonlocal thread_id, failure, steering_request_pending request_id = payload.get("id") method = pending.pop(request_id, None) if type(request_id) is int else None steering_request = steering_requests.pop(request_id, None) if type(request_id) is int else None child_thread_id = child_thread_read_requests.pop(request_id, None) if type(request_id) is int else None if method is None: await send_server_error(request_id) failure = failure or "unexpected_server_request" return if "error" in payload: failure = failure or f"app_server_{method.replace('/', '_')}_error" if method == "turn/steer": steering_request_pending = False events.append({"type": "evaluation.steering.failed", "status": "failed", "message_id": steering_request[1] if steering_request else NATIVE_STEERING_MESSAGE_ID}) elif method == "thread/read": events.append({"type": "evaluation.child_thread_read.failed", "thread_id": child_thread_id}) return if method == "initialize": await send_notification("initialized") await send_request( "thread/start", {"cwd": str(work_dir), "model": model, "sandbox": "read-only", "approvalPolicy": "never", "ephemeral": True}, ) elif method == "thread/start": result = payload.get("result") thread = result.get("thread") if isinstance(result, dict) else None thread_id = thread.get("id") if isinstance(thread, dict) else None if not isinstance(thread_id, str) or not thread_id: failure = failure or "app_server_thread_id_missing" return await send_request( "turn/start", { "threadId": thread_id, "input": [{"type": "text", "text": prompt}], "model": model, "effort": reasoning_effort, }, ) elif method == "turn/steer": steering_request_pending = False events.append({"type": "evaluation.steering.applied", "status": "applied", "message_id": steering_request[1] if steering_request else NATIVE_STEERING_MESSAGE_ID}) await send_next_steering_if_ready() elif method == "thread/read": result = payload.get("result") thread = result.get("thread") if isinstance(result, dict) else None _append_thread_history_events(events, thread) try: async with asyncio.timeout(timeout): await send_request( "initialize", {"clientInfo": {"name": "moviepilot-evaluation", "version": "1"}, "capabilities": {"experimentalApi": True}}, ) while True: line = await process.stdout.readline() if not line: break stdout_bytes += len(line) if stdout_bytes + len(stderr) > MAX_PROCESS_OUTPUT_BYTES: raise RuntimeError("原生 app-server 输出超过 8 MiB") try: payload = json.loads(line) except (TypeError, ValueError): events.append({"type": "evaluation.invalid_event", "sha256": hashlib.sha256(line).hexdigest()}) failure = failure or "invalid_native_events" continue if not isinstance(payload, dict): failure = failure or "invalid_native_events" continue if "id" in payload and "method" not in payload: await handle_response(payload) if turn_completed and pending == {}: break continue method = payload.get("method") if not isinstance(method, str): failure = failure or "invalid_native_events" continue event = _normalize_app_server_notification(method, payload.get("params")) if event is not None: events.append(event) item = event.get("item") if ( isinstance(item, dict) and item.get("type") == "collab_tool_call" and item.get("tool") == "spawn_agent" ): child_thread_ids.update( child_id for child_id in item.get("receiver_thread_ids", []) if isinstance(child_id, str) and child_id ) if ( event.get("type") == "item.completed" and isinstance(item, dict) and item.get("type") == "mcpToolCall" and str(item.get("tool") or "").rsplit(".", 1)[-1] == "moviepilot_api" ): business_tool_count += 1 await send_next_steering_if_ready() if method == "turn/started": active_turn_id = event.get("turn_id") if isinstance(event.get("turn_id"), str) else None if method == "turn/completed": turn_completed = True if schedule and any(value == "turn/steer" for value in pending.values()): continue await send_child_thread_reads_if_ready() if not child_thread_read_requests: break continue if "id" in payload: await send_server_error(payload["id"]) failure = failure or "unexpected_server_request" except (RuntimeError, TimeoutError, OSError) as error: failure = failure or type(error).__name__ finally: if process.returncode is None: if turn_completed and failure is None: process.stdin.close() try: await asyncio.wait_for(process.wait(), timeout=10) except TimeoutError: failure = failure or "app_server_shutdown_timeout" await _stop_process(process) else: await _stop_process(process) if not stderr_reader.done(): stderr_reader.cancel() reader_result = (await asyncio.gather(stderr_reader, return_exceptions=True))[0] if isinstance(reader_result, BaseException) and not isinstance(reader_result, asyncio.CancelledError): failure = failure or type(reader_result).__name__ return { "returncode": process.returncode, "stdout": "\n".join(json.dumps(event, ensure_ascii=False) for event in events) + ("\n" if events else ""), "stderr": stderr.decode("utf-8", errors="replace"), "error_type": failure, } def _events(text: str) -> tuple[list[dict[str, Any]], str, bool]: """ 保留 JSONL 原生事件；只有完成事件之后的有效最终报告才由 oracle 判断。 """ events, final_text, completed, malformed = [], "", False, False for line in text.splitlines(): try: event = json.loads(line) if not isinstance(event, dict): raise ValueError("原生客户端输出非事件对象") except (ValueError, TypeError): malformed = True events.append({"type": "evaluation.invalid_event", "sha256": hashlib.sha256(line.encode()).hexdigest()}) continue events.append(event) if event.get("type") == "turn.started": completed, final_text = False, "" item = event.get("item") if event.get("type") == "item.completed" and isinstance(item, dict) and item.get("type") == "agent_message": final_text = str(item.get("text") or "") if event.get("type") == "turn.completed": completed = True elif event.get("type") in {"turn.failed", "error"}: completed = False return events, final_text, completed and not malformed def _record_native_command_events(world: EvaluationWorld, events: list[dict[str, Any]]) -> None: """ 把 CLI 的已完成命令事件映射为同一场景账本，避免只看模型文字。 """ seen: set[str] = set() streamed_output: dict[str, str] = {} streamed_input: set[str] = set() child_thread_ids: set[str] = set() for event in events: if event.get("type") not in {"item.started", "item.completed"}: continue item = event.get("item") if not isinstance(item, dict) or item.get("type") != "collab_tool_call" or item.get("tool") != "spawn_agent": continue child_thread_ids.update( receiver_thread_id for receiver_thread_id in item.get("receiver_thread_ids", []) if isinstance(receiver_thread_id, str) ) for event in events: event_type = event.get("type") if event_type == "item.command_execution.output_delta": item_id = event.get("item_id") delta = event.get("delta") if isinstance(item_id, str) and isinstance(delta, str): streamed_output[item_id] = streamed_output.get(item_id, "") + delta continue if event_type == "item.command_execution.terminal_interaction": item_id = event.get("item_id") stdin = event.get("stdin") if isinstance(item_id, str) and isinstance(stdin, str) and "MOVIEPILOT_TERMINAL_OK" in stdin: streamed_input.add(item_id) continue if event_type != "item.completed": continue item = event.get("item") if not isinstance(item, dict) or item.get("type") not in {"command_execution", "command_execution_output"}: continue command = item.get("command") if not isinstance(command, str) or not command.strip(): continue identity = str(item.get("id") or command) if identity in seen: continue seen.add(identity) output = item.get("aggregated_output", item.get("output", "")) if not isinstance(output, str): output = str(output or "") if not output: output = streamed_output.get(identity, "") exit_code = item.get("exit_code", item.get("exitCode")) if type(exit_code) is not int: exit_code = None outcome = "succeeded" if exit_code == 0 else ("unknown" if exit_code is None else "failed") # 终端输入必须有 app-server 的 terminalInteraction 事件作为证据。 # 聚合输出可能只是命令自身打印了相同字符串，不能据此冒充 stdin 写入。 terminal_input_observed = world.scenario.kind == "terminal" and identity in streamed_input thread_id = event.get("thread_id") scope_kind = "subagent" if thread_id in child_thread_ids else "conversation" world.record_command(command.strip(), { "action": "start" if world.scenario.kind == "terminal" else "run", "success": outcome == "succeeded", "execution_outcome": outcome, "status": "exited" if exit_code is not None else "unknown", "exit_code": exit_code, "timed_out": False, "output": output, "terminal_input_observed": terminal_input_observed, }, action="start" if world.scenario.kind == "terminal" else "run", scope_kind=scope_kind, scope_task_id=thread_id if isinstance(thread_id, str) else None) def _probe_ready(usage: dict[str, Any], server_stats: dict[str, Any]) -> bool: """ 探针必须到达实际业务工具目录且没有真实模型或业务调用，仅有计划工具不算就绪。 """ retained = {name for record in usage.get("model_requests", []) for name in record.get("retained_tools", [])} fixture = {f"mcp__evaluation.{name}" for name in ("moviepilot_api", "read_skill", "read_tool_result")} return bool(usage.get("probe_requests") == 1 and not usage.get("model_calls") and not usage.get("rejected_requests") and not server_stats.get("world_calls") and ("tool_search" in retained or fixture <= retained)) async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str, *, probe_only: bool) -> dict[str, Any]: """ 每次运行创建新目录、新模型预算和新世界，源码与 oracle 从不进入模型工作目录。 """ from scripts.evaluation.__main__ import _provenance world = EvaluationWorld(scenario_id) fixture_context = _native_browser_fixture() if world.scenario.kind == "browser" else nullcontext(None) with fixture_context as fixture: if fixture is not None: world.configure_browser_url(fixture[0]) provenance = _provenance(world) started = time.monotonic() result: dict[str, Any] = {} failure = None with tempfile.TemporaryDirectory(prefix="moviepilot-native-evaluation-") as directory: control_dir = Path(directory) work_dir = control_dir / "workspace" work_dir.mkdir() catalog_result = await _execute([executable, "debug", "models", "--bundled"], "", _worker_environment(), work_dir, 10) if catalog_result.get("error_type") or catalog_result.get("returncode") != 0: raise RuntimeError("无法读取当前原生客户端自带模型目录") catalog, catalog_metadata = _catalog(json.loads(catalog_result["stdout"]), settings.model) (control_dir / "models.json").write_text(json.dumps(catalog), encoding="utf-8") browser_runtime = ( _browser_runtime_configuration(executable, control_dir) if scenario_id in NATIVE_BROWSER_SCENARIOS else None ) extra_native_tools = NATIVE_SHELL_TOOLS if scenario_id in { "command_execution", "terminal_session", "terminal_pty_session", "subagent_terminal_share", } else frozenset() if browser_runtime is not None: extra_native_tools |= NATIVE_BROWSER_TOOLS async with EvaluationMcpServer(world) as server, EvaluationModelProxy( settings, probe_only=probe_only, extra_native_tools=extra_native_tools, ) as proxy: config = _configuration( settings, proxy, server, control_dir, scenario_id=scenario_id, browser_runtime=browser_runtime, ) use_app_server = scenario_id in (NATIVE_TERMINAL_SCENARIOS | NATIVE_STEERING_SCENARIOS) command = ( [executable, "app-server", "--listen", "stdio://", "--strict-config"] if use_app_server else [executable, "exec", "--ignore-user-config", "--ignore-rules", "--strict-config", "--ephemeral", "--skip-git-repo-check", "--sandbox", "read-only", "--json", "--color", "never", "-C", str(work_dir)] ) for key, value in config.items(): command.extend(["-c", f"{key}={_toml(value)}"]) environment = _worker_environment() environment.update(MOVIEPILOT_EVAL_MODEL_TOKEN=proxy.bearer_token, MOVIEPILOT_EVAL_MCP_TOKEN=server.bearer_token) if use_app_server: # app-server 必须使用临时状态根，避免读取或写入调用方的桌面会话和插件配置。 environment["CODEX_HOME"] = str(control_dir) else: command.append("-") try: result = await ( _execute_app_server( command, world.model_input(), environment, work_dir, settings.timeout_seconds, model=settings.model, reasoning_effort=settings.reasoning_effort, steering_plan=(world.scenario.steering_schedule() if scenario_id in NATIVE_STEERING_SCENARIOS else ()), ) if use_app_server else _execute(command, world.model_input(), environment, work_dir, settings.timeout_seconds) ) failure = result.get("error_type") except (RuntimeError, TimeoutError, OSError) as error: failure = type(error).__name__ local_tokens = tuple( value for value in (proxy.bearer_token, server.bearer_token, settings.account_id) if isinstance(value, str) and value ) usage = proxy.snapshot() server_stats = server.stats skill_sha256 = server.skill_sha256 if fixture is not None and fixture[1].get("clicked"): # 原生浏览器插件绕过评测 MCP 服务；回环回调独立证明按钮点击已执行。 world.record_browser("native_browser", { "success": True, "execution_outcome": "succeeded", "rendered": "BROWSER_OK", }) events, final_text, completed = _events(result.get("stdout", "")) if scenario_id in { "command_execution", "terminal_session", "terminal_pty_session", "subagent_terminal_share", }: _record_native_command_events(world, events) if any(event.get("type") == "evaluation.invalid_event" for event in events): failure = failure or "invalid_native_events" final_report = _parse_final(final_text) source_unchanged = _provenance(world) == provenance if not source_unchanged: failure = failure or "source_changed_during_run" stderr = result.get("stderr", "") report = { **evaluate( world, final_report, events, [event for event in events if str(event.get("type", "")).startswith("evaluation.steering.")], ).to_dict(), **provenance, **usage, "source_unchanged": source_unchanged, "evidence_kind": "codex_native_probe" if probe_only else "codex_native_controlled", "intelligence_evaluated": usage["completed_model_calls"] > 0, "codex_comparison": False, "model": settings.public_metadata(), "elapsed_seconds": round(time.monotonic() - started, 3), "cli_version": SUPPORTED_CLI_VERSION, "native_exit_code": result.get("returncode"), "native_model_catalog": catalog_metadata, "native_turn_completed": completed, "runner_error_type": failure, "native_events": events, "native_stderr": stderr, "native_stderr_sha256": hashlib.sha256(stderr.encode()).hexdigest(), "final_text": final_text, "final_report": final_report, "ledger": world.ledger, "fixture_server": server_stats, "exposed_skill_sha256": skill_sha256, "tool_catalog_scope": ( "native_control_tools_shared_fixture_mcp_and_browser_plugin" if scenario_id in NATIVE_BROWSER_SCENARIOS and browser_runtime is not None else "native_control_tools_and_shared_fixture_mcp" ), "browser_runtime_enabled": bool(scenario_id in NATIVE_BROWSER_SCENARIOS and browser_runtime is not None), "native_transport": "app_server" if use_app_server else "exec", } report["task_passed"] = report["passed"] report["passed"] = bool(report["passed"] and completed and result.get("returncode") == 0 and not failure) report["probe_ready"] = bool(probe_only and not failure and _probe_ready(usage, server_stats)) checked: dict[str, Any] = _validate_report( _redact(report, local_tokens), settings.api_key, settings.account_id, ) checked["native_stderr_sha256"] = hashlib.sha256(checked["native_stderr"].encode()).hexdigest() return checked def run_codex(scenario_id: str, settings: ModelSettings, *, executable: str = "codex", probe_only: bool = False) -> dict[str, Any]: """ 仅运行已核对的原生版本，探针不会向真实供应商发送请求。 """ if settings.wire_api != "responses": raise RuntimeError("原生 Codex 对照目前要求 Responses provider；当前配置是 Chat Completions") binary = shutil.which(executable) if binary is None: raise RuntimeError("未找到原生 Codex CLI") version = subprocess.run([binary, "--version"], capture_output=True, text=True, check=False, env=_worker_environment(), timeout=10) if version.returncode != 0 or version.stdout.strip() != SUPPORTED_CLI_VERSION: raise RuntimeError(f"此适配器只核对过 {SUPPORTED_CLI_VERSION}，请先验证当前 CLI 合同") return asyncio.run(_run_codex(scenario_id, settings, binary, probe_only=probe_only))