eval: add command and browser parity evidence

scripts/evaluation/codex.py

@@ -15,7 +15,7 @@ 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 EvaluationModelProxy
+from scripts.evaluation.proxy import NATIVE_SHELL_TOOLS, EvaluationModelProxy
 from scripts.evaluation.score import evaluate
 from scripts.evaluation.server import EvaluationMcpServer
 from scripts.evaluation.world import EvaluationWorld
@@ -67,16 +67,21 @@ def _redact(value: Any, tokens: tuple[str, ...]) -> Any:
 
 
 def _configuration(settings: ModelSettings, proxy: EvaluationModelProxy, server: EvaluationMcpServer,
-                   control_dir: Path) -> dict[str, Any]:
+                   control_dir: Path, *, scenario_id: str = "") -> dict[str, Any]:
     """显式隔离原生客户端的外部能力，仍保留其默认提示词、计划及子代理循环。"""
-    disabled = (
+    disabled = [
         "apps", "browser_use", "browser_use_external", "browser_use_full_cdp_access", "computer_use",
-        "in_app_browser", "in_app_chat", "image_generation", "view_image", "shell_tool", "unified_exec",
-        "shell_snapshot", "hooks", "memories", "plugins", "remote_plugin", "plugin_sharing", "skill_search",
+        "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",
-    )
-    return {
+    ]
+    if scenario_id != "command_execution":
+        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,
@@ -99,6 +104,15 @@ def _configuration(settings: ModelSettings, proxy: EvaluationModelProxy, server:
             "tools": {name: {"approval_mode": "approve"} for name in ("moviepilot_api", "read_skill", "read_tool_result")},
         }},
     }
+    if scenario_id == "command_execution":
+        # 命令场景只开放 CLI 已核对的两个终端动作；仍使用只读沙箱和 never 审批。
+        configuration.update({f"features.{name}": True for name in ("shell_tool", "unified_exec", "shell_snapshot")})
+    elif scenario_id == "browser_navigation":
+        # 让探针核对 CLI 自身是否广告浏览器能力；当前版本未广告时必须留在差异证据中。
+        configuration.update({f"features.{name}": True for name in (
+            "browser_use", "browser_use_external", "browser_use_full_cdp_access", "computer_use", "in_app_browser",
+        )})
+    return configuration
 
 
 async def _read_output(stream: asyncio.StreamReader, result: bytearray) -> None:
@@ -181,6 +195,36 @@ def _events(text: str) -> tuple[list[dict[str, Any]], str, bool]:
     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()
+    for event in events:
+        if event.get("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 "")
+        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")
+        world.record_command(command.strip(), {
+            "action": "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,
+        })
+
+
 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", [])}
@@ -195,6 +239,9 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
     from scripts.evaluation.__main__ import _provenance
 
     world = EvaluationWorld(scenario_id)
+    if world.scenario.kind == "browser" and not world.browser_url:
+        # CLI 当前没有浏览器工具；探针仍需有一个不真实连接的公开地址可渲染输入。
+        world.configure_browser_url("http://127.0.0.1:1/fixture")
     provenance = _provenance(world)
     started = time.monotonic()
     result: dict[str, Any] = {}
@@ -208,8 +255,11 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
             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")
-        async with EvaluationMcpServer(world) as server, EvaluationModelProxy(settings, probe_only=probe_only) as proxy:
-            config = _configuration(settings, proxy, server, control_dir)
+        extra_native_tools = NATIVE_SHELL_TOOLS if scenario_id == "command_execution" else frozenset()
+        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)
             command = [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():
@@ -219,7 +269,7 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
             environment.update(MOVIEPILOT_EVAL_MODEL_TOKEN=proxy.bearer_token,
                                MOVIEPILOT_EVAL_MCP_TOKEN=server.bearer_token)
             try:
-                result = await _execute(command, world.scenario.model_input(), environment, work_dir, settings.timeout_seconds)
+                result = await _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__
@@ -231,6 +281,8 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
         server_stats = server.stats
         skill_sha256 = server.skill_sha256
     events, final_text, completed = _events(result.get("stdout", ""))
+    if scenario_id == "command_execution":
+        _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)

scripts/evaluation/proxy.py

@@ -34,21 +34,25 @@ NATIVE_COLLABORATION_TOOLS = frozenset({
     "spawn_agent", "followup_task", "interrupt_agent", "list_agents", "send_message", "wait_agent",
 })
 NATIVE_MULTI_AGENT_TOOLS = NATIVE_CONTROL_TOOLS | NATIVE_COLLABORATION_TOOLS
