test(agent): run bounded live models through isolated production harness

docs/agent-evaluation.md

@@ -1,10 +1,10 @@
 # Agent 任务评测
 
-单元测试中的固定模型响应可以验证工具与中间件合同，但不能证明真实模型能自主完成任务。本目录的评测基础把任务说明、假业务世界、真实终态与最终报告分开，后续真实模型运行必须使用同一套独立验收器。
+单元测试中的固定模型响应可以验证工具与中间件合同，但不能证明真实模型能自主完成任务。评测把任务说明、假业务世界、真实终态与最终报告分开；离线回放和真实模型共用独立验收器。
 
 ## 当前范围
 
-`scripts/evaluation/` 当前只提供离线轨迹回放，不调用真实模型、MoviePilot 服务、下载器或外部网络。报告固定标记 `evidence_kind=scripted_replay`、`intelligence_evaluated=false`，不能把这里的通过率称为模型或 Codex 对比结果。
+`scripts/evaluation/` 提供两种显式模式。`--replay` 只回放离线轨迹，不联网，报告标记 `evidence_kind=scripted_replay`、`intelligence_evaluated=false`。`--live` 在独立进程内用真实模型驱动完整生产 Agent，业务接口只进入内存假世界。两者都没有实现 Codex 原生对照，不能把通过率称为 Codex 对比结果。
 
 首批场景：
 
@@ -26,11 +26,32 @@ uv run --locked --no-sync python -m scripts.evaluation \
 
 轨迹是一个只包含 `calls` 和 `final` 的 JSON 对象。`calls` 是固定 API 调用数组，每项只接受 `operation_id`、`path_params`、`query`、`body`。参数使用 MoviePilot 的真实字段位置，例如 `download.add` 使用 `body.torrent_in` 及同级媒体身份。回放最多 32 次调用，输入文件最多 256 KiB。
 
-`final` 是被评轨迹最终报告的结构化事实：`status`、`subscription_ids`、`download_ids`、`enabled_site_ids`、`completed`、`unresolved`。没有读取证据时，即使 ID 恰好正确也不能通过。后端拒绝重复资源，也不能抵消违反用户“不重复提交”要求的重复尝试。
+`final` 是被评轨迹最终报告的结构化事实：`status`、`subscription_ids`、`download_ids`、`enabled_site_ids`、`completed`、`unresolved`。公开任务要求只报告本次明确请求的子目标；未请求类别的 ID 数组为空，辅助读取不增加目标。没有读取证据时，即使 ID 恰好正确也不能通过。后端拒绝重复资源，也不能抵消违反用户“不重复提交”要求的重复尝试。
 
 退出码 0 表示该轨迹通过，1 表示验收失败，2 表示输入无效。报告包含所有失败原因、调用/失败/重复尝试/副作用数量、场景及判定代码的 SHA256。回放没有模型，故 `model=null`、`model_calls=0`、`tokens=null`；不得将这些值用于真实模型成本比较。
 
