eval: add Codex OAuth parity comparison

scripts/evaluation/__main__.py

@@ -64,12 +64,16 @@ def main(argv: list[str] | None = None) -> int:
     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 配置和工具目录，不调用真实模型")
+    mode.add_argument("--compare", nargs=2, type=Path, metavar=("MOVIEPILOT_REPORT", "CODEX_REPORT"),
+                      help="严格比较同场景的 MoviePilot live 与 Codex native 报告")
     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")
     parser.add_argument("--model", help="真实评测的模型名称，默认沿用显式 Codex provider 配置")
     parser.add_argument("--reasoning-effort", help="真实评测推理预算，默认沿用配置")
+    parser.add_argument("--use-codex-auth", action="store_true",
+                        help="显式使用本机 Codex OAuth 作为 MoviePilot/native 配对评测凭据")
     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)
@@ -77,6 +81,18 @@ def main(argv: list[str] | None = None) -> int:
     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.compare:
+        try:
+            from scripts.evaluation.compare import compare_report_files
+
+            result = compare_report_files(*args.compare)
+        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:
+            args.output.write_text(rendered + "\n", encoding="utf-8")
+        print(rendered)
+        return 0 if result["pair_valid"] else 1
     if args.scenario is None:
         parser.error("评测需要 --scenario")
     try:
@@ -86,7 +102,8 @@ def main(argv: list[str] | None = None) -> int:
             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,
+                timeout_seconds=args.timeout_seconds, probe_only=args.native_probe,
+                use_codex_auth=args.use_codex_auth,
             )
             if args.live:
                 from scripts.evaluation.live import run_live

scripts/evaluation/codex.py

@@ -223,7 +223,10 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
                 failure = result.get("error_type")
             except (RuntimeError, TimeoutError, OSError) as error:
                 failure = type(error).__name__
-            local_tokens = (proxy.bearer_token, server.bearer_token)
+            local_tokens = tuple(
+                value for value in (proxy.bearer_token, server.bearer_token, settings.account_id)
+                if isinstance(value, str) and value
+            )
         usage = proxy.snapshot()
         server_stats = server.stats
         skill_sha256 = server.skill_sha256
@@ -252,7 +255,9 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
     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: dict[str, Any] = _validate_report(
+        _redact(report, local_tokens), settings.api_key, settings.account_id,
+    )
     checked["native_stderr_sha256"] = hashlib.sha256(checked["native_stderr"].encode()).hexdigest()
     return checked
 

scripts/evaluation/compare.py

