diff --git a/docs/agent-evaluation.md b/docs/agent-evaluation.md
index cc3d61d80..a1df2b087 100644
--- a/docs/agent-evaluation.md
+++ b/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。自建模型循环只能叫受控运行时对照。
diff --git a/scripts/evaluation/__main__.py b/scripts/evaluation/__main__.py
index a17343e3e..32a8862a4 100644
--- a/scripts/evaluation/__main__.py
+++ b/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__":
diff --git a/scripts/evaluation/codex.py b/scripts/evaluation/codex.py
new file mode 100644
index 000000000..43fe99fc0
--- /dev/null
+++ b/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))
diff --git a/scripts/evaluation/proxy.py b/scripts/evaluation/proxy.py
new file mode 100644
index 000000000..fc933c73d
--- /dev/null
+++ b/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\n[\s\S]*\n\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, "", "", "", "",
+ )):
+ 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)}
diff --git a/scripts/evaluation/server.py b/scripts/evaluation/server.py
new file mode 100644
index 000000000..f5e29e257
--- /dev/null
+++ b/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))
diff --git a/tests/test_agent_evaluation_codex.py b/tests/test_agent_evaluation_codex.py
new file mode 100644
index 000000000..72bd6d9e0
--- /dev/null
+++ b/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
diff --git a/tests/test_agent_evaluation_proxy.py b/tests/test_agent_evaluation_proxy.py
new file mode 100644
index 000000000..2c9a387ce
--- /dev/null
+++ b/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\nsynthetic host preference\n"
+ 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\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\nbody\n"},
+ {"type": "text", "kind": "foreign.kind", "text": "# AGENTS.md instructions\n\n\nbody\n"},
+ {"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\nhost\n"
+ 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", ["host skills",
+ "ordinary unrecognized host"])
+@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
diff --git a/tests/test_agent_evaluation_server.py b/tests/test_agent_evaluation_server.py
new file mode 100644
index 000000000..762ada6c6
--- /dev/null
+++ b/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"