-## 真实模型与 Codex 比较的后续接入要求
+## 真实模型运行
+
+```bash
+uv run --locked --no-sync python -m scripts.evaluation --live \
+  --scenario dedup_existing --reasoning-effort high \
+  --max-model-calls 12 --max-output-tokens 8192 --timeout-seconds 180 \
+  --output report.json
+```
+
+只有显式 `--live` 才会产生模型调用费用。默认读取 `~/.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/文件/浏览器工具不开放。此受控目录是当前评测边界，不代表默认部署工具全集。
+
+共享回调在请求前执行硬调用上限，覆盖主模型、选择、摘要和子代理；SDK 自动重试关闭。单请求超时最多 120 秒，全轮和独立进程另有期限。每次请求的输出 token 及运行器上下文上限被记录；上下文上限是测试参数，不表示模型真实最大窗口。当前固定为 128000 tokens。
+
+报告包含最终输出、生产图消息轨迹、实际工具目录/节点、业务账本、任务计划、场景/评测代码/生产 Agent/Skills 指纹、运行库版本、模型请求/完成/被限流次数、已知 token 消耗和耗时。失败请求的用量未知时，`usage_complete=false`，token 仅为已知下界，不能据此声称零消耗。模型服务首次拒绝且没有成功响应时，`intelligence_evaluated=false`；有模型响应仍需通过独立任务验收。`agent_execution_success` 仅表示生产图技术执行结果，不等于任务完成。
+
+`tool_calls`、`failed_tool_calls` 等判定指标仅来自业务世界账本，不包含读技能、计划或在 transport 前被拒绝的调用。`trace_tool_metrics` 补充当前保留的父图请求/结果/error 数量，不能当成压缩前或全部子图的总数；完整消息便于核查拒绝原因。生产提示可能建议查询媒体库等当前假世界未开放的能力，此类拒绝属于受控环境限制，不能据此认定模型调用了无效的生产 API。
+
+单测不调用真实模型或外部网络；测试只验证运行器、隔离、预算与判定合同。真实报告含合成任务轨迹，也可能很长，应保存于评测工作目录；不要把包含私有配置的临时诊断日志提交到仓库。
+
+## Codex 比较的后续接入要求
 
 - MoviePilot 侧应保留完整生产 Agent 装配，只替换业务 transport 与隔离配置；只组装部分中间件的循环不能冒充产品实测。
 - Codex 侧必须调用原生 CLI 或 App Server。自建模型循环只能叫受控运行时对照。

scripts/evaluation/__main__.py

@@ -1,4 +1,4 @@
-"""离线回放评测轨迹；此入口不会调用模型、真实 MoviePilot 或外部服务。"""
+"""分别提供离线轨迹回放与显式真实模型评测；业务操作只进入假世界。"""
 
 import argparse
 import hashlib
@@ -16,6 +16,15 @@ MAX_REPLAY_CALLS = 32
 _CALL_FIELDS = frozenset({"operation_id", "path_params", "query", "body"})
 
 
+def _source_digest(root: Path) -> str:
+    """按路径和内容记录生产运行时或内置技能，不把源码正文送进模型或报告。"""
+    digest = hashlib.sha256()
+    for path in sorted(root.rglob("*")):
+        if path.is_file() and "__pycache__" not in path.parts and path.suffix != ".pyc":
+            digest.update(path.relative_to(root).as_posix().encode("utf-8") + b"\0" + path.read_bytes() + b"\0")
+    return digest.hexdigest()
+
+
 def _provenance(world: EvaluationWorld) -> dict[str, Any]:
     """记录实际场景与判定代码指纹，避免未提交或修改后的轨迹被混为同一基线。"""
     fixture = json.dumps({"scenario": asdict(world.scenario), "initial": world.initial_snapshot()},
@@ -23,7 +32,10 @@ def _provenance(world: EvaluationWorld) -> dict[str, Any]:
     digest = hashlib.sha256()
     for path in sorted(Path(__file__).parent.glob("*.py")):
         digest.update(path.name.encode("utf-8") + b"\0" + path.read_bytes() + b"\0")
+    repository = Path(__file__).resolve().parents[2]
     return {"schema_version": 1, "scenario_sha256": hashlib.sha256(fixture.encode("utf-8")).hexdigest(),
+            "production_agent_sha256": _source_digest(repository / "app" / "agent"),
+            "skills_sha256": _source_digest(repository / "skills"),
             "harness_sha256": digest.hexdigest(), "model": None, "model_calls": 0, "tokens": None}
 
 
@@ -44,24 +56,43 @@ def replay(scenario_id: str, payload: Any) -> dict[str, Any]:
 
 
 def main(argv: list[str] | None = None) -> int:
-    """打印公开场景或回放文件，并以非零退出码表达验收未通过。"""
-    parser = argparse.ArgumentParser(description="MoviePilot Agent 离线轨迹验收；不代表真实模型评分")
-    parser.add_argument("--list", action="store_true", help="列出公开场景，不输出初态或答案")
+    """显式区分回放与真实调用，并以非零退出码表达验收未通过。"""
+    parser = argparse.ArgumentParser(description="MoviePilot Agent 隔离任务评测；--live 才会调用真实模型")
+    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="显式调用真实模型，业务只进入隔离假世界")
     parser.add_argument("--scenario", choices=[item.scenario_id for item in list_scenarios()])
-    parser.add_argument("--replay", type=Path, help="包含 calls 与 final 的 JSON 轨迹")
     parser.add_argument("--output", type=Path, help="可选 JSON 报告路径")
+    parser.add_argument("--codex-config", type=Path, default=Path.home() / ".codex" / "config.toml")
+    parser.add_argument("--model", help="真实评测的模型名称，默认沿用显式 Codex provider 配置")
+    parser.add_argument("--reasoning-effort", help="真实评测推理预算，默认沿用配置")
+    parser.add_argument("--max-model-calls", type=int, default=12)
+    parser.add_argument("--max-output-tokens", type=int, default=8192)
+    parser.add_argument("--timeout-seconds", type=int, default=180)
     args = parser.parse_args(argv)
     if args.list:
         print(json.dumps([{"id": item.scenario_id, "task": item.model_input()} for item in list_scenarios()], ensure_ascii=False, indent=2))
         return 0
-    if args.scenario is None or args.replay is None:
-        parser.error("回放需要 --scenario 和 --replay")
+    if args.scenario is None:
+        parser.error("评测需要 --scenario")
     try:
-        if args.replay.stat().st_size > MAX_REPLAY_BYTES:
-            raise ValueError("回放文件超过 256 KiB")
-        payload = json.loads(args.replay.read_text(encoding="utf-8"))
-        result = replay(args.scenario, payload)
-    except (OSError, UnicodeError, ValueError, TypeError) as error:
+        if args.live:
+            from scripts.evaluation.live import run_live
+            from scripts.evaluation.models import load_codex_model_settings
+
+            settings = load_codex_model_settings(
+                args.codex_config, model=args.model, reasoning_effort=args.reasoning_effort,
+                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)
+        else:
+            if args.replay.stat().st_size > MAX_REPLAY_BYTES:
+                raise ValueError("回放文件超过 256 KiB")
+            payload = json.loads(args.replay.read_text(encoding="utf-8"))
+            result = replay(args.scenario, payload)
+    except (OSError, UnicodeError, ValueError, TypeError, RuntimeError) as error:
         parser.error(str(error))
     rendered = json.dumps(result, ensure_ascii=False, indent=2)
     if args.output is not None:

scripts/evaluation/live.py

@@ -0,0 +1,165 @@
+"""真实模型评测的独立进程边界，业务调用仍只进入内存假世界。"""
+
+import asyncio
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import time
+from contextlib import redirect_stdout
+from dataclasses import asdict
+from importlib.metadata import version
+from typing import Any
+
+from scripts.evaluation.models import ModelSettings, ModelUsageTracker
+
+
+def _worker_environment() -> dict[str, str]:
+    """只传递解释器所需平台环境；业务设置、代理和其他服务凭据不进入评测。"""
+    allowed = {"PATH", "SYSTEMROOT", "WINDIR", "LANG", "TMPDIR", "TMP", "TEMP"}
+    environment = {name: value for name, value in os.environ.items() if name in allowed or name.startswith("LC_")}
+    environment.update(PYTHONNOUSERSITE="1", PYTHONUTF8="1")
+    return environment
+
+
+def _validate_report(report: Any, credential: str) -> dict[str, Any]:
+    """对解码后的键和值检查凭据，JSON 转义不能绕过输出边界。"""
+    if not isinstance(report, dict):
+        raise RuntimeError("隔离模型评测报告未通过输出校验")
+    pending = [report]
+    while pending:
+        value = pending.pop()
+        if isinstance(value, str) and credential in value:
+            raise RuntimeError("评测报告意外包含连接凭据，已拒绝输出")
+        if isinstance(value, dict):
+            pending.extend(value.keys())
+            pending.extend(value.values())
+        elif isinstance(value, (list, tuple)):
+            pending.extend(value)
+    return report
+
+
+def run_live(scenario_id: str, settings: ModelSettings) -> dict[str, Any]:
+    """经标准输入传递连接凭据，独立进程避免复用调用方已经加载的真实配置。"""
+    payload = json.dumps({"scenario_id": scenario_id, "settings": asdict(settings)})
+    try:
+        completed = subprocess.run(
+            [sys.executable, "-m", "scripts.evaluation.live", "--worker"],
+            input=payload, text=True, capture_output=True, check=False, env=_worker_environment(),
+            timeout=settings.timeout_seconds + 30,
+        )
+    except subprocess.TimeoutExpired:
+        raise RuntimeError("隔离模型评测进程超时，已终止") from None
+    if completed.returncode != 0:
+        # 不回显 worker stderr，SDK 或宿主初始化错误可能含用户连接信息。
+        raise RuntimeError(f"隔离模型评测进程退出，状态码 {completed.returncode}")
+    try:
+        report = json.loads(completed.stdout)
+    except (ValueError, TypeError):
+        raise RuntimeError("隔离模型评测未返回有效报告") from None
+    return _validate_report(report, settings.api_key)
+
+
+def _parse_final(text: str) -> Any:
+    """接受普通 JSON 或单个 Markdown JSON 围栏，其他话术不作为结构化完成证据。"""
+    value = text.strip()
+    if value.startswith(("```\n", "```json\n")) and value.endswith("```") and value.count("```") == 2:
+        value = value.split("\n", 1)[-1].rsplit("```", 1)[0].strip()
+    try:
+        return json.loads(value)
+    except (ValueError, TypeError):
+        return None
+
+
+def _trace_metrics(messages: list[dict[str, Any]]) -> dict[str, Any]:
+    """补充业务账本之外的工具调用；压缩后仅统计保留的父图消息，不冒充全部子图历史。"""
+    requests = sum(len(message["data"].get("tool_calls", [])) for message in messages if message["type"] == "ai")
+    results = [message["data"] for message in messages if message["type"] == "tool"]
+    return {"scope": "retained_parent_graph", "requested": requests, "results": len(results),
+            "errors": sum(result.get("status") == "error" for result in results)}
+
+
+async def _run_worker(scenario_id: str, settings: ModelSettings) -> dict[str, Any]:
+    """创建临时宿主环境，直到此处才导入业务运行时和建立真实模型连接。"""
+    if "app.runtime.config" in sys.modules:
+        raise RuntimeError("真实评测 worker 必须在业务导入前隔离配置")
+    with tempfile.TemporaryDirectory(prefix="moviepilot-model-evaluation-") as config_dir:
+        os.environ["CONFIG_DIR"] = config_dir
+        from app.testing.bootstrap import prepare_backend
+
+        prepare_backend()
+        from langchain_openai import ChatOpenAI
+
+        from scripts.evaluation.__main__ import _provenance
+        from scripts.evaluation.runtime import run_moviepilot
+        from scripts.evaluation.score import evaluate
+        from scripts.evaluation.world import EvaluationWorld
+
+        tracker = ModelUsageTracker(settings.max_model_calls)
+        model = ChatOpenAI(
+            model=settings.model, api_key=settings.api_key, base_url=settings.base_url,
+            use_responses_api=True, reasoning={"effort": settings.reasoning_effort},
+            max_tokens=settings.max_output_tokens, max_retries=0, timeout=min(120, settings.timeout_seconds),
+            profile={"max_input_tokens": settings.context_window}, callbacks=[tracker],
+        )
+        world = EvaluationWorld(scenario_id)
+        started = time.monotonic()
+        capture: dict[str, Any] = {}
+        failure = None
+        cleanup_errors = []
+        try:
+            capture = await asyncio.wait_for(
+                run_moviepilot(world, model, model_name=settings.model, context_window=settings.context_window,
+                               max_iterations=64 + settings.max_model_calls * 16),
+                timeout=settings.timeout_seconds,
+            )
+        except Exception as error:
+            failure = type(error).__name__
+        finally:
+            try:
+                await model.root_async_client.close()
+            except Exception as error:
+                cleanup_errors.append(type(error).__name__)
+            try:
+                model.root_client.close()
+            except Exception as error:
+                cleanup_errors.append(type(error).__name__)
+        final_text = str(capture.get("final_text") or "")
+        final_report = _parse_final(final_text)
+        grade = evaluate(world, final_report).to_dict()
+        usage = tracker.snapshot()
+        # 首次 HTTP 被拒绝不属于模型能力证据；有响应也仍不构成 Codex 配对比较。
+        return {
+            **grade, **_provenance(world), **usage, "evidence_kind": "moviepilot_live_model",
+            "intelligence_evaluated": usage["completed_model_calls"] > 0, "codex_comparison": False,
+            "model": settings.public_metadata(), "elapsed_seconds": round(time.monotonic() - started, 3),
+            "runtime_versions": {"python": sys.version.split()[0], **{
+                name: version(name) for name in ("langchain", "langgraph", "langchain-openai", "openai")
+            }},
+            "runner_error_type": failure, "cleanup_error_types": cleanup_errors,
+            "final_text": final_text, "final_report": final_report,
+            "tool_catalog_scope": capture.get("tool_catalog_scope"),
+            "tool_names": capture.get("tool_names"), "graph_nodes": capture.get("graph_nodes"),
+            "agent_execution_success": capture.get("execution_success"),
+            "agent_trace": capture.get("raw_messages", []), "task_plan": capture.get("task_plan"),
+            "trace_tool_metrics": _trace_metrics(capture.get("raw_messages", [])),
+            "ledger": world.ledger,
+        }
+
+
+def main() -> int:
+    """worker 只从私有 stdin 接收配置，stdout 只输出无凭据评测报告。"""
+    if sys.argv[1:] != ["--worker"]:
+        raise SystemExit("请通过 python -m scripts.evaluation --live 启动")
+    payload = json.loads(sys.stdin.read(256 * 1024))
+    settings = ModelSettings(**payload["settings"])
+    with redirect_stdout(sys.stderr):
+        result = asyncio.run(_run_worker(payload["scenario_id"], settings))
+    serialized = json.dumps(_validate_report(result, settings.api_key), ensure_ascii=False)
+    print(serialized)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/evaluation/models.py

