feat(agent): add controlled native Codex evaluation runner

docs/agent-evaluation.md

@@ -4,7 +4,7 @@
 
 ## 当前范围
 
-`scripts/evaluation/` 提供两种显式模式。`--replay` 只回放离线轨迹，不联网，报告标记 `evidence_kind=scripted_replay`、`intelligence_evaluated=false`。`--live` 在独立进程内用真实模型驱动完整生产 Agent，业务接口只进入内存假世界。两者都没有实现 Codex 原生对照，不能把通过率称为 Codex 对比结果。
+`scripts/evaluation/` 显式区分离线回放、MoviePilot 真实运行和原生 Codex 受控运行。`--replay` 不联网，标记 `evidence_kind=scripted_replay`、`intelligence_evaluated=false`；`--live` 用真实模型驱动完整生产 Agent；`--native` 调用原生 Codex CLI。两种真实运行共用内存假业务世界与独立验收器。单条报告仍标记 `codex_comparison=false`，只有固定条件下的配对重复结果才能形成比较。
 
 首批场景：
 
@@ -39,7 +39,7 @@ uv run --locked --no-sync python -m scripts.evaluation --live \
   --output report.json
 ```
 
-只有显式 `--live` 才会产生模型调用费用。默认读取 `~/.codex/config.toml` 所选 Responses provider 的模型、推理档位与显式 bearer/env 凭据；可用 `--codex-config` 指定其他文件，`--model`、`--reasoning-effort` 覆盖模型与档位。不会借用其他服务的登录凭据，也不会自动降低被供应商拒绝的参数。报告同时保留请求模型与供应商返回的模型标识；本地 Codex 配置不能证明运行中的 MoviePilot 使用相同配置。
+只有显式 `--live` 或 `--native` 才会调用真实模型并产生费用。默认读取 `~/.codex/config.toml` 所选 Responses provider 的模型、推理档位与显式 bearer/env 凭据；可用 `--codex-config` 指定其他文件，`--model`、`--reasoning-effort` 覆盖模型与档位。不会借用其他服务的登录凭据，也不会自动降低被供应商拒绝的参数。报告同时保留请求模型与供应商返回的模型标识；本地 Codex 配置不能证明运行中的 MoviePilot 使用相同配置。
 
 调用配置经私有标准输入传给 worker，凭据不进入命令行、提示词或报告。worker 只继承必要的平台环境，先创建临时 `CONFIG_DIR`，再导入后端；每轮拥有独立回执库、记忆、会话和工具实例。生产 `process/_create_agent`、Skills、计划、权限、持久回执、工具输出预算、压缩和子代理仍按真实路径执行。主工具目录限定为假业务 API 和生产内部工具，API transport 拒绝任何外部目标及场景外 operation；插件、外部 MCP、通知和任意 shell/文件/浏览器工具不开放。此受控目录是当前评测边界，不代表默认部署工具全集。
 
@@ -51,7 +51,30 @@ uv run --locked --no-sync python -m scripts.evaluation --live \
 
 单测不调用真实模型或外部网络；测试只验证运行器、隔离、预算与判定合同。真实报告含合成任务轨迹，也可能很长，应保存于评测工作目录；不要把包含私有配置的临时诊断日志提交到仓库。
 
-## Codex 比较的后续接入要求
+## 原生 Codex 受控运行
+
+先使用无真实模型请求的目录探针，再运行真实场景：
+
+```bash
+uv run --locked --no-sync python -m scripts.evaluation --native-probe \
+  --scenario dedup_existing --model gpt-6-astra --reasoning-effort max \
+  --output probe.json
+uv run --locked --no-sync python -m scripts.evaluation --native \
+  --scenario dedup_existing --model gpt-6-astra --reasoning-effort max \
+  --max-model-calls 12 --timeout-seconds 180 --output codex-report.json
+```
+
+适配器目前只核对 `codex-cli 0.153.4`，可用 `--codex-executable` 指定文件。其他版本或二进制目录中没有指定模型时明确拒绝，不换模型或套用其他模型的元数据。`--native-probe` 的 `probe_ready` 只代表配置和目录检查，不是任务通过；原生进程预期收到本地探针错误并退出，模型调用数仍为零。
+
+每次创建空白临时工作目录；场景源码、初态、账本、oracle、模型凭据留在控制器。原生客户端使用局部随机令牌连接回环模型代理和 MCP 假世界，不继承业务配置、真实模型令牌或其他服务环境。客户端忽略用户配置和规则文件，关闭宿主技能、插件、浏览器、文件查看及 shell 等能力，采用只读沙箱、never 审批与有界退出；仅对当前回环 evaluation 服务的三个假工具显式设置 `approval_mode=approve`，避免原生客户端拒绝已授权的假业务动作。不修改用户 HOME、CODEX_HOME 或现有配置。全局 AGENTS 仍可能由原生客户端加载，因此代理在发送给模型前仅移除规范的独立 AGENTS 用户块，保留原生基础说明和任务文本，并记录被移除块的长度与哈希。
+
+原生循环、计划和协作工具仍由 Codex 执行。当前模型可固定使用 Code Mode，单独关闭 feature 无法改变它；适配器通过 `codex debug models --bundled` 读取当前二进制自带目录，只投影指定模型的 `tool_mode=direct`，保持其余字段、`base_instructions` 和 `model_messages`，记录前后指纹。随后代理限制实际广告目录为计划/协作与 evaluation MCP 工具，兼顾请求中的动态目录，并在每个完整响应事件交给客户端前再次拒绝目录以外的调用。这是明确投影过工具模式与目录的原生 Codex 对照，不代表默认部署的完整产品工具环境。
+
+MCP 服务只公开 `moviepilot_api`、完整的 `moviepilot-api` Skill 和有界结果续读；不提前给出场景支持 operation 清单。所有原生客户端与子代理 MCP 会话共享同一个世界及 32 次业务调用预算。服务使用已锁定的 Starlette/uvicorn 实现本评测所需 Streamable HTTP 子集；不作为通用生产 MCP 服务或对外部署入口。
+
+模型代理固定供应商目标，禁止不同模型或不同推理档位静默替换，主调用和子代理共同消耗硬调用预算，HTTP 与 SSE 自动重试为零。报告记录原生 JSONL 事件、退出码、完成事件、独立业务账本、实际目录投影、模型目录/输入指纹、逐请求用量及失败。超时和输出超限保留已有事件并终止进程组；技术失败即使已有正确 JSON 也不会标记整轮通过。`task_passed` 仅为独立 oracle 的业务判断，`passed` 还要求原生运行完整退出。
+
+## 配对比较的验收要求
 
 - MoviePilot 侧应保留完整生产 Agent 装配，只替换业务 transport 与隔离配置；只组装部分中间件的循环不能冒充产品实测。
 - Codex 侧必须调用原生 CLI 或 App Server。自建模型循环只能叫受控运行时对照。

scripts/evaluation/__main__.py

@@ -57,11 +57,14 @@ def replay(scenario_id: str, payload: Any) -> dict[str, Any]:
 
 def main(argv: list[str] | None = None) -> int:
     """显式区分回放与真实调用，并以非零退出码表达验收未通过。"""
-    parser = argparse.ArgumentParser(description="MoviePilot Agent 隔离任务评测；--live 才会调用真实模型")
+    parser = argparse.ArgumentParser(description="MoviePilot Agent 隔离任务评测；--live/--native 才会调用真实模型")
     mode = parser.add_mutually_exclusive_group(required=True)
     mode.add_argument("--list", action="store_true", help="列出公开场景，不输出初态或答案")
     mode.add_argument("--replay", type=Path, help="包含 calls 与 final 的 JSON 轨迹")
     mode.add_argument("--live", action="store_true", help="显式调用真实模型，业务只进入隔离假世界")
+    mode.add_argument("--native", action="store_true", help="用原生 Codex CLI 操作同一假世界并调用真实模型")
+    mode.add_argument("--native-probe", action="store_true", help="核对原生 Codex 配置和工具目录，不调用真实模型")
+    parser.add_argument("--codex-executable", default="codex", help="原生模式使用的 Codex CLI 可执行文件")
     parser.add_argument("--scenario", choices=[item.scenario_id for item in list_scenarios()])
     parser.add_argument("--output", type=Path, help="可选 JSON 报告路径")
     parser.add_argument("--codex-config", type=Path, default=Path.home() / ".codex" / "config.toml")
@@ -77,8 +80,7 @@ def main(argv: list[str] | None = None) -> int:
     if args.scenario is None:
         parser.error("评测需要 --scenario")
     try:
-        if args.live:
-            from scripts.evaluation.live import run_live
+        if args.live or args.native or args.native_probe:
             from scripts.evaluation.models import load_codex_model_settings
 
             settings = load_codex_model_settings(
@@ -86,7 +88,14 @@ def main(argv: list[str] | None = None) -> int:
                 max_model_calls=args.max_model_calls, max_output_tokens=args.max_output_tokens,
                 timeout_seconds=args.timeout_seconds,
             )
-            result = run_live(args.scenario, settings)
+            if args.live:
+                from scripts.evaluation.live import run_live
+
+                result = run_live(args.scenario, settings)
+            else:
+                from scripts.evaluation.codex import run_codex
+
+                result = run_codex(args.scenario, settings, executable=args.codex_executable, probe_only=args.native_probe)
         else:
             if args.replay.stat().st_size > MAX_REPLAY_BYTES:
                 raise ValueError("回放文件超过 256 KiB")
@@ -98,7 +107,7 @@ def main(argv: list[str] | None = None) -> int:
     if args.output is not None:
         args.output.write_text(rendered + "\n", encoding="utf-8")
     print(rendered)
-    return 0 if result["passed"] else 1
+    return 0 if (result.get("probe_ready") if args.native_probe else result["passed"]) else 1
 
 
 if __name__ == "__main__":

scripts/evaluation/codex.py

@@ -0,0 +1,269 @@
+"""让原生 Codex CLI 操作同一假世界，模型循环与独立评分器不共用状态。"""
+
+import asyncio
+import copy
+import hashlib
+import json
+import os
+import shutil
+import signal
+import subprocess
+import tempfile
+import time
+from pathlib import Path
+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.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
+
+
+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) -> 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", "shell_tool", "unified_exec",
+        "shell_snapshot", "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 {
+        "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": min(settings.timeout_seconds, 120) * 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")},
+        }},
+    }
+
+
+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",
+    )
+    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 _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 _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)
+    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")
+        async with EvaluationMcpServer(world) as server, EvaluationModelProxy(settings, probe_only=probe_only) as proxy:
+            config = _configuration(settings, proxy, server, control_dir)
+            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():
+                command.extend(["-c", f"{key}={_toml(value)}"])
+            command.append("-")
+            environment = _worker_environment()
+            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)
+                failure = result.get("error_type")
+            except (RuntimeError, TimeoutError, OSError) as error:
+                failure = type(error).__name__
+            local_tokens = (proxy.bearer_token, server.bearer_token)
+        usage = proxy.snapshot()
+        server_stats = server.stats
+        skill_sha256 = server.skill_sha256
+    events, final_text, completed = _events(result.get("stdout", ""))
+    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).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_and_shared_fixture_mcp",
+    }
+    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)
+    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]:
+    """仅运行已核对的原生版本，探针不会向真实供应商发送请求。"""
+    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))

scripts/evaluation/proxy.py

@@ -0,0 +1,569 @@
+"""给原生客户端提供有调用上限的模型连接，凭据和业务答案留在控制器。"""
+
+import asyncio
+import copy
+import hashlib
+import json
+import re
+import secrets
+import socket
+from collections.abc import AsyncGenerator, Iterator
+from contextlib import contextmanager, suppress
+from typing import Any, Optional
+
+import anyio
+import httpx
+import uvicorn
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response, StreamingResponse
+from starlette.routing import Route
+from starlette.types import Receive, Scope, Send
+
+from scripts.evaluation.models import ModelSettings
+
+MAX_REQUEST_BYTES = 4 * 1024 * 1024
+MAX_RESPONSE_BYTES = 4 * 1024 * 1024
+MAX_EVENT_BYTES = 1024 * 1024
+TOKEN_KEYS = ("input_tokens", "output_tokens", "total_tokens")
+NATIVE_CONTROL_TOOLS = frozenset({
+    "update_plan", "spawn_agent", "send_input", "resume_agent", "close_agent", "wait", "list_agents",
+    "request_user_input", "search_tools", "tool_search", "search_tool_bm25",
+})
+NATIVE_COLLABORATION_TOOLS = frozenset({
+    "spawn_agent", "followup_task", "interrupt_agent", "list_agents", "send_message", "wait_agent",
+})
+KNOWN_NAMESPACES = frozenset({"functions", "clock", "collaboration", "mcp__evaluation"})
+FIXTURE_TOOLS = frozenset({"moviepilot_api", "read_skill", "read_tool_result"})
+
+
+def allowed_tool(name: str) -> bool:
+    """只允许原生计划/协作控制与本地固定 MCP 目录，不接入真实系统工具。"""
+    normalized = name.removeprefix("functions.").replace(".", "__")
+    return (normalized in NATIVE_CONTROL_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})
+
+
+def project_tools(tools: Any, prefix: str = "") -> tuple[list[dict[str, Any]], list[str], list[str]]:
+    """保留允许工具的原始 schema；外部 MCP 配置污染必须中止，不能静默投影成合格评测。"""
+    if not isinstance(tools, list) or prefix.count(".") > 8:
+        raise ValueError("工具目录结构无效")
+    projected, retained, removed = [], [], []
+    for tool in tools:
+        if not isinstance(tool, dict) or not isinstance(tool.get("type"), str):
+            raise ValueError("工具定义无效")
+        if tool["type"] in {"function", "custom", "namespace"} and "name" not in tool:
+            raise ValueError("工具缺少名称")
+        if "parameters" in tool and not isinstance(tool["parameters"], dict):
+            raise ValueError("工具参数 schema 无效")
+        name = tool.get("name", tool["type"])
+        if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_.-]{0,255}", name):
+            raise ValueError("工具名称无效")
+        qualified = f"{prefix}.{name}" if prefix else name
+        normalized = qualified.removeprefix("functions.").replace(".", "__")
+        if tool["type"] == "mcp" or (normalized.startswith("mcp__") and not (
+            normalized == "mcp__evaluation" or normalized.startswith("mcp__evaluation__")
+        )):
+            raise ValueError("评测目录包含外部 MCP")
+        if tool["type"] == "namespace":
+            if qualified not in KNOWN_NAMESPACES:
+                raise ValueError("评测目录包含未知命名空间")
+            if "tools" not in tool:
+                raise ValueError("命名空间没有工具目录")
+            children, kept, dropped = project_tools(tool["tools"], qualified)
+            if children:
+                projected.append({**tool, "tools": children})
+            retained.extend(kept)
+            removed.extend(dropped)
+        elif tool["type"] == "tool_search":
+            description = tool.get("description")
+            sources = re.findall(r"(?m)^- ([^:\n]+):", description) if isinstance(description, str) else []
+            if prefix or tool.get("execution") != "client" or sources != ["evaluation"]:
+                raise ValueError("原生工具搜索不是固定评测目录")
+            projected.append(copy.deepcopy(tool))
+            retained.append(qualified)
+        elif allowed_tool(qualified) and tool["type"] in {"function", "custom"}:
+            projected.append(copy.deepcopy(tool))
+            retained.append(qualified)
+        else:
+            removed.append(qualified)
+    return projected, retained, removed
+
+
+def _project_catalogs(payload: dict[str, Any]) -> 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"])
+        value["tools"] = projected
+        retained.extend(kept)
+        removed.extend(dropped)
+        catalogs.append({"location": location, "retained_tools": kept, "removed_tools": dropped})
+
+    def visit(value: Any, location: str) -> None:
+        """递归检查输入及子代理历史，未知额外目录封装必须拒绝。"""
+        if isinstance(value, dict):
+            kind = value.get("type")
+            if isinstance(kind, str) and kind.startswith("additional_tools"):
+                if (kind != "additional_tools" or value.get("role") != "developer" or "tools" not in value
+                        or set(value) - {"type", "id", "role", "tools"}):
+                    raise ValueError("原生额外工具目录结构无效")
+                project(value, f"{location}.tools")
+                return
+            if kind == "tool_search_output":
+                _validate_search_output(value)
+                project(value, f"{location}.tools")
+                return
+            for key, child in value.items():
+                visit(child, f"{location}.{key}")
+        elif isinstance(value, list):
+            for index, child in enumerate(value):
+                visit(child, f"{location}[{index}]")
+
+    if "tools" in payload:
+        project(payload, "tools")
+    visit(payload.get("input", []), "input")
+    return retained, removed, catalogs
+
+
+def project_host_instructions(payload: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+    """只去除原生独立 AGENTS 用户说明块，保留系统提示及同消息内的业务文本。"""
+    removed: list[dict[str, Any]] = []
+    envelope = re.compile(r"\A# AGENTS\.md instructions(?: for [^\r\n]+)?\n\n<INSTRUCTIONS>\n[\s\S]*\n</INSTRUCTIONS>\Z")
+    marker = "# AGENTS.md instructions"
+    omitted = object()
+
+    def visit(value: Any, *, top_level: bool = False) -> Any:
+        """递归处理子代理历史，未知封装不能靠关键词删除后继续发送。"""
+        if isinstance(value, dict):
+            result = {}
+            removed_content = False
+            for key, child in value.items():
+                if top_level and key == "instructions":
+                    result[key] = copy.deepcopy(child)
+                elif key == "content" and value.get("role") == "user" and isinstance(child, list):
+                    blocks = []
+                    for block in child:
+                        content = block.get("text") if isinstance(block, dict) else None
+                        kind = block.get("kind") if isinstance(block, dict) else None
+                        if isinstance(content, str) and (marker in content or kind == "agents_md.instructions"):
+                            if (block.get("type") not in {"input_text", "text"}
+                                    or kind not in {None, "agents_md.instructions"} or not envelope.fullmatch(content)):
+                                raise ValueError("无法确认宿主说明块边界")
+                            encoded = content.encode("utf-8")
+                            removed.append({"sha256": hashlib.sha256(encoded).hexdigest(), "bytes": len(encoded)})
+                            removed_content = True
+                        else:
+                            blocks.append(visit(block))
+                    result[key] = blocks
+                else:
+                    result[key] = visit(child)
+            return omitted if removed_content and not result.get("content") else result
+        if isinstance(value, list):
+            values = [visit(child) for child in value]
+            return [child for child in values if child is not omitted]
+        if isinstance(value, str) and any(boundary in value for boundary in (
+            marker, "<skills_instructions>", "</skills_instructions>", "<INSTRUCTIONS>", "</INSTRUCTIONS>",
+        )):
+            raise ValueError("宿主说明出现在未知位置")
+        return value
+
+    return visit(payload, top_level=True), removed
+
+
+def _input_metadata(payload: Any) -> list[dict[str, Any]]:
+    """仅审计主输入文本块的角色、类型和指纹，不保存宿主偏好或用户任务正文。"""
+    if not isinstance(payload, dict):
+        return []
+    result = []
+    for item in payload.get("input", []) if isinstance(payload.get("input"), list) else []:
+        if not isinstance(item, dict):
+            continue
+        role = item.get("role")
+        content = item.get("content")
+        blocks = content if isinstance(content, list) else [{"type": "text", "text": content}]
+        for block in blocks:
+            text = block.get("text") if isinstance(block, dict) else None
+            if isinstance(text, str):
+                encoded = text.encode("utf-8")
+                result.append({"role": role if isinstance(role, str) and role in {"user", "assistant", "developer", "system", "tool"} else None,
+                               "type": block.get("type") if isinstance(block.get("type"), str) and block["type"] in {"text", "input_text", "output_text"} else None,
+                               "bytes": len(encoded), "sha256": hashlib.sha256(encoded).hexdigest()})
+    return result
+
+
+def _strings(payload: Any) -> Iterator[str]:
+    """遍历解析后的字段和值，转义后的凭据也不能从响应或审计元数据泄露。"""
+    if isinstance(payload, str):
+        yield payload
+    elif isinstance(payload, dict):
+        for key, value in payload.items():
+            yield from _strings(key)
+            yield from _strings(value)
+    elif isinstance(payload, list):
+        for value in payload:
+            yield from _strings(value)
+
+
+def _validate_search_call(payload: dict[str, Any], *, partial: bool = False) -> None:
+    """只接受原生客户端的查询参数，服务端搜索或自选来源不能进入受控运行。"""
+    if payload.get("execution") != "client" or not isinstance(payload.get("call_id"), str) or not payload["call_id"]:
+        raise ValueError("原生工具搜索调用边界无效")
+    arguments = payload.get("arguments")
+    if partial and arguments in (None, "", {}):
+        return
+    if not isinstance(arguments, dict) or set(arguments) - {"query", "limit"}:
+        raise ValueError("原生工具搜索参数无效")
+    if not isinstance(arguments.get("query"), str) or not arguments["query"].strip():
+        raise ValueError("原生工具搜索缺少查询")
+    limit = arguments.get("limit", 8)
+    if limit is not None and (type(limit) is not int or limit <= 0):
+        raise ValueError("原生工具搜索数量无效")
+
+
+def _validate_search_output(payload: dict[str, Any]) -> 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"))
+    fixture_names = {f"mcp__evaluation__{tool}" for tool in FIXTURE_TOOLS}
+    if removed or any(name.removeprefix("functions.").replace(".", "__") not in fixture_names for name in retained):
+        raise ValueError("搜索结果包含评测目录以外的定义")
+
+
+def _reject_unsafe_calls(payload: Any, *, partial: bool = False) -> 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)
+        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):
+                raise ValueError("模型返回了受控目录以外的工具调用")
+        for value in payload.values():
+            _reject_unsafe_calls(value, partial=partial or kind == "response.output_item.added")
+    elif isinstance(payload, list):
+        for value in payload:
+            _reject_unsafe_calls(value, partial=partial)
+
+
+class _LocalServer(uvicorn.Server):
+    """回环服务的取消由评测控制器统一管理，不占用宿主进程的信号处理器。"""
+
+    @contextmanager
+    def capture_signals(self) -> Iterator[None]:
+        """控制器并行管理 MCP 和模型连接，服务不得互相覆盖全局信号处理。"""
+        yield
+
+
+class _ModelStreamingResponse(StreamingResponse):
+    """客户端断开时显式等待生成器收尾，确保连接和审计结果不依赖垃圾回收。"""
+
+    def __init__(self, body: AsyncGenerator[str, None]) -> None:
+        """保留本次响应独占的生成器，后续关闭不会影响其他模型请求。"""
+        super().__init__(body, media_type="text/event-stream")
+        self._body = body
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        """流式发送被取消也要完成局部清理，之后才让控制器读取终态。"""
+        try:
+            await super().__call__(scope, receive, send)
+        finally:
+            with anyio.CancelScope(shield=True):
+                await self._body.aclose()
+
+
+class EvaluationModelProxy:
+    """原生 Codex 使用本地随机令牌，真正供应商凭据只由此固定目标代理持有。"""
+
+    def __init__(self, settings: ModelSettings, *, probe_only: bool = False,
+                 transport: Optional[httpx.AsyncBaseTransport] = None) -> None:
+        """创建独立计数和目录记录；probe_only 只记录目录，绝不真正出站。"""
+        self.settings = settings
+        self.probe_only = probe_only
+        self.bearer_token = secrets.token_urlsafe(32)
+        self.endpoint = ""
+        self.requests: list[dict[str, Any]] = []
+        self.blocked_calls = 0
+        self._rejected_requests: list[dict[str, Any]] = []
+        self._client = httpx.AsyncClient(transport=transport, trust_env=False, timeout=min(120, settings.timeout_seconds))
+        self._socket: Optional[socket.socket] = None
+        self._server: Optional[uvicorn.Server] = None
+        self._task: Optional[asyncio.Task[Any]] = None
+        self._closed = False
+        self._server_error: Optional[str] = None
+        self.app = Starlette(routes=[Route("/v1/responses", self._respond, methods=["POST"])])
+
+    async def __aenter__(self) -> "EvaluationModelProxy":
+        """显式启动本地 HTTP 服务，不访问真实模型。"""
+        await self.start()
+        return self
+
+    async def __aexit__(self, *_args: Any) -> None:
+        """退出时回收服务和连接，不把运行任务交给解释器退出。"""
+        await self.close()
+
+    async def start(self) -> None:
+        """绑定随机回环端口；启动失败或取消也必须关闭连接、监听和子任务。"""
+        if self._task is not None or self._closed:
+            raise RuntimeError("模型代理已经启动或关闭")
+        try:
+            self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            self._socket.bind(("127.0.0.1", 0))
+            self.endpoint = f"http://127.0.0.1:{self._socket.getsockname()[1]}/v1"
+            self._server = _LocalServer(uvicorn.Config(self.app, log_config=None, access_log=False, log_level="error"))
+            self._task = asyncio.create_task(self._server.serve(sockets=[self._socket]))
+            async with asyncio.timeout(10):
+                while not self._server.started:
+                    if self._task.done():
+                        await self._task
+                        raise RuntimeError("模型代理未能启动")
+                    await asyncio.sleep(0.01)
+        except BaseException:
+            await self.close()
+            raise
+
+    async def close(self) -> None:
+        """有界等待退出，幂等回收；子任务异常也不能跳过后续资源关闭。"""
+        if self._server is not None:
+            self._server.should_exit = True
+        try:
+            if self._task is not None:
+                try:
+                    await asyncio.wait_for(asyncio.shield(self._task), 10)
+                except asyncio.TimeoutError:
+                    self._task.cancel()
+                except Exception as error:
+                    self._server_error = type(error).__name__
+                finally:
+                    if not self._task.done():
+                        self._task.cancel()
+                    with suppress(asyncio.CancelledError, Exception):
+                        await self._task
+        finally:
+            self._closed = True
+            try:
+                await self._client.aclose()
+            finally:
+                if self._socket is not None:
+                    self._socket.close()
+
+    @staticmethod
+    def _error(message: str, status: int = 400) -> JSONResponse:
+        """只返回固定控制错误，不回显原始供应商响应、密钥或提示词。"""
+        return JSONResponse({"error": {"message": message, "type": "evaluation_error"}}, status_code=status)
+
+    def _validate_payload(self, payload: Any) -> tuple[dict[str, Any], dict[str, Any]]:
+        """请求边界同时约束模型、预算和目录，子模型不能静默替换对照条件。"""
+        if not isinstance(payload, dict) or payload.get("model") != self.settings.model:
+            raise ValueError("模型与评测配置不符")
+        if self.settings.api_key in "".join(_strings(payload)):
+            raise ValueError("请求包含私有凭据")
+        payload, host_instructions = project_host_instructions(payload)
+        reasoning = payload.get("reasoning")
+        if reasoning is None:
+            reasoning = {}
+        if not isinstance(reasoning, dict) or reasoning.get("effort") not in {None, self.settings.reasoning_effort}:
+            raise ValueError("推理预算与评测配置不符")
+        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", []))
+        choice = payload.get("tool_choice")
+        if isinstance(choice, dict) and not (choice.get("type") in {"function", "custom"}
+                                             and allowed_tool(str(choice.get("name", "")))):
+            raise ValueError("强制工具不在受控目录")
+        record = {"requested_model": payload["model"], "reasoning_effort": reasoning.get("effort"),
+                  "retained_tools": retained, "removed_tools": removed, "tool_catalogs": catalogs, "forwarded": False,
+                  "removed_host_instructions": host_instructions, "input_blocks": _input_metadata(payload),
+                  "request_field_types": {key: {"type": type(value).__name__,
+                                                "length": len(value) if isinstance(value, (str, list, dict)) else None}
+                                          for key, value in payload.items()},
+                  "completed": False, "usage": None, "response_model": None, "error_type": None}
+        payload["max_output_tokens"] = min(tokens, self.settings.max_output_tokens)
+        return payload, record
+
+    async def _respond(self, request: Request) -> Response:
+        """认证与原子预算判断先于真正出站，探测请求单独计数。"""
+        if request.headers.get("authorization") != f"Bearer {self.bearer_token}":
+            return self._error("Local evaluation authentication failed", 401)
+        if self._closed:
+            return self._error("Local evaluation proxy is closed", 503)
+        origin = request.headers.get("origin")
+        if origin is not None and origin != self.endpoint.removesuffix("/v1"):
+            return self._error("Unexpected request origin", 403)
+        body = bytearray()
+        async for chunk in request.stream():
+            body.extend(chunk)
+            if len(body) > MAX_REQUEST_BYTES:
+                return self._error("Evaluation request is too large", 413)
+        parsed = None
+        try:
+            parsed = json.loads(body)
+            payload, record = self._validate_payload(parsed)
+        except (ValueError, TypeError, RecursionError) as error:
+            self._rejected_requests.append({"error_type": type(error).__name__, "input_blocks": _input_metadata(parsed)})
+            return self._error("Invalid evaluation model request")
+        if self.probe_only:
+            self.requests.append(record)
+            return self._error("Evaluation probe complete; no model request was forwarded")
+        if len(self.requests) >= self.settings.max_model_calls:
+            self.blocked_calls += 1
+            return self._error("Evaluation model call budget exhausted")
+        record["forwarded"] = True
+        self.requests.append(record)
+        try:
+            upstream = await self._client.send(self._client.build_request(
+                "POST", f"{self.settings.base_url.rstrip('/')}/responses", json=payload,
+                headers={"Authorization": f"Bearer {self.settings.api_key}", "Content-Type": "application/json"},
+            ), stream=True)
+        except httpx.HTTPError as error:
+            record["error_type"] = type(error).__name__
+            return self._error("Configured evaluation model provider could not be reached", 502)
+        if upstream.status_code != 200:
+            record["http_status"] = upstream.status_code
+            record["error_type"] = "upstream_rejected"
+            await upstream.aclose()
+            return self._error("Configured evaluation model provider rejected the request")
+        if payload.get("stream"):
+            return _ModelStreamingResponse(self._stream(upstream, record))
+        return await self._response(upstream, record)
+
+    async def _response(self, upstream: httpx.Response, record: dict[str, Any]) -> Response:
+        """非流式响应同样有大小和工具边界，错误正文始终留在代理内部。"""
+        body = bytearray()
+        try:
+            async for chunk in upstream.aiter_bytes():
+                body.extend(chunk)
+                if len(body) > MAX_RESPONSE_BYTES:
+                    raise ValueError("响应超限")
+            response = json.loads(body)
+            self._validate_response(response)
+            self._capture(response, record)
+            return JSONResponse(response)
+        except (ValueError, TypeError, RecursionError, httpx.HTTPError):
+            record["error_type"] = "invalid_response"
+            return self._error("Model response violated the evaluation contract")
+        finally:
+            await upstream.aclose()
+
+    def _validate_response(self, value: Any) -> None:
+        """在任何正文交给原生客户端之前拒绝凭据、非法调用与非对象响应。"""
+        if not isinstance(value, dict) or self.settings.api_key in "".join(_strings(value)):
+            raise ValueError("响应格式或凭据边界无效")
+        _reject_unsafe_calls(value)
+
+    @staticmethod
+    def _capture(response: dict[str, Any], record: dict[str, Any]) -> None:
+        """只记录完整非负用量；响应声明完成与流后续失败分别保留。"""
+        if not isinstance(response, dict):
+            raise ValueError("响应完成对象无效")
+        record["completed"] = response.get("status") == "completed"
+        usage = response.get("usage")
+        record["usage"] = ({key: usage[key] for key in TOKEN_KEYS} if isinstance(usage, dict) and all(
+            type(usage.get(key)) is int and usage[key] >= 0 for key in TOKEN_KEYS
+        ) else None)
+        model = response.get("model")
+        record["response_model"] = model if isinstance(model, str) else None
+
+    def _event(self, raw: bytes, record: dict[str, Any], tails: dict[str, str]) -> str:
+        """校验完整 SSE 帧以及跨帧文本片段，非法调用所在帧不向客户端发送。"""
+        text = raw.decode("utf-8")
+        data = "\n".join(line[5:].removeprefix(" ") for line in text.splitlines() if line.startswith("data:"))
+        if self.settings.api_key in text:
+            raise ValueError("事件包含私有凭据")
+        if data and data != "[DONE]":
+            value = json.loads(data)
+            self._validate_response(value)
+            delta = value.get("delta")
+            if isinstance(delta, str):
+                identity = str(value.get("item_id", ""))
+                combined = tails.get(identity, "") + delta
+                if self.settings.api_key in combined:
+                    raise ValueError("跨事件包含私有凭据")
+                tails[identity] = combined[-max(1, len(self.settings.api_key) - 1):]
+            kind = value.get("type")
+            if kind in {"response.completed", "response.incomplete", "response.failed"}:
+                self._capture(value.get("response"), record)
+                record["terminal_event"] = kind
+            elif kind == "error":
+                record["error_type"] = "upstream_stream_error"
+        return text + "\n\n"
+
+    async def _stream(self, upstream: httpx.Response, record: dict[str, Any]) -> AsyncGenerator[str, None]:
+        """完整事件校验后转发，终态交付后即关闭上游，不再等待客户端通常忽略的尾帧。"""
+        pending = bytearray()
+        tails: dict[str, str] = {}
+        received = 0
+        try:
+            async for chunk in upstream.aiter_bytes():
+                received += len(chunk)
+                if received > MAX_RESPONSE_BYTES:
+                    raise ValueError("事件流总量超限")
+                pending.extend(chunk)
+                while True:
+                    match = re.search(rb"\r?\n\r?\n", pending)
+                    if match is None:
+                        break
+                    if match.start() > MAX_EVENT_BYTES:
+                        raise ValueError("事件超限")
+                    raw = bytes(pending[:match.start()])
+                    del pending[:match.end()]
+                    yield self._event(raw, record, tails)
+                    if "terminal_event" in record:
+                        return
+                if len(pending) > MAX_EVENT_BYTES:
+                    raise ValueError("事件超限")
+            if pending:
+                yield self._event(bytes(pending), record, tails)
+                if "terminal_event" in record:
+                    return
+            if "terminal_event" not in record and record["error_type"] is None:
+                raise ValueError("事件流缺少终态")
+        except (ValueError, TypeError, RecursionError, httpx.HTTPError):
+            record["error_type"] = "invalid_stream"
+            record["completed"] = False
+            yield 'event: error\ndata: {"error":{"type":"evaluation_error","message":"Model stream violated the evaluation contract"}}\n\n'
+        except (asyncio.CancelledError, GeneratorExit):
+            self._record_disconnect(record)
+            raise
+        finally:
+            try:
+                await upstream.aclose()
+            except asyncio.CancelledError:
+                self._record_disconnect(record)
+                raise
+
+    @staticmethod
+    def _record_disconnect(record: dict[str, Any]) -> None:
+        """客户端收到终态后关闭属于正常完成，终态前断开则保留未完成事实。"""
+        if "terminal_event" in record:
+            record["client_closed_after_terminal"] = True
+        else:
+            record["error_type"] = "stream_cancelled"
+            record["completed"] = False
+
+    def snapshot(self) -> dict[str, Any]:
+        """探测不计真实消耗；缺失用量保持未知，已知总量是可审计下界。"""
+        records = copy.deepcopy(self.requests)
+        forwarded = [record for record in records if record["forwarded"]]
+        known = [record["usage"] for record in forwarded if isinstance(record["usage"], dict)]
+        return {"model_calls": len(forwarded), "probe_requests": len(records) - len(forwarded),
+                "completed_model_calls": sum(bool(record["completed"]) for record in forwarded),
+                "blocked_model_calls": self.blocked_calls, "usage_complete": len(known) == len(forwarded),
+                "reported_models": sorted({record["response_model"] for record in forwarded
+                                           if isinstance(record["response_model"], str)}),
+                "tokens": {key: sum(usage[key] for usage in known) for key in TOKEN_KEYS} if known else None,
+                "model_requests": records, "server_error": self._server_error,
+                "rejected_requests": copy.deepcopy(self._rejected_requests)}

scripts/evaluation/server.py

@@ -0,0 +1,459 @@
+"""仅绑定回环地址的受控 MCP JSON-RPC 子集，不加载 MoviePilot 运行时或真实业务服务。"""
+
+import asyncio
+import hashlib
+import json
+import re
+import secrets
+import socket
+import time
+from collections import Counter, OrderedDict
+from collections.abc import Iterator
+from contextlib import contextmanager
+from copy import deepcopy
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Optional
+
+import uvicorn
+import yaml  # type: ignore[import-untyped]  # 锁定的 PyYAML 不提供类型声明。
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.routing import Route
+
+from scripts.evaluation.world import EvaluationWorld
+
+PROTOCOL_VERSIONS = ("2025-11-25", "2025-06-18", "2025-03-26")
+MAX_REQUEST_BYTES = 1024 * 1024
+MAX_SKILL_BYTES = 512 * 1024
+MAX_RESULT_BYTES = 1024 * 1024
+MAX_TOTAL_RESULT_BYTES = 4 * MAX_RESULT_BYTES
+MAX_RESULTS = 8
+RESULT_TTL_SECONDS = 900
+FIRST_PAGE_CHARS = 8192
+NEXT_PAGE_CHARS = 16000
+
+# 与 Agent 的 MoviePilotApiInput 保持相同字段，不提前暴露场景支持操作或 oracle 判据。
+API_INPUT_SCHEMA: dict[str, Any] = {
+    "type": "object", "required": ["operation_id"], "additionalProperties": False,
+    "properties": {
+        "operation_id": {"type": "string", "description": (
+            "Exact allowlisted MoviePilot operation ID selected from the loaded domain Skill. "
+            "Never supply a URL, authentication header, or API token."
+        )},
+        "path_params": {"type": "object", "additionalProperties": True, "description": "Route placeholder values declared by the selected operation."},
+        "query": {"type": "object", "additionalProperties": True, "description": "Query-string fields declared by the selected operation."},
+        "body": {"default": None, "description": (
+            "JSON request value declared by the selected operation and its loaded Skill contract. "
+            "Most operations use an object; a oneOf branch may require an exact scalar."
+        )},
+    },
+}
+
+
+@dataclass(frozen=True)
+class _StoredResult:
+    """本次评测独占的有界结果归档，MCP 客户端会话共享但其他服务实例不可读取。"""
+
+    text: str
+    tool_name: str
+    byte_size: int
+    expires_at: float
+
+
+@dataclass
+class _Session:
+    """只保存 MCP 握手状态，不拥有或重置业务世界。"""
+
+    protocol: str
+    initialized: bool = False
+
+
+class _LocalServer(uvicorn.Server):  # type: ignore[misc]  # follow_imports=skip 不展开外部 Server 基类。
+    """让控制器拥有服务启停，嵌入运行时不替换进程信号处理器。"""
+
+    def __init__(self, config: uvicorn.Config, ready: asyncio.Event) -> None:
+        """通过事件通知实际套接字监听已就绪。"""
+        super().__init__(config)
+        self.ready = ready
+
+    @contextmanager
+    def capture_signals(self) -> Iterator[None]:
+        """主控制器处理取消和信号，MCP 服务不接管全局状态。"""
+        yield
+
+    async def startup(self, sockets: Optional[list[socket.socket]] = None) -> None:
+        """完成 uvicorn 的真实监听后才允许模型开始握手。"""
+        await super().startup(sockets=sockets)
+        self.ready.set()
+
+
+class EvaluationMcpServer:
+    """为一个独立世界提供受认证的本地 Streamable HTTP 必需子集。"""
+
+    def __init__(self, world: EvaluationWorld, *, root_dir: Optional[Path] = None, max_world_calls: int = 32) -> None:
+        """固定世界与可信源码目录；模型不能通过请求选择场景、文件或真实后端。"""
+        if type(max_world_calls) is not int or not 1 <= max_world_calls <= 32:
+            raise ValueError("max_world_calls 必须为 1 到 32 的整数")
+        self._world = world
+        self._max_world_calls = max_world_calls
+        self._world_calls = 0
+        self._blocked_world_calls = 0
+        self._tool_calls: Counter[str] = Counter()
+        self._bearer_token = secrets.token_urlsafe(32)
+        self._endpoint: Optional[str] = None
+        self._listener: Optional[socket.socket] = None
+        self._server: Optional[_LocalServer] = None
+        self._task: Optional[asyncio.Task[None]] = None
+        self._ready = asyncio.Event()
+        self._lifecycle_lock = asyncio.Lock()
+        self._closed = False
+        self._sessions: dict[str, _Session] = {}
+        self._results: OrderedDict[str, _StoredResult] = OrderedDict()
+        self._skill = self._load_skill(Path(root_dir) if root_dir is not None else Path(__file__).resolve().parents[2])
+
+    @property
+    def endpoint(self) -> str:
+        """控制器在 start 完成后取得回环 MCP 地址，不包含认证秘密。"""
+        if self._endpoint is None or self._closed:
+            raise RuntimeError("MCP 评测服务尚未运行")
+        return self._endpoint
+
+    @property
+    def bearer_token(self) -> str:
+        """仅供控制器放入客户端局部环境，不会出现在任何工具或协议响应中。"""
+        return self._bearer_token
+
+    @property
+    def stats(self) -> dict[str, Any]:
+        """提供无场景答案、无参数正文的控制器计数快照。"""
+        return {"world_calls": self._world_calls, "blocked_world_calls": self._blocked_world_calls,
+                "tool_calls": dict(self._tool_calls), "max_world_calls": self._max_world_calls}
+
+    @property
+    def skill_sha256(self) -> str:
+        """控制器可记录实际公开技能正文的指纹，模型工具不暴露额外上下文。"""
+        return hashlib.sha256(self._skill["content"].encode("utf-8")).hexdigest()
+
+    async def __aenter__(self) -> "EvaluationMcpServer":
+        """异步上下文进入时等待真实端点就绪。"""
+        return await self.start()
+
+    async def __aexit__(self, *_args: Any) -> None:
+        """控制器正常退出、失败或取消时均关闭端点。"""
+        await self.close()
+
+    async def start(self) -> "EvaluationMcpServer":
+        """绑定系统分配的 127.0.0.1 端口，不读取任何业务运行配置。"""
+        async with self._lifecycle_lock:
+            if self._closed:
+                raise RuntimeError("已关闭的评测服务不能重新启动")
+            if self._task is not None:
+                return self
+            listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            self._listener = listener
+            try:
+                listener.bind(("127.0.0.1", 0))
+                listener.setblocking(False)
+                self._endpoint = f"http://127.0.0.1:{listener.getsockname()[1]}/mcp"
+                app = Starlette(routes=[Route("/mcp", self._handle, methods=["POST", "GET", "DELETE"])])
+                config = uvicorn.Config(app, host="127.0.0.1", log_config=None, log_level=None, access_log=False,
+                                        lifespan="off", proxy_headers=False, ws="none", timeout_graceful_shutdown=2)
+                self._server = _LocalServer(config, self._ready)
+                self._task = asyncio.create_task(self._server.serve(sockets=[listener]))
+                self._task.add_done_callback(lambda _task: self._ready.set())
+                await asyncio.wait_for(self._ready.wait(), timeout=10)
+                if self._task.done() or not self._server.started:
+                    raise RuntimeError("本地 MCP 服务启动失败")
+            except BaseException:
+                await self._stop()
+                raise
+        return self
+
+    async def close(self) -> None:
+        """显式、幂等关闭 HTTP 服务，保留业务世界供控制器独立评分。"""
+        async with self._lifecycle_lock:
+            await self._stop()
+
+    async def _stop(self) -> None:
+        """有限等待服务退出，异常路径也回收监听套接字与归档。"""
+        self._closed = True
+        try:
+            if self._server is not None:
+                self._server.should_exit = True
+            if self._task is not None:
+                try:
+                    await asyncio.wait_for(asyncio.shield(self._task), timeout=5)
+                except TimeoutError:
+                    self._task.cancel()
+                    await asyncio.gather(self._task, return_exceptions=True)
+                except asyncio.CancelledError:
+                    self._task.cancel()
+                    await asyncio.gather(self._task, return_exceptions=True)
+                    raise
+        finally:
+            if self._listener is not None:
+                self._listener.close()
+            self._sessions.clear()
+            self._results.clear()
+
+    @staticmethod
+    def _error(request_id: Any, code: int, message: str, status: int = 400) -> JSONResponse:
+        """错误只使用固定协议文案，不回显异常、环境、参数或私有文件路径。"""
+        return JSONResponse({"jsonrpc": "2.0", "id": request_id, "error": {"code": code, "message": message}}, status_code=status)
+
+    def _http_error(self, request: Request) -> Optional[Response]:
+        """先校验认证、Host 与 Origin，防止回环服务被浏览器或代理借用。"""
+        expected = f"Bearer {self._bearer_token}"
+        if len(request.headers.getlist("authorization")) != 1 or not secrets.compare_digest(
+            request.headers.get("authorization", "").encode("utf-8"), expected.encode("ascii"),
+        ):
+            return self._error(None, -32001, "Unauthorized", 401)
+        origin = self.endpoint.removesuffix("/mcp")
+        if request.headers.get("host") != origin.removeprefix("http://") or request.headers.get("origin", origin) != origin:
+            return self._error(None, -32001, "Forbidden", 403)
+        if request.client is None or request.client.host != "127.0.0.1":
+            return self._error(None, -32001, "Forbidden", 403)
+        version = request.headers.get("mcp-protocol-version")
+        if version is not None and version not in PROTOCOL_VERSIONS:
+            return self._error(None, -32600, "Unsupported protocol version")
+        return None
+
+    async def _handle(self, request: Request) -> Response:
+        """处理 JSON-RPC POST；本服务不提供服务器推送 SSE 或远程关闭能力。"""
+        rejected = self._http_error(request)
+        if rejected is not None:
+            return rejected
+        if request.method != "POST":
+            return Response(status_code=405, headers={"Allow": "POST"})
+        accepted = {item.split(";", 1)[0].strip().lower() for item in request.headers.get("accept", "").split(",")}
+        if not {"application/json", "text/event-stream"} <= accepted:
+            return self._error(None, -32600, "Accept must include application/json and text/event-stream", 406)
+        if request.headers.get("content-type", "").split(";", 1)[0].lower() != "application/json":
+            return self._error(None, -32600, "Content-Type must be application/json", 415)
+        try:
+            data = bytearray()
+            async for chunk in request.stream():
+                data.extend(chunk)
+                if len(data) > MAX_REQUEST_BYTES:
+                    return self._error(None, -32600, "Request too large", 413)
+            payload = json.loads(data.decode("utf-8"), parse_constant=self._invalid_json_constant)
+        except (ValueError, UnicodeError):
+            return self._error(None, -32700, "Parse error")
+        if not self._valid_request(payload):
+            return self._error(None, -32600, "Invalid Request")
+        try:
+            return self._dispatch(request, payload)
+        except Exception:
+            return self._error(payload.get("id"), -32603, "Internal error", 500)
+
+    @staticmethod
+    def _invalid_json_constant(_value: str) -> None:
+        """JSON-RPC 仅接受标准 JSON，拒绝 Python 默认允许的 NaN/Infinity 扩展。"""
+        raise ValueError("Non-finite JSON number")
+
+    @staticmethod
+    def _valid_request(payload: Any) -> bool:
+        """仅接受单条 JSON-RPC 请求或通知，拒绝 batch 和可混淆的标量类型。"""
+        return (type(payload) is dict and payload.get("jsonrpc") == "2.0"
+                and not set(payload) - {"jsonrpc", "id", "method", "params"}
+                and type(payload.get("method")) is str and bool(payload["method"])
+                and type(payload.get("params", {})) is dict
+                and ("_meta" not in payload.get("params", {}) or type(payload["params"]["_meta"]) is dict)
+                and ("id" not in payload or type(payload["id"]) in (str, int)))
+
+    def _dispatch(self, request: Request, payload: dict[str, Any]) -> Response:
+        """握手会话只管理协议状态，所有合法工具调用始终进入同一评测世界。"""
+        method, params, request_id = payload["method"], payload.get("params", {}), payload.get("id")
+        if method == "initialize":
+            return self._initialize(params, request_id) if "id" in payload else self._error(None, -32600, "initialize requires an id")
+        session = self._sessions.get(request.headers.get("mcp-session-id", ""))
+        if session is None:
+            return self._error(request_id, -32000, "Unknown MCP session", 404 if request.headers.get("mcp-session-id") else 400)
+        if request.headers.get("mcp-protocol-version", session.protocol) != session.protocol:
+            return self._error(request_id, -32600, "Protocol version differs from initialized session")
+        if method == "notifications/initialized":
+            if "id" in payload or set(params) - {"_meta"}:
+                return self._error(request_id, -32600, "Invalid initialized notification")
+            session.initialized = True
+            return Response(status_code=202)
+        if method == "notifications/cancelled" and "id" not in payload:
+            if (set(params) - {"requestId", "reason", "_meta"} or type(params.get("requestId")) not in (str, int)
+                    or ("reason" in params and type(params["reason"]) is not str)):
+                return self._error(None, -32600, "Invalid cancellation notification")
+            return Response(status_code=202)
+        if "id" not in payload:
+            return self._error(None, -32600, "Unsupported notification")
+        if not session.initialized and method != "ping":
+            return self._error(request_id, -32000, "MCP session is not initialized")
+        if method in {"ping", "tools/list"}:
+            allowed = {"_meta", "cursor"} if method == "tools/list" else {"_meta"}
+            if set(params) - allowed or params.get("cursor") not in (None, ""):
+                return self._error(request_id, -32602, "Invalid params")
+            result: Any = {} if method == "ping" else {"tools": self._tool_catalog()}
+        elif method == "tools/call":
+            if set(params) - {"name", "arguments", "_meta"} or type(params.get("name")) is not str or type(params.get("arguments", {})) is not dict:
+                return self._error(request_id, -32602, "Invalid tools/call params")
+            result = self._call_tool(params["name"], params.get("arguments", {}))
+        else:
+            return self._error(request_id, -32601, "Method not found", 200)
+        return JSONResponse({"jsonrpc": "2.0", "id": request_id, "result": result})
+
+    def _initialize(self, params: dict[str, Any], request_id: Any) -> Response:
+        """按正式握手版本协商，新客户端会话不会重置业务状态或调用预算。"""
+        client = params.get("clientInfo")
+        if (set(params) - {"protocolVersion", "capabilities", "clientInfo", "_meta"}
+                or type(params.get("protocolVersion")) is not str
+                or not re.fullmatch(r"\d{4}-\d{2}-\d{2}", params["protocolVersion"])
+                or type(params.get("capabilities")) is not dict or type(client) is not dict
+                or any(type(client.get(key)) is not str or not client[key] for key in ("name", "version"))):
+            return self._error(request_id, -32602, "Invalid initialize params")
+        if len(self._sessions) >= 64:
+            return self._error(request_id, -32000, "Session limit reached", 429)
+        protocol = params["protocolVersion"] if params["protocolVersion"] in PROTOCOL_VERSIONS else PROTOCOL_VERSIONS[0]
+        session_id = secrets.token_hex(24)
+        self._sessions[session_id] = _Session(protocol)
+        result = {"protocolVersion": protocol, "capabilities": {"tools": {"listChanged": False}},
+                  "serverInfo": {"name": "MoviePilot Evaluation", "version": "1"},
+                  "instructions": "Use the MoviePilot domain Skill to select API operations and interpret results."}
+        return JSONResponse({"jsonrpc": "2.0", "id": request_id, "result": result}, headers={"Mcp-Session-Id": session_id})
+
+    @staticmethod
+    def _tool_catalog() -> list[dict[str, Any]]:
+        """只公开与 MoviePilot 任务输入等价的工具合同，不注入场景答案。"""
+        return [
+            {"name": "moviepilot_api", "description": (
+                "Call allowlisted MoviePilot business APIs. Use the domain Skill to select operation_id, parameters, and failure handling. "
+                "For collection counts, use the smallest documented page and read collection.total_count instead of querying the database after item truncation. "
+                "Arbitrary URLs, commands, and authentication endpoints are forbidden."
+            ), "inputSchema": deepcopy(API_INPUT_SCHEMA)},
+            {"name": "read_skill", "description": "Read the named MoviePilot Skill including its full body and supporting file paths. Available skill: moviepilot-api.", "inputSchema": {
+                "type": "object", "required": ["name"], "additionalProperties": False, "properties": {"name": {"type": "string"}},
+            }},
+            {"name": "read_tool_result", "description": "Read an archived tool result using result_id and next_offset. Offsets count Unicode characters. Results expire after 15 minutes or eviction; never guess result IDs.", "inputSchema": {
+                "type": "object", "required": ["result_id"], "additionalProperties": False, "properties": {
+                    "result_id": {"type": "string", "minLength": 32, "maxLength": 32},
+                    "offset": {"type": "integer", "minimum": 0, "default": 0},
+                    "limit": {"type": "integer", "minimum": 1, "maximum": NEXT_PAGE_CHARS, "default": 4000},
+                },
+            }},
+        ]
+
+    @staticmethod
+    def _failure(error: str, message: str) -> dict[str, Any]:
+        """工具失败使用固定四态字段，异常细节不交给模型。"""
+        return {"success": False, "execution_outcome": "failed", "error": error, "message": message}
+
+    def _call_tool(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]:
+        """精确工具白名单与全局预算均在进入业务世界前检查。"""
+        self._tool_calls[name if name in {"moviepilot_api", "read_skill", "read_tool_result"} else "unknown"] += 1
+        try:
+            if name == "moviepilot_api":
+                if set(arguments) - set(API_INPUT_SCHEMA["properties"]) or type(arguments.get("operation_id")) is not str:
+                    result: Any = self._failure("invalid_arguments", "Invalid moviepilot_api arguments")
+                elif any(key in arguments and type(arguments[key]) is not dict for key in ("path_params", "query")):
+                    result = self._failure("invalid_arguments", "path_params and query must be objects")
+                elif self._world_calls >= self._max_world_calls:
+                    self._blocked_world_calls += 1
+                    result = self._failure("world_call_limit", "World tool call budget exhausted")
+                else:
+                    self._world_calls += 1
+                    result = self._world.execute(**arguments)
+            elif name == "read_skill":
+                result = self._read_skill(arguments)
+            elif name == "read_tool_result":
+                result = self._read_result(arguments)
+            else:
+                result = self._failure("tool_not_found", "Unknown tool")
+        except Exception:
+            result = self._failure("evaluation_tool_error", "Controlled tool execution failed")
+        text = result if isinstance(result, str) else json.dumps(result, ensure_ascii=False, indent=2)
+        outcome = json.loads(text).get("execution_outcome")
+        return {"content": [{"type": "text", "text": text}], "isError": outcome in {"failed", "unknown"}}
+
+    @staticmethod
+    def _load_skill(root: Path) -> dict[str, Any]:
+        """读取唯一可信技能文件，不从请求中接受路径，也不加载宿主元数据解析器。"""
+        skill_path = root / "skills/moviepilot-api/SKILL.md"
+        with skill_path.open("rb") as handle:
+            data = handle.read(MAX_SKILL_BYTES + 1)
+        truncated = len(data) > MAX_SKILL_BYTES
+        content = data[:MAX_SKILL_BYTES].decode("utf-8", errors="ignore" if truncated else "replace")
+        match = re.match(r"^---\s*\n(.*?)\n---\s*\n", content, re.DOTALL)
+        metadata = yaml.safe_load(match.group(1)) if match else None
+        if not isinstance(metadata, dict) or metadata.get("name") != "moviepilot-api":
+            raise ValueError("moviepilot-api Skill 元数据无效")
+        return {"success": True, "skill": {
+            "id": "moviepilot-api", "name": "moviepilot-api", "description": str(metadata.get("description", ""))[:1024],
+            "path": "skills/moviepilot-api/SKILL.md", "allowed_tools": str(metadata.get("allowed-tools", "")).split(),
+            "allowed_api_operations": str(metadata.get("allowed-api-operations", "")).split(),
+        }, "content": content, "content_limit_bytes": MAX_SKILL_BYTES, "supporting_files": sorted(
+            str(path.relative_to(skill_path.parent)) for path in skill_path.parent.rglob("*") if path.is_file() and path != skill_path
+        ), "truncated": truncated,
+            "truncation_message": "SKILL.md exceeds 512 KiB; content contains only the first 512 KiB." if truncated else None}
+
+    def _prune_results(self) -> None:
+        """匹配 MoviePilot 的 15 分钟、8 项、4MiB 会话归档约束。"""
+        now = time.monotonic()
+        for key in list(self._results):
+            if self._results[key].expires_at <= now:
+                del self._results[key]
+        while self._results and (len(self._results) > MAX_RESULTS or sum(value.byte_size for value in self._results.values()) > MAX_TOTAL_RESULT_BYTES):
+            self._results.popitem(last=False)
+
+    @staticmethod
+    def _bounded_page(build: Any, limit: int, length: int) -> str:
+        """按最终 JSON 文本长度二分页边界，游标只覆盖实际交付的字符。"""
+        low, high = 0, length
+        best = json.dumps(build(0), ensure_ascii=False, indent=2)
+        while low <= high:
+            middle = (low + high) // 2
+            candidate = json.dumps(build(middle), ensure_ascii=False, indent=2)
+            if len(candidate) <= limit:
+                best, low = candidate, middle + 1
+            else:
+                high = middle - 1
+        return best
+
+    def _read_skill(self, arguments: dict[str, Any]) -> Any:
+        """返回完整真实技能的首屏或可续读归档，不提供精简场景专用说明。"""
+        if set(arguments) != {"name"} or arguments.get("name") != "moviepilot-api":
+            return self._failure("skill_not_found", "Unknown Skill")
+        text = json.dumps(self._skill, ensure_ascii=False, indent=2)
+        if len(text) <= FIRST_PAGE_CHARS:
+            return text
+        if len(text.encode("utf-8")) > MAX_RESULT_BYTES:
+            return self._failure("result_too_large", "Skill result exceeds archive limit")
+        result_id = secrets.token_hex(16)
+        self._results[result_id] = _StoredResult(text, "read_skill", len(text.encode("utf-8")), time.monotonic() + RESULT_TTL_SECONDS)
+        self._prune_results()
+
+        def page(length: int) -> dict[str, Any]:
+            """首屏与 MoviePilot 归档预览使用相同的定位字段。"""
+            return {"result_id": result_id, "read_tool": "read_tool_result", "expires_in_seconds": RESULT_TTL_SECONDS,
+                    "offset_unit": "unicode_characters", "next_offset": length, "execution_outcome": "succeeded",
+                    "tool_result_truncated": True, "tool_name": "read_skill", "total_chars": len(text),
+                    "returned_chars": length, "content_preview": text[:length],
+                    "message": "完整结果已在当前会话临时保存；使用 read_tool_result 的 result_id 和 next_offset 继续读取，无需重复执行原工具。"}
+        return self._bounded_page(page, FIRST_PAGE_CHARS, len(text))
+
+    def _read_result(self, arguments: dict[str, Any]) -> Any:
+        """按 Unicode 字符续读本次服务归档，非法位置不得生成伪观察内容。"""
+        offset, limit, result_id = arguments.get("offset", 0), arguments.get("limit", 4000), arguments.get("result_id")
+        if (set(arguments) - {"result_id", "offset", "limit"} or type(result_id) is not str or len(result_id) != 32
+                or type(offset) is not int or offset < 0 or type(limit) is not int or not 1 <= limit <= NEXT_PAGE_CHARS):
+            return self._failure("invalid_arguments", "Invalid result cursor or limit")
+        self._prune_results()
+        result = self._results.get(result_id)
+        if result is None:
+            return self._failure("result_unavailable", "Result unavailable")
+        if offset > len(result.text):
+            return self._failure("offset_out_of_range", "Result offset out of range")
+
+        def page(length: int) -> dict[str, Any]:
+            """最终响应预算包含转义开销，next_offset 始终对应实际返回的正文。"""
+            end = offset + length
+            return {"success": True, "result_id": result_id, "tool_name": result.tool_name, "offset": offset,
+                    "next_offset": end if end < len(result.text) else None, "total_chars": len(result.text), "content": result.text[offset:end]}
+        return self._bounded_page(page, NEXT_PAGE_CHARS, min(limit, len(result.text) - offset))

tests/test_agent_evaluation_codex.py

@@ -0,0 +1,353 @@
+"""原生适配器只运行假进程或控制边界，核验生命周期、目录隔离和独立评分真实性。"""
+
+import asyncio
+import json
+import os
+import subprocess
+import sys
+import tomllib
+from copy import deepcopy
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, Optional
+from unittest.mock import AsyncMock
+
+import pytest
+
+from scripts.evaluation import codex
+from scripts.evaluation.models import ModelSettings
+
+
+def _program(tmp_path: Path, code: str) -> list[str]:
+    """把无害 Python 假程序放在测试目录，不调用真实 Codex 或模型。"""
+    script = tmp_path / "fake_native.py"
+    script.write_text(code, encoding="utf-8")
+    return [sys.executable, "-u", str(script)]
+
+
+def _jsonl(events: list[dict[str, Any]]) -> str:
+    """构造与原生客户端 stdout 相同的一行一个事件格式。"""
+    return "\n".join(json.dumps(event, ensure_ascii=False) for event in events) + "\n"
+
+
+def _completed_events(final: dict[str, Any]) -> list[dict[str, Any]]:
+    """完成证据来自终态事件，最终业务报告仍交给真实 oracle。"""
+    return [{"type": "turn.started"}, {"type": "item.completed", "item": {
+        "type": "agent_message", "text": json.dumps(final),
+    }}, {"type": "turn.completed"}]
+
+
+def test_new_turn_cannot_reuse_previous_completion_or_final_text() -> None:
+    """上一轮成功后开始了新轮但尚未完成，不能继续标成已完成。"""
+    events, final, completed = codex._events(_jsonl([*_completed_events({"status": "completed"}), {"type": "turn.started"}]))
+    assert len(events) == 4
+    assert completed is False
+    assert final == ""
+
+
+def test_invalid_jsonl_preserves_valid_events_without_claiming_completion() -> None:
+    """损坏日志用摘要保留诊断，不泄露非事件正文，也不能被后续 completed 抹平。"""
+    text = _jsonl(_completed_events({"status": "completed"})) + "private-broken-event\n"
+    events, final, completed = codex._events(text)
+    assert completed is False
+    assert json.loads(final)["status"] == "completed"
+    assert events[-1]["type"] == "evaluation.invalid_event"
+    assert len(events[-1]["sha256"]) == 64
+    assert "private-broken-event" not in json.dumps(events)
+
+
+@pytest.mark.asyncio
+async def test_execute_sends_exact_prompt_and_preserves_nonzero_exit_output(tmp_path: Path) -> None:
+    """真实假程序从 stdin 接收公开输入，非零退出与两路输出不得被丢弃。"""
+    command = _program(tmp_path, "import sys\ntext=sys.stdin.read()\nprint(text)\nprint('stderr-marker',file=sys.stderr)\nraise SystemExit(7)\n")
+    result = await codex._execute(command, "公开任务：保持原样 $HOME `echo no`", codex._worker_environment(), tmp_path, 5)
+    assert result["returncode"] == 7
+    assert result["error_type"] is None
+    assert result["stdout"].strip() == "公开任务：保持原样 $HOME `echo no`"
+    assert "stderr-marker" in result["stderr"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(os.name != "posix", reason="继承进程组回收是 POSIX 合同")
+@pytest.mark.parametrize(("cancel", "parent_exits"), [(False, False), (False, True), (True, False)])
+async def test_execute_timeout_or_cancellation_closes_descendant_pipes(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, cancel: bool, parent_exits: bool,
+) -> None:
+    """继承 stdout 的后代不关闭管道就不会出现 EOF，用真实 EOF 验证整组回收。"""
+    command = _program(tmp_path, (
+        "import subprocess,sys,time\n"
+        "sys.stdin.read()\n"
+        "child=subprocess.Popen([sys.executable,'-c','import time; time.sleep(30)'],stdin=subprocess.DEVNULL)\n"
+        "print('READY',flush=True)\n"
+        + ("raise SystemExit(0)\n" if parent_exits else "time.sleep(30)\n")
+    ))
+    ready = asyncio.Event()
+    captured: list[asyncio.subprocess.Process] = []
+    original_create = asyncio.create_subprocess_exec
+
+    async def capture(*args: Any, **kwargs: Any) -> asyncio.subprocess.Process:
+        """在真实 stdout 读取边界发布握手，取消测试不依靠固定休眠。"""
+        process = await original_create(*args, **kwargs)
+        captured.append(process)
+        original_read = process.stdout.read
+
+        async def read(size: int = -1) -> bytes:
+            """确认父进程已创建后代并输出 READY 后才允许测试发起取消。"""
+            data = await original_read(size)
+            if b"READY" in data:
+                ready.set()
+            return data
+
+        monkeypatch.setattr(process.stdout, "read", read)
+        return process
+
+    monkeypatch.setattr(codex.asyncio, "create_subprocess_exec", capture)
+    task = asyncio.create_task(codex._execute(command, "task", codex._worker_environment(), tmp_path, 10 if cancel else 1))
+    try:
+        await asyncio.wait_for(ready.wait(), 5)
+        if cancel:
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+        else:
+            result = await task
+            assert result["error_type"] == "TimeoutError"
+            assert "READY" in result["stdout"]
+            assert result["returncode"] == (0 if parent_exits else -9)
+        assert captured[0].returncode is not None
+        assert await asyncio.wait_for(captured[0].stdout.read(), 2) == b""
+    finally:
+        if not task.done():
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+        for process in captured:
+            if process.returncode is None:
+                await codex._stop_process(process)
+
+
+@pytest.mark.asyncio
+async def test_output_limit_preserves_bounded_prefix_and_stops_process(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """超过输出预算时保留已取得的前缀，记录控制器失败并终止假进程。"""
+    monkeypatch.setattr(codex, "MAX_PROCESS_OUTPUT_BYTES", 128)
+    command = _program(tmp_path, "import sys,time\nsys.stdout.write('x'*4096)\nsys.stdout.flush()\ntime.sleep(30)\n")
+    result = await codex._execute(command, "", codex._worker_environment(), tmp_path, 5)
+    assert result["error_type"] == "RuntimeError"
+    assert result["stdout"] == "x" * 128
+    assert result["returncode"] is not None
+
+
+@pytest.mark.asyncio
+async def test_cleanup_failure_still_cancels_and_joins_output_readers(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """组终止失败不能跳过其他读取器的 finally 回收。"""
+    stdout, stderr = object(), object()
+    readers: list[asyncio.Task[Any]] = []
+    process = SimpleNamespace(stdin=SimpleNamespace(write=lambda _data: None, drain=AsyncMock(), close=lambda: None),
+                              stdout=stdout, stderr=stderr, returncode=None)
+
+    async def read(stream: Any, _buffer: bytearray) -> None:
+        """一条读取器失败，另一条保持挂起以验证 gather 不会自动替控制器清理。"""
+        readers.append(asyncio.current_task())
+        if stream is stdout:
+            raise RuntimeError("read failed")
+        await asyncio.Event().wait()
+
+    monkeypatch.setattr(codex.asyncio, "create_subprocess_exec", AsyncMock(return_value=process))
+    monkeypatch.setattr(codex, "_read_output", read)
+    monkeypatch.setattr(codex, "_stop_process", AsyncMock(side_effect=RuntimeError("cleanup failed")))
+    try:
+        with pytest.raises(RuntimeError, match="cleanup failed"):
+            await codex._execute(["fake"], "", {}, tmp_path, 5)
+        assert len(readers) == 2 and all(reader.done() for reader in readers)
+    finally:
+        for reader in readers:
+            if not reader.done():
+                reader.cancel()
+        await asyncio.gather(*readers, return_exceptions=True)
+
+
+def _model_catalog() -> dict[str, Any]:
+    """模拟二进制自带的模型记录，保留与工具模式无关的嵌套原生配置。"""
+    return {"models": [{"slug": "gpt-test", "tool_mode": "code_mode_only", "base_instructions": "native instructions",
+                        "model_messages": {"instructions_template": "native message template"}, "limits": [1, 2]},
+                       {"slug": "unrelated", "tool_mode": "direct"}]}
+
+
+def test_catalog_projection_changes_only_selected_tool_mode_and_is_independent() -> None:
+    """控制条件投影不能替换模型提示词，也不能原地改写从二进制取得的目录。"""
+    original = _model_catalog()
+    before = deepcopy(original)
+    projected, metadata = codex._catalog(original, "gpt-test")
+    assert projected == {"models": [{**original["models"][0], "tool_mode": "direct"}]}
+    assert metadata["original_tool_mode"] == "code_mode_only"
+    assert metadata["projected_tool_mode"] == "direct"
+    assert metadata["original_model_sha256"] != metadata["projected_model_sha256"]
+    projected["models"][0]["model_messages"]["instructions_template"] = "changed copy"
+    assert original == before
+
+
+@pytest.mark.parametrize("catalog", [{}, {"models": []}, {"models": [{"slug": "gpt-test"}, {"slug": "gpt-test"}]}])
+def test_missing_or_ambiguous_selected_model_never_falls_back(catalog: dict[str, Any]) -> None:
+    """模型元数据缺失或重复时必须中止，不能换另一模型制造成功对照。"""
+    with pytest.raises(ValueError):
+        codex._catalog(catalog, "gpt-test")
+
+
+async def _run_case(monkeypatch: pytest.MonkeyPatch, *, returncode: int = 0, completed: bool = True,
+                    error_type: Optional[str] = None, echo_tokens: bool = False) -> dict[str, Any]:
+    """只替换原生进程与模型边界，业务状态和 oracle 仍使用真正内存世界。"""
+    worlds = []
+    settings = ModelSettings("gpt-test", "https://provider.invalid/v1", "private-provider-test-key")
+
+    class Server:
+        """捕获同一世界，等价模拟真实 MCP 已读取的独立业务事实。"""
+
+        endpoint = "http://127.0.0.1:49997/mcp"
+        bearer_token = "local-mcp-test-token"
+        skill_sha256 = "skill-digest"
+        stats: dict[str, Any] = {}
+
+        def __init__(self, world: Any) -> None:
+            """保留控制器传入的真实世界，模拟器不能替换 oracle。"""
+            worlds.append(world)
+
+        async def __aenter__(self) -> "Server":
+            """不启动服务，所有验证都在本进程的模拟边界内。"""
+            return self
+
+        async def __aexit__(self, *_args: Any) -> None:
+            """没有外部监听或进程需要清理。"""
+
+    class Proxy(Server):
+        """模型调用统计由模拟控制边界给出，不产生真实请求。"""
+
+        endpoint = "http://127.0.0.1:49998/v1"
+        bearer_token = "local-model-test-token"
+
+        def __init__(self, _settings: ModelSettings, **_kwargs: Any) -> None:
+            """接受真实控制器参数，但不把连接凭据送到子进程。"""
+            self.closed = False
+
+        async def __aexit__(self, *_args: Any) -> None:
+            """模拟客户端断开后的最终资源收敛，统计必须在此之后取得。"""
+            self.closed = True
+
+        def snapshot(self) -> dict[str, Any]:
+            """返回最终模拟用量，关闭前不能冻结仍可能变化的完成/取消状态。"""
+            assert self.closed
+            return {"model_calls": 1, "completed_model_calls": 1, "probe_requests": 0, "model_requests": [],
+                    "tokens": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}}
+
+    async def execute(command: list[str], prompt: str, environment: dict[str, str], work_dir: Path, _timeout: int) -> dict[str, Any]:
+        """检查 argv、私有环境与中立目录，然后生成原生 JSONL 事件。"""
+        assert settings.api_key not in json.dumps(environment)
+        assert "OPENAI_API_KEY" not in environment and "CONFIG_DIR" not in environment
+        assert list(work_dir.iterdir()) == []
+        if command[1:] == ["debug", "models", "--bundled"]:
+            assert not prompt
+            return {"returncode": 0, "stdout": json.dumps(_model_catalog()), "stderr": "", "error_type": None}
+        assert command[1] == "exec" and command[-1] == "-"
+        for flag in ("--ignore-user-config", "--ignore-rules", "--strict-config", "--ephemeral", "--json", "--skip-git-repo-check"):
+            assert flag in command
+        assert command[command.index("--sandbox") + 1] == "read-only"
+        assert command[command.index("-C") + 1] == str(work_dir)
+        assert 'approval_policy="never"' in command
+        mcp = tomllib.loads(next(value for value in command if value.startswith("mcp_servers=")))["mcp_servers"]
+        assert set(mcp) == {"evaluation"}
+        assert set(mcp["evaluation"]["enabled_tools"]) == {"moviepilot_api", "read_skill", "read_tool_result"}
+        assert mcp["evaluation"]["tools"] == {name: {"approval_mode": "approve"}
+                                                for name in mcp["evaluation"]["enabled_tools"]}
+        assert settings.api_key not in json.dumps(command)
+        assert "oracle" not in prompt and "dedup_existing" not in prompt
+        world = worlds[0]
+        subscription = world.execute("subscription.find", path_params={"media_id": world.scenario.media_id},
+                                     query={"media_source": world.scenario.media_source})["data"]
+        downloads = world.execute("download.tasks.active")["data"]
+        final = {"status": "completed", "subscription_ids": [subscription["id"]],
+                 "download_ids": [row["id"] for row in downloads if row["infohash"] == world.scenario.infohash],
+                 "enabled_site_ids": [], "completed": ["subscription", "download"], "unresolved": []}
+        if echo_tokens:
+            final["debug"] = [Server.bearer_token, Proxy.bearer_token]
+        events = _completed_events(final)
+        if not completed:
+            events[-1] = {"type": "turn.failed", "error": {"message": "native failed"}}
+        stdout = _jsonl(events)
+        if echo_tokens:
+            for token in (Server.bearer_token, Proxy.bearer_token):
+                stdout = stdout.replace(token, "".join(f"\\u{ord(char):04x}" for char in token))
+        return {"returncode": returncode, "stdout": stdout, "stderr": "", "error_type": error_type}
+
+    monkeypatch.setattr(codex, "EvaluationMcpServer", Server)
+    monkeypatch.setattr(codex, "EvaluationModelProxy", Proxy)
+    monkeypatch.setattr(codex, "_execute", execute)
+    monkeypatch.setenv("OPENAI_API_KEY", "unrelated-private-key")
+    return await codex._run_codex("dedup_existing", settings, "fake-codex", probe_only=False)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("returncode", "completed", "error_type"), [(7, True, None), (0, False, None), (0, True, "TimeoutError"), (0, True, None)])
+async def test_native_success_is_required_even_when_real_oracle_passes(monkeypatch: pytest.MonkeyPatch, returncode: int,
+                                                                    completed: bool, error_type: Optional[str]) -> None:
+    """真实状态和读取证据正确也不能掩盖原生退出、未完成或控制器异常。"""
+    report = await _run_case(monkeypatch, returncode=returncode, completed=completed, error_type=error_type)
+    assert report["task_passed"] is True
+    assert report["passed"] is (returncode == 0 and completed and error_type is None)
+    assert report["native_exit_code"] == returncode
+    assert report["native_turn_completed"] is completed
+
+
+@pytest.mark.asyncio
+async def test_local_tokens_are_removed_after_native_json_unicode_decoding(monkeypatch: pytest.MonkeyPatch) -> None:
+    """JSON Unicode 转义不能绕过本地代理和 MCP 令牌的最终报告脱敏。"""
+    report = await _run_case(monkeypatch, echo_tokens=True)
+    serialized = json.dumps(report)
+    assert "local-mcp-test-token" not in serialized
+    assert "local-model-test-token" not in serialized
+
+
+@pytest.mark.parametrize(("names", "calls", "rejected", "world_calls", "ready"), [
+    (["functions.update_plan"], 0, [], 0, False),
+    (["tool_search"], 0, [], 0, True),
+    (["tool_search"], 1, [], 0, False),
+    (["tool_search"], 0, [{}], 0, False),
+    (["tool_search"], 0, [], 1, False),
+])
+def test_probe_requires_fixture_discovery_without_model_or_world_execution(names: list[str], calls: int,
+                                                                          rejected: list[Any], world_calls: int,
+                                                                          ready: bool) -> None:
+    """空目录和仅计划目录不能被作为可执行对照入口，探针也不能消耗真实预算。"""
+    usage = {"probe_requests": 1, "model_calls": calls, "rejected_requests": rejected,
+             "model_requests": [{"retained_tools": names}]}
+    assert codex._probe_ready(usage, {"world_calls": world_calls}) is ready
+
+
+@pytest.mark.parametrize("version", ["codex-cli 0.153.3", codex.SUPPORTED_CLI_VERSION])
+def test_cli_version_gate_uses_isolated_environment_before_any_execution(monkeypatch: pytest.MonkeyPatch, version: str) -> None:
+    """只在已核对版本上进入适配器；版本查询本身也不继承真实模型凭据。"""
+    calls = []
+
+    def check(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]:
+        """模拟真实 --version 响应，不运行原生客户端。"""
+        calls.append((command, kwargs))
+        return subprocess.CompletedProcess(command, 0, stdout=version + "\n", stderr="")
+
+    execute = AsyncMock(return_value={"checked": True})
+    monkeypatch.setattr(codex.shutil, "which", lambda _name: "/fake/codex")
+    monkeypatch.setattr(codex.subprocess, "run", check)
+    monkeypatch.setattr(codex, "_run_codex", execute)
+    monkeypatch.setenv("OPENAI_API_KEY", "private-key-not-for-native")
+    settings = ModelSettings("gpt-test", "https://provider.invalid/v1", "provider-key")
+    if version != codex.SUPPORTED_CLI_VERSION:
+        with pytest.raises(RuntimeError, match="只核对过"):
+            codex.run_codex("dedup_existing", settings)
+        execute.assert_not_awaited()
+    else:
+        assert codex.run_codex("dedup_existing", settings) == {"checked": True}
+    assert calls[0][0] == ["/fake/codex", "--version"]
+    assert calls[0][1]["timeout"] == 10
+    assert "OPENAI_API_KEY" not in calls[0][1]["env"]
+
+
+def test_toml_arguments_preserve_literals_without_shell_interpretation() -> None:
+    """带引号、换行和 shell 字符的模型配置仍是单个正确 TOML 值。"""
+    value = {"name": "引号\"与换行\n以及$(command)", "enabled": False, "budget": 12}
+    assert tomllib.loads("value=" + codex._toml(value))["value"] == value

tests/test_agent_evaluation_proxy.py

@@ -0,0 +1,645 @@
+"""模型代理的离线协议、工具隔离、预算、凭据和资源回收回归。"""
+
+import asyncio
+import json
+import signal
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from typing import Any, Optional
+
+import httpx
+import pytest
+
+from scripts.evaluation import proxy as proxy_module
+from scripts.evaluation.models import ModelSettings
+from scripts.evaluation.proxy import EvaluationModelProxy, project_host_instructions, project_tools
+
+SETTINGS = ModelSettings("test-model", "https://model.invalid/v1", "private-test-provider-key", max_model_calls=2)
+USAGE = {"input_tokens": 12, "output_tokens": 8, "total_tokens": 20}
+
+
+def _payload(**updates: Any) -> dict[str, Any]:
+    """构造与原生 Responses 客户端相同形态的受控工具请求。"""
+    return {"model": SETTINGS.model, "tools": [
+        {"type": "function", "name": "update_plan", "parameters": {}},
+        {"type": "function", "name": "mcp__evaluation__moviepilot_api", "parameters": {}},
+        {"type": "custom", "name": "apply_patch", "format": {"type": "text"}},
+    ], **updates}
+
+
+def _completed(**updates: Any) -> dict[str, Any]:
+    """提供真实协议的完成信号，用量缺失与返回模型可以独立变化。"""
+    return {"id": "response-test", "status": "completed", "model": SETTINGS.model, "usage": USAGE, "output": [], **updates}
+
+
+def _event(value: Any) -> bytes:
+    """生成标准 SSE 帧，测试仍可刻意跨任意字节拆分。"""
+    return f"data: {json.dumps(value)}\n\n".encode()
+
+
+class _ByteStream(httpx.AsyncByteStream):
+    """可检查关闭状态的离线字节流，允许模拟帧间故障。"""
+
+    def __init__(self, chunks: list[Any]) -> None:
+        """保存测试字节及异常，不持有套接字。"""
+        self.chunks = chunks
+        self.closed = False
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        """按固定顺序返回片段，以覆盖真正的增量解析路径。"""
+        for chunk in self.chunks:
+            if isinstance(chunk, Exception):
+                raise chunk
+            yield chunk
+
+    async def aclose(self) -> None:
+        """记录代理是否在成功、拒绝和中途失败时回收上游。"""
+        self.closed = True
+
+
+@asynccontextmanager
+async def _client(proxy: EvaluationModelProxy) -> AsyncIterator[httpx.AsyncClient]:
+    """ASGI 与 MockTransport 两侧都留在内存，不访问真实网络。"""
+    try:
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(proxy.app), base_url="http://127.0.0.1",
+                                    headers={"Authorization": f"Bearer {proxy.bearer_token}"}) as client:
+            yield client
+    finally:
+        await proxy.close()
+
+
+def test_namespace_projection_preserves_allowed_schema_and_audits_removed_tools() -> None:
+    """目录投影保留函数原定义，不能把原生文件能力混入对照世界。"""
+    tools = [{"type": "namespace", "name": "functions", "description": "native controls", "tools": _payload()["tools"]}]
+    projected, retained, removed = project_tools(tools)
+    assert retained == ["functions.update_plan", "functions.mcp__evaluation__moviepilot_api"]
+    assert removed == ["functions.apply_patch"]
+    assert projected[0]["description"] == "native controls"
+    assert projected[0]["tools"] == tools[0]["tools"][:2]
+    projected[0]["tools"][0]["parameters"]["changed"] = True
+    assert tools[0]["tools"][0]["parameters"] == {}
+
+
+@pytest.mark.asyncio
+async def test_probe_records_catalog_without_forwarding_or_spending_budget() -> None:
+    """探测首个请求目录即可继续调试，统计不能把探测伪装为真实模型调用。"""
+    outbound = []
+    proxy = EvaluationModelProxy(SETTINGS, probe_only=True, transport=httpx.MockTransport(lambda request: outbound.append(request)))
+    async with _client(proxy) as client:
+        for _ in range(3):
+            assert (await client.post("/v1/responses", json=_payload())).status_code == 400
+    result = proxy.snapshot()
+    assert not outbound and result["model_calls"] == 0 and result["probe_requests"] == 3
+    assert result["tokens"] is None and result["blocked_model_calls"] == 0
+    result["model_requests"][0]["retained_tools"].append("forged")
+    assert "forged" not in proxy.snapshot()["model_requests"][0]["retained_tools"]
+
+
+@pytest.mark.parametrize("updates", [
+    {"tools": None}, {"tools": [None]}, {"tools": [{"type": "function"}]},
+    {"tools": [{"type": "function", "name": 42}]}, {"tools": [{"type": "namespace", "name": "functions", "tools": {}}]},
+    {"tools": [{"type": "function", "name": "update_plan", "parameters": []}]},
+    {"tools": [{"type": "function", "name": "mcp__private__message"}]},
+    {"tools": [{"type": "mcp", "server_url": "https://private.invalid"}]},
+    {"max_output_tokens": "100"}, {"max_output_tokens": 0}, {"max_output_tokens": -1},
+    {"max_output_tokens": True}, {"max_output_tokens": 0.5}, {"max_output_tokens": None},
+    {"stream": "true"}, {"model": "different-model"}, {"reasoning": {"effort": "ultra"}},
+    {"reasoning": ["high"]}, {"reasoning": []}, {"reasoning": False}, {"tool_choice": {"type": "function", "name": "exec_command"}},
+    {"input": [{"type": "custom_tool_call", "name": "apply_patch"}]},
+    {"metadata": {SETTINGS.api_key: "hidden-in-key"}},
+])
+@pytest.mark.asyncio
+async def test_invalid_request_is_rejected_before_outbound_or_accounting(updates: dict[str, Any]) -> None:
+    """恶意目录、错模型和非法预算必须在真正调用前明确失败。"""
+    outbound = []
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda request: outbound.append(request)))
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=_payload(**updates))
+    assert response.status_code == 400 and not outbound and proxy.snapshot()["model_calls"] == 0
+    assert SETTINGS.api_key not in response.text + json.dumps(proxy.snapshot())
+
+
+@pytest.mark.asyncio
+async def test_request_authentication_origin_and_size_fail_without_forwarding(monkeypatch) -> None:
+    """本地地址也需要随机认证，错误来源与过大正文不消耗模型预算。"""
+    proxy = EvaluationModelProxy(SETTINGS, probe_only=True)
+    monkeypatch.setattr(proxy_module, "MAX_REQUEST_BYTES", 20)
+    async with _client(proxy) as client:
+        assert (await client.post("/v1/responses", json={}, headers={"Authorization": "Bearer wrong"})).status_code == 401
+        assert (await client.post("/v1/responses", json={}, headers={"Origin": "https://foreign.invalid"})).status_code == 403
+        assert (await client.post("/v1/responses", json=_payload())).status_code == 413
+        assert (await client.post("/v1/responses", content="malformed")).status_code == 400
+    assert proxy.snapshot()["probe_requests"] == 0
+
+
+@pytest.mark.asyncio
+async def test_forwarding_uses_only_fixed_provider_and_real_key_inside_transport() -> None:
+    """真实凭据只进入固定供应商请求，原生客户端看到的令牌不会被转发。"""
+    outbound = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        """捕获私有连接边界并返回完成响应。"""
+        outbound.append(request)
+        return httpx.Response(200, json=_completed(model="provider-reported-model"))
+
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(handler))
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=_payload(max_output_tokens=999999))
+    assert response.status_code == 200 and len(outbound) == 1
+    request = outbound[0]
+    assert str(request.url) == "https://model.invalid/v1/responses"
+    assert request.headers["Authorization"] == f"Bearer {SETTINGS.api_key}"
+    assert proxy.bearer_token not in str(request.headers)
+    payload = json.loads(request.content)
+    assert payload["max_output_tokens"] == SETTINGS.max_output_tokens
+    assert [tool["name"] for tool in payload["tools"]] == ["update_plan", "mcp__evaluation__moviepilot_api"]
+    result = proxy.snapshot()
+    assert result["tokens"] == USAGE and result["usage_complete"] is True
+    assert result["reported_models"] == ["provider-reported-model"]
+    assert SETTINGS.api_key not in response.text + json.dumps(result)
+
+
+@pytest.mark.asyncio
+async def test_parallel_requests_share_atomic_hard_limit_and_unknown_failure_cost() -> None:
+    """多子代理同时请求也不能超额，已发送失败请求保留为用量未知。"""
+    outbound = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        """让请求挂起一个调度点以模拟重叠调用。"""
+        outbound.append(request)
+        await asyncio.sleep(0)
+        return httpx.Response(500, text=SETTINGS.api_key)
+
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(handler))
+    async with _client(proxy) as client:
+        responses = await asyncio.gather(*(client.post("/v1/responses", json=_payload()) for _ in range(5)))
+    result = proxy.snapshot()
+    assert len(outbound) == result["model_calls"] == 2 and result["blocked_model_calls"] == 3
+    assert result["usage_complete"] is False and result["tokens"] is None
+    assert all(SETTINGS.api_key not in response.text for response in responses)
+
+
+@pytest.mark.parametrize("usage", [None, {}, {"input_tokens": 2}, {**USAGE, "input_tokens": True},
+                                          {**USAGE, "input_tokens": -1}, {**USAGE, "input_tokens": 2.5}])
+@pytest.mark.asyncio
+async def test_partial_or_invalid_usage_stays_unknown(usage: Any) -> None:
+    """完成响应缺少完整合法用量时，不能输出虚构的零 token。"""
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda _: httpx.Response(200, json=_completed(usage=usage))))
+    async with _client(proxy) as client:
+        assert (await client.post("/v1/responses", json=_payload())).status_code == 200
+    result = proxy.snapshot()
+    assert result["completed_model_calls"] == 1 and result["tokens"] is None and result["usage_complete"] is False
+
+
+@pytest.mark.parametrize("output", [
+    [{"type": "function_call", "name": "exec_command", "arguments": "{}"}],
+    [{"type": "custom_tool_call", "name": "apply_patch", "input": "arbitrary"}],
+    [{"type": "computer_call", "action": {"type": "click"}}],
+    [{"type": "message", "content": [{"type": "output_text", "text": SETTINGS.api_key}]}],
+])
+@pytest.mark.asyncio
+async def test_non_streaming_forbidden_calls_and_credentials_never_reach_client(output: Any) -> None:
+    """非流式同样验证整个完成对象，不能依赖模型遵守已过滤目录。"""
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda _: httpx.Response(200, json=_completed(output=output))))
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=_payload())
+    assert response.status_code == 400 and proxy.snapshot()["completed_model_calls"] == 0
+    assert SETTINGS.api_key not in response.text and "exec_command" not in response.text
+
+
+@pytest.mark.parametrize("ending", [b"\n\n", b"\r\n\r\n", b""])
+@pytest.mark.asyncio
+async def test_sse_fragmented_frames_and_final_unterminated_event_are_captured(ending: bytes) -> None:
+    """帧可以跨 TCP 片段，末尾无空行的完整 JSON 事件也必须处理。"""
+    completed = _event({"type": "response.completed", "response": _completed()}).rstrip(b"\n") + ending
+    stream = _ByteStream([completed[:3], completed[3:19], completed[19:]])
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda _: httpx.Response(200, stream=stream)))
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=_payload(stream=True))
+    assert response.status_code == 200 and "response.completed" in response.text and stream.closed
+    assert proxy.snapshot()["tokens"] == USAGE and proxy.snapshot()["completed_model_calls"] == 1
+
+
+@pytest.mark.parametrize("bad_event", [
+    {"type": "response.output_item.added", "item": {"type": "function_call", "name": "exec_command"}},
+    {"type": "response.output_item.done", "item": {"type": "custom_tool_call", "name": "apply_patch"}},
+    {"type": "response.completed", "response": _completed(output=[{"type": "function_call", "name": "exec_command"}])},
+    {"type": "response.output_text.delta", "delta": SETTINGS.api_key},
+])
+@pytest.mark.asyncio
+async def test_sse_blocks_unsafe_frame_before_native_execution(bad_event: dict[str, Any]) -> None:
+    """已经转发安全进度不代表后续帧可信，非法调用帧须阻断并标记失败。"""
+    safe = _event({"type": "response.created", "response": {"id": "safe-progress"}})
+    stream = _ByteStream([safe, _event(bad_event), _event({"type": "response.completed", "response": _completed()})])
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda _: httpx.Response(200, stream=stream)))
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=_payload(stream=True))
+    assert "safe-progress" in response.text and "evaluation_error" in response.text and stream.closed
+    assert "exec_command" not in response.text and "apply_patch" not in response.text and SETTINGS.api_key not in response.text
+    result = proxy.snapshot()
+    assert result["completed_model_calls"] == 0 and result["model_requests"][0]["error_type"] == "invalid_stream"
+
+
+@pytest.mark.asyncio
+async def test_credentials_split_across_deltas_cannot_be_reassembled_by_client() -> None:
+    """密钥拆成多个合法 JSON 帧时，形成完整秘密的帧仍必须被扣留。"""
+    key = SETTINGS.api_key
+    frames = [_event({"type": "response.output_text.delta", "item_id": "message-one", "delta": part})
+              for part in (key[:7], key[7:15], key[15:])]
+    stream = _ByteStream(frames)
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda _: httpx.Response(200, stream=stream)))
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=_payload(stream=True))
+    events = [json.loads(line[6:]) for line in response.text.splitlines() if line.startswith("data: ")]
+    assert key not in "".join(event.get("delta", "") for event in events)
+    assert stream.closed and proxy.snapshot()["completed_model_calls"] == 0
+
+
+@pytest.mark.parametrize("chunks", [[b"data: invalid\n\n"], [b"data: {"],
+                                     [_event({"type": "response.created"})], [httpx.ReadError("private failure")]])
+@pytest.mark.asyncio
+async def test_malformed_truncated_or_failed_stream_has_unknown_completion(chunks: list[Any]) -> None:
+    """没有完整终态或网络中断的流必须保留失败证据，且关闭上游。"""
+    stream = _ByteStream(chunks)
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda _: httpx.Response(200, stream=stream)))
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=_payload(stream=True))
+    assert "evaluation_error" in response.text and "private failure" not in response.text and stream.closed
+    assert proxy.snapshot()["completed_model_calls"] == 0 and proxy.snapshot()["tokens"] is None
+
+
+@pytest.mark.asyncio
+async def test_oversized_event_and_nonstream_response_are_bounded(monkeypatch) -> None:
+    """无分隔符流和大响应都不能无限累计内存。"""
+    monkeypatch.setattr(proxy_module, "MAX_EVENT_BYTES", 16)
+    monkeypatch.setattr(proxy_module, "MAX_RESPONSE_BYTES", 16)
+    for streaming in (False, True):
+        stream = _ByteStream([b"data: " + b"x" * 30])
+        proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda _: httpx.Response(200, stream=stream)))
+        async with _client(proxy) as client:
+            response = await client.post("/v1/responses", json=_payload(stream=streaming))
+        assert "evaluation_error" in response.text and stream.closed
+
+
+@pytest.mark.asyncio
+async def test_startup_failure_closes_client_and_socket(monkeypatch) -> None:
+    """服务启动异常发生在上下文进入前，也必须清理预先分配的资源。"""
+    async def fail(_server: Any, **_kwargs: Any) -> None:
+        """模拟监听服务子任务在报告就绪前失败。"""
+        raise RuntimeError("offline-startup-failure")
+
+    monkeypatch.setattr(proxy_module.uvicorn.Server, "serve", fail)
+    proxy = EvaluationModelProxy(SETTINGS)
+    with pytest.raises(RuntimeError, match="offline-startup-failure"):
+        await proxy.start()
+    assert proxy._client.is_closed and proxy._socket.fileno() == -1 and proxy._task.done()
+    await proxy.close()
+    assert proxy.snapshot()["server_error"] == "RuntimeError"
+
+
+@pytest.mark.asyncio
+async def test_native_host_instructions_are_removed_without_changing_core_or_user_task() -> None:
+    """只移除精确原生用户说明块，不能改写原生系统提示或邻接的任务内容。"""
+    removed = "# AGENTS.md instructions for /synthetic/private\n\n<INSTRUCTIONS>\nsynthetic host preference\n</INSTRUCTIONS>"
+    outbound = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        """保留供应商请求以断言正文仅含允许的评测上下文。"""
+        outbound.append(json.loads(request.content))
+        return httpx.Response(200, json=_completed())
+
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(handler))
+    core = "native core instructions including literal # AGENTS.md instructions example"
+    ordinary = {"type": "input_text", "text": "business task must remain identical"}
+    nested = {"role": "user", "content": [{"type": "text", "kind": "agents_md.instructions", "text": removed}, ordinary]}
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=_payload(
+            instructions=core, input=[{"role": "user", "content": [{"type": "input_text", "text": removed}, ordinary]},
+                                      {"nested_history": [nested]}],
+        ))
+    assert response.status_code == 200 and outbound[0]["instructions"] == core
+    assert outbound[0]["input"][0]["content"] == [ordinary]
+    assert outbound[0]["input"][1]["nested_history"][0]["content"] == [ordinary]
+    records = proxy.snapshot()["model_requests"][0]["removed_host_instructions"]
+    assert len(records) == 2 and records[0]["bytes"] == len(removed.encode()) and len(records[0]["sha256"]) == 64
+    assert "synthetic host preference" not in json.dumps(proxy.snapshot())
+
+
+@pytest.mark.parametrize("block", [
+    {"type": "input_text", "text": "# AGENTS.md instructions\n\n<INSTRUCTIONS>\nmissing closing tag"},
+    {"type": "input_text", "text": "quoted # AGENTS.md instructions should not be silently removed"},
+    {"type": "output_text", "text": "# AGENTS.md instructions\n\n<INSTRUCTIONS>\nbody\n</INSTRUCTIONS>"},
+    {"type": "text", "kind": "foreign.kind", "text": "# AGENTS.md instructions\n\n<INSTRUCTIONS>\nbody\n</INSTRUCTIONS>"},
+    {"type": "input_text", "kind": "agents_md.instructions", "text": "missing known envelope"},
+])
+@pytest.mark.asyncio
+async def test_ambiguous_host_instructions_abort_before_model_request(block: dict[str, Any]) -> None:
+    """无法确认的宿主文本必须停止评测，不能任意删改内容或泄漏给供应商。"""
+    outbound = []
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda request: outbound.append(request)))
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=_payload(input=[{"role": "user", "content": [block]}]))
+    assert response.status_code == 400 and not outbound and proxy.snapshot()["model_calls"] == 0
+
+
+@pytest.mark.asyncio
+async def test_standalone_host_message_is_omitted_in_probe_audit() -> None:
+    """完整用户消息只有宿主块时必须连空消息一起去除，避免产生无效 Responses 输入。"""
+    text = "# AGENTS.md instructions\n\n<INSTRUCTIONS>\nhost\n</INSTRUCTIONS>"
+    payload = _payload(input=[{"role": "user", "content": [{"type": "input_text", "text": text}]},
+                              {"role": "user", "content": [{"type": "input_text", "text": "business"}]}])
+    projected, audit = project_host_instructions(payload)
+    assert len(projected["input"]) == 1 and len(audit) == 1
+    proxy = EvaluationModelProxy(SETTINGS, probe_only=True)
+    async with _client(proxy) as client:
+        assert (await client.post("/v1/responses", json=payload)).status_code == 400
+    assert len(proxy.snapshot()["model_requests"][0]["removed_host_instructions"]) == 1
+
+
+@pytest.mark.parametrize("text", ["<skills_instructions>host skills</skills_instructions>",
+                                  "ordinary <INSTRUCTIONS>unrecognized host</INSTRUCTIONS>"])
+@pytest.mark.asyncio
+async def test_unknown_host_skill_boundaries_abort_with_text_fingerprint_only(text: str) -> None:
+    """原生宿主技能残留未识别时拒绝出站，保留可私下比对的指纹证据。"""
+    outbound = []
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda request: outbound.append(request)))
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=_payload(input=[{"role": "user", "content": [{"type": "input_text", "text": text}]}]))
+    result = proxy.snapshot()
+    assert response.status_code == 400 and not outbound
+    metadata = result["rejected_requests"][0]["input_blocks"][0]
+    assert metadata["role"] == "user" and metadata["type"] == "input_text" and metadata["bytes"] == len(text.encode())
+    assert text not in json.dumps(result) and len(metadata["sha256"]) == 64
+
+
+@pytest.mark.asyncio
+async def test_clean_probe_reports_only_projected_input_fingerprints() -> None:
+    """探测报告可核对送出任务完全相同，又不会持久化任务和宿主说明正文。"""
+    proxy = EvaluationModelProxy(SETTINGS, probe_only=True)
+    async with _client(proxy) as client:
+        await client.post("/v1/responses", json=_payload(input=[{"role": "user", "content": [{"type": "input_text", "text": "business task"}]}]))
+    result = proxy.snapshot()
+    assert result["model_requests"][0]["input_blocks"][0]["bytes"] == 13
+    fields = result["model_requests"][0]["request_field_types"]
+    assert fields["tools"] == {"type": "list", "length": 2}
+    assert fields["model"] == {"type": "str", "length": len(SETTINGS.model)}
+    assert "business task" not in json.dumps(result)
+
+
+def _search_tool(description: str = 'Tools from the following sources:\n- evaluation: Controlled world') -> dict[str, Any]:
+    """构造已在原生线协议观察到的客户端工具搜索定义。"""
+    return {"type": "tool_search", "execution": "client", "description": description,
+            "parameters": {"type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"]}}
+
+
+def _native_payload(**updates: Any) -> dict[str, Any]:
+    """原生工具随 additional_tools 输入传递，不应被代理改成空的顶层 tools。"""
+    return {"model": SETTINGS.model, "input": [{"type": "additional_tools", "id": "synthetic-catalog", "role": "developer", "tools": [
+        {"type": "namespace", "name": "functions", "tools": _payload()["tools"]},
+        {"type": "namespace", "name": "clock", "tools": [{"type": "function", "name": "curr_time", "parameters": {}}]},
+        {"type": "namespace", "name": "collaboration", "tools": [
+            {"type": "function", "name": name, "parameters": {}} for name in
+            ("spawn_agent", "followup_task", "interrupt_agent", "list_agents", "send_message", "wait_agent")
+        ]}, _search_tool(),
+    ]}], **updates}
+
+
+@pytest.mark.asyncio
+async def test_native_additional_tools_are_projected_in_place_and_audited() -> None:
+    """真实原生目录位置必须执行同样隔离，不能只检查不存在的顶层字段。"""
+    outbound = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        """捕获投影后的 wire 结构，不连接任何外部服务。"""
+        outbound.append(json.loads(request.content))
+        return httpx.Response(200, json=_completed())
+
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(handler))
+    async with _client(proxy) as client:
+        assert (await client.post("/v1/responses", json=_native_payload())).status_code == 200
+    assert "tools" not in outbound[0]
+    catalog = outbound[0]["input"][0]["tools"]
+    assert [tool.get("name", tool["type"]) for tool in catalog] == ["functions", "collaboration", "tool_search"]
+    record = proxy.snapshot()["model_requests"][0]
+    assert record["removed_tools"] == ["functions.apply_patch", "clock.curr_time"]
+    assert len(record["retained_tools"]) == 9 and "collaboration.spawn_agent" in record["retained_tools"]
+    assert record["tool_catalogs"][0]["location"] == "input[0].tools"
+
+
+@pytest.mark.parametrize("catalog", [
+    [{"type": "namespace", "name": "unknown", "tools": [{"type": "function", "name": "update_plan"}]}],
+    [_search_tool('Tools from sources:\n- evaluation: fake\n- private: external')],
+    [_search_tool('No explicit source list')],
+    [{**_search_tool(), "execution": "server"}],
+])
+@pytest.mark.asyncio
+async def test_native_unknown_namespace_or_foreign_discovery_aborts_locally(catalog: list[dict[str, Any]]) -> None:
+    """原生未知目录与搜索源污染必须拒绝，而非静默缩减后宣称隔离成功。"""
+    outbound = []
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda request: outbound.append(request)))
+    payload = _native_payload()
+    payload["input"][0]["tools"] = catalog
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=payload)
+    assert response.status_code == 400 and not outbound
+
+
+@pytest.mark.parametrize("updates", [{"type": "additional_tools_v2"}, {"role": "user"}, {"unexpected": "field"}, {"tools": None}])
+@pytest.mark.asyncio
+async def test_native_unknown_additional_tools_shape_aborts_locally(updates: dict[str, Any]) -> None:
+    """额外工具消息未知变体不能绕过目录投影。"""
+    outbound = []
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda request: outbound.append(request)))
+    payload = _native_payload()
+    payload["input"][0].update(updates)
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=payload)
+    assert response.status_code == 400 and not outbound
+
+
+def _search_call(**updates: Any) -> dict[str, Any]:
+    """原生工具搜索调用使用独立响应变体，参数不是 function_call.arguments 字符串。"""
+    return {"type": "tool_search_call", "id": "tool-search-one", "call_id": "search-one", "execution": "client",
+            "arguments": {"query": "MoviePilot API Skill", "limit": 3}, **updates}
+
+
+def _search_output(**updates: Any) -> dict[str, Any]:
+    """模拟原生搜索注册目录后合并出的 MCP 命名空间结果。"""
+    return {"type": "tool_search_output", "call_id": "search-one", "status": "completed", "execution": "client", "tools": [
+        {"type": "namespace", "name": "mcp__evaluation", "tools": [
+            {"type": "function", "name": name, "parameters": {}} for name in ("moviepilot_api", "read_skill", "read_tool_result")
+        ]},
+    ], **updates}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("limit", [3, None])
+async def test_native_client_discovery_call_and_namespace_output_are_allowed(limit: Optional[int]) -> None:
+    """真实搜索调用返回完整命名空间后，下一次模型请求仍只收到同一评测目录。"""
+    call = _search_call(arguments={"query": "MoviePilot API Skill", "limit": limit})
+    frames = [_event({"type": "response.output_item.added", "item": _search_call(arguments={})}),
+              _event({"type": "response.output_item.done", "item": call}),
+              _event({"type": "response.completed", "response": _completed(output=[call])})]
+    stream = _ByteStream(frames)
+    outbound = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        """第一次返回搜索指令，第二次确认已发现工具的后续请求。"""
+        outbound.append(json.loads(request.content))
+        return httpx.Response(200, stream=stream) if len(outbound) == 1 else httpx.Response(200, json=_completed())
+
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(handler))
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=_native_payload(stream=True))
+        assert "tool_search_call" in response.text and "evaluation_error" not in response.text
+        payload = _native_payload()
+        payload["input"].extend([call, _search_output()])
+        assert (await client.post("/v1/responses", json=payload)).status_code == 200
+    assert "tools" not in outbound[1]
+    record = proxy.snapshot()["model_requests"][1]
+    assert record["tool_catalogs"][-1]["location"] == "input[2].tools"
+    assert record["tool_catalogs"][-1]["retained_tools"] == [
+        "mcp__evaluation.moviepilot_api", "mcp__evaluation.read_skill", "mcp__evaluation.read_tool_result",
+    ]
+
+
+@pytest.mark.parametrize("updates", [
+    {"execution": "server"}, {"call_id": None}, {"call_id": ""}, {"arguments": {"query": "  "}},
+    {"arguments": {"query": "x", "sources": ["external"]}}, {"arguments": {"query": "x", "limit": 0}},
+    {"arguments": {"query": "x", "limit": True}}, {"arguments": {"query": "x", "limit": 1.5}},
+    {"arguments": '{"query":"x"}'},
+])
+@pytest.mark.asyncio
+async def test_native_discovery_rejects_server_execution_and_unknown_arguments(updates: dict[str, Any]) -> None:
+    """工具搜索只能操作已注册本地目录，不能让模型提供外部来源或服务端执行指令。"""
+    stream = _ByteStream([_event({"type": "response.output_item.done", "item": _search_call(**updates)})])
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda _: httpx.Response(200, stream=stream)))
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=_native_payload(stream=True))
+    assert "evaluation_error" in response.text and '"type": "tool_search_call"' not in response.text
+
+
+@pytest.mark.parametrize("updates", [
+    {"execution": "server"}, {"call_id": None}, {"status": "unknown"}, {"tools": None},
+    {"tools": [{"type": "function", "name": "mcp__private__read_secret", "parameters": {}}]},
+    {"tools": [{"type": "custom", "name": "apply_patch"}]},
+    {"tools": [{"type": "function", "name": "update_plan", "parameters": {}}]},
+])
+@pytest.mark.asyncio
+async def test_native_discovery_output_cannot_inject_other_tool_definitions(updates: dict[str, Any]) -> None:
+    """工具搜索输出也是目录入口，不能通过历史结果或模型返回另行引入工具。"""
+    outbound = []
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda request: outbound.append(request)))
+    payload = _native_payload()
+    payload["input"].append(_search_output(**updates))
+    async with _client(proxy) as client:
+        response = await client.post("/v1/responses", json=payload)
+    assert response.status_code == 400 and not outbound
+
+
+class _WaitingStream(_ByteStream):
+    """发出首帧后等待，复现客户端不等供应商结束连接的真实 SSE 行为。"""
+
+    def __init__(self, first: bytes) -> None:
+        """保留尾部读取事实，用于证明终态后代理主动停止读取。"""
+        super().__init__([first])
+        self.tail_read = False
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        """尾部故意不结束，由客户端断开或代理主动关闭来结束本次响应。"""
+        yield self.chunks[0]
+        self.tail_read = True
+        await asyncio.Event().wait()
+
+
+@pytest.mark.asyncio
+async def test_terminal_frame_closes_upstream_without_waiting_for_done_or_eof() -> None:
+    """原生收到完成事件即结束，代理不能继续等待尾帧并把正常结束误记为取消。"""
+    stream = _WaitingStream(_event({"type": "response.completed", "response": _completed()}))
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda _: httpx.Response(200, stream=stream)))
+    async with _client(proxy) as client:
+        async with asyncio.timeout(1):
+            response = await client.post("/v1/responses", json=_payload(stream=True))
+    assert "response.completed" in response.text and stream.closed and not stream.tail_read
+    result = proxy.snapshot()
+    assert result["completed_model_calls"] == 1 and result["tokens"] == USAGE
+    assert result["model_requests"][0]["error_type"] is None
+
+
+@pytest.mark.parametrize("completed", [True, False])
+@pytest.mark.parametrize("cancel", [True, False])
+@pytest.mark.asyncio
+async def test_generator_close_or_cancellation_preserves_only_verified_terminal(completed: bool, cancel: bool) -> None:
+    """生成器取消与关闭都区分已观察终态和提前中断，不抹掉已完成模型的真实消耗。"""
+    event = {"type": "response.completed", "response": _completed()} if completed else {"type": "response.created"}
+    stream = _WaitingStream(_event(event))
+    upstream = httpx.Response(200, stream=stream)
+    proxy = EvaluationModelProxy(SETTINGS)
+    record = {"completed": False, "usage": None, "error_type": None}
+    body = proxy._stream(upstream, record)
+    await anext(body)
+    if cancel:
+        with pytest.raises(asyncio.CancelledError):
+            await body.athrow(asyncio.CancelledError())
+    else:
+        await body.aclose()
+    await proxy.close()
+    assert stream.closed and record["completed"] is completed
+    if completed:
+        assert record["usage"] == USAGE and record["error_type"] is None and record["client_closed_after_terminal"] is True
+    else:
+        assert record["usage"] is None and record["error_type"] == "stream_cancelled"
+
+
+@pytest.mark.parametrize("completed", [True, False])
+@pytest.mark.asyncio
+async def test_real_asgi_client_disconnect_keeps_terminal_and_rejects_early_close(completed: bool) -> None:
+    """通过真实 ASGI 断开事件取消流式传输，验证客户端结束时的最终审计结果。"""
+    event = {"type": "response.completed", "response": _completed()} if completed else {"type": "response.created"}
+    stream = _WaitingStream(_event(event))
+    proxy = EvaluationModelProxy(SETTINGS, transport=httpx.MockTransport(lambda _: httpx.Response(200, stream=stream)))
+    disconnect = asyncio.Event()
+    request_sent = False
+    delivered = []
+
+    async def receive() -> dict[str, Any]:
+        """首个事件传入请求，客户端收到首帧后立刻报告断开。"""
+        nonlocal request_sent
+        if not request_sent:
+            request_sent = True
+            return {"type": "http.request", "body": json.dumps(_payload(stream=True)).encode(), "more_body": False}
+        await disconnect.wait()
+        return {"type": "http.disconnect"}
+
+    async def send(message: dict[str, Any]) -> None:
+        """完成交付首帧后让断开监听器抢占，覆盖真正传输生命周期中的取消。"""
+        if message["type"] == "http.response.body" and message.get("body"):
+            delivered.append(message["body"])
+            disconnect.set()
+            await asyncio.sleep(0)
+
+    scope = {"type": "http", "asgi": {"version": "3.0", "spec_version": "2.3"}, "method": "POST",
+             "scheme": "http", "path": "/v1/responses", "raw_path": b"/v1/responses", "query_string": b"",
+             "root_path": "", "http_version": "1.1", "server": ("127.0.0.1", 80), "client": ("127.0.0.1", 1234),
+             "headers": [(b"authorization", f"Bearer {proxy.bearer_token}".encode())]}
+    try:
+        async with asyncio.timeout(1):
+            await proxy.app(scope, receive, send)
+        await asyncio.sleep(0)
+    finally:
+        await proxy.close()
+    assert delivered and stream.closed
+    result = proxy.snapshot()
+    assert result["completed_model_calls"] == int(completed)
+    assert result["tokens"] == (USAGE if completed else None)
+    assert result["model_requests"][0]["error_type"] == (None if completed else "stream_cancelled")
+
+
+@pytest.mark.asyncio
+async def test_proxy_start_and_close_do_not_replace_host_signal_handlers() -> None:
+    """本地模型服务与 MCP 共用控制器时，启动和关闭都不能接管进程全局信号。"""
+    signals = (signal.SIGINT, signal.SIGTERM)
+    previous = {item: signal.getsignal(item) for item in signals}
+    async with EvaluationModelProxy(SETTINGS, probe_only=True):
+        assert {item: signal.getsignal(item) for item in signals} == previous
+    assert {item: signal.getsignal(item) for item in signals} == previous

tests/test_agent_evaluation_server.py

@@ -0,0 +1,299 @@
+"""通过真实回环 HTTP 验证受控 MCP 握手、世界共享、结果分页和关闭，不访问外部服务。"""
+
+import ast
+import asyncio
+import json
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from dataclasses import replace
+from pathlib import Path
+from typing import Any
+
+import httpx
+import pytest
+
+from scripts.evaluation.server import API_INPUT_SCHEMA, EvaluationMcpServer
+from scripts.evaluation.world import EvaluationWorld
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+
+
+def _headers(server: EvaluationMcpServer) -> dict[str, str]:
+    """客户端局部传入随机认证值，不读取进程凭据或代理设置。"""
+    return {"Authorization": f"Bearer {server.bearer_token}", "Accept": "application/json, text/event-stream"}
+
+
+async def _rpc(client: httpx.AsyncClient, method: str, params: Any = None, request_id: Any = 1) -> httpx.Response:
+    """通过实际 TCP 端点发送单条 JSON-RPC 请求。"""
+    payload = {"jsonrpc": "2.0", "id": request_id, "method": method}
+    if params is not None:
+        payload["params"] = params
+    return await client.post("/mcp", json=payload)
+
+
+@asynccontextmanager
+async def _client(server: EvaluationMcpServer, protocol: str = "2025-11-25") -> AsyncIterator[httpx.AsyncClient]:
+    """完成真实 initialize/initialized 握手后交付一个独立 MCP 客户端会话。"""
+    async with httpx.AsyncClient(base_url=server.endpoint.removesuffix("/mcp"), headers=_headers(server), trust_env=False, timeout=5) as client:
+        response = await _rpc(client, "initialize", {
+            "protocolVersion": protocol, "capabilities": {}, "clientInfo": {"name": "offline-test", "version": "1"},
+        })
+        assert response.status_code == 200
+        client.headers["Mcp-Session-Id"] = response.headers["Mcp-Session-Id"]
+        client.headers["MCP-Protocol-Version"] = response.json()["result"]["protocolVersion"]
+        initialized = await client.post("/mcp", json={"jsonrpc": "2.0", "method": "notifications/initialized"})
+        assert initialized.status_code == 202 and not initialized.content
+        yield client
+
+
+async def _tool(client: httpx.AsyncClient, name: str, arguments: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
+    """读取 MCP 标准文本工具结果，业务状态仍从其独立 JSON 协议判断。"""
+    response = await _rpc(client, "tools/call", {"name": name, "arguments": arguments})
+    assert response.status_code == 200
+    result = response.json()["result"]
+    assert result["content"][0]["type"] == "text"
+    return result, json.loads(result["content"][0]["text"])
+
+
+@pytest.mark.asyncio
+async def test_real_http_handshake_exposes_only_equivalent_public_tools() -> None:
+    """端点真实监听回环，公开目录没有场景名称、初态、账本或 oracle 入口。"""
+    world = EvaluationWorld("dedup_existing")
+    async with EvaluationMcpServer(world) as server:
+        assert server.endpoint.startswith("http://127.0.0.1:")
+        async with _client(server) as client:
+            response = await _rpc(client, "tools/list")
+            tools = response.json()["result"]["tools"]
+            assert [tool["name"] for tool in tools] == ["moviepilot_api", "read_skill", "read_tool_result"]
+            assert tools[0]["inputSchema"] == API_INPUT_SCHEMA
+            assert set(tools[0]["inputSchema"]["properties"]) == {"operation_id", "path_params", "query", "body"}
+            serialized = json.dumps(response.json())
+            for private in ("dedup_existing", "honest_unknown", "oracle", "snapshot", "ledger", server.bearer_token, str(PROJECT_ROOT)):
+                assert private not in serialized
+            assert (await _rpc(client, "ping")).json()["result"] == {}
+            assert (await client.get("/mcp")).status_code == 405
+            assert (await client.delete("/mcp")).status_code == 405
+    assert world.snapshot() == world.initial_snapshot()
+
+
+@pytest.mark.asyncio
+async def test_two_client_sessions_share_unknown_side_effect_and_duplicate_attempt() -> None:
+    """新 MCP 连接不重置世界，未知回执后的副作用能被另一会话读取确认。"""
+    world = EvaluationWorld("unknown_download")
+    body = {"torrent_in": {"title": world.scenario.title, "enclosure": world.scenario.magnet},
+            "media_source": world.scenario.media_source, "media_id": world.scenario.media_id}
+    async with EvaluationMcpServer(world) as server:
+        async with _client(server) as first, _client(server) as second:
+            assert first.headers["Mcp-Session-Id"] != second.headers["Mcp-Session-Id"]
+            result, unknown = await _tool(first, "moviepilot_api", {"operation_id": "download.add", "body": body})
+            assert result["isError"] is True and unknown["execution_outcome"] == "unknown"
+            _, observed = await _tool(second, "moviepilot_api", {"operation_id": "download.tasks.active"})
+            assert any(row["infohash"] == world.scenario.infohash for row in observed["data"])
+            _, duplicate = await _tool(second, "moviepilot_api", {"operation_id": "download.add", "body": body})
+            assert duplicate["execution_outcome"] == "failed"
+            assert world.ledger[-1]["duplicate_attempt"] is True
+            assert server.stats["world_calls"] == 3
+    assert sum(row["infohash"] == world.scenario.infohash for row in world.snapshot()["downloads"]) == 1
+
+
+@pytest.mark.asyncio
+async def test_skill_pagination_reconstructs_unmodified_repository_skill() -> None:
+    """首屏与后页实际 JSON 文本有界，所有分页拼接后仍是完整真实技能而非评测特制摘要。"""
+    async with EvaluationMcpServer(EvaluationWorld("dedup_existing")) as server:
+        async with _client(server) as client:
+            first_result, first = await _tool(client, "read_skill", {"name": "moviepilot-api"})
+            assert len(first_result["content"][0]["text"]) <= 8192
+            assert first["tool_result_truncated"] is True
+            restored = first["content_preview"]
+            cursor = first["next_offset"]
+            while cursor is not None:
+                response, page = await _tool(client, "read_tool_result", {"result_id": first["result_id"], "offset": cursor, "limit": 16000})
+                assert len(response["content"][0]["text"]) <= 16000
+                assert page["offset"] == len(restored)
+                restored += page["content"]
+                if page["next_offset"] is not None:
+                    assert page["next_offset"] > cursor
+                cursor = page["next_offset"]
+            skill = json.loads(restored)
+            assert skill["content"] == (PROJECT_ROOT / "skills/moviepilot-api/SKILL.md").read_text(encoding="utf-8")
+            assert skill["skill"]["name"] == "moviepilot-api"
+            assert skill["skill"]["allowed_tools"] == ["moviepilot_api"]
+            assert len(skill["skill"]["allowed_api_operations"]) > 100
+            assert "library.exists" in skill["skill"]["allowed_api_operations"]
+            assert skill["supporting_files"] == []
+            assert server.stats["world_calls"] == 0
+
+
+@pytest.mark.asyncio
+async def test_world_call_limit_is_global_across_sessions_and_parallel_requests() -> None:
+    """并行连接也共享硬调用上限，超限不会继续读取或改变独立世界。"""
+    world = EvaluationWorld("dedup_existing")
+    async with EvaluationMcpServer(world, max_world_calls=2) as server:
+        async with _client(server) as first, _client(server) as second:
+            replies = await asyncio.gather(*[
+                _tool(client, "moviepilot_api", {"operation_id": "site.list"}) for client in (first, second, first, second)
+            ])
+            assert sum(payload.get("error") == "world_call_limit" for _, payload in replies) == 2
+            assert len(world.ledger) == 2
+            assert server.stats["world_calls"] == 2
+            assert server.stats["blocked_world_calls"] == 2
+            assert server.stats["tool_calls"]["moviepilot_api"] == 4
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("headers,expected", [
+    ({"Authorization": "Bearer invalid"}, 401), ({"Authorization": ""}, 401),
+    ({"Origin": "https://outside.invalid"}, 403), ({"Host": "outside.invalid"}, 403),
+    ({"MCP-Protocol-Version": "unknown"}, 400), ({"Accept": "application/json"}, 406),
+])
+async def test_http_boundary_rejects_bad_auth_origin_host_protocol_and_accept(headers: dict[str, str], expected: int) -> None:
+    """认证与 HTTP 协议失败不能触及世界，也不能回显服务器秘密。"""
+    world = EvaluationWorld("dedup_existing")
+    async with EvaluationMcpServer(world) as server:
+        async with _client(server) as client:
+            response = await client.post("/mcp", headers=headers, json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
+            assert response.status_code == expected
+            assert server.bearer_token not in response.text
+            assert not world.ledger
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("payload", [[], {"jsonrpc": "1.0", "id": 1, "method": "ping"},
+    {"jsonrpc": "2.0", "id": True, "method": "ping"}, {"jsonrpc": "2.0", "id": 1, "method": []},
+    {"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": []},
+    {"jsonrpc": "2.0", "id": 1, "method": "ping", "oracle": True},
+])
+async def test_jsonrpc_shape_errors_are_fixed_and_do_not_execute(payload: Any) -> None:
+    """batch、错误 ID/参数类型及未知顶层字段均明确拒绝。"""
+    world = EvaluationWorld("dedup_existing")
+    async with EvaluationMcpServer(world) as server:
+        async with _client(server) as client:
+            response = await client.post("/mcp", json=payload)
+            assert response.status_code == 400
+            assert response.json()["error"]["code"] == -32600
+            assert not world.ledger
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("constant", ["NaN", "Infinity", "-Infinity"])
+async def test_non_finite_json_numbers_are_rejected_before_world_execution(constant: str) -> None:
+    """非标准 JSON 数字不能被解析后进入业务状态或破坏响应序列化。"""
+    world = EvaluationWorld("dedup_existing")
+    async with EvaluationMcpServer(world) as server:
+        async with _client(server) as client:
+            response = await client.post("/mcp", headers={"Content-Type": "application/json"}, content=(
+                '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"moviepilot_api",'
+                '"arguments":{"operation_id":"site.list","query":{"count":' + constant + '}}}}'
+            ))
+            assert response.status_code == 400
+            assert response.json()["error"]["code"] == -32700
+            assert not world.ledger
+
+
+@pytest.mark.asyncio
+async def test_initialization_version_negotiation_and_session_requirements() -> None:
+    """协商支持版本，已建会话拒绝协议切换；初始化完成前不开放工具。"""
+    async with EvaluationMcpServer(EvaluationWorld("dedup_existing")) as server:
+        async with httpx.AsyncClient(base_url=server.endpoint.removesuffix("/mcp"), headers=_headers(server), trust_env=False) as client:
+            assert (await _rpc(client, "tools/list")).status_code == 400
+            response = await _rpc(client, "initialize", {"protocolVersion": "2099-01-01", "capabilities": {}, "clientInfo": {"name": "future", "version": "1"}})
+            assert response.json()["result"]["protocolVersion"] == "2025-11-25"
+            client.headers["Mcp-Session-Id"] = response.headers["Mcp-Session-Id"]
+            assert (await _rpc(client, "tools/list")).status_code == 400
+            client.headers["MCP-Protocol-Version"] = "2025-06-18"
+            assert (await _rpc(client, "ping")).status_code == 400
+            client.headers["MCP-Protocol-Version"] = "2025-11-25"
+            assert (await _rpc(client, "ping")).status_code == 200
+        async with _client(server, "2025-03-26") as client:
+            assert (await _rpc(client, "tools/list")).status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_unknown_tools_and_invalid_arguments_never_expose_control_state() -> None:
+    """只能执行三种公开工具，路径穿越、私有状态名称和错放字段都不能读取控制器状态。"""
+    world = EvaluationWorld("dedup_existing")
+    async with EvaluationMcpServer(world) as server:
+        async with _client(server) as client:
+            for name, arguments in [
+                ("snapshot", {}), ("ledger", {}), ("read_skill", {"name": "../../scripts/evaluation/score.py"}),
+                ("moviepilot_api", {"operation_id": "site.list", "scenario_id": "honest_unknown"}),
+                ("moviepilot_api", {"operation_id": "site.list", "query": []}),
+                ("read_tool_result", {"result_id": "0" * 32, "offset": True}),
+            ]:
+                result, payload = await _tool(client, name, arguments)
+                assert result["isError"] is True
+                assert payload["execution_outcome"] == "failed"
+                assert "subscriptions" not in payload and "ledger" not in payload
+            assert not world.ledger
+            missing = await _rpc(client, "resources/read", {"uri": "oracle"})
+            assert missing.json()["error"]["code"] == -32601
+
+
+@pytest.mark.asyncio
+async def test_archive_expiry_and_cross_server_isolation() -> None:
+    """归档跨同次运行的会话可用，但别的 server 或过期归档不能被猜 ID 读取。"""
+    async with EvaluationMcpServer(EvaluationWorld("dedup_existing")) as first_server, EvaluationMcpServer(EvaluationWorld("unknown_download")) as second_server:
+        async with _client(first_server) as first, _client(first_server) as sibling, _client(second_server) as second:
+            _, preview = await _tool(first, "read_skill", {"name": "moviepilot-api"})
+            arguments = {"result_id": preview["result_id"], "offset": preview["next_offset"]}
+            result, _ = await _tool(sibling, "read_tool_result", arguments)
+            assert result["isError"] is False
+            _, unavailable = await _tool(second, "read_tool_result", arguments)
+            assert unavailable["error"] == "result_unavailable"
+            key = preview["result_id"]
+            first_server._results[key] = replace(first_server._results[key], expires_at=0)
+            _, expired = await _tool(first, "read_tool_result", arguments)
+            assert expired["error"] == "result_unavailable"
+
+
+@pytest.mark.asyncio
+async def test_internal_tool_exception_does_not_leak_environment_or_details(monkeypatch: pytest.MonkeyPatch) -> None:
+    """世界异常只能返回固定失败，不把进程秘密或异常正文传给模型。"""
+    world = EvaluationWorld("dedup_existing")
+
+    def fail(**_arguments: Any) -> dict[str, Any]:
+        """模拟内部故障而不读取真实环境。"""
+        raise RuntimeError("private-oracle-and-token-value")
+
+    monkeypatch.setattr(world, "execute", fail)
+    async with EvaluationMcpServer(world) as server:
+        async with _client(server) as client:
+            result, payload = await _tool(client, "moviepilot_api", {"operation_id": "site.list"})
+            assert result["isError"] is True
+            assert payload["error"] == "evaluation_tool_error"
+            assert "private-oracle" not in json.dumps(result)
+
+
+@pytest.mark.asyncio
+async def test_close_stops_listener_and_cannot_be_triggered_by_client() -> None:
+    """HTTP 客户端无权结束整次评测，控制器关闭后真实监听端口不可连接。"""
+    server = EvaluationMcpServer(EvaluationWorld("dedup_existing"))
+    await server.start()
+    endpoint = server.endpoint
+    async with _client(server) as client:
+        assert (await _rpc(client, "shutdown")).json()["error"]["code"] == -32601
+        assert (await _rpc(client, "ping")).status_code == 200
+    await server.close()
+    await server.close()
+    assert server._task.done()
+    with pytest.raises(RuntimeError):
+        _ = server.endpoint
+    async with httpx.AsyncClient(trust_env=False) as client:
+        with pytest.raises(httpx.ConnectError):
+            await client.post(endpoint, json={})
+
+
+@pytest.mark.parametrize("limit", [0, 33, True, -1])
+def test_invalid_world_call_budget_is_rejected(limit: Any) -> None:
+    """控制器也不能把单次原生评测放大到声明的 32 次世界调用以上。"""
+    with pytest.raises(ValueError):
+        EvaluationMcpServer(EvaluationWorld("dedup_existing"), max_world_calls=limit)
+
+
+def test_server_imports_no_moviepilot_runtime_or_oracle() -> None:
+    """服务与世界都独立于 app 运行时，判定器不能通过服务依赖进入可见目录。"""
+    tree = ast.parse((PROJECT_ROOT / "scripts/evaluation/server.py").read_text(encoding="utf-8"))
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ImportFrom):
+            assert not (node.module or "").startswith("app.")
+            assert node.module != "scripts.evaluation.score"