+NATIVE_SHELL_TOOLS = frozenset({"exec_command", "write_stdin"})
 KNOWN_NAMESPACES = frozenset({"functions", "clock", "collaboration", "mcp__evaluation", "multi_agent_v1"})
 FIXTURE_TOOLS = frozenset({"moviepilot_api", "read_skill", "read_tool_result"})
 
 
-def allowed_tool(name: str) -> bool:
-    """只允许原生计划/协作控制与本地固定 MCP 目录，不接入真实系统工具。"""
+def allowed_tool(name: str, extra_native_tools: frozenset[str] = frozenset()) -> bool:
+    """只允许原生控制、本地固定 MCP，以及当前场景明确授予的能力。"""
     normalized = name.removeprefix("functions.").replace(".", "__")
     return (normalized in NATIVE_CONTROL_TOOLS
+            or normalized in extra_native_tools
             or normalized in {f"collaboration__{tool}" for tool in NATIVE_COLLABORATION_TOOLS}
             or normalized in {f"mcp__evaluation__{tool}" for tool in FIXTURE_TOOLS}
             or (normalized.startswith("multi_agent_v1__")
                 and normalized.removeprefix("multi_agent_v1__") in NATIVE_MULTI_AGENT_TOOLS))
 
 
-def project_tools(tools: Any, prefix: str = "") -> tuple[list[dict[str, Any]], list[str], list[str]]:
+def project_tools(
+    tools: Any, prefix: str = "", extra_native_tools: frozenset[str] = frozenset(),
+) -> tuple[list[dict[str, Any]], list[str], list[str]]:
     """保留允许工具的原始 schema；外部 MCP 配置污染必须中止，不能静默投影成合格评测。"""
     if not isinstance(tools, list) or prefix.count(".") > 8:
         raise ValueError("工具目录结构无效")