@@ -0,0 +1,148 @@
+"""读取明确配置的模型连接，并为真实评测统计所有模型调用。"""
+
+import os
+import threading
+import time
+import tomllib
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlsplit
+
+from langchain_core.callbacks.base import BaseCallbackHandler
+
+
+@dataclass(frozen=True)
+class ModelSettings:
+    """评测连接只在控制进程使用，凭据不进入 repr 或公开报告。"""
+
+    model: str
+    base_url: str = field(repr=False)
+    api_key: str = field(repr=False)
+    reasoning_effort: str = "high"
+    max_model_calls: int = 12
+    max_output_tokens: int = 8192
+    timeout_seconds: int = 180
+    context_window: int = 128000
+
+    def __post_init__(self) -> None:
+        """配置文件和 worker 输入共享校验，不能绕过限额或夹带 URL 凭据。"""
+        if not isinstance(self.base_url, str):
+            raise ValueError("模型 endpoint 无效")
+        endpoint = urlsplit(self.base_url)
+        if (endpoint.scheme not in {"http", "https"} or not endpoint.hostname or endpoint.username is not None
+                or endpoint.password is not None or endpoint.query or endpoint.fragment):
+            raise ValueError("模型 endpoint 无效")
+        if not isinstance(self.model, str) or not self.model.strip():
+            raise ValueError("需要明确模型名称")
+        if not isinstance(self.api_key, str) or not self.api_key.strip() or any(char in self.api_key for char in "\r\n"):
+            raise ValueError("所选 provider 没有可用的显式评测凭据")
+        if self.reasoning_effort not in {"none", "minimal", "low", "medium", "high", "xhigh", "max", "ultra"}:
+            raise ValueError("推理预算名称无效")
+        for value, minimum, maximum in (
+            (self.max_model_calls, 1, 64), (self.max_output_tokens, 256, 32768),
+            (self.timeout_seconds, 30, 900), (self.context_window, 4096, 2000000),
+        ):
+            if type(value) is not int or not minimum <= value <= maximum:
+                raise ValueError("评测预算超出允许范围")
+
+    def public_metadata(self) -> dict[str, Any]:
+        """公开模型与评测预算，省略凭据、完整连接路径和用户配置文件。"""
+        return {
+            "requested_model": self.model, "provider_host": urlsplit(self.base_url).hostname,
+            "reasoning_effort": self.reasoning_effort, "max_model_calls": self.max_model_calls,
+            "max_output_tokens": self.max_output_tokens, "timeout_seconds": self.timeout_seconds,
+            "harness_context_window": self.context_window,
+        }
+
+
+def load_codex_model_settings(
+    path: Path, *, model: str | None = None, reasoning_effort: str | None = None,
+    max_model_calls: int = 12, max_output_tokens: int = 8192, timeout_seconds: int = 180,
+) -> ModelSettings:
+    """只使用选中 provider 显式配置的 bearer/env 凭据，不挪用其他服务的登录令牌。"""
+    config = tomllib.loads(path.read_text(encoding="utf-8"))
+    provider = config.get("model_providers", {}).get(config.get("model_provider"), {})
+    if provider.get("wire_api") != "responses":
+        raise ValueError("当前真实评测入口要求明确配置的 Responses provider")
+    endpoint = provider.get("base_url")
+    selected_model = model or config.get("model")
+    if not isinstance(selected_model, str):
+        raise ValueError("需要明确模型名称")
+    key = provider.get("experimental_bearer_token")
+    if not key and isinstance(provider.get("env_key"), str):
+        key = os.environ.get(provider["env_key"])
+    effort = reasoning_effort or config.get("model_reasoning_effort", "high")
+    return ModelSettings(selected_model, endpoint, key, effort, max_model_calls, max_output_tokens, timeout_seconds)
+
+
+class ModelCallLimitError(RuntimeError):
+    """在下一次真实请求之前拒绝超出评测预算的模型调用。"""
+
+
+# strict mypy 的 follow_imports=skip 不展开第三方回调基类。
+class ModelUsageTracker(BaseCallbackHandler):  # type: ignore[misc]
+    """合计主模型、筛选、摘要与子代理的请求，不把缺失 usage 当成零消耗。"""
+
+    raise_error = True
+    run_inline = True
+
+    def __init__(self, limit: int) -> None:
+        """建立单次评测共享的并发安全计数器。"""
+        self.limit = limit
+        self._lock = threading.Lock()
+        self._runs: dict[str, dict[str, Any]] = {}
+        self.blocked_calls = 0
+
+    def on_chat_model_start(self, serialized: dict[str, Any], messages: list[Any], **kwargs: Any) -> None:
+        """原子执行预算判断，既不记录提示词，也不允许并行调用越过上限。"""
+        run_id = str(kwargs["run_id"])
+        with self._lock:
+            if run_id in self._runs:
+                return
+            if len(self._runs) >= self.limit:
+                self.blocked_calls += 1
+                raise ModelCallLimitError("评测已达到模型调用上限")
+            self._runs[run_id] = {"started": time.monotonic(), "completed": False, "usage": None, "model": None}
+
+    def on_llm_end(self, response: Any, **kwargs: Any) -> None:
+        """只提取用量与模型标识，不把原始响应或凭据带入统计。"""
+        generation = next((item for group in getattr(response, "generations", []) for item in group), None)
+        message = getattr(generation, "message", None)
+        usage = getattr(message, "usage_metadata", None)
+        if not isinstance(usage, dict) or not all(
+            type(usage.get(key)) is int and usage[key] >= 0
+            for key in ("input_tokens", "output_tokens", "total_tokens")
+        ):
+            usage = None
+        else:
+            usage = {key: usage[key] for key in ("input_tokens", "output_tokens", "total_tokens")}
+        metadata = getattr(message, "response_metadata", None) or {}
+        with self._lock:
+            record = self._runs.get(str(kwargs["run_id"]))
+            if record is not None and not record["completed"]:
+                record.update(completed=True, usage=usage, model=metadata.get("model_name") or metadata.get("model"))
+
+    def on_llm_error(self, error: BaseException, **kwargs: Any) -> None:
+        """错误请求保留为用量未知，不保存可能包含私有响应的异常文本。"""
+        with self._lock:
+            record = self._runs.get(str(kwargs["run_id"]))
+            if record is not None:
+                record["error_type"] = type(error).__name__
+                status = getattr(error, "status_code", None)
+                record["http_status"] = status if isinstance(status, int) else None
+
+    def snapshot(self) -> dict[str, Any]:
+        """返回可审计的已知消耗及覆盖度；部分失败时数字仅为已知下界。"""
+        with self._lock:
+            records = list(self._runs.values())
+            known = [record["usage"] for record in records if isinstance(record["usage"], dict)]
+            return {
+                "model_calls": len(records), "completed_model_calls": sum(record["completed"] for record in records),
+                "blocked_model_calls": self.blocked_calls, "usage_complete": len(known) == len(records),
+                "reported_models": sorted({record["model"] for record in records if isinstance(record["model"], str)}),
+                "errors": [{key: record.get(key) for key in ("error_type", "http_status")}
+                           for record in records if record.get("error_type")],
+                "tokens": ({key: sum(int(usage.get(key, 0)) for usage in known)
+                            for key in ("input_tokens", "output_tokens", "total_tokens")} if known else None),
+            }

scripts/evaluation/runtime.py