@@ -0,0 +1,122 @@
+"""校验 MoviePilot 与原生 Codex 的同条件评测报告并生成成对摘要。"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+MAX_REPORT_BYTES = 8 * 1024 * 1024
+_MODEL_CONDITION_FIELDS = (
+    "requested_model", "reasoning_effort", "max_model_calls", "max_output_tokens",
+    "timeout_seconds", "harness_context_window", "auth_mode",
+)
+_EVIDENCE_KINDS = {
+    "moviepilot_live_model": "moviepilot",
+    "codex_native_controlled": "codex",
+}
+
+
+def _read_report(path: Path) -> dict[str, Any]:
+    """读取有限大小的 JSON 报告，拒绝目录、超限或非对象正文。"""
+    if not path.is_file():
+        raise ValueError(f"评测报告不存在：{path}")
+    if path.stat().st_size > MAX_REPORT_BYTES:
+        raise ValueError("评测报告超过 8 MiB")
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeError, ValueError) as error:
+        raise ValueError(f"评测报告不是有效 JSON：{path}") from error
+    if not isinstance(payload, dict):
+        raise ValueError(f"评测报告必须是 JSON 对象：{path}")
+    return payload
+
+
+def _conditions(report: dict[str, Any]) -> dict[str, Any]:
+    """提取必须一致的模型条件，不把令牌或完整工具正文带入比较结果。"""
+    model = report.get("model")
+    if not isinstance(model, dict):
+        raise ValueError("评测报告缺少模型条件")
+    return {field: model.get(field) for field in _MODEL_CONDITION_FIELDS}
+
+
+def _metric(report: dict[str, Any], key: str) -> Any:
+    """读取报告中的公开轨迹指标，未知用量保持 None 而不是补零。"""
+    if key == "total_tokens":
+        tokens = report.get("tokens")
+        return tokens.get("total_tokens") if isinstance(tokens, dict) else None
+    return report.get(key)
+
+
+def _side(report: dict[str, Any], role: str) -> dict[str, Any]:
+    """生成不含最终正文、工具参数和凭据的单侧摘要。"""
+    return {
+        "role": role,
+        "passed": report.get("passed"),
+        "violations": report.get("violations", []),
+        "intelligence_evaluated": report.get("intelligence_evaluated"),
+        "usage_complete": report.get("usage_complete"),
+        "model_calls": _metric(report, "model_calls"),
+        "completed_model_calls": _metric(report, "completed_model_calls"),
+        "failed_tool_calls": _metric(report, "failed_tool_calls"),
+        "tool_calls": _metric(report, "tool_calls"),
+        "duplicate_attempts": _metric(report, "duplicate_attempts"),
+        "side_effects": _metric(report, "side_effects"),
+        "elapsed_seconds": _metric(report, "elapsed_seconds"),
+        "total_tokens": _metric(report, "total_tokens"),
+    }
+
+
+def _delta(moviepilot: dict[str, Any], codex: dict[str, Any]) -> dict[str, Any]:
+    """计算两个方向的数值差；任一侧未知时保留 unknown。"""
+    result: dict[str, Any] = {}
+    for key in ("model_calls", "completed_model_calls", "failed_tool_calls", "tool_calls",
+                "duplicate_attempts", "side_effects", "elapsed_seconds", "total_tokens"):
+        left, right = _metric(moviepilot, key), _metric(codex, key)
+        result[key] = None if not isinstance(left, (int, float)) or not isinstance(right, (int, float)) else left - right
+    return result
+
+
+def compare_reports(moviepilot: dict[str, Any], codex: dict[str, Any]) -> dict[str, Any]:
+    """仅在同场景、同源码和同模型预算成立时生成 Codex 成对比较摘要。"""
+    roles = {}
+    for report in (moviepilot, codex):
+        evidence_kind = report.get("evidence_kind")
+        role = _EVIDENCE_KINDS.get(evidence_kind)
+        if role is None:
+            raise ValueError("成对比较需要 MoviePilot live 与 Codex native 报告")
+        if role in roles:
+            raise ValueError("成对比较不能重复使用同一侧报告")
+        roles[role] = report
+    moviepilot, codex = roles["moviepilot"], roles["codex"]
+
+    for field in ("scenario_id", "scenario_sha256", "production_agent_sha256", "skills_sha256", "harness_sha256"):
+        if moviepilot.get(field) != codex.get(field):
+            raise ValueError(f"成对报告的 {field} 不一致")
+    moviepilot_conditions, codex_conditions = _conditions(moviepilot), _conditions(codex)
+    if moviepilot_conditions != codex_conditions:
+        raise ValueError("成对报告的模型、推理档位或预算不一致")
+    if not all(report.get("intelligence_evaluated") and report.get("usage_complete") for report in (moviepilot, codex)):
+        raise ValueError("成对比较需要两侧都完成真实模型调用并取得完整用量")
+
+    moviepilot_side, codex_side = _side(moviepilot, "moviepilot"), _side(codex, "codex")
+    return {
+        "schema_version": 1,
+        "codex_comparison": True,
+        "pair_valid": True,
+        "both_passed": bool(moviepilot.get("passed") and codex.get("passed")),
+        "scenario_id": moviepilot["scenario_id"],
+        "scenario_sha256": moviepilot["scenario_sha256"],
+        "production_agent_sha256": moviepilot["production_agent_sha256"],
+        "skills_sha256": moviepilot["skills_sha256"],
+        "harness_sha256": moviepilot["harness_sha256"],
+        "model": moviepilot_conditions,
+        "moviepilot": moviepilot_side,
+        "codex": codex_side,
+        "delta_moviepilot_minus_codex": _delta(moviepilot, codex),
+    }
+
+
+def compare_report_files(moviepilot_path: Path, codex_path: Path) -> dict[str, Any]:
+    """从两个路径读取报告并执行严格成对校验。"""
+    return compare_reports(_read_report(moviepilot_path), _read_report(codex_path))

scripts/evaluation/live.py

@@ -21,6 +21,40 @@ def _build_model(settings: ModelSettings, tracker: ModelUsageTracker) -> tuple[A
     """按显式 provider 选择评测模型，并让官方 Gemini 走原生工具协议。"""
     from urllib.parse import urlsplit
 
+    if settings.auth_mode == "codex_oauth":
+        from langchain_openai import ChatOpenAI
+
+        from app.agent.llm.helper import (
+            LLMHelper,
+            _patch_openai_responses_instructions_support,
+        )
+
+        _patch_openai_responses_instructions_support()
+        headers = {"originator": "moviepilot"}
+        if settings.account_id:
+            headers["ChatGPT-Account-Id"] = settings.account_id
+        reasoning_effort = LLMHelper._normalize_openai_reasoning_effort(settings.reasoning_effort)
+        options: dict[str, Any] = {
+            "model": settings.model,
+            "api_key": settings.api_key,
+            "base_url": settings.base_url,
+            "max_retries": 0,
+            "timeout": min(120, settings.timeout_seconds),
+            "profile": {"max_input_tokens": settings.context_window},
+            "callbacks": [tracker],
+            "streaming": True,
+            "stream_usage": True,
+            "default_headers": headers,
+            "use_responses_api": True,
+            "output_version": "responses/v1",
+            "store": False,
+        }
+        if reasoning_effort:
+            options["reasoning_effort"] = reasoning_effort
+        return ChatOpenAI(
+            **options,
+        ), "chatgpt_codex_oauth"
+
     endpoint_host = (urlsplit(settings.base_url).hostname or "").lower()
     if endpoint_host == "generativelanguage.googleapis.com":
         # Gemini 3 的 OpenAI 兼容层会把 thought_signature 放进扩展字段；
@@ -98,14 +132,15 @@ def _worker_environment() -> dict[str, str]:
     return environment
 
 
-def _validate_report(report: Any, credential: str) -> dict[str, Any]:
+def _validate_report(report: Any, credential: str, *additional_credentials: str | None) -> dict[str, Any]:
     """对解码后的键和值检查凭据，JSON 转义不能绕过输出边界。"""
+    credentials = tuple(value for value in (credential, *additional_credentials) if isinstance(value, str) and value)
     if not isinstance(report, dict):
         raise RuntimeError("隔离模型评测报告未通过输出校验")
     pending = [report]
     while pending:
         value = pending.pop()
-        if isinstance(value, str) and credential in value:
+        if isinstance(value, str) and any(secret in value for secret in credentials):
             raise RuntimeError("评测报告意外包含连接凭据，已拒绝输出")
         if isinstance(value, dict):
             pending.extend(value.keys())
@@ -133,7 +168,7 @@ def run_live(scenario_id: str, settings: ModelSettings) -> dict[str, Any]:
         report = json.loads(completed.stdout)
     except (ValueError, TypeError):
         raise RuntimeError("隔离模型评测未返回有效报告") from None
-    return _validate_report(report, settings.api_key)
+    return _validate_report(report, settings.api_key, settings.account_id)
 
 
 def _parse_final(text: str) -> Any:
@@ -223,7 +258,7 @@ def main() -> int:
     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)
+    serialized = json.dumps(_validate_report(result, settings.api_key, settings.account_id), ensure_ascii=False)
     print(serialized)
     return 0
 

scripts/evaluation/models.py

@@ -1,5 +1,6 @@
 """读取明确配置的模型连接，并为真实评测统计所有模型调用。"""
 
+import json
 import os
 import threading
 import time
@@ -11,6 +12,10 @@ from urllib.parse import urlsplit
 
 from langchain_core.callbacks.base import BaseCallbackHandler
 
+_PROBE_BASE_URL = "https://evaluation-probe.invalid/v1"
+_PROBE_API_KEY = "evaluation-probe-only"
+_CODEX_OAUTH_BASE_URL = "https://chatgpt.com/backend-api/codex"
+
 
 @dataclass(frozen=True)
 class ModelSettings:
@@ -25,6 +30,8 @@ class ModelSettings:
     timeout_seconds: int = 180
     context_window: int = 128000
     wire_api: str = "responses"
+    auth_mode: str = "explicit"
+    account_id: str | None = field(default=None, repr=False)
 
     def __post_init__(self) -> None:
         """配置文件和 worker 输入共享校验，不能绕过限额或夹带 URL 凭据。"""
@@ -42,6 +49,16 @@ class ModelSettings:
             raise ValueError("推理预算名称无效")
         if self.wire_api not in {"responses", "chat_completions"}:
             raise ValueError("模型 provider 协议无效")
+        if self.auth_mode not in {"explicit", "codex_oauth"}:
+            raise ValueError("模型鉴权模式无效")
+        if self.account_id is not None and (
+                not isinstance(self.account_id, str)
+                or not self.account_id.strip()
+                or any(char in self.account_id for char in "\r\n")
+        ):
+            raise ValueError("模型账户标识无效")
+        if self.auth_mode == "codex_oauth" and self.wire_api != "responses":
+            raise ValueError("Codex OAuth 只支持 Responses provider")
         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),
@@ -54,6 +71,7 @@ class ModelSettings:
         return {
             "requested_model": self.model, "provider_host": urlsplit(self.base_url).hostname,
             "wire_api": self.wire_api,
+            "auth_mode": self.auth_mode,
             "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,
@@ -63,8 +81,9 @@ class ModelSettings:
 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,
+    probe_only: bool = False, use_codex_auth: bool = False,
 ) -> ModelSettings:
-    """只使用选中 provider 显式配置的 bearer/env 凭据，不挪用其他服务的登录令牌。"""
+    """读取选中 provider；真实运行只接受显式凭据或用户明确选择的 Codex OAuth。"""
     config = tomllib.loads(path.read_text(encoding="utf-8"))
     provider = config.get("model_providers", {}).get(config.get("model_provider"), {})
     wire_api = provider.get("wire_api")
@@ -78,8 +97,30 @@ def load_codex_model_settings(
     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")
+    auth_mode = "explicit"
+    account_id = None
+    if use_codex_auth:
+        if probe_only:
+            raise ValueError("原生探针不能使用 Codex OAuth")
+        if endpoint or key or provider.get("env_key") or not provider.get("requires_openai_auth"):
+            raise ValueError("Codex OAuth 只适用于未自定义 endpoint 的官方 OpenAI provider")
+        auth_path = Path(os.environ.get("CODEX_HOME") or Path.home() / ".codex") / "auth.json"
+        try:
+            auth = json.loads(auth_path.read_text(encoding="utf-8"))
+            tokens = auth.get("tokens") if isinstance(auth, dict) else None
+            key = tokens.get("access_token") if isinstance(tokens, dict) else None
+            account_id = tokens.get("account_id") if isinstance(tokens, dict) else None
+        except (OSError, UnicodeError, ValueError, TypeError):
+            raise ValueError("当前 Codex OAuth 凭据不可读") from None
+        if not isinstance(key, str) or not key.strip() or not isinstance(account_id, str) or not account_id.strip():
+            raise ValueError("当前 Codex OAuth 凭据不完整")
+        endpoint = _CODEX_OAUTH_BASE_URL
+        auth_mode = "codex_oauth"
+    if probe_only:
+        endpoint = endpoint or _PROBE_BASE_URL
+        key = key or _PROBE_API_KEY
     return ModelSettings(selected_model, endpoint, key, effort, max_model_calls, max_output_tokens,
-                         timeout_seconds, wire_api=wire_api)
+                         timeout_seconds, wire_api=wire_api, auth_mode=auth_mode, account_id=account_id)
 
 
 class ModelCallLimitError(RuntimeError):

scripts/evaluation/proxy.py

@@ -33,7 +33,7 @@ NATIVE_CONTROL_TOOLS = frozenset({
 NATIVE_COLLABORATION_TOOLS = frozenset({
     "spawn_agent", "followup_task", "interrupt_agent", "list_agents", "send_message", "wait_agent",
 })
-KNOWN_NAMESPACES = frozenset({"functions", "clock", "collaboration", "mcp__evaluation"})
+KNOWN_NAMESPACES = frozenset({"functions", "clock", "collaboration", "mcp__evaluation", "multi_agent_v1"})
 FIXTURE_TOOLS = frozenset({"moviepilot_api", "read_skill", "read_tool_result"})
 
 
@@ -42,7 +42,9 @@ def allowed_tool(name: str) -> bool:
     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})
+            or normalized in {f"mcp__evaluation__{tool}" for tool in FIXTURE_TOOLS}
+            or (normalized.startswith("multi_agent_v1__")
+                and normalized.removeprefix("multi_agent_v1__") in NATIVE_CONTROL_TOOLS))
 
 
 def project_tools(tools: Any, prefix: str = "") -> tuple[list[dict[str, Any]], list[str], list[str]]:
@@ -65,10 +67,10 @@ def project_tools(tools: Any, prefix: str = "") -> tuple[list[dict[str, Any]], l
         if tool["type"] == "mcp" or (normalized.startswith("mcp__") and not (
             normalized == "mcp__evaluation" or normalized.startswith("mcp__evaluation__")
         )):
-            raise ValueError("评测目录包含外部 MCP")
+            raise ValueError(f"评测目录包含外部 MCP：{qualified}")
         if tool["type"] == "namespace":
             if qualified not in KNOWN_NAMESPACES:
-                raise ValueError("评测目录包含未知命名空间")
+                raise ValueError(f"评测目录包含未知命名空间：{qualified}")
             if "tools" not in tool:
                 raise ValueError("命名空间没有工具目录")
             children, kept, dropped = project_tools(tool["tools"], qualified)
@@ -232,9 +234,9 @@ def _validate_search_output(payload: dict[str, Any]) -> None:
             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("搜索结果包含评测目录以外的定义")
+    if removed or any(not allowed_tool(name) for name in retained):
+        names = ", ".join(removed[:8]) or ", ".join(retained[:8])
+        raise ValueError(f"搜索结果包含评测目录以外的定义：{names}")
 
 
 def _reject_unsafe_calls(payload: Any, *, partial: bool = False) -> None:
@@ -393,7 +395,12 @@ class EvaluationModelProxy:
                                                 "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)
+        if self.settings.auth_mode == "codex_oauth":
+            # ChatGPT Codex Responses 不接受 max_output_tokens；原生 Codex
+            # 客户端本来也不发送该字段，调用预算由代理计数和超时控制。
+            payload.pop("max_output_tokens", None)
+        else:
+            payload["max_output_tokens"] = min(tokens, self.settings.max_output_tokens)
         return payload, record
 
     async def _respond(self, request: Request) -> Response:
@@ -415,8 +422,12 @@ class EvaluationModelProxy:
             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")
+            # 校验异常来自本文件的固定合同文本，可安全反馈给模型以便纠正
+            # 输入；请求正文、密钥和供应商响应仍不会回显。
+            message = str(error) or "请求字段不符合评测合同"
+            self._rejected_requests.append({"error_type": type(error).__name__, "message": message,
+                                             "input_blocks": _input_metadata(parsed)})
+            return self._error(f"Invalid evaluation model request: {message}")
         if self.probe_only:
             self.requests.append(record)
             return self._error("Evaluation probe complete; no model request was forwarded")
@@ -426,9 +437,14 @@ class EvaluationModelProxy:
         record["forwarded"] = True
         self.requests.append(record)
         try:
+            headers = {"Authorization": f"Bearer {self.settings.api_key}", "Content-Type": "application/json"}
+            if self.settings.auth_mode == "codex_oauth":
+                headers["originator"] = "moviepilot"
+                if self.settings.account_id:
+                    headers["ChatGPT-Account-Id"] = self.settings.account_id
             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"},
+                headers=headers,
             ), stream=True)
         except httpx.HTTPError as error:
             record["error_type"] = type(error).__name__

scripts/evaluation/server.py

@@ -33,6 +33,7 @@ MAX_RESULTS = 8
 RESULT_TTL_SECONDS = 900
 FIRST_PAGE_CHARS = 8192
 NEXT_PAGE_CHARS = 16000
+DIRECT_RESULT_MAX_CHARS = 64 * 1024
 
 # 与 Agent 的 MoviePilotApiInput 保持相同字段，不提前暴露场景支持操作或 oracle 判据。
 API_INPUT_SCHEMA: dict[str, Any] = {
@@ -532,7 +533,9 @@ class EvaluationMcpServer:
             )
 
         text = json.dumps(skill_payload, ensure_ascii=False, indent=2)
-        if len(text) <= FIRST_PAGE_CHARS:
+        # 生产 read_skill 先走统一工具结果上限 64 KiB；当前拆分后的主文档
+        # 与分类文档都应一次交付，只有更大的文档才进入 8 KiB 首屏归档。
+        if len(text) <= DIRECT_RESULT_MAX_CHARS:
             return text
         if len(text.encode("utf-8")) > MAX_RESULT_BYTES:
             return self._failure("result_too_large", "Skill result exceeds archive limit")