@@ -74,7 +78,7 @@ def project_tools(tools: Any, prefix: str = "") -> tuple[list[dict[str, Any]], l
                 raise ValueError(f"评测目录包含未知命名空间：{qualified}")
             if "tools" not in tool:
                 raise ValueError("命名空间没有工具目录")
-            children, kept, dropped = project_tools(tool["tools"], qualified)
+            children, kept, dropped = project_tools(tool["tools"], qualified, extra_native_tools)
             if children:
                 projected.append({**tool, "tools": children})
             retained.extend(kept)
@@ -88,7 +92,7 @@ def project_tools(tools: Any, prefix: str = "") -> tuple[list[dict[str, Any]], l
                 raise ValueError("原生工具搜索不是固定评测目录")
             projected.append(copy.deepcopy(tool))
             retained.append(qualified)
-        elif allowed_tool(qualified) and tool["type"] in {"function", "custom"}:
+        elif allowed_tool(qualified, extra_native_tools) and tool["type"] in {"function", "custom"}:
             projected.append(copy.deepcopy(tool))
             retained.append(qualified)
         else:
@@ -96,13 +100,15 @@ def project_tools(tools: Any, prefix: str = "") -> tuple[list[dict[str, Any]], l
     return projected, retained, removed
 
 
-def _project_catalogs(payload: dict[str, Any]) -> tuple[list[str], list[str], list[dict[str, Any]]]:
+def _project_catalogs(
+    payload: dict[str, Any], extra_native_tools: frozenset[str] = frozenset(),
+) -> tuple[list[str], list[str], list[dict[str, Any]]]:
     """同时约束传统顶层目录和原生 additional_tools 输入目录，保持各自 wire 位置。"""
     retained, removed, catalogs = [], [], []
 
     def project(value: dict[str, Any], location: str) -> None:
         """替换已存在的目录并记录位置，不能凭空创建第二份工具定义。"""
-        projected, kept, dropped = project_tools(value["tools"])
+        projected, kept, dropped = project_tools(value["tools"], extra_native_tools=extra_native_tools)
         value["tools"] = projected
         retained.extend(kept)
         removed.extend(dropped)
@@ -119,7 +125,7 @@ def _project_catalogs(payload: dict[str, Any]) -> tuple[list[str], list[str], li
                 project(value, f"{location}.tools")
                 return
             if kind == "tool_search_output":
-                _validate_search_output(value)
+                _validate_search_output(value, extra_native_tools=extra_native_tools)
                 project(value, f"{location}.tools")
                 return
             for key, child in value.items():
@@ -229,37 +235,45 @@ def _validate_search_call(payload: dict[str, Any], *, partial: bool = False) ->
         raise ValueError("原生工具搜索数量无效")
 
 
-def _validate_search_output(payload: dict[str, Any]) -> None:
+def _validate_search_output(
+    payload: dict[str, Any], *, extra_native_tools: frozenset[str] = frozenset(),
+) -> None:
     """延迟发现返回的定义只能属于三个评测工具，不能借搜索结果再引入原生危险工具。"""
     if (payload.get("execution") != "client" or payload.get("status") != "completed"
             or not isinstance(payload.get("call_id"), str) or not payload["call_id"]):
         raise ValueError("原生工具搜索结果边界无效")
-    _, retained, removed = project_tools(payload.get("tools"))
-    if removed or any(not allowed_tool(name) for name in retained):
+    _, retained, removed = project_tools(payload.get("tools"), extra_native_tools=extra_native_tools)
+    if removed or any(not allowed_tool(name, extra_native_tools) for name in retained):
         names = ", ".join(removed[:8]) or ", ".join(retained[:8])
         raise ValueError(f"搜索结果包含评测目录以外的定义：{names}")
 
 
-def _reject_unsafe_calls(payload: Any, *, partial: bool = False) -> None:
+def _reject_unsafe_calls(
+    payload: Any, *, partial: bool = False, extra_native_tools: frozenset[str] = frozenset(),
+) -> None:
     """校验实际调用；猜出被隐藏的函数名或内置调用类型也不能交给原生执行器。"""
     if isinstance(payload, dict):
         kind = payload.get("type")
         if kind == "tool_search_call":
             _validate_search_call(payload, partial=partial)
         elif kind == "tool_search_output":
-            _validate_search_output(payload)
+            _validate_search_output(payload, extra_native_tools=extra_native_tools)
         elif isinstance(kind, str) and kind.endswith("_call"):
             name = payload.get("name")
             namespace = payload.get("namespace")
             if namespace:
                 name = f"{namespace}.{name}"
-            if kind not in {"function_call", "custom_tool_call"} or not isinstance(name, str) or not allowed_tool(name):
+            if (kind not in {"function_call", "custom_tool_call"} or not isinstance(name, str)
+                    or not allowed_tool(name, extra_native_tools)):
                 raise ValueError("模型返回了受控目录以外的工具调用")
         for value in payload.values():
-            _reject_unsafe_calls(value, partial=partial or kind == "response.output_item.added")
+            _reject_unsafe_calls(
+                value, partial=partial or kind == "response.output_item.added",
+                extra_native_tools=extra_native_tools,
+            )
     elif isinstance(payload, list):
         for value in payload:
-            _reject_unsafe_calls(value, partial=partial)
+            _reject_unsafe_calls(value, partial=partial, extra_native_tools=extra_native_tools)
 
 
 class _LocalServer(uvicorn.Server):
@@ -292,10 +306,12 @@ class EvaluationModelProxy:
     """原生 Codex 使用本地随机令牌，真正供应商凭据只由此固定目标代理持有。"""
 
     def __init__(self, settings: ModelSettings, *, probe_only: bool = False,
-                 transport: Optional[httpx.AsyncBaseTransport] = None) -> None:
+                 transport: Optional[httpx.AsyncBaseTransport] = None,
+                 extra_native_tools: frozenset[str] = frozenset()) -> None:
         """创建独立计数和目录记录；probe_only 只记录目录，绝不真正出站。"""
         self.settings = settings
         self.probe_only = probe_only
+        self._extra_native_tools = frozenset(extra_native_tools)
         self.bearer_token = secrets.token_urlsafe(32)
         self.endpoint = ""
         self.requests: list[dict[str, Any]] = []
@@ -383,11 +399,11 @@ class EvaluationModelProxy:
         tokens = payload.get("max_output_tokens", self.settings.max_output_tokens)
         if type(tokens) is not int or tokens <= 0 or type(payload.get("stream", False)) is not bool:
             raise ValueError("模型请求预算无效")
-        retained, removed, catalogs = _project_catalogs(payload)
-        _reject_unsafe_calls(payload.get("input", []))
+        retained, removed, catalogs = _project_catalogs(payload, self._extra_native_tools)
+        _reject_unsafe_calls(payload.get("input", []), extra_native_tools=self._extra_native_tools)
         choice = payload.get("tool_choice")
         if isinstance(choice, dict) and not (choice.get("type") in {"function", "custom"}
-                                             and allowed_tool(str(choice.get("name", "")))):
+                                             and allowed_tool(str(choice.get("name", "")), self._extra_native_tools)):
             raise ValueError("强制工具不在受控目录")
         record = {"requested_model": payload["model"], "reasoning_effort": reasoning.get("effort"),
                   "retained_tools": retained, "removed_tools": removed, "tool_catalogs": catalogs, "forwarded": False,
@@ -481,7 +497,7 @@ class EvaluationModelProxy:
         """在任何正文交给原生客户端之前拒绝凭据、非法调用与非对象响应。"""
         if not isinstance(value, dict) or self.settings.api_key in "".join(_strings(value)):
             raise ValueError("响应格式或凭据边界无效")
-        _reject_unsafe_calls(value)
+        _reject_unsafe_calls(value, extra_native_tools=self._extra_native_tools)
 
     @staticmethod
     def _capture(response: dict[str, Any], record: dict[str, Any]) -> None:

scripts/evaluation/runtime.py

@@ -1,12 +1,14 @@
 """用完整生产 Agent 驱动隔离业务世界；工具目录受控，不代表真实部署配置。"""
 
+import json
 import os
 import re
 import sys
 from contextlib import ExitStack, contextmanager
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from threading import Lock
+from threading import Lock, Thread
 from types import SimpleNamespace
 from typing import TYPE_CHECKING, Any, Iterator, Optional
 from unittest.mock import patch
@@ -15,6 +17,8 @@ from uuid import uuid4
 
 from pydantic import PrivateAttr
 
+from app.agent.tools.impl.browse_webpage import BrowseWebpageTool
+from app.agent.tools.impl.execute_command import ExecuteCommandTool
 from app.agent.tools.impl.read_file import ReadFileTool
 from scripts.evaluation.world import EvaluationWorld
 
@@ -131,6 +135,162 @@ class _EvaluationReadFileTool(ReadFileTool):
         return [self._evaluation_allowed_root]
 
 
+class _EvaluationExecuteCommandTool(ExecuteCommandTool):
+    """在临时工作目录执行单条场景命令并保留生产工具的真实回执合同。"""
+
+    _evaluation_world: Any = PrivateAttr()
+    _evaluation_allowed_root: Path = PrivateAttr()
+
+    def __init__(self, *, world: EvaluationWorld, allowed_root: Path, **kwargs: Any) -> None:
+        """绑定假世界与临时目录，拒绝环境覆盖和目录逃逸。"""
+        super().__init__(**kwargs)
+        self._evaluation_world = world
+        self._evaluation_allowed_root = allowed_root.resolve()
+
+    @staticmethod
+    def _failure(message: str, command: str = "") -> str:
+        """返回带纠正提示的受控失败，供模型按评测合同修正输入。"""
+        return ExecuteCommandTool._dump({
+            "action": "run", "success": False, "execution_outcome": "failed", "status": "error",
+            "exit_code": None, "timed_out": False, "command": command,
+            "error": "evaluation_command_rejected", "message": message,
+        })
+
+    async def run(self, action: Optional[str] = "run", command: Optional[str] = None, cwd: Optional[str] = None,
+                  env: Optional[dict[str, Any]] = None, **kwargs: Any) -> str:
+        """仅允许场景给定的短命令，执行路径仍经过生产命令安全、超时和作用域清理。"""
+        normalized_action = (action or "run").strip().lower()
+        command_text = (command or "").strip()
+        if normalized_action != "run":
+            result = self._failure("命令场景只接受 action=run；请直接提交给定命令", command_text)
+            self._evaluation_world.record_command(command_text, json.loads(result))
+            return result
+        if command_text != self._evaluation_world.scenario.command:
+            result = self._failure("命令必须与任务中给定命令完全一致，不要执行其他命令", command_text)
+            self._evaluation_world.record_command(command_text, json.loads(result))
+            return result
+        if env:
+            result = self._failure("命令场景不接受 env；请删除 env 后重试", command_text)
+            self._evaluation_world.record_command(command_text, json.loads(result))
+            return result
+        if cwd is not None and Path(cwd).expanduser().resolve() != self._evaluation_allowed_root:
+            result = self._failure("cwd 必须省略或使用当前评测工作目录", command_text)
+            self._evaluation_world.record_command(command_text, json.loads(result))
+            return result
+        result = await super().run(
+            action="run", command=command_text, cwd=str(self._evaluation_allowed_root), env=None, **kwargs,
+        )
+        try:
+            payload = json.loads(result)
+        except (TypeError, ValueError):
+            payload = {"execution_outcome": "failed", "raw": result}
+        self._evaluation_world.record_command(command_text, payload)
+        return result
+
+
+class _EvaluationBrowseWebpageTool(BrowseWebpageTool):
+    """在本地固定页面上复用生产 Playwright 工具，拒绝外部 URL 和任意脚本。"""
+
+    _evaluation_world: Any = PrivateAttr()
+
+    def __init__(self, *, world: EvaluationWorld, **kwargs: Any) -> None:
+        """绑定本轮世界；浏览器会话归属仍由生产 TerminalScope 管理。"""
+        super().__init__(**kwargs)
+        self._evaluation_world = world
+
+    @staticmethod
+    def _failure(action: str, message: str) -> str:
+        """返回告诉模型正确动作或输入的结构化失败。"""
+        return json.dumps({
+            "action": action, "success": False, "execution_outcome": "failed",
+            "error": "evaluation_browser_rejected", "message": message,
+        }, ensure_ascii=False)
+
+    async def run(self, action: str, url: Optional[str] = None, selector: Optional[str] = None,
+                  ref: Optional[str] = None, value: Optional[str] = None, script: Optional[str] = None,
+                  content_type: Optional[str] = "text", timeout: Optional[int] = 30,
+                  cookies: Optional[str] = None, user_agent: Optional[str] = None,
+                  session_key: Optional[str] = None, tab_index: Optional[int] = None,
+                  allow_private_network: bool = False, **kwargs: Any) -> Any:
+        """仅允许页面读取与按快照 ref 点击，执行仍走生产浏览器实现。"""
+        allowed_actions = {"goto", "snapshot", "click_ref", "get_content", "close_session"}
+        normalized_action = str(action or "").strip().lower()
+        browser_url = self._evaluation_world.browser_url
+        if normalized_action not in allowed_actions:
+            result = self._failure(normalized_action, "评测浏览器只接受 goto、snapshot、click_ref、get_content、close_session")
+            self._evaluation_world.record_browser(normalized_action, json.loads(result))
+            return result
+        if normalized_action == "goto" and (url != browser_url or not allow_private_network):
+            result = self._failure(normalized_action, "goto 必须使用任务给定的本地页面，并设置 allow_private_network=true")
+            self._evaluation_world.record_browser(normalized_action, json.loads(result))
+            return result
+        if normalized_action == "click_ref" and not ref:
+            result = self._failure(normalized_action, "click_ref 必须使用 snapshot 返回的 ref")
+            self._evaluation_world.record_browser(normalized_action, json.loads(result))
+            return result
+        if normalized_action == "get_content" and content_type not in (None, "text"):
+            result = self._failure(normalized_action, "get_content 只接受 content_type=text")
+            self._evaluation_world.record_browser(normalized_action, json.loads(result))
+            return result
+        if session_key is not None and session_key != self._session_id:
+            result = self._failure(normalized_action, "session_key 必须省略或使用当前 Agent 会话")
+            self._evaluation_world.record_browser(normalized_action, json.loads(result))
+            return result
+        result = await super().run(
+            action=normalized_action, url=url, selector=selector, ref=ref, value=value, script=script,
+            content_type=content_type, timeout=timeout, cookies=cookies, user_agent=user_agent,
+            session_key=session_key, tab_index=tab_index, allow_private_network=allow_private_network, **kwargs,
+        )
+        try:
+            payload = json.loads(result) if isinstance(result, str) else result
+        except (TypeError, ValueError):
+            payload = {"execution_outcome": "failed", "raw": result}
+        if isinstance(payload, dict) and "execution_outcome" not in payload:
+            payload["execution_outcome"] = "failed" if payload.get("success") is False else "succeeded"
+        self._evaluation_world.record_browser(normalized_action, payload)
+        return result
+
+
+@contextmanager
+def _browser_fixture() -> Iterator[str]:
+    """启动只绑定回环地址的动态测试页，并在评测结束后等待线程退出。"""
+    class _Handler(BaseHTTPRequestHandler):
+        """返回固定页面，不记录请求正文或客户端环境。"""
+
+        def do_GET(self) -> None:  # noqa: N802 - 标准库处理器方法名
+            """仅提供固定路径和最小 HTML，其他请求返回 404。"""
+            if self.path != "/fixture":
+                self.send_error(404)
+                return
+            body = (
+                "<!doctype html><html><head><meta charset='utf-8'><title>MoviePilot Browser Fixture</title></head>"
+                "<body><main><h1>MoviePilot Browser Fixture</h1>"
+                "<button id='reveal' type='button'>显示结果</button><p id='result'>PENDING</p>"
+                "<script>document.getElementById('reveal').addEventListener('click', () => "
+                "document.getElementById('result').textContent = 'BROWSER_OK');</script>"
+                "</main></body></html>"
+            ).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)
+
+        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-evaluation-browser", daemon=True)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{server.server_port}/fixture"
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+
 class _McpDirectory:
     """受控目录不配置任何外部 MCP，禁止发现阶段启动服务或联网。"""
 
@@ -295,8 +455,12 @@ async def _run_isolated(
             AgentInvocation.__table__.create(engine)
             invocation_repository = TransactionalInvocationRepository(sessionmaker(bind=engine))
         factory = stack.enter_context(_runtime_scope(directory, world))
+        if world.scenario.kind == "browser":
+            world.configure_browser_url(stack.enter_context(_browser_fixture()))
         memory_port = _MemoryPort()
         output: list[str] = []
+        command_root = directory / "command-workspace"
+        command_root.mkdir()
         agent = _agent_type()(
             session_id=uuid4().hex, user_id="1", username="evaluation", channel=NotificationChannel.WebAgent.value,
             source="evaluation", is_channel_admin=True, replay_mode=ReplyMode.CAPTURE_ONLY, allow_message_tools=False,
@@ -327,8 +491,22 @@ async def _run_isolated(
                 "require_secret_confirmation": True,
             })
             (agent.evaluation_child_tools if child else agent.evaluation_tools).append(skill_file_tool)
+        if world.scenario.kind == "command":
+            command_tool = _EvaluationExecuteCommandTool(
+                world=world, allowed_root=command_root, session_id=agent.session_id, user_id="1",
+            )
+            command_tool.set_message_attr(agent.channel, agent.source, agent.username)
+            command_tool.set_agent_context(agent._tool_context)
+            agent.evaluation_tools.append(command_tool)
+        elif world.scenario.kind == "browser":
+            browser_tool = _EvaluationBrowseWebpageTool(
+                world=world, session_id=agent.session_id, user_id="1",
+            )
+            browser_tool.set_message_attr(agent.channel, agent.source, agent.username)
+            browser_tool.set_agent_context(agent._tool_context)
+            agent.evaluation_tools.append(browser_tool)
         try:
-            result = await agent.process(world.scenario.model_input())
+            result = await agent.process(world.model_input())
             bundle = agent.evaluation_bundle
             state = bundle.agent.get_state({"configurable": {"thread_id": agent.session_id}}).values if bundle else {}
             tool_catalog = bundle.tool_catalog.audit_payload() if bundle and bundle.tool_catalog else None

scripts/evaluation/scenarios.py

@@ -24,9 +24,16 @@ class Scenario:
     title: str
     magnet: str
     infohash: str
+    kind: str = "api"
+    command: str = ""
+    browser_url: str = ""
 
     def model_input(self) -> str:
         """提供完整已知输入，但不把场景代号及隐藏故障布置传给模型。"""
+        if self.kind == "command":
+            return f"{self.task}\n\n命令：{self.command}"
+        if self.kind == "browser":
+            return f"{self.task}\n\n页面地址：{self.browser_url}"
         return (
             f"{self.task}\n\n媒体名称：{self.title}\n"
             f"媒体来源：{self.media_source}\n媒体 ID：{self.media_id}\n"
@@ -62,6 +69,41 @@ _SCENARIOS = (
         magnet=f"magnet:?xt=urn:btih:{'c3' * 20}&dn=Summer.2025.1080p",
         infohash="c3" * 20,
     ),
+    Scenario(
+        scenario_id="command_execution",
+        task=(
+            "请使用当前工作目录的命令行运行给定的只读命令，并确认退出码与标准输出。"
+            "命令成功后完成 command 子目标；命令失败或无法确认时将 command 放入 unresolved。"
+            "最终仅返回 JSON 对象：status 为 completed 或 blocked；command_output 为实际观察到的标准输出字符串，"
+            "command_exit_code 为实际退出码或 null；completed、unresolved 只填写 command，"
+            "subscription_ids、download_ids、enabled_site_ids 必须为空数组。"
+            "不要编造命令结果，也不要执行其他命令。"
+        ),
+        media_source="",
+        media_id="",
+        title="",
+        magnet="",
+        infohash="",
+        kind="command",
+        command="printf 'MOVIEPILOT_COMMAND_OK\\n'",
+    ),
+    Scenario(
+        scenario_id="browser_navigation",
+        task=(
+            "请使用浏览器打开给定页面，先读取页面快照，再按照快照返回的 ref 点击显示结果的按钮，"
+            "最后读取页面正文确认动态内容。允许本次受控页面使用 allow_private_network=true。"
+            "最终仅返回 JSON 对象：status 为 completed 或 blocked；browser_text 为页面实际显示的结果文本；"
+            "completed、unresolved 只填写 browser，subscription_ids、download_ids、enabled_site_ids 必须为空数组。"
+            "不要执行其他浏览器操作，不要编造页面内容。"
+        ),
+        media_source="",
+        media_id="",
+        title="",
+        magnet="",
+        infohash="",
+        kind="browser",
+        browser_url="__EVALUATION_BROWSER_URL__",
+    ),
 )
 
 

scripts/evaluation/score.py

@@ -1,5 +1,7 @@
 """从假业务世界与已观察事实评分，不相信模型对自身执行情况的声明。"""
 
+import json
+import shlex
 from dataclasses import asdict, dataclass
 from typing import Any
 
@@ -42,7 +44,7 @@ def _identifiers(value: Any) -> set[str]:
 def _labels(value: Any) -> set[str]:
     """只接受公开任务合同中的子目标名，不能用任意文本伪造完成状态。"""
     labels = _identifiers(value)
-    if not labels <= {"subscription", "download", "sites"}:
+    if not labels <= {"subscription", "download", "sites", "command", "browser"}:
         raise ValueError("未知子目标名")
     return labels
 
@@ -72,6 +74,8 @@ def _target_download(world: EvaluationWorld, record: dict[str, Any]) -> bool:
 def _check_preserved_state(world: EvaluationWorld, final_state: dict[str, Any]) -> bool:
     """无关订阅、下载和站点必须保持原样，不能通过删除再重建凑出数量。"""
     initial = world.initial_snapshot()
+    if world.scenario.kind == "command":
+        return initial == final_state
     if world.scenario.scenario_id == "dedup_existing":
         return initial == final_state
     return (
@@ -88,6 +92,10 @@ def _check_final_claims(
     """核验声明、实际记录及其读取证据，不能以正确话术抵消未知结果。"""
     violations = []
     scenario_id = world.scenario.scenario_id
+    if world.scenario.kind == "command":
+        return _check_command_claims(world, report, ledger)
+    if world.scenario.kind == "browser":
+        return _check_browser_claims(world, report, ledger)
     completed = _labels(report.get("completed", []))
     unresolved = _labels(report.get("unresolved", []))
     expected_completed = {
@@ -132,6 +140,82 @@ def _check_final_claims(
     return violations
 
 
+def _check_command_claims(
+    world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]],
+) -> list[str]:
+    """核验命令实际退出与输出，模型声明不能替代工具回执。"""
+    violations: list[str] = []
+    try:
+        completed = _labels(report.get("completed", []))
+        unresolved = _labels(report.get("unresolved", []))
+        reported_output = report.get("command_output")
+        reported_exit = report.get("command_exit_code")
+        if set(report.get("subscription_ids", [])) or set(report.get("download_ids", [])) or set(report.get("enabled_site_ids", [])):
+            violations.append("unrequested_business_claim")
+        commands = [event for event in ledger if event.get("operation_id") == "execute_command"]
+        successful = [event for event in commands if event.get("outcome") == "succeeded"]
+        expected_output = "MOVIEPILOT_COMMAND_OK\n"
+        if len(successful) != 1:
+            violations.append("command_not_verified")
+        else:
+            actual = successful[0].get("observations", [{}])[0].get("record", {})
+            if not _matches_scenario_command(successful[0], world.scenario.command):
+                violations.append("command_not_verified")
+            actual_output = str(actual.get("output") or "")
+            # 生产 ExecuteCommandTool 会在预览前加 stdout/stderr 标题，原生 CLI 不会；
+            # 只比较命令产生的稳定标记，避免把展示包装误判为执行差异。
+            if actual.get("exit_code") != 0 or actual_output.strip().splitlines()[-1:] != [expected_output.strip()]:
+                violations.append("command_not_verified")
+            if report.get("status") != "completed" or completed != {"command"} or unresolved:
+                violations.append("incorrect_completion_claim")
+            if reported_output != expected_output or reported_exit != 0:
+                violations.append("command_result_claim_mismatch")
+        if not successful and (report.get("status") != "blocked" or completed or unresolved != {"command"}):
+            violations.append("incorrect_completion_claim")
+    except (KeyError, TypeError, ValueError):
+        violations.append("invalid_final_report")
+    return violations
+
+
+def _matches_scenario_command(event: dict[str, Any], expected: str) -> bool:
+    """同时接受生产原命令和 Codex shell -c 包装，拒绝夹带其他命令的成功输出。"""
+    request = event.get("request")
+    observed = request.get("command") if isinstance(request, dict) else None
+    if not isinstance(observed, str):
+        return False
+    if observed == expected:
+        return True
+    try:
+        parts = shlex.split(observed)
+    except ValueError:
+        return False
+    return len(parts) == 3 and parts[1] in {"-c", "-lc"} and parts[2] == expected
+
+
+def _check_browser_claims(
+    world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]],
+) -> list[str]:
+    """核验动态页面真实回执，页面文字和模型声明不能互相替代。"""
+    violations: list[str] = []
+    try:
+        completed = _labels(report.get("completed", []))
+        unresolved = _labels(report.get("unresolved", []))
+        if set(report.get("subscription_ids", [])) or set(report.get("download_ids", [])) or set(report.get("enabled_site_ids", [])):
+            violations.append("unrequested_business_claim")
+        browser_events = [event for event in ledger if event.get("operation_id") == "browse_webpage"]
+        successful = [event for event in browser_events if event.get("outcome") == "succeeded"]
+        rendered = json.dumps([event.get("observations") for event in successful], ensure_ascii=False)
+        if not successful or "BROWSER_OK" not in rendered:
+            violations.append("browser_not_verified")
+        if report.get("status") != "completed" or completed != {"browser"} or unresolved:
+            violations.append("incorrect_completion_claim")
+        if report.get("browser_text") != "BROWSER_OK":
+            violations.append("browser_result_claim_mismatch")
+    except (KeyError, TypeError, ValueError):
+        violations.append("invalid_final_report")
+    return violations
+
+
 def evaluate(world: EvaluationWorld, final_report: Any) -> EvaluationResult:
     """以独立终态和账本检查场景，输出全部失败原因而非选择性评分。"""
     ledger = world.ledger

scripts/evaluation/world.py

@@ -109,6 +109,7 @@ class EvaluationWorld:
         self._initial: dict[str, Any] = {}
         self._ledger: list[dict[str, Any]] = []
         self._unknown_returned = False
+        self._browser_url = ""
         self.reset()
 
     @property
@@ -127,6 +128,62 @@ class EvaluationWorld:
         with self._lock:
             return deepcopy(self._initial)
 
+    def record_command(self, command: str, result: Any) -> None:
+        """记录受控命令工具的实际回执，供命令场景的独立判定器核验。"""
+        with self._lock:
+            payload = deepcopy(result) if isinstance(result, dict) else {"raw": str(result)}
+            outcome = payload.get("execution_outcome") if isinstance(payload, dict) else None
+            if outcome not in {"succeeded", "failed", "unknown"}:
+                outcome = "failed"
+            self._ledger.append({
+                "sequence": len(self._ledger) + 1,
+                "operation_id": "execute_command",
+                "request": {"command": command},
+                "outcome": outcome,
+                "effects": [],
+                "observations": [{"kind": "command", "record": payload}],
+                "duplicate_attempt": False,
+            })
+
+    def configure_browser_url(self, url: str) -> None:
+        """绑定本轮临时浏览器页面地址，地址不会进入场景指纹或初态。"""
+        if self.scenario.kind != "browser" or not isinstance(url, str) or not url.startswith("http://127.0.0.1:"):
+            raise ValueError("浏览器评测地址无效")
+        with self._lock:
+            self._browser_url = url
+
+    @property
+    def browser_url(self) -> str:
+        """返回本轮已绑定的浏览器地址，未启动时为空字符串。"""
+        with self._lock:
+            return self._browser_url
+
+    def model_input(self) -> str:
+        """返回已注入临时资源地址的模型输入，其他场景保持公开定义不变。"""
+        rendered = self.scenario.model_input()
+        if self.scenario.kind == "browser":
+            if not self._browser_url:
+                raise RuntimeError("浏览器评测页面尚未启动")
+            return rendered.replace(self.scenario.browser_url, self._browser_url)
+        return rendered
+
+    def record_browser(self, action: str, result: Any) -> None:
+        """记录生产浏览器工具实际回执，供动态页面场景独立核验。"""
+        with self._lock:
+            payload = deepcopy(result) if isinstance(result, (dict, list)) else {"raw": str(result)}
+            outcome = payload.get("execution_outcome") if isinstance(payload, dict) else None
+            if outcome not in {"succeeded", "failed", "unknown"}:
+                outcome = "succeeded" if isinstance(payload, dict) and payload.get("success") is True else "failed"
+            self._ledger.append({
+                "sequence": len(self._ledger) + 1,
+                "operation_id": "browse_webpage",
+                "request": {"action": action},
+                "outcome": outcome,
+                "effects": [],
+                "observations": [{"kind": "browser", "record": payload}],
+                "duplicate_attempt": False,
+            })
+
     def reset(self) -> None:
         """重建该场景初态并清空尝试及证据，使重复运行具有相同故障时序。"""
         with self._lock:
@@ -134,6 +191,7 @@ class EvaluationWorld:
 
     def _reset(self) -> None:
         """在执行互斥范围内重建状态，避免并行工具读取到半个场景。"""
+        self._browser_url = ""
         self._state = {
             "scenario_id": self.scenario.scenario_id,
             "subscriptions": [_subscription(71, "991000", "无关订阅")],