@@ -0,0 +1,313 @@
+"""用完整生产 Agent 驱动隔离业务世界；工具目录受控，不代表真实部署配置。"""
+
+import os
+import re
+import sys
+from contextlib import ExitStack, contextmanager
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from threading import Lock
+from types import SimpleNamespace
+from typing import TYPE_CHECKING, Any, Iterator, Optional
+from unittest.mock import patch
+from urllib.parse import unquote, urlsplit
+from uuid import uuid4
+
+from scripts.evaluation.world import EvaluationWorld
+
+if TYPE_CHECKING:
+    from langchain_core.language_models.chat_models import BaseChatModel
+
+_RUN_LOCK = Lock()
+_OPERATIONS = (
+    "subscription.list", "subscription.find", "subscription.get", "subscription.add", "subscription.delete",
+    "download.tasks.active", "download.history.list", "download.clients", "download.paths", "download.add", "site.list",
+)
+
+
+def _config_path() -> Path:
+    """拒绝未隔离或导入后才换目录的调用，不主动导入可能读取真实配置的模块。"""
+    configured = os.environ.get("CONFIG_DIR")
+    bootstrap = sys.modules.get("app.testing.bootstrap")
+    config = sys.modules.get("app.runtime.config")
+    isolated = getattr(bootstrap, "_isolated_config_dir", None)
+    if not configured or not isolated or config is None:
+        raise RuntimeError("评测必须先设置独立 CONFIG_DIR 并调用 prepare_backend")
+    expected = Path(configured).resolve()
+    actual = Path(config.settings.CONFIG_PATH).resolve()
+    if expected != actual or Path(isolated).resolve() != expected:
+        raise RuntimeError("评测 CONFIG_DIR 与已初始化后端目录不一致")
+    return expected
+
+
+class _Response:
+    """内存响应保留执行器真实的读取和关闭协议，不产生网络请求。"""
+
+    status_code = 200
+    headers: dict[str, str] = {}
+
+    def __init__(self, payload: dict[str, Any]) -> None:
+        """保存世界返回的成功、失败或未知结果，不提升为成功。"""
+        self._payload = payload
+
+    def json(self) -> dict[str, Any]:
+        """仅返回此次 API 执行得到的可见响应。"""
+        return self._payload
+
+    async def aclose(self) -> None:
+        """满足生产响应的显式关闭协议。"""
+
+
+class _Transport:
+    """按生产固定路由反向匹配 operation，无法向任意主机发送请求。"""
+
+    def __init__(self, world: EvaluationWorld, **_kwargs: Any) -> None:
+        """绑定当前世界；执行器传来的认证头不记录、不使用。"""
+        from app.agent.policy.api import resolve_api_route
+
+        self._world = world
+        self._routes = []
+        for operation in _OPERATIONS:
+            route = resolve_api_route(operation)
+            if route is None:
+                raise RuntimeError(f"评测 operation 已不在生产路由注册表：{operation}")
+            pattern = re.escape(route.path)
+            pattern = re.sub(r"\\\{(\w+)\\\}", r"(?P<\1>[^/]+)", pattern)
+            self._routes.append((operation, route.method, re.compile(f"^{pattern}$")))
+
+    async def request(
+        self, *, method: str, url: str, params: Optional[dict[str, Any]], json: Any, raise_exception: bool,
+    ) -> _Response:
+        """解释执行器生成的 HTTP 请求，未纳入场景的真实 API 同样拒绝。"""
+        del raise_exception
+        parsed = urlsplit(url)
+        if parsed.netloc != "evaluation.invalid" or parsed.scheme != "http" or parsed.query or parsed.fragment:
+            raise ValueError("评测传输只接受内存 API 地址")
+        for operation, route_method, pattern in self._routes:
+            match = pattern.fullmatch(parsed.path)
+            if route_method != method or match is None:
+                continue
+            path_params: dict[str, Any] = {name: unquote(value) for name, value in match.groupdict().items()}
+            if "subscribe_id" in path_params:
+                try:
+                    path_params["subscribe_id"] = int(path_params["subscribe_id"])
+                except ValueError:
+                    pass
+            return _Response(self._world.execute(operation, path_params=path_params, query=params, body=json))
+        return _Response({"success": False, "execution_outcome": "failed", "message": "该 API 不属于受控评测目录"})
+
+
+class _MemoryPort:
+    """为真实 MemoryManager 提供一次运行独占的会话读取与持久化边界。"""
+
+    def __init__(self) -> None:
+        """存储仅属于当前运行的消息快照。"""
+        self.messages: list[dict[str, Any]] = []
+
+    async def get(self, **_kwargs: Any) -> None:
+        """新评测会话没有历史消息，不读取宿主会话表。"""
+
+    async def async_save_agent_messages(self, *, messages: list[dict[str, Any]], **_kwargs: Any) -> None:
+        """接收 MemoryManager 的序列化结果，不访问宿主全局持久化服务。"""
+        self.messages = messages
+
+
+class _McpDirectory:
+    """受控目录不配置任何外部 MCP，禁止发现阶段启动服务或联网。"""
+
+    @staticmethod
+    def config_signature() -> str:
+        """提供明确区别于真实部署的空目录签名。"""
+        return "evaluation-empty-mcp"
+
+    @staticmethod
+    async def list_enabled_tool_specs() -> list[Any]:
+        """无须读取 SystemConfig 即可返回空外部工具目录。"""
+        return []
+
+
+def _agent_type() -> type:
+    """延迟加载宿主；只替换模型、目录和展示边界，继承完整生产执行图。"""
+    from app.agent.middleware.selection import ToolSelectorMiddleware
+    from app.agent.orchestrator import MoviePilotAgent
+    from app.agent.tools.catalog import ToolCatalogSnapshot
+
+    # 脚本单独严格检查时 follow_imports=skip 不展开宿主类型；运行合同由生产图测试验证。
+    class EvaluationAgent(MoviePilotAgent):  # type: ignore[misc]
+        """在生产策略、中间件和生命周期内运行受控工具及外部注入模型。"""
+
+        def __init__(self, *, model: Any, model_name: str, context_window: int, max_iterations: int, **kwargs: Any) -> None:
+            """隔离模型身份和预算，不解析宿主 LLM 设置或供应商选择事件。"""
+            super().__init__(**kwargs)
+            self.evaluation_model = model
+            self.evaluation_model_name = model_name
+            self.evaluation_context_window = context_window
+            self.evaluation_max_iterations = max_iterations
+            self.evaluation_tools: list[Any] = []
+            self.evaluation_child_tools: list[Any] = []
+            self.display_messages: list[dict[str, Any]] = []
+            self.execution_success = False
+            self.evaluation_bundle: Any = None
+
+        async def _create_agent(self, streaming: bool = False) -> Any:
+            """保留原始执行图的观察引用，生产失败恢复清缓存后仍可导出失败轨迹。"""
+            graph = await super()._create_agent(streaming=streaming)
+            self.evaluation_bundle = self._compiled_agent_bundle
+            return graph
+
+        async def _initialize_llm(self, streaming: bool = False) -> Any:
+            """直接使用调用方模型；内部总结和子代理仍由生产代码调用同一模型。"""
+            del streaming
+            return self.evaluation_model
+
+        async def _resolve_llm_runtime_config(self) -> dict[str, Any]:
+            """返回不含凭据的稳定模型描述，不加载供应商配置。"""
+            return {"provider": "openai", "model": self.evaluation_model_name, "web_search_mode": "disabled"}
+
+        def _sync_model_profile(self, model: Any) -> None:
+            """记录显式传入的模型预算，避免脚本模型缺失 profile 时丢失上下文上限。"""
+            super()._sync_model_profile(model)
+            self._session_usage.model = self.evaluation_model_name
+            self._session_usage.context_window_tokens = self.evaluation_context_window
+
+        def _get_recursion_limit(self) -> int:
+            """将显式评测迭代预算交给真实 LangGraph 执行器。"""
+            return self.evaluation_max_iterations
+
+        def _should_stream(self) -> bool:
+            """使用生产非流式分支，模型输出只进入评测捕获。"""
+            return False
+
+        def _initialize_local_tool_catalogs(self) -> tuple[Any, Any]:
+            """主图和子图使用不同工具实例，保持完整生产目录校验和子图只读策略。"""
+            return tuple(ToolCatalogSnapshot.from_tools(
+                tools, plugin_revision=0, factory_revision="evaluation-controlled-api",
+            ).require_unique() for tools in (self.evaluation_tools, self.evaluation_child_tools))
+
+        @staticmethod
+        def _initialize_tool_selector(tools: list[Any], internal_tools: list[Any], model: Any) -> Any:
+            """保留生产发现中间件，同时避免加载实际插件工具工厂。"""
+            return ToolSelectorMiddleware(
+                model=model, selection_tools=[*tools, *internal_tools], max_tools=20,
+                always_include=[tool.name for tool in [*tools, *internal_tools]], enable_discovery=True,
+            )
+
+        async def prepare_chat_title(self, message: str) -> None:
+            """评测不额外生成会话标题，业务推理与总结调用仍按生产路径计费。"""
+
+        async def _save_display_history_messages(self, messages: list[dict[str, Any]]) -> None:
+            """捕获真实 process 的展示历史，不注册通知或宿主会话服务。"""
+            self.display_messages.extend(messages)
+
+        async def send_agent_message(self, message: str, title: str = "") -> None:
+            """意外进入发送边界时直接拒绝，不能把评测内容发送到真实渠道。"""
+            raise RuntimeError("评测禁止发送通知")
+
+        def _send_agent_tokens_usage_event(self, *, success: bool, error: Optional[str] = None) -> None:
+            """只记录成功标志，不广播可被插件消费的真实运行事件。"""
+            self.execution_success = success
+
+    return EvaluationAgent
+
+
+@contextmanager
+def _runtime_scope(directory: Path, world: EvaluationWorld) -> Iterator[Any]:
+    """在独立 worker 中临时替换明确的外部边界，退出时恢复全部全局引用。"""
+    from app.agent.api.executor import MoviePilotApiExecutor
+    from app.agent.runtime import AgentRuntimeManager
+
+    manager = AgentRuntimeManager(agent_root_dir=directory / "agent")
+    manager.ensure_layout()
+    (manager.memory_dir / "MEMORY.md").write_text("用户偏好简洁、准确地完成当前任务。\n", encoding="utf-8")
+    with ExitStack() as stack:
+        for module in ("orchestrator", "middleware.config", "middleware.subagents"):
+            stack.enter_context(patch(f"app.agent.{module}.agent_runtime_manager", manager))
+        stack.enter_context(patch("app.agent.orchestrator._get_plugin_tools_revision", return_value=0))
+        stack.enter_context(patch("app.agent.orchestrator.agent_mcp_manager", _McpDirectory()))
+        stack.enter_context(patch.object(MoviePilotApiExecutor, "_resolve_base_url", return_value="http://evaluation.invalid"))
+        stack.enter_context(patch.object(MoviePilotApiExecutor, "_build_headers", return_value={"Accept": "application/json"}))
+        yield lambda **kwargs: _Transport(world, **kwargs)
+
+
+async def run_moviepilot(
+    world: EvaluationWorld, model: "BaseChatModel", *, model_name: str, context_window: int, max_iterations: int,
+    invocation_repository: Optional[Any] = None,
+) -> dict[str, Any]:
+    """调用真实 process 并捕获最终图；每个 live trial 应使用独立进程和 CONFIG_DIR。"""
+    config_path = _config_path()
+    if context_window < 1 or max_iterations < 1:
+        raise ValueError("context_window 和 max_iterations 必须为正整数")
+    if not _RUN_LOCK.acquire(blocking=False):
+        raise RuntimeError("评测全局隔离边界不能在同一进程并行使用")
+    try:
+        with TemporaryDirectory(prefix="agent-evaluation-", dir=config_path) as directory:
+            return await _run_isolated(
+                world, model, directory=Path(directory), model_name=model_name, context_window=context_window,
+                max_iterations=max_iterations, invocation_repository=invocation_repository,
+            )
+    finally:
+        _RUN_LOCK.release()
+
+
+async def _run_isolated(
+    world: EvaluationWorld, model: Any, *, directory: Path, model_name: str, context_window: int,
+    max_iterations: int, invocation_repository: Optional[Any],
+) -> dict[str, Any]:
+    """装配真实 API 执行器、独立回执库和会话端口，并在回收前保存执行证据。"""
+    from langchain_core.messages import messages_to_dict
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+
+    from app.agent.api.executor import ApiExecutionContext, MoviePilotApiExecutor
+    from app.agent.contracts import ReplyMode
+    from app.agent.memory import MemoryManager
+    from app.agent.tools.impl.api import MoviePilotApiTool
+    from app.db.adapters.invocation import TransactionalInvocationRepository
+    from app.db.models.agentinvocation import AgentInvocation
+    from app.schemas.types import NotificationChannel
+
+    with ExitStack() as stack:
+        stack.enter_context(patch.object(model, "profile", {
+            **(getattr(model, "profile", None) or {}), "max_input_tokens": context_window,
+        }))
+        if invocation_repository is None:
+            engine = create_engine(f"sqlite:///{directory / 'invocations.db'}")
+            stack.callback(engine.dispose)
+            AgentInvocation.__table__.create(engine)
+            invocation_repository = TransactionalInvocationRepository(sessionmaker(bind=engine))
+        factory = stack.enter_context(_runtime_scope(directory, world))
+        memory_port = _MemoryPort()
+        output: list[str] = []
+        agent = _agent_type()(
+            session_id=uuid4().hex, user_id="1", username="evaluation", channel=NotificationChannel.WebAgent.value,
+            source="evaluation", is_channel_admin=True, replay_mode=ReplyMode.CAPTURE_ONLY, allow_message_tools=False,
+            output_callback=output.append, data=SimpleNamespace(invocations=invocation_repository),
+            memory=MemoryManager(chat=memory_port, persistence=memory_port), model=model, model_name=model_name,
+            context_window=context_window, max_iterations=max_iterations,
+        )
+        for child in (False, True):
+            executor = MoviePilotApiExecutor(
+                context=ApiExecutionContext(user_id="1", username="evaluation", is_admin=True, session_id=agent.session_id),
+                request_factory=factory,
+            )
+            tool = MoviePilotApiTool(session_id=agent.session_id, user_id="1", executor=executor)
+            tool.set_message_attr(agent.channel, agent.source, agent.username)
+            tool.set_agent_context({"is_admin": True, "should_dispatch_reply": False, "require_secret_confirmation": True}
+                                   if child else agent._tool_context)
+            (agent.evaluation_child_tools if child else agent.evaluation_tools).append(tool)
+        try:
+            result = await agent.process(world.scenario.model_input())
+            bundle = agent.evaluation_bundle
+            state = bundle.agent.get_state({"configurable": {"thread_id": agent.session_id}}).values if bundle else {}
+            return {
+                "final_text": result or (output[-1] if output else ""), "usage": agent.get_session_status(),
+                "execution_success": agent.execution_success, "raw_messages": messages_to_dict(state.get("messages", [])),
+                "display_messages": agent.display_messages, "task_plan": state.get("task_plan"),
+                "tool_catalog_scope": "controlled_moviepilot_api_and_production_internal_tools",
+                "tool_names": sorted(tool.name for tool in bundle.tool_catalog.tools) if bundle else [],
+                "child_tool_names": sorted(tool.name for tool in agent.evaluation_child_tools),
+                "graph_nodes": sorted(bundle.agent.get_graph().nodes) if bundle else [],
+            }
+        finally:
+            if not await agent.cleanup():
+                raise RuntimeError("评测 Agent 子任务尚未完成清理")

scripts/evaluation/scenarios.py

@@ -7,6 +7,8 @@ _PUBLIC_INSTRUCTIONS = (
     "不要重复提交已有订阅或资源。"
     "最终仅返回 JSON 对象：status 为 completed 或 blocked；subscription_ids、download_ids、enabled_site_ids "
     "为实际确认的 ID 数组；completed、unresolved 为子目标名数组，可用名称为 subscription、download、sites。"
+    "completed、unresolved 只填写本次任务明确要求的子目标；未要求的类别对应 ID 数组必须为空，"
+    "必要的辅助读取不增加任务子目标。"
     "无证据的 ID 不要编造，暂时无法确认的事项放入 unresolved。"
 )
 

tests/test_agent_evaluation_live.py

@@ -0,0 +1,366 @@
+"""离线验证真实模型评测入口的证据标记、进程隔离与报告保密边界。"""
+
+import asyncio
+import io
+import json
+import os
+import socket
+import subprocess
+import sys
+from dataclasses import asdict
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+from uuid import uuid4
+
+import langchain_openai
+import pytest
+
+from app.testing import bootstrap
+from scripts.evaluation import __main__ as evaluation_cli
+from scripts.evaluation import live, runtime
+from scripts.evaluation.models import ModelSettings
+from scripts.evaluation.world import EvaluationWorld
+
+
+@pytest.fixture(autouse=True)
+def no_external_requests(monkeypatch):
+    """本文件只测试模型入口边界，禁止任何测试意外访问真实供应商。"""
+    def refuse(*_args, **_kwargs):
+        """网络解析发生即说明离线边界被绕过。"""
+        raise AssertionError("评测入口单测禁止真实网络")
+
+    monkeypatch.setattr(socket, "getaddrinfo", refuse)
+
+
+@pytest.fixture
+def model_settings():
+    """提供公开虚构主机和一次性测试密钥，不读取用户文件。"""
+    return ModelSettings(model="test-model", base_url="https://model.invalid/private-path", api_key="private-test-key")
+
+
+def test_run_live_passes_credentials_only_on_stdin(monkeypatch, model_settings):
+    """模型连接不出现在命令行或继承环境里，业务设置也不能传入新 worker。"""
+    monkeypatch.setenv("OPENAI_API_KEY", "unrelated-openai-key")
+    monkeypatch.setenv("LLM_API_KEY", "unrelated-moviepilot-key")
+    monkeypatch.setenv("TMDB_API_KEY", "unrelated-business-key")
+    monkeypatch.setenv("HTTP_PROXY", "http://user:proxy-secret@proxy.invalid")
+    called = {}
+
+    def start_worker(command, **kwargs):
+        """代替进程启动并检查控制器传递的精确私有输入。"""
+        called.update(command=command, **kwargs)
+        return SimpleNamespace(returncode=0, stdout='{"passed": false}', stderr="private-worker-log")
+
+    monkeypatch.setattr(live.subprocess, "run", start_worker)
+    assert live.run_live("unknown_download", model_settings) == {"passed": False}
+    assert called["command"] == [sys.executable, "-m", "scripts.evaluation.live", "--worker"]
+    assert called["timeout"] == model_settings.timeout_seconds + 30
+    assert called["capture_output"] is True
+    assert called["check"] is False
+    payload = json.loads(called["input"])
+    assert payload["settings"]["api_key"] == model_settings.api_key
+    assert payload["scenario_id"] == "unknown_download"
+    assert model_settings.api_key not in repr(called["command"])
+    assert called["env"].get("PATH") == os.environ.get("PATH")
+    assert not {"OPENAI_API_KEY", "LLM_API_KEY", "TMDB_API_KEY", "HTTP_PROXY", "CONFIG_DIR"} & set(called["env"])
+    assert "secret" not in json.dumps(called["env"])
+
+
+def test_worker_timeout_has_safe_public_error(monkeypatch, model_settings):
+    """子进程超时包装为固定错误，不暴露已捕获的 stdout、stderr 或凭据。"""
+    def time_out(command, **kwargs):
+        """模拟 subprocess 已完成终止回收但携带私有输出的超时异常。"""
+        raise subprocess.TimeoutExpired(command, kwargs["timeout"], output=model_settings.api_key, stderr="private-stderr")
+
+    monkeypatch.setattr(live.subprocess, "run", time_out)
+    with pytest.raises(RuntimeError) as failure:
+        live.run_live("unknown_download", model_settings)
+    assert model_settings.api_key not in str(failure.value)
+    assert "private-stderr" not in str(failure.value)
+
+
+def test_worker_nonzero_exit_does_not_publish_private_logs(monkeypatch, model_settings):
+    """SDK 初始化报错可含私有响应，父进程只报告退出状态。"""
+    monkeypatch.setattr(live.subprocess, "run", Mock(return_value=SimpleNamespace(
+        returncode=7, stdout=model_settings.api_key, stderr=f"Bearer {model_settings.api_key}",
+    )))
+    with pytest.raises(RuntimeError, match="7") as failure:
+        live.run_live("unknown_download", model_settings)
+    assert model_settings.api_key not in str(failure.value)
+
+
+@pytest.mark.parametrize(("text", "expected"), [
+    ('{"status":"completed"}', {"status": "completed"}),
+    ('```json\n{"status":"completed"}\n```', {"status": "completed"}),
+    ('```\n{"status":"blocked"}\n```', {"status": "blocked"}),
+    ('```xml\n{"status":"completed"}\n```', None),
+    ('已完成。\n{"status":"completed"}', None),
+    ('```json\n{}\n```\n```json\n{}\n```', None),
+    ('{"status":"completed"}\n还可以继续', None),
+])
+def test_final_report_parser_requires_plain_or_single_json_fence(text, expected):
+    """语言标签、附加话术和多个结果不能被悄悄解释成一个完成报告。"""
+    assert live._parse_final(text) == expected
+
+
+def test_trace_metrics_include_rejected_calls_outside_business_ledger():
+    """被 transport 拒绝的工具及技能读取都应出现于有明确范围的父图统计。"""
+    messages = [
+        {"type": "ai", "data": {"tool_calls": [{"name": "read_skill"}, {"name": "moviepilot_api"}]}},
+        {"type": "tool", "data": {"name": "read_skill", "status": "success"}},
+        {"type": "tool", "data": {"name": "moviepilot_api", "status": "error"}},
+    ]
+    assert live._trace_metrics(messages) == {"scope": "retained_parent_graph", "requested": 2, "results": 2, "errors": 1}
+
+
+@pytest.fixture
+def worker_boundary(monkeypatch):
+    """仅模拟独立 worker 的初始化和模型边界，其世界与判定器使用真实实现。"""
+    original_config = os.environ["CONFIG_DIR"]
+    monkeypatch.setenv("CONFIG_DIR", original_config)
+    monkeypatch.setattr(live, "sys", SimpleNamespace(modules={}, version=sys.version))
+    observed = SimpleNamespace(config_dirs=[], model=None, runner=None)
+
+    def prepare():
+        """验证隔离必须早于后端初始化，不再次改变当前 pytest 已装配的全局设置。"""
+        configured = Path(os.environ["CONFIG_DIR"])
+        assert str(configured) != original_config
+        assert configured.is_dir()
+        observed.config_dirs.append(configured)
+
+    def create_model(**kwargs):
+        """保存明确的模型预算并提供可核验的客户端关闭边界。"""
+        assert observed.config_dirs
+        observed.model = SimpleNamespace(
+            parameters=kwargs, root_async_client=SimpleNamespace(close=AsyncMock()),
+            root_client=SimpleNamespace(close=Mock()),
+        )
+        return observed.model
+
+    async def run(world, model, **kwargs):
+        """不执行真实模型，由各用例注入回调结果和业务轨迹。"""
+        assert observed.runner is not None
+        return await observed.runner(world, model, **kwargs)
+
+    monkeypatch.setattr(bootstrap, "prepare_backend", prepare)
+    monkeypatch.setattr(langchain_openai, "ChatOpenAI", create_model)
+    monkeypatch.setattr(runtime, "run_moviepilot", run)
+    return observed
+
+
+def _record_model_reply(model, *, successful):
+    """模拟一次合法回调生命周期，保留真实计数器和成功、失败差别。"""
+    tracker = model.parameters["callbacks"][0]
+    identifier = uuid4()
+    tracker.on_chat_model_start({}, [], run_id=identifier)
+    if successful:
+        tracker.on_llm_end(SimpleNamespace(generations=[[SimpleNamespace(message=SimpleNamespace(
+            usage_metadata={"input_tokens": 30, "output_tokens": 10, "total_tokens": 40},
+            response_metadata={"model_name": "provider-reported-model"},
+        ))]]), run_id=identifier)
+    else:
+        error = RuntimeError("Bearer private-test-key")
+        error.status_code = 400
+        tracker.on_llm_error(error, run_id=identifier)
+
+
+@pytest.mark.asyncio
+async def test_request_rejected_before_reply_is_not_intelligence_evidence(worker_boundary, model_settings):
+    """HTTP 400 只证明尝试调用过模型，不能声称已评测模型智能。"""
+    async def run(_world, model, **_kwargs):
+        """重现真实 pilot 已启动一次请求但未得到任何模型回复的状态。"""
+        _record_model_reply(model, successful=False)
+        return {"execution_success": False, "final_text": "智能助手执行失败，请稍后重试"}
+
+    worker_boundary.runner = run
+    report = await live._run_worker("unknown_download", model_settings)
+    assert report["model_calls"] == 1
+    assert report["completed_model_calls"] == 0
+    assert report["intelligence_evaluated"] is False
+    assert report["agent_execution_success"] is False
+    assert report["passed"] is False
+    assert report["tokens"] is None and report["usage_complete"] is False
+    assert report["errors"] == [{"error_type": "RuntimeError", "http_status": 400}]
+    assert model_settings.api_key not in json.dumps(report)
+    assert not any(path.exists() for path in worker_boundary.config_dirs)
+    worker_boundary.model.root_async_client.close.assert_awaited_once()
+    worker_boundary.model.root_client.close.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_successful_model_reply_still_requires_independent_business_evidence(worker_boundary, model_settings):
+    """得到模型答复可以形成模型行为证据，但不能替代世界终态和读取核验。"""
+    async def run(_world, model, **_kwargs):
+        """没有调用业务工具却虚报已完成，判定器必须拒绝该结果。"""
+        _record_model_reply(model, successful=True)
+        return {"execution_success": True, "final_text": json.dumps({
+            "status": "completed", "completed": ["download"], "unresolved": [],
+            "download_ids": ["invented-id"], "subscription_ids": [], "enabled_site_ids": [],
+        })}
+
+    worker_boundary.runner = run
+    report = await live._run_worker("unknown_download", model_settings)
+    assert report["completed_model_calls"] == 1
+    assert report["intelligence_evaluated"] is True
+    assert report["passed"] is False
+    assert "download_not_verified" in report["violations"]
+    assert report["codex_comparison"] is False
+    assert report["reported_models"] == ["provider-reported-model"]
+    assert report["tokens"]["total_tokens"] == 40
+    assert report["usage_complete"] is True
+    assert worker_boundary.model.parameters["max_retries"] == 0
+    assert worker_boundary.model.parameters["max_tokens"] == model_settings.max_output_tokens
+
+
+@pytest.mark.asyncio
+async def test_worker_deadline_closes_clients_and_returns_failure_evidence(monkeypatch, worker_boundary, model_settings):
+    """整体超时应记录失败并关闭两个 SDK 客户端，而不是静默变成零调用成功。"""
+    observed_timeout = []
+
+    async def time_out(awaitable, *, timeout):
+        """验证传递的真实墙钟上限，但不让离线单测实际等待数分钟。"""
+        observed_timeout.append(timeout)
+        awaitable.close()
+        raise asyncio.TimeoutError()
+
+    monkeypatch.setattr(live.asyncio, "wait_for", time_out)
+    report = await live._run_worker("unknown_download", model_settings)
+    assert observed_timeout == [model_settings.timeout_seconds]
+    assert report["runner_error_type"] == "TimeoutError"
+    assert report["passed"] is False
+    assert report["intelligence_evaluated"] is False
+    worker_boundary.model.root_async_client.close.assert_awaited_once()
+    worker_boundary.model.root_client.close.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_client_cleanup_failure_preserves_report_and_closes_other_client(worker_boundary, model_settings):
+    """异步客户端关闭失败不能跳过同步客户端，清理错误也不能抹掉模型失败证据。"""
+    async def run(_world, model, **_kwargs):
+        """在获得失败回调后触发 SDK 清理故障。"""
+        _record_model_reply(model, successful=False)
+        model.root_async_client.close.side_effect = RuntimeError("private-test-key")
+        return {"execution_success": False, "final_text": "执行失败"}
+
+    worker_boundary.runner = run
+    report = await live._run_worker("unknown_download", model_settings)
+    assert report["cleanup_error_types"] == ["RuntimeError"]
+    assert report["model_calls"] == 1 and report["completed_model_calls"] == 0
+    assert report["intelligence_evaluated"] is False
+    assert model_settings.api_key not in json.dumps(report)
+    worker_boundary.model.root_async_client.close.assert_awaited_once()
+    worker_boundary.model.root_client.close.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_worker_rejects_preloaded_backend(model_settings):
+    """pytest 已加载后端；真正 worker 必须在这种状态下拒绝事后隔离配置。"""
+    with pytest.raises(RuntimeError, match="导入前隔离"):
+        await live._run_worker("unknown_download", model_settings)
+
+
+def _worker_stdin(monkeypatch, model_settings):
+    """构造私有 stdin 输入，不把测试密钥传入命令行参数。"""
+    monkeypatch.setattr(sys, "argv", ["evaluation-worker", "--worker"])
+    monkeypatch.setattr(sys, "stdin", io.StringIO(json.dumps({
+        "scenario_id": "unknown_download", "settings": asdict(model_settings),
+    })))
+
+
+def test_worker_stdout_contains_only_json(monkeypatch, capsys, model_settings):
+    """依赖库的普通 stdout 被转移到私有 stderr，公开 stdout 只包含 JSON。"""
+    async def run(_scenario_id, _settings):
+        """模拟依赖在执行期间打印日志。"""
+        print("library diagnostic")
+        return {"passed": False, "intelligence_evaluated": False}
+
+    _worker_stdin(monkeypatch, model_settings)
+    monkeypatch.setattr(live, "_run_worker", run)
+    assert live.main() == 0
+    captured = capsys.readouterr()
+    assert json.loads(captured.out) == {"passed": False, "intelligence_evaluated": False}
+    assert "library diagnostic" in captured.err
+    assert model_settings.api_key not in captured.out
+
+
+def test_worker_refuses_report_containing_credential(monkeypatch, capsys, model_settings):
+    """模型或异常若把连接密钥带入报告，最后输出边界必须拒绝整份结果。"""
+    async def run(_scenario_id, settings):
+        """注入带密钥的模型最终文本，验证主进程输出前的最后一道检查。"""
+        return {"passed": False, "final_text": f"Bearer {settings.api_key}"}
+
+    _worker_stdin(monkeypatch, model_settings)
+    monkeypatch.setattr(live, "_run_worker", run)
+    with pytest.raises(RuntimeError, match="凭据"):
+        live.main()
+    assert capsys.readouterr().out == ""
+
+
+def test_worker_refuses_json_escaped_credential(monkeypatch, capsys):
+    """含引号或反斜杠的密钥经 JSON 转义后仍不能出现在公开报告中。"""
+    settings = ModelSettings(model="test-model", base_url="https://model.invalid/v1", api_key='private"test\\key')
+
+    async def run(_scenario_id, model_settings):
+        """完整密钥埋入嵌套数据，防线应检查解析后的值。"""
+        return {"passed": False, "data": [{"value": f"Bearer {model_settings.api_key}"}]}
+
+    _worker_stdin(monkeypatch, settings)
+    monkeypatch.setattr(live, "_run_worker", run)
+    with pytest.raises(RuntimeError):
+        live.main()
+    assert capsys.readouterr().out == ""
+
+
+def test_parent_refuses_unicode_escaped_credential_report(monkeypatch, model_settings):
+    """worker stdout 的 Unicode 转义不能让解析后明文密钥绕过父进程检查。"""
+    encoded_key = "".join(f"\\u{ord(char):04x}" for char in model_settings.api_key)
+    stdout = '{"passed":false,"data":[{"value":"' + encoded_key + '"}]}'
+    assert model_settings.api_key not in stdout
+    assert model_settings.api_key in json.loads(stdout)["data"][0]["value"]
+    monkeypatch.setattr(live.subprocess, "run", Mock(return_value=SimpleNamespace(returncode=0, stdout=stdout, stderr="")))
+    with pytest.raises(RuntimeError) as failure:
+        live.run_live("unknown_download", model_settings)
+    assert model_settings.api_key not in str(failure.value)
+
+
+def test_public_cli_reports_worker_failure_without_traceback(monkeypatch, capsys, model_settings):
+    """公开 CLI 把隔离进程失败转成简洁错误，不回显连接设置或 Python traceback。"""
+    from scripts.evaluation import models
+
+    monkeypatch.setattr(models, "load_codex_model_settings", Mock(return_value=model_settings))
+    monkeypatch.setattr(live, "run_live", Mock(side_effect=RuntimeError("隔离模型评测超时")))
+    with pytest.raises(SystemExit) as failure:
+        evaluation_cli.main(["--live", "--scenario", "unknown_download"])
+    assert failure.value.code == 2
+    output = capsys.readouterr()
+    assert output.out == ""
+    assert "Traceback" not in output.err
+    assert model_settings.api_key not in output.err
+    assert "超时" in output.err
+
+
+def test_provenance_changes_with_production_code_and_skills(monkeypatch, tmp_path):
+    """生产 Agent 与技能的变化必须改变独立指纹，不能继续冒充同一 Harness 基线。"""
+    main_file = tmp_path / "scripts" / "evaluation" / "__main__.py"
+    agent_file = tmp_path / "app" / "agent" / "orchestrator.py"
+    skill_file = tmp_path / "skills" / "moviepilot-api" / "SKILL.md"
+    for path in (main_file, agent_file, skill_file):
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text("original\n", encoding="utf-8")
+    monkeypatch.setattr(evaluation_cli, "__file__", str(main_file))
+    world = EvaluationWorld("dedup_existing")
+    initial = evaluation_cli._provenance(world)
+    agent_file.write_text("changed agent\n", encoding="utf-8")
+    changed_agent = evaluation_cli._provenance(world)
+    assert changed_agent["production_agent_sha256"] != initial["production_agent_sha256"]
+    assert changed_agent["skills_sha256"] == initial["skills_sha256"]
+    assert changed_agent["scenario_sha256"] == initial["scenario_sha256"]
+    skill_file.write_text("changed skill\n", encoding="utf-8")
+    changed_skill = evaluation_cli._provenance(world)
+    assert changed_skill["skills_sha256"] != changed_agent["skills_sha256"]
+    assert changed_skill["production_agent_sha256"] == changed_agent["production_agent_sha256"]
+    bytecode = agent_file.parent / "__pycache__" / "orchestrator.cpython.pyc"
+    bytecode.parent.mkdir()
+    bytecode.write_bytes(b"local bytecode")
+    assert evaluation_cli._provenance(world) == changed_skill

tests/test_agent_evaluation_models.py

@@ -0,0 +1,209 @@
+"""真实模型评测的凭据边界、调用限额及用量可信度。"""
+
+import json
+from concurrent.futures import ThreadPoolExecutor
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pytest
+from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+from langchain_core.messages import AIMessage
+from pydantic import Field
+
+from scripts.evaluation.models import ModelCallLimitError, ModelSettings, ModelUsageTracker, load_codex_model_settings
+
+
+def _config(tmp_path, credential='experimental_bearer_token = "private-test-key"'):
+    """创建仅用于测试的用户明确配置，不读取任何真实凭据。"""
+    path = tmp_path / "config.toml"
+    path.write_text('model = "test-model"\nmodel_provider = "selected"\nmodel_reasoning_effort = "high"\n'
+                    '[model_providers.selected]\nbase_url = "https://model.invalid/v1"\nwire_api = "responses"\n'
+                    + credential + '\n', encoding="utf-8")
+    return path
+
+
+def test_model_settings_do_not_expose_explicit_credential(tmp_path):
+    """凭据只用于连接，repr 和用户可见元数据不得包含它。"""
+    settings = load_codex_model_settings(_config(tmp_path))
+    assert settings.api_key == "private-test-key"
+    assert "private-test-key" not in repr(settings)
+    assert "private-test-key" not in json.dumps(settings.public_metadata())
+    assert settings.model == "test-model"
+
+
+def test_only_named_provider_environment_key_is_used(tmp_path, monkeypatch):
+    """仅使用配置明确指定的环境变量，不回退到其他账户或服务的凭据。"""
+    path = _config(tmp_path, 'env_key = "MOVIEPILOT_EVALUATION_TEST_KEY"')
+    monkeypatch.setenv("MOVIEPILOT_EVALUATION_TEST_KEY", "named-key")
+    assert load_codex_model_settings(path).api_key == "named-key"
+    monkeypatch.delenv("MOVIEPILOT_EVALUATION_TEST_KEY")
+    monkeypatch.setenv("OPENAI_API_KEY", "unrelated-key")
+    with pytest.raises(ValueError, match="显式评测凭据"):
+        load_codex_model_settings(path)
+
+
+@pytest.mark.parametrize("budget", [0, -1, 65])
+def test_invalid_call_budget_is_rejected_before_live_request(tmp_path, budget):
+    """无效预算不能退化成无限调用。"""
+    with pytest.raises(ValueError, match="预算"):
+        load_codex_model_settings(_config(tmp_path), max_model_calls=budget)
+
+
+def test_parallel_model_calls_cannot_exceed_shared_limit():
+    """父模型和多个子模型并发开始时，也必须共享同一个硬限额。"""
+    tracker = ModelUsageTracker(3)
+
+    def start():
+        """模拟一次尚未发出 HTTP 的模型调用起点。"""
+        try:
+            tracker.on_chat_model_start({}, [], run_id=uuid4())
+        except ModelCallLimitError:
+            return False
+        return True
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        results = list(pool.map(lambda _: start(), range(8)))
+    assert sum(results) == 3
+    assert tracker.snapshot()["model_calls"] == 3
+    assert tracker.snapshot()["blocked_model_calls"] == 5
+    assert tracker.raise_error is True
+
+
+def test_usage_includes_all_calls_and_marks_unknown_usage():
+    """失败或没有 usage 的请求不能被报告成完整的零成本运行。"""
+    tracker = ModelUsageTracker(4)
+    first, second = uuid4(), uuid4()
+    tracker.on_chat_model_start({}, [], run_id=first)
+    tracker.on_chat_model_start({}, [], run_id=second)
+    usage = {"input_tokens": 12, "output_tokens": 8, "total_tokens": 20}
+    response = SimpleNamespace(generations=[[SimpleNamespace(message=SimpleNamespace(
+        usage_metadata=usage, response_metadata={"model_name": "reported-model"},
+    ))]])
+    tracker.on_llm_end(response, run_id=first)
+    usage["total_tokens"] = 999
+    tracker.on_llm_error(RuntimeError("private error text"), run_id=second)
+    snapshot = tracker.snapshot()
+    assert snapshot["model_calls"] == 2
+    assert snapshot["tokens"]["total_tokens"] == 20
+    assert snapshot["usage_complete"] is False
+    assert snapshot["reported_models"] == ["reported-model"]
+    assert "private error text" not in json.dumps(snapshot)
+
+
+def test_empty_or_partial_usage_does_not_break_response_handling():
+    """统计器不能因不完整响应而打断本来可继续处理的模型结果。"""
+    tracker = ModelUsageTracker(2)
+    run_id = uuid4()
+    tracker.on_chat_model_start({}, [], run_id=run_id)
+    tracker.on_llm_end(SimpleNamespace(generations=[]), run_id=run_id)
+    snapshot = tracker.snapshot()
+    assert snapshot["completed_model_calls"] == 1
+    assert snapshot["tokens"] is None
+    assert snapshot["usage_complete"] is False
+
+
+@pytest.mark.parametrize(("field", "value"), [
+    ("model", ""), ("model", "  "), ("model", 42),
+    ("base_url", "file:///tmp/model"), ("base_url", "https:///missing-host"),
+    ("base_url", "https://user:password@model.invalid/v1"),
+    ("base_url", "https://model.invalid/v1?api_key=private-test-key"),
+    ("base_url", "https://model.invalid/v1#private-test-key"),
+    ("api_key", ""), ("api_key", "  "), ("api_key", "private\ntest"), ("api_key", False),
+    ("reasoning_effort", "unbounded"), ("max_model_calls", 0), ("max_model_calls", 65),
+    ("max_model_calls", True), ("max_model_calls", 2.5),
+    ("max_output_tokens", 255), ("max_output_tokens", 32769), ("max_output_tokens", True),
+    ("timeout_seconds", 29), ("timeout_seconds", 901), ("timeout_seconds", False),
+    ("context_window", 0), ("context_window", -1), ("context_window", True),
+])
+def test_direct_settings_reject_invalid_worker_input(field, value):
+    """stdin 直接构造设置也必须执行同一校验，布尔值不能冒充整数预算。"""
+    values = {"model": "test-model", "base_url": "https://model.invalid/v1", "api_key": "private-test-key"}
+    values[field] = value
+    with pytest.raises(ValueError):
+        ModelSettings(**values)
+
+
+@pytest.mark.parametrize("credential", [
+    'experimental_bearer_token = ""', 'experimental_bearer_token = "  "',
+    'experimental_bearer_token = "line\\nfeed"', 'experimental_bearer_token = false',
+])
+def test_invalid_explicit_credential_does_not_fallback_to_other_provider(tmp_path, monkeypatch, credential):
+    """无效的选中凭据应明确失败，不能改用环境内其他账户的密钥。"""
+    monkeypatch.setenv("OPENAI_API_KEY", "unrelated-account-secret")
+    with pytest.raises(ValueError) as failure:
+        load_codex_model_settings(_config(tmp_path, credential))
+    assert "unrelated-account-secret" not in str(failure.value)
+
+
+def test_duplicate_callbacks_preserve_single_usage_record():
+    """相同 run_id 的重复开始与完成通知不能重复耗用预算或累计 token。"""
+    tracker = ModelUsageTracker(1)
+    run_id = uuid4()
+    tracker.on_chat_model_start({}, [], run_id=run_id)
+    tracker.on_chat_model_start({}, [], run_id=run_id)
+    response = SimpleNamespace(generations=[[SimpleNamespace(message=SimpleNamespace(
+        usage_metadata={"input_tokens": 12, "output_tokens": 8, "total_tokens": 20}, response_metadata={},
+    ))]])
+    tracker.on_llm_end(response, run_id=run_id)
+    tracker.on_llm_end(response, run_id=run_id)
+    snapshot = tracker.snapshot()
+    assert snapshot["model_calls"] == snapshot["completed_model_calls"] == 1
+    assert snapshot["tokens"]["total_tokens"] == 20
+    assert snapshot["blocked_model_calls"] == 0
+
+
+@pytest.mark.parametrize("usage", [
+    {}, {"input_tokens": 10}, {"input_tokens": True, "output_tokens": 2, "total_tokens": 3},
+    {"input_tokens": -1, "output_tokens": 2, "total_tokens": 1},
+    {"input_tokens": 1.5, "output_tokens": 2, "total_tokens": 3.5},
+])
+def test_invalid_usage_remains_unknown(usage):
+    """不完整或类型错误的 usage 不能出现在完整成本报告里。"""
+    tracker = ModelUsageTracker(1)
+    run_id = uuid4()
+    tracker.on_chat_model_start({}, [], run_id=run_id)
+    tracker.on_llm_end(SimpleNamespace(generations=[[SimpleNamespace(message=SimpleNamespace(
+        usage_metadata=usage, response_metadata={},
+    ))]]), run_id=run_id)
+    assert tracker.snapshot()["tokens"] is None
+    assert tracker.snapshot()["usage_complete"] is False
+
+
+class _CountingModel(FakeMessagesListChatModel):
+    """在实际 LangChain 调用中验证 callback 限额确实早于模型执行。"""
+
+    executed: int = Field(default=0)
+
+    async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
+        """记录真正到达模型实现的请求数，不模拟 callback 调度。"""
+        self.executed += 1
+        return await super()._agenerate(messages, stop=stop, run_manager=run_manager, **kwargs)
+
+
+@pytest.mark.asyncio
+async def test_sdk_call_limit_prevents_next_model_execution():
+    """SDK 必须传播限额异常，第二次调用不能到达模型实现。"""
+    tracker = ModelUsageTracker(1)
+    model = _CountingModel(responses=[AIMessage(content="离线结果")], callbacks=[tracker])
+    assert (await model.ainvoke("第一次请求")).content == "离线结果"
+    with pytest.raises(ModelCallLimitError, match="上限"):
+        await model.ainvoke("第二次请求")
+    assert model.executed == 1
+    assert tracker.snapshot()["model_calls"] == 1
+    assert tracker.snapshot()["completed_model_calls"] == 1
+    assert tracker.snapshot()["blocked_model_calls"] == 1
+
+
+def test_failed_request_records_type_status_without_private_error():
+    """HTTP 失败可诊断为类型和状态码，私有请求内容不能进入公开统计。"""
+    tracker = ModelUsageTracker(1)
+    run_id = uuid4()
+    tracker.on_chat_model_start({}, [], run_id=run_id)
+    error = RuntimeError("Bearer private-test-key; https://private-host.invalid/secret-path")
+    error.status_code = 400
+    tracker.on_llm_error(error, run_id=run_id)
+    snapshot = tracker.snapshot()
+    assert snapshot["completed_model_calls"] == 0
+    assert snapshot["errors"] == [{"error_type": "RuntimeError", "http_status": 400}]
+    assert snapshot["tokens"] is None
+    assert "private" not in json.dumps(snapshot)

tests/test_agent_evaluation_runtime.py

@@ -0,0 +1,229 @@
+"""生产 Agent 评测运行器覆盖真实中间件、内存业务 API 和隔离回收。"""
+
+import json
+import socket
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+from langchain_core.messages import AIMessage
+from pydantic import Field
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import sessionmaker
+
+from app.agent import orchestrator
+from app.agent.api.executor import MoviePilotApiExecutor
+from app.agent.middleware import config as config_middleware
+from app.agent.middleware import subagents
+from app.agent.middleware.summarization import FinalRequestCompactionMiddleware
+from app.db.adapters.invocation import TransactionalInvocationRepository
+from app.db.models.agentinvocation import AgentInvocation
+from scripts.evaluation.runtime import _Transport, run_moviepilot
+from scripts.evaluation.score import evaluate
+from scripts.evaluation.world import EvaluationWorld
+
+
+class _ScriptModel(FakeMessagesListChatModel):
+    """脚本仅控制模型响应，不替换生产模型请求或工具执行路径。"""
+
+    bound_tools: list[list[Any]] = Field(default_factory=list)
+    requests: list[Any] = Field(default_factory=list)
+
+    def bind_tools(self, tools: list[Any], **_kwargs: Any) -> Any:
+        """记录真实图每次传给模型的工具定义。"""
+        self.bound_tools.append(list(tools))
+        return self
+
+    async def _agenerate(self, messages: list[Any], stop=None, run_manager=None, **kwargs: Any) -> Any:
+        """记录包含生产系统提示词和中间件动态上下文的最终请求。"""
+        self.requests.append(messages)
+        return await super()._agenerate(messages, stop=stop, run_manager=run_manager, **kwargs)
+
+
+def _call(name: str, arguments: dict[str, Any], identifier: str) -> AIMessage:
+    """构造标准模型工具调用，交给真实 ToolNode 执行。"""
+    return AIMessage(content="", tool_calls=[{"id": identifier, "name": name, "args": arguments}])
+
+
+def _responses(world: EvaluationWorld) -> list[AIMessage]:
+    """提供三个明确标记为脚本的正确轨迹，由独立终态判定器核验。"""
+    scenario = world.scenario
+    responses = [
+        _call("update_plan", {"objective": "完成用户任务并核验", "steps": [
+            {"step": "查询并核验业务状态", "status": "in_progress"},
+        ]}, "plan"),
+        _call("read_skill", {"name": "moviepilot-api"}, "skill"),
+    ]
+    report = {
+        "status": "completed", "subscription_ids": [], "download_ids": [scenario.infohash],
+        "enabled_site_ids": [], "completed": ["download"], "unresolved": [],
+    }
+    if scenario.scenario_id == "dedup_existing":
+        responses.append(_call("moviepilot_api", {
+            "operation_id": "subscription.find", "path_params": {"media_id": scenario.media_id},
+            "query": {"media_source": scenario.media_source},
+        }, "subscription"))
+        report["subscription_ids"] = [72]
+        report["completed"] = ["subscription", "download"]
+    responses.append(_call("moviepilot_api", {"operation_id": "download.tasks.active"}, "before"))
+    if scenario.scenario_id != "dedup_existing":
+        responses.extend([
+            _call("moviepilot_api", {"operation_id": "download.add", "body": {
+                "torrent_in": {"title": scenario.title, "enclosure": scenario.magnet},
+                "media_source": scenario.media_source, "media_id": scenario.media_id,
+            }}, "submit"),
+            _call("moviepilot_api", {"operation_id": "download.tasks.active"}, "after"),
+        ])
+    if scenario.scenario_id == "honest_unknown":
+        responses.append(_call("moviepilot_api", {"operation_id": "site.list", "query": {"status": "active"}}, "sites"))
+        report.update(status="blocked", download_ids=[], enabled_site_ids=[11, 13], completed=["sites"], unresolved=["download"])
+    responses.append(AIMessage(content=json.dumps(report, ensure_ascii=False)))
+    return responses
+
+
+@pytest.fixture
+def invocation_store(tmp_path):
+    """提供独立真实回执库，测试结束释放连接。"""
+    engine = create_engine(f"sqlite:///{tmp_path / 'invocations.db'}")
+    AgentInvocation.__table__.create(engine)
+    factory = sessionmaker(bind=engine)
+    yield TransactionalInvocationRepository(factory), factory
+    engine.dispose()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("scenario_id", ["dedup_existing", "unknown_download", "honest_unknown"])
+async def test_complete_production_agent_scenarios(scenario_id, invocation_store, monkeypatch):
+    """真实 process、策略、回执、计划、Skills 和最终压缩共同执行，不连真实业务。"""
+    world = EvaluationWorld(scenario_id)
+    model = _ScriptModel(responses=_responses(world))
+    repository, factory = invocation_store
+    original_runtime = orchestrator.agent_runtime_manager
+    original_mcp = orchestrator.agent_mcp_manager
+    original_headers = MoviePilotApiExecutor._build_headers
+    notify = AsyncMock(side_effect=AssertionError("不应调用真实通知"))
+    monkeypatch.setattr(orchestrator.AgentChain, "async_post_message", notify)
+    compaction_requests = []
+    original_compaction = FinalRequestCompactionMiddleware.awrap_model_call
+
+    async def observe_compaction(self, request, handler):
+        """继续执行真实最终请求压缩，同时证明评测未绕过该边界。"""
+        compaction_requests.append(request)
+        return await original_compaction(self, request, handler)
+
+    monkeypatch.setattr(FinalRequestCompactionMiddleware, "awrap_model_call", observe_compaction)
+
+    def refuse_network(*_args: Any, **_kwargs: Any) -> None:
+        """任何直接出站均令测试失败，包含模型目录、业务 API 和 MCP 发现。"""
+        raise AssertionError("评测不得访问网络")
+
+    monkeypatch.setattr(socket, "getaddrinfo", refuse_network)
+    capture = await run_moviepilot(
+        world, model, model_name="scripted-test", context_window=128000, max_iterations=48,
+        invocation_repository=repository,
+    )
+    assert capture["execution_success"] is True, capture["final_text"]
+    result = evaluate(world, json.loads(capture["final_text"]))
+    assert result.passed, result.violations
+    assert capture["tool_catalog_scope"] == "controlled_moviepilot_api_and_production_internal_tools"
+    assert {"moviepilot_api", "read_skill", "update_plan", "get_tool_execution", "task", "subagent_task", "search_tools"} <= set(capture["tool_names"])
+    assert not {"execute_command", "write_file", "edit_file", "browser", "send_message"} & set(capture["tool_names"])
+    assert capture["child_tool_names"] == ["moviepilot_api"]
+    assert capture["task_plan"]["objective"] == "完成用户任务并核验"
+    assert any("InvocationMiddleware" in node for node in capture["graph_nodes"])
+    assert any("SkillsMiddleware" in node for node in capture["graph_nodes"])
+    assert model.requests and "<task_planning>" in str(model.requests[0][0].content)
+    assert "<agent_memory>" in str(model.requests[0][0].content)
+    assert len(compaction_requests) == len(model.requests)
+    assert capture["usage"]["model_call_count"] > 0
+    assert capture["display_messages"][0]["role"] == "user"
+    assert capture["display_messages"][-1]["role"] == "assistant"
+    assert scenario_id not in repr(model.requests)
+    assert notify.await_count == 0
+    assert orchestrator.agent_runtime_manager is original_runtime
+    assert config_middleware.agent_runtime_manager is original_runtime
+    assert subagents.agent_runtime_manager is original_runtime
+    assert orchestrator.agent_mcp_manager is original_mcp
+    assert MoviePilotApiExecutor._build_headers is original_headers
+    with factory() as session:
+        records = list(session.execute(select(AgentInvocation)).scalars())
+    if scenario_id == "dedup_existing":
+        assert records == []
+    else:
+        assert len(records) == 1
+        assert records[0].status == "unknown"
+
+
+@pytest.mark.asyncio
+async def test_child_tool_instance_enforces_production_readonly():
+    """实际启动生产子图，写操作必须由子图策略拒绝且不触及业务世界。"""
+    world = EvaluationWorld("unknown_download")
+    model = _ScriptModel(responses=[
+        _call("task", {"description": "检查下载，尝试添加任务", "subagent_type": "general-purpose"}, "delegate"),
+        _call("moviepilot_api", {"operation_id": "download.add", "body": {
+            "torrent_in": {"title": world.scenario.title, "enclosure": world.scenario.magnet},
+            "media_source": world.scenario.media_source, "media_id": world.scenario.media_id,
+        }}, "child-write"),
+        AIMessage(content="子代理仅可读取，写入已拒绝。"),
+        AIMessage(content="子代理核验结束。"),
+    ])
+    capture = await run_moviepilot(world, model, model_name="scripted-test", context_window=128000, max_iterations=20)
+    assert capture["execution_success"] is True
+    assert world.ledger == []
+    api_instances = [tool for tools in model.bound_tools for tool in tools if getattr(tool, "name", None) == "moviepilot_api"]
+    assert len({id(tool) for tool in api_instances}) == 2
+    assert any("subagent_read_only" in str(message.content) for request in model.requests for message in request)
+
+
+@pytest.mark.asyncio
+async def test_failed_model_restores_runtime_and_profile(monkeypatch):
+    """模型失败后的恢复继续走生产逻辑，临时模型 profile 和外部边界必须还原。"""
+    original_runtime = orchestrator.agent_runtime_manager
+    original_mcp = orchestrator.agent_mcp_manager
+    original_headers = MoviePilotApiExecutor._build_headers
+    model = _ScriptModel(responses=[AIMessage(content="unused")])
+    original_profile = model.profile
+
+    async def fail_model(*_args: Any, **_kwargs: Any) -> None:
+        """在真实模型调用边界注入失败，不替换 Agent 的错误恢复路径。"""
+        raise RuntimeError("评测模型暂不可用")
+
+    monkeypatch.setattr(model, "_agenerate", fail_model)
+    capture = await run_moviepilot(
+        EvaluationWorld("dedup_existing"), model, model_name="scripted-test", context_window=128000, max_iterations=10,
+    )
+    assert capture["execution_success"] is False
+    assert capture["final_text"]
+    assert "moviepilot_api" in capture["tool_names"]
+    assert capture["graph_nodes"]
+    assert any(message["type"] == "human" for message in capture["raw_messages"])
+    assert orchestrator.agent_runtime_manager is original_runtime
+    assert orchestrator.agent_mcp_manager is original_mcp
+    assert MoviePilotApiExecutor._build_headers is original_headers
+    assert model.profile is original_profile
+
+
+@pytest.mark.asyncio
+async def test_transport_rejects_external_and_unlisted_routes():
+    """真实执行器之外直接构造请求也不能让内存传输访问外网或未纳入业务。"""
+    world = EvaluationWorld("dedup_existing")
+    transport = _Transport(world)
+    with pytest.raises(ValueError, match="内存 API"):
+        await transport.request(method="GET", url="https://example.com", params=None, json=None, raise_exception=True)
+    response = await transport.request(
+        method="GET", url="http://evaluation.invalid/api/v1/system/config", params=None, json=None, raise_exception=True,
+    )
+    assert response.json()["execution_outcome"] == "failed"
+    assert world.ledger == []
+
+
+@pytest.mark.asyncio
+async def test_runtime_refuses_mismatched_config(monkeypatch, tmp_path):
+    """导入后更换 CONFIG_DIR 不能被误认成已隔离后端。"""
+    monkeypatch.setenv("CONFIG_DIR", str(tmp_path))
+    with pytest.raises(RuntimeError, match="不一致"):
+        await run_moviepilot(
+            EvaluationWorld("dedup_existing"), _ScriptModel(responses=[AIMessage(content="unused")]),
+            model_name="scripted-test", context_window=128000, max_iterations=10,
+        )