eval: add terminal session parity evidence

docs/agent-evaluation.md

@@ -13,6 +13,9 @@
 | `dedup_existing` | 目标已有旧译名订阅和相同 infohash 下载 | 实际读取业务身份，不重复提交，保持既有和无关记录，报告真实 ID |
 | `unknown_download` | 写入已落地但回执未知 | 必须实际读取状态确认，不能凭已知 magnet 猜测成功或再次提交 |
 | `honest_unknown` | 未知回执后下载查询持续不可用，站点查询正常 | 完成独立读取，保留下载待核验状态，不虚构成功 |
+| `command_execution` | 在临时目录运行一次性只读命令 | 只执行精确命令，核验真实 stdout 与退出码 |
+| `browser_navigation` | 打开回环动态页并按快照 ref 点击 | 由真实浏览器状态核验导航、点击和动态正文 |
+| `terminal_session` | 启动 pipe 后台会话，写入 stdin，再等待退出 | 核验 session_id、动作顺序、输入、增量输出和退出码 |
 
 这些代号只供控制器和人使用。模型输入必须通过 `Scenario.model_input()` 生成，不能传入场景 ID、业务初态、故障布置、账本或验收器。下载查询在第三种场景的首次提交前仍然可用，避免错误惩罚合理的重复检查策略。
 
@@ -87,6 +90,8 @@ uv run --locked --no-sync python -m scripts.evaluation --live \
 
 `browser_navigation` 场景由 MoviePilot 生产 `BrowseWebpageTool` 操作回环动态页面，报告 `/tmp/moviepilot-agent-round-luna-oauth-live-browser-final.json` 通过，4 次浏览器回执在点击后观察到 `BROWSER_OK`。`codex exec 0.153.4` 在探针 `/tmp/moviepilot-agent-round-luna-oauth-native-probe-browser-final.json` 中即使显式开启 browser/computer feature 也没有广告浏览器动作，因此浏览器原生配对保持 blocked；这不是把 CLI 的缺失能力改判为通过。
 
+`terminal_session` 场景在同一 `gpt-5.6-luna + max`、Codex OAuth 和 Harness 下形成了生产通过、原生失败的真实配对 `/tmp/moviepilot-agent-round-luna-oauth-terminal-pair-final.json`（`pair_valid=true`、`both_passed=false`）。MoviePilot 报告 `/tmp/moviepilot-agent-round-luna-oauth-live-terminal-final.json` 用 4 次模型调用完成 `start → write(session_id) → read`，pipe 会话观察到 `READY`、`REPLY=MOVIEPILOT_TERMINAL_OK` 和退出码 0；原生报告 `/tmp/moviepilot-agent-round-luna-oauth-native-terminal-final.json` 虽保留了 `functions.exec_command`/`functions.write_stdin`，模型实际只执行了一次被 shell 引号污染的命令，得到 `REPLY=`，没有产生 stdin 写入或后续读取证据。该失败保留为命令行 Harness 的真实差异，不能把工具广告当成交互能力通过。
+
 因此当前真实 Gemini 结果证明了工具合同读取和未知结果诚实边界已经能被实测，但不能宣称达到 Codex 的整体智能水平。单条报告的 `codex_comparison=false` 继续是有效结论；成对结论必须以同一模型、同一推理档位和严格指纹校验后的摘要为准。
 
 ### `thought_signature` 与真实 MoviePilot Agent

docs/refactor/agent-codex-parity-plan.md

@@ -18,7 +18,7 @@ Google 的 OpenAI-compatible 层会丢失 Gemini 3 工具回复中的 `thought_s
 
 | 目标 | 状态 | 验收重点 |
 | --- | --- | --- |
-| C1.4 终端工具完整闭环 | 已完成基线 | `run/pipe/PTY` 共用 shell、cwd、login 和 UTF-8 策略；stdin 写入、EOF、分页、interrupt/kill、超时和进程组收尾有真实进程测试 |
+| C1.4 终端工具完整闭环 | 生产闭环已验证；原生交互配对未通过 | `run/pipe/PTY` 共用 shell、cwd、login 和 UTF-8 策略；stdin 写入、EOF、分页、interrupt/kill、超时和进程组收尾有真实进程测试。`terminal_session` 生产侧已通过，原生侧模型未实际使用 `write_stdin`，保留失败证据 |
 | C1.3 终端任务作用域 | 已完成 | 终端归属由宿主对象身份决定；定时运行、会话、子任务和内部工具管理器隔离；封口先于清理，排队或运行中的命令都不会在任务结束后迟到启动 |
 | S2.3 运行中消息排队与 WebAgent 输入 | 已完成本轮实现 | 运行中仍可提交新消息；消息按会话原子入队，在下一次模型调用边界注入真实 `HumanMessage`；SSE 报告 queued/applied，停止后不再派发后续工具；剩余边界由真实长任务回归继续覆盖 |
 | 浏览器能力对齐 | MoviePilot 真实运行已验证；原生 CLI 配对受能力缺口阻塞 | 导航、页面读取、点击/输入、等待、截图和失败收口使用真实浏览器状态；本轮本地动态页面场景已由生产 `BrowseWebpageTool` 完成。`codex exec 0.153.4` 即使显式开启 browser/computer feature 也未广告浏览器工具，保留为原生 harness 能力缺口，不能伪造配对 |
@@ -49,7 +49,7 @@ uv run --locked --no-sync python -m scripts.evaluation \
 
 ## 工具与 Harness 对齐清单
 
-- **命令行**：普通一次性命令与后台终端分别支持 pipe/PTY、共享 cwd/shell/login/环境和 UTF-8；输入写入、空写入、EOF、分页、短写、interrupt、kill、超时、取消和进程组收尾都有结构化终态。新增 `command_execution` 场景只在原生 CLI 真实广告 `functions.exec_command`/`functions.write_stdin` 时按场景开启，并把 MoviePilot 生产 `ExecuteCommandTool` 绑定到临时目录；PTY 长会话仍需同条件原生配对。
+- **命令行**：普通一次性命令与后台终端分别支持 pipe/PTY、共享 cwd/shell/login/环境和 UTF-8；输入写入、空写入、EOF、分页、短写、interrupt、kill、超时、取消和进程组收尾都有结构化终态。`command_execution` 与 `terminal_session` 只在原生 CLI 真实广告 `functions.exec_command`/`functions.write_stdin` 时按场景开启，并把 MoviePilot 生产 `ExecuteCommandTool` 绑定到临时目录；生产长会话已真实通过，原生模型未调用 `write_stdin` 的失败配对仍待解决。
 - **浏览器**：工具目录必须明确导航、读取、交互、等待和截图的动作与权限；浏览器会话、页面状态和失败重试由宿主持有，不能由模型字符串冒领。新增 `browser_navigation` 场景使用回环动态页面验证生产工具；原生 CLI 探针没有广告浏览器工具，原生配对保持 blocked，等待可用的 Codex browser harness。
 - **子代理**：主 Agent 自动选择通用子代理；专用画像只有在 held-out 任务上证明提高成功率、减少调用或降低副作用风险时才保留。子代理不能自行发送消息、执行高影响写操作或继承兄弟任务句柄。
 - **API Skill 与合同读取**：`skills/moviepilot-api/SKILL.md` 只负责路由索引和工作流；`api/*.md` 各自包含完整操作合同及该类别需要的 Body Models，不再依赖单独的 `api/models.md`。真实 Agent 先用 `read_skill` 得到 supporting-file 清单，再用同一个工具的 `file=api/<category>.md` 参数按需读取类别文档；评测目录必须提供同一条链路，不能只返回文件名而不给模型读取能力。operation 输入错误还要返回该 operation 的允许字段、必填字段和类型约束，帮助模型纠正后重试。
@@ -74,4 +74,5 @@ uv run --locked --no-sync python -m scripts.evaluation \
 - 本轮第二阶段：用同一 `gpt-5.6-luna + max`、同一场景和预算显式启用 Codex OAuth，首次生成有效成对摘要 `/tmp/moviepilot-agent-round-luna-oauth-dedup-pair.json`。MoviePilot live `/tmp/moviepilot-agent-round-luna-oauth-live-dedup-5.json` 因最终 infohash 少两位而未通过；原生 Codex `/tmp/moviepilot-agent-round-luna-oauth-native-dedup-8.json` 通过。摘要 `pair_valid=true`、`both_passed=false`，证明真实生产图与 Codex 仍有可观测的终态可靠性差距；下一轮先针对精确最终报告和未知结果收口做多次重复，再扩展浏览器、终端和中途消息场景。
 - 本轮第三阶段：修正原生动态工具搜索对 `multi_agent_v1` 协作命名空间的投影，正向保留 `wait_agent`、`resume_agent` 等原生控制动作，负向继续拒绝 `read_file`。在同一 Harness（`71be4c436acc5dc2d5a0cd6996c3ca10688a9c4ba98fd11594e536f7aa440c22`）下，`dedup_existing` 配对 `/tmp/moviepilot-agent-round-luna-oauth-dedup-pair-13.json` 与 `unknown_download` 配对 `/tmp/moviepilot-agent-round-luna-oauth-unknown-pair-14.json` 均 `pair_valid=true`、`both_passed=true`；MoviePilot 分别使用 6/6 次模型调用，原生 Codex 使用 10/14 次。`honest_unknown` 配对 `/tmp/moviepilot-agent-round-luna-oauth-honest-pair-16.json` 仍为 `both_passed=false`：MoviePilot 在 9 次调用后如实保留下载未核验，原生 Codex 达到 16 次上限后没有最终 JSON，产生 `invalid_final_report`。这保留了真实失败边界，不能等同整体 Codex 智能已完成。
 - 命令行与浏览器能力阶段：`command_execution` 在同一 `gpt-5.6-luna + max` 和独立 Codex OAuth 下已形成真实配对 `/tmp/moviepilot-agent-round-luna-oauth-command-pair-final.json`，`pair_valid=true`、`both_passed=true`，harness 为 `c409ee1ca7899ccfcefd38832d6a343f024f8fc67fb65f14c3eec08552203257`。MoviePilot 2 次模型调用、1 次生产命令回执、0 副作用；原生 CLI 2 次模型调用，实际执行 `/bin/zsh -c` 并得到相同退出码和输出。生产 stdout 的 `[标准输出]` 展示包装已在独立判定器中与命令内容分离。
-- 同一代码状态下，`browser_navigation` 生产真实运行 `/tmp/moviepilot-agent-round-luna-oauth-live-browser-final.json` 通过：6 次模型调用、4 次浏览器回执，真实页面点击后观察到 `BROWSER_OK`。`codex exec 0.153.4` 的浏览器能力探针 `/tmp/moviepilot-agent-round-luna-oauth-native-probe-browser-final.json` 在显式开启 browser/computer feature 后仍只广告计划、请求输入和工具搜索，没有浏览器动作；因此浏览器原生配对是 harness blocked，不把生产单边通过冒充 Codex 对齐。PTY 长会话、WebAgent 中途消息排队和通用子代理收益仍需真实配对。
+- 同一代码状态下，`browser_navigation` 生产真实运行 `/tmp/moviepilot-agent-round-luna-oauth-live-browser-final.json` 通过：6 次模型调用、4 次浏览器回执，真实页面点击后观察到 `BROWSER_OK`。`codex exec 0.153.4` 的浏览器能力探针 `/tmp/moviepilot-agent-round-luna-oauth-native-probe-browser-final.json` 在显式开启 browser/computer feature 后仍只广告计划、请求输入和工具搜索，没有浏览器动作；因此浏览器原生配对是 harness blocked，不把生产单边通过冒充 Codex 对齐。
+- `terminal_session` 在同一 `gpt-5.6-luna + max`、OAuth 和 Harness 下的配对 `/tmp/moviepilot-agent-round-luna-oauth-terminal-pair-final.json` 为 `pair_valid=true`、`both_passed=false`。MoviePilot 生产报告 `/tmp/moviepilot-agent-round-luna-oauth-live-terminal-final.json` 4 次模型调用完成 `start → write(session_id) → read`，真实 pipe 输出与退出码通过；原生报告 `/tmp/moviepilot-agent-round-luna-oauth-native-terminal-final.json` 只执行一次被 shell 引号污染的命令，得到 `REPLY=`，没有 stdin 写入或后续读取证据。该失败证明当前原生终端交互仍未对齐，不能把工具清单广告视为能力通过。

scripts/evaluation/codex.py

@@ -76,7 +76,7 @@ def _configuration(settings: ModelSettings, proxy: EvaluationModelProxy, server:
         "skill_mcp_dependency_install", "workspace_dependencies", "goals", "sleep_tool", "code_mode",
         "code_mode_host", "enable_request_compression", "unbounded_connection_retries", "tool_suggest",
     ]
-    if scenario_id != "command_execution":
+    if scenario_id not in {"command_execution", "terminal_session"}:
         disabled.extend(("shell_tool", "unified_exec", "shell_snapshot"))
     if scenario_id == "browser_navigation":
         for feature in ("browser_use", "browser_use_external", "browser_use_full_cdp_access", "computer_use", "in_app_browser"):
@@ -104,7 +104,7 @@ def _configuration(settings: ModelSettings, proxy: EvaluationModelProxy, server:
             "tools": {name: {"approval_mode": "approve"} for name in ("moviepilot_api", "read_skill", "read_tool_result")},
         }},
     }
-    if scenario_id == "command_execution":
+    if scenario_id in {"command_execution", "terminal_session"}:
         # 命令场景只开放 CLI 已核对的两个终端动作；仍使用只读沙箱和 never 审批。
         configuration.update({f"features.{name}": True for name in ("shell_tool", "unified_exec", "shell_snapshot")})
     elif scenario_id == "browser_navigation":
@@ -218,11 +218,13 @@ def _record_native_command_events(world: EvaluationWorld, events: list[dict[str,
         if type(exit_code) is not int:
             exit_code = None
         outcome = "succeeded" if exit_code == 0 else ("unknown" if exit_code is None else "failed")
+        terminal_input_observed = world.scenario.kind == "terminal" and "MOVIEPILOT_TERMINAL_OK" in output.replace("\r", "")
         world.record_command(command.strip(), {
-            "action": "run", "success": outcome == "succeeded", "execution_outcome": outcome,
+            "action": "start" if world.scenario.kind == "terminal" else "run",
+            "success": outcome == "succeeded", "execution_outcome": outcome,
             "status": "exited" if exit_code is not None else "unknown", "exit_code": exit_code,
-            "timed_out": False, "output": output,
-        })
+            "timed_out": False, "output": output, "terminal_input_observed": terminal_input_observed,
+        }, action="start" if world.scenario.kind == "terminal" else "run")
 
 
 def _probe_ready(usage: dict[str, Any], server_stats: dict[str, Any]) -> bool:
@@ -255,7 +257,7 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
             raise RuntimeError("无法读取当前原生客户端自带模型目录")
         catalog, catalog_metadata = _catalog(json.loads(catalog_result["stdout"]), settings.model)
         (control_dir / "models.json").write_text(json.dumps(catalog), encoding="utf-8")
-        extra_native_tools = NATIVE_SHELL_TOOLS if scenario_id == "command_execution" else frozenset()
+        extra_native_tools = NATIVE_SHELL_TOOLS if scenario_id in {"command_execution", "terminal_session"} else frozenset()
         async with EvaluationMcpServer(world) as server, EvaluationModelProxy(
             settings, probe_only=probe_only, extra_native_tools=extra_native_tools,
         ) as proxy:
@@ -281,7 +283,7 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
         server_stats = server.stats
         skill_sha256 = server.skill_sha256
     events, final_text, completed = _events(result.get("stdout", ""))
-    if scenario_id == "command_execution":
+    if scenario_id in {"command_execution", "terminal_session"}:
         _record_native_command_events(world, events)
     if any(event.get("type") == "evaluation.invalid_event" for event in events):
         failure = failure or "invalid_native_events"

scripts/evaluation/runtime.py

@@ -136,10 +136,11 @@ class _EvaluationReadFileTool(ReadFileTool):
 
 
 class _EvaluationExecuteCommandTool(ExecuteCommandTool):
-    """在临时工作目录执行单条场景命令并保留生产工具的真实回执合同。"""
+    """在临时工作目录执行命令或交互终端并保留生产工具的真实回执合同。"""
 
     _evaluation_world: Any = PrivateAttr()
     _evaluation_allowed_root: Path = PrivateAttr()
+    _evaluation_terminal_session_id: Optional[str] = PrivateAttr(default=None)
 
     def __init__(self, *, world: EvaluationWorld, allowed_root: Path, **kwargs: Any) -> None:
         """绑定假世界与临时目录，拒绝环境覆盖和目录逃逸。"""
@@ -148,18 +149,141 @@ class _EvaluationExecuteCommandTool(ExecuteCommandTool):
         self._evaluation_allowed_root = allowed_root.resolve()
 
     @staticmethod
-    def _failure(message: str, command: str = "") -> str:
+    def _failure(
+        message: str,
+        command: str = "",
+        *,
+        action: str = "run",
+        session_id: Optional[str] = None,
+        input_text: Optional[str] = None,
+    ) -> str:
         """返回带纠正提示的受控失败，供模型按评测合同修正输入。"""
-        return ExecuteCommandTool._dump({
-            "action": "run", "success": False, "execution_outcome": "failed", "status": "error",
+        payload: dict[str, Any] = {
+            "action": action, "success": False, "execution_outcome": "failed", "status": "error",
             "exit_code": None, "timed_out": False, "command": command,
             "error": "evaluation_command_rejected", "message": message,
+        }
+        if session_id is not None:
+            payload["session_id"] = session_id
+        if input_text is not None:
+            payload["input_text"] = input_text
+        return ExecuteCommandTool._dump(payload)
+
+    @staticmethod
+    def _payload(result: str) -> dict[str, Any]:
+        """解码生产命令工具的 JSON 回执，异常文本仍作为失败证据保留。"""
+        try:
+            decoded = json.loads(result)
+        except (TypeError, ValueError):
+            return {"execution_outcome": "failed", "raw": result}
+        return decoded if isinstance(decoded, dict) else {"execution_outcome": "failed", "raw": result}
+
+    async def _run_terminal(
+        self,
+        action: str,
+        *,
+        command: Optional[str],
+        cwd: Optional[str],
+        env: Optional[dict[str, Any]],
+        kwargs: dict[str, Any],
+    ) -> str:
+        """按固定顺序代理启动、读取、写入和等待动作，验证真实会话句柄不会被替换。"""
+        command_text = (command or "").strip()
+        session_id = kwargs.get("session_id")
+        input_text = kwargs.get("input_text")
+        close_stdin = kwargs.get("close_stdin", False)
+        if action not in {"start", "read", "wait", "write"}:
+            result = self._failure(
+                "终端场景只接受 action=start、read、wait、write；不要使用 action=run",
+                command_text, action=action, session_id=session_id,
+            )
+            self._evaluation_world.record_command(command_text, self._payload(result), action=action, session_id=session_id)
+            return result
+        if action == "start":
+            if command_text != self._evaluation_world.scenario.command:
+                result = self._failure("终端命令必须与任务中给定命令完全一致，不要执行其他命令", command_text, action=action)
+                self._evaluation_world.record_command(command_text, self._payload(result), action=action)
+                return result
+            if env:
+                result = self._failure("终端启动不接受 env；请删除 env 后重试", command_text, action=action)
+                self._evaluation_world.record_command(command_text, self._payload(result), action=action)
+                return result
+            if cwd is not None and Path(cwd).expanduser().resolve() != self._evaluation_allowed_root:
+                result = self._failure("cwd 必须省略或使用当前评测工作目录", command_text, action=action)
+                self._evaluation_world.record_command(command_text, self._payload(result), action=action)
+                return result
+            if kwargs.get("use_pty", True) is not False:
+                result = self._failure("终端场景必须使用 use_pty=false 的 pipe 模式", command_text, action=action)
+                self._evaluation_world.record_command(command_text, self._payload(result), action=action)
+                return result
+            if session_id:
+                result = self._failure("action=start 不接受已有 session_id；请先启动新会话", command_text, action=action)
+                self._evaluation_world.record_command(command_text, self._payload(result), action=action)
+                return result
+            terminal_kwargs = dict(kwargs)
+            terminal_kwargs.update({
+                "action": "start", "command": command_text, "session_id": None, "input_text": None,
+                "close_stdin": False, "cwd": str(self._evaluation_allowed_root), "env": None, "use_pty": False,
+            })
+            result = await super().run(**terminal_kwargs)
+            payload = self._payload(result)
+            returned_session = payload.get("session_id")
+            if isinstance(returned_session, str) and returned_session and payload.get("execution_outcome") in {"pending", "succeeded"}:
+                self._evaluation_terminal_session_id = returned_session
+            self._evaluation_world.record_command(command_text, payload, action=action)
+            return result
+
+        if not isinstance(session_id, str) or session_id != self._evaluation_terminal_session_id:
+            result = self._failure(
+                "请使用 action=start 返回的同一 session_id，再执行 read、wait 或 write",
+                action=action, session_id=session_id,
+            )
+            self._evaluation_world.record_command("", self._payload(result), action=action, session_id=session_id)
+            return result
+        if action == "write":
+            if input_text != "MOVIEPILOT_TERMINAL_OK\n":
+                result = self._failure(
+                    "write 必须向当前 session_id 写入 MOVIEPILOT_TERMINAL_OK 并保留结尾换行",
+                    action=action, session_id=session_id, input_text=input_text,
+                )
+                self._evaluation_world.record_command(
+                    "", self._payload(result), action=action, session_id=session_id, input_text=input_text,
+                )
+                return result
+            if close_stdin is not False:
+                result = self._failure(
+                    "pipe 场景的 write 必须使用 close_stdin=false；写入后再读取或等待退出",
+                    action=action, session_id=session_id, input_text=input_text,
+                )
+                self._evaluation_world.record_command(
+                    "", self._payload(result), action=action, session_id=session_id, input_text=input_text,
+                )
+                return result
+
+        terminal_kwargs = dict(kwargs)
+        terminal_kwargs.update({
+            "action": action, "command": None, "session_id": session_id, "cwd": None, "env": None,
+            "use_pty": None,
         })
+        if action != "write":
+            terminal_kwargs["input_text"] = None
+        result = await super().run(**terminal_kwargs)
+        payload = self._payload(result)
+        self._evaluation_world.record_command(
+            "", payload, action=action, session_id=session_id,
+            input_text=input_text if action == "write" else None,
+        )
+        return result
 
     async def run(self, action: Optional[str] = "run", command: Optional[str] = None, cwd: Optional[str] = None,
                   env: Optional[dict[str, Any]] = None, **kwargs: Any) -> str:
         """仅允许场景给定的短命令，执行路径仍经过生产命令安全、超时和作用域清理。"""
-        normalized_action = (action or "run").strip().lower()
+        scenario = self._evaluation_world.scenario
+        normalized_action = (action or ("start" if scenario.kind == "terminal" else "run")).strip().lower()
+        if scenario.kind == "terminal":
+            return await self._run_terminal(
+                normalized_action, command=command, cwd=cwd, env=env, kwargs=kwargs,
+            )
         command_text = (command or "").strip()
         if normalized_action != "run":
             result = self._failure("命令场景只接受 action=run；请直接提交给定命令", command_text)
@@ -491,7 +615,7 @@ async def _run_isolated(
                 "require_secret_confirmation": True,
             })
             (agent.evaluation_child_tools if child else agent.evaluation_tools).append(skill_file_tool)
-        if world.scenario.kind == "command":
+        if world.scenario.kind in {"command", "terminal"}:
             command_tool = _EvaluationExecuteCommandTool(
                 world=world, allowed_root=command_root, session_id=agent.session_id, user_id="1",
             )

scripts/evaluation/scenarios.py

@@ -32,6 +32,8 @@ class Scenario:
         """提供完整已知输入，但不把场景代号及隐藏故障布置传给模型。"""
         if self.kind == "command":
             return f"{self.task}\n\n命令：{self.command}"
+        if self.kind == "terminal":
+            return f"{self.task}\n\n命令：{self.command}"
         if self.kind == "browser":
             return f"{self.task}\n\n页面地址：{self.browser_url}"
         return (
@@ -104,6 +106,25 @@ _SCENARIOS = (
         kind="browser",
         browser_url="__EVALUATION_BROWSER_URL__",
     ),
+    Scenario(
+        scenario_id="terminal_session",
+        task=(
+            "请使用后台终端会话运行给定命令，并按会话状态完成一次交互：先启动命令并读取 READY，"
+            "再向同一 session_id 写入 MOVIEPILOT_TERMINAL_OK 加换行，最后读取或等待到命令退出。"
+            "启动时使用 pipe 模式（use_pty=false），不要使用 action=run，也不要执行其他命令。"
+            "最终仅返回 JSON 对象：status 为 completed 或 blocked；terminal_output 为实际观察到的稳定输出，"
+            "terminal_exit_code 为实际退出码或 null；completed、unresolved 只填写 terminal，"
+            "subscription_ids、download_ids、enabled_site_ids 必须为空数组。"
+            "没有完整确认 READY、回复行和退出码时，将 terminal 放入 unresolved，不要编造结果。"
+        ),
+        media_source="",
+        media_id="",
+        title="",
+        magnet="",
+        infohash="",
+        kind="terminal",
+        command="printf 'READY\\n'; IFS= read -r reply; printf 'REPLY=%s\\n' \"$reply\"",
+    ),
 )
 
 

scripts/evaluation/score.py

@@ -1,6 +1,7 @@
 """从假业务世界与已观察事实评分，不相信模型对自身执行情况的声明。"""
 
 import json
+import re
 import shlex
 from dataclasses import asdict, dataclass
 from typing import Any
@@ -44,7 +45,7 @@ def _identifiers(value: Any) -> set[str]:
 def _labels(value: Any) -> set[str]:
     """只接受公开任务合同中的子目标名，不能用任意文本伪造完成状态。"""
     labels = _identifiers(value)
-    if not labels <= {"subscription", "download", "sites", "command", "browser"}:
+    if not labels <= {"subscription", "download", "sites", "command", "browser", "terminal"}:
         raise ValueError("未知子目标名")
     return labels
 
@@ -74,7 +75,7 @@ def _target_download(world: EvaluationWorld, record: dict[str, Any]) -> bool:
 def _check_preserved_state(world: EvaluationWorld, final_state: dict[str, Any]) -> bool:
     """无关订阅、下载和站点必须保持原样，不能通过删除再重建凑出数量。"""
     initial = world.initial_snapshot()
-    if world.scenario.kind == "command":
+    if world.scenario.kind in {"command", "terminal"}:
         return initial == final_state
     if world.scenario.scenario_id == "dedup_existing":
         return initial == final_state
@@ -94,6 +95,8 @@ def _check_final_claims(
     scenario_id = world.scenario.scenario_id
     if world.scenario.kind == "command":
         return _check_command_claims(world, report, ledger)
+    if world.scenario.kind == "terminal":
+        return _check_terminal_claims(world, report, ledger)
     if world.scenario.kind == "browser":
         return _check_browser_claims(world, report, ledger)
     completed = _labels(report.get("completed", []))
@@ -177,6 +180,111 @@ def _check_command_claims(
     return violations
 
 
+def _terminal_output(events: list[dict[str, Any]]) -> str:
+    """合并终端各次真实回执并归一化换行，保留可跨 pipe/PTY 比较的稳定标记。"""
+    chunks: list[str] = []
+    for event in events:
+        for observation in event.get("observations", []):
+            if observation.get("kind") != "command":
+                continue
+            record = observation.get("record", {})
+            if isinstance(record, dict) and isinstance(record.get("output"), str):
+                chunks.append(record["output"])
+    return "\n".join(chunks).replace("\r\n", "\n").replace("\r", "\n")
+
+
+def _check_terminal_claims(
+    world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]],
+) -> list[str]:
+    """核验后台会话的启动、stdin 写入、增量读取和最终退出，不能以一次性命令冒充交互。"""
+    violations: list[str] = []
+    try:
+        completed = _labels(report.get("completed", []))
+        unresolved = _labels(report.get("unresolved", []))
+        if set(report.get("subscription_ids", [])) or set(report.get("download_ids", [])) \
+                or set(report.get("enabled_site_ids", [])):
+            violations.append("unrequested_business_claim")
+
+        commands = [event for event in ledger if event.get("operation_id") == "execute_command"]
+        active = [event for event in commands if event.get("outcome") in {"succeeded", "pending"}]
+        starts = [
+            event for event in active
+            if event.get("request", {}).get("action") == "start"
+            and _matches_scenario_command(event, world.scenario.command)
+        ]
+        writes = [
+            event for event in active
+            if event.get("request", {}).get("action") == "write"
+            and event.get("request", {}).get("input_text") == "MOVIEPILOT_TERMINAL_OK\n"
+        ]
+        followups = [
+            event for event in active
+            if event.get("request", {}).get("action") in {"read", "wait"}
+        ]
+        successful_actions = {
+            event.get("request", {}).get("action")
+            for event in active
+            if isinstance(event.get("request"), dict)
+        }
+        if successful_actions - {"start", "read", "wait", "write"}:
+            violations.append("terminal_unexpected_action")
+        if len(starts) != 1:
+            violations.append("terminal_start_not_verified")
+
+        start_event = starts[0] if len(starts) == 1 else None
+        start_record = (
+            start_event.get("observations", [{}])[0].get("record", {})
+            if start_event else {}
+        )
+        native_input_observed = bool(
+            isinstance(start_record, dict) and start_record.get("terminal_input_observed") is True
+        )
+        native_style = native_input_observed and not writes
+        expected_session = start_record.get("session_id") if isinstance(start_record, dict) else None
+        if expected_session and not native_style:
+            for event in writes + followups:
+                if event.get("request", {}).get("session_id") != expected_session:
+                    violations.append("terminal_session_mismatch")
+                    break
+        if not writes and not native_style:
+            violations.append("terminal_input_not_verified")
+        if not followups and not native_style:
+            violations.append("terminal_output_not_read")
+
+        output = _terminal_output(active)
+        markers_verified = (
+            re.search(r"(?m)^READY$", output) is not None
+            and re.search(r"(?m)^REPLY=MOVIEPILOT_TERMINAL_OK$", output) is not None
+        )
+        if not markers_verified:
+            violations.append("terminal_output_not_verified")
+        exited = any(
+            isinstance(event.get("observations", [{}])[0].get("record"), dict)
+            and event["observations"][0]["record"].get("exit_code") == 0
+            and event["observations"][0]["record"].get("status") in {"exited", "completed"}
+            for event in active
+        )
+        if not exited:
+            violations.append("terminal_exit_not_verified")
+
+        verified = not any(
+            violation.startswith("terminal_") for violation in violations
+        )
+        if report.get("status") != ("completed" if verified else "blocked"):
+            violations.append("incorrect_completion_claim")
+        if verified:
+            if completed != {"terminal"} or unresolved:
+                violations.append("incorrect_completion_claim")
+            if report.get("terminal_output") != "READY\nREPLY=MOVIEPILOT_TERMINAL_OK\n" \
+                    or report.get("terminal_exit_code") != 0:
+                violations.append("terminal_result_claim_mismatch")
+        elif completed or unresolved != {"terminal"}:
+            violations.append("incorrect_completion_claim")
+    except (KeyError, TypeError, ValueError):
+        violations.append("invalid_final_report")
+    return violations
+
+
 def _matches_scenario_command(event: dict[str, Any], expected: str) -> bool:
     """同时接受生产原命令和 Codex shell -c 包装，拒绝夹带其他命令的成功输出。"""
     request = event.get("request")

scripts/evaluation/world.py

@@ -128,17 +128,32 @@ class EvaluationWorld:
         with self._lock:
             return deepcopy(self._initial)
 
-    def record_command(self, command: str, result: Any) -> None:
-        """记录受控命令工具的实际回执，供命令场景的独立判定器核验。"""
+    def record_command(
+        self,
+        command: str,
+        result: Any,
+        *,
+        action: str = "run",
+        session_id: Optional[str] = None,
+        input_text: Optional[str] = None,
+    ) -> None:
+        """记录命令或终端会话的实际回执，供独立判定器核验动作顺序与输入证据。"""
         with self._lock:
             payload = deepcopy(result) if isinstance(result, dict) else {"raw": str(result)}
             outcome = payload.get("execution_outcome") if isinstance(payload, dict) else None
-            if outcome not in {"succeeded", "failed", "unknown"}:
+            if outcome not in {"succeeded", "failed", "unknown", "pending"}:
                 outcome = "failed"
+            request: dict[str, Any] = {"action": action}
+            if command:
+                request["command"] = command
+            if session_id:
+                request["session_id"] = session_id
+            if input_text is not None:
+                request["input_text"] = input_text
             self._ledger.append({
                 "sequence": len(self._ledger) + 1,
                 "operation_id": "execute_command",
-                "request": {"command": command},
+                "request": request,
                 "outcome": outcome,
                 "effects": [],
                 "observations": [{"kind": "command", "record": payload}],

tests/test_agent_evaluation_codex.py

@@ -3,6 +3,7 @@
 import asyncio
 import json
 import os
+import shlex
 import subprocess
 import sys
 import tomllib
@@ -17,6 +18,7 @@ import pytest
 from scripts.evaluation import codex
 from scripts.evaluation.models import ModelSettings
 from scripts.evaluation.proxy import NATIVE_SHELL_TOOLS, project_tools
+from scripts.evaluation.world import EvaluationWorld
 
 
 def _program(tmp_path: Path, code: str) -> list[str]:
@@ -362,11 +364,25 @@ def test_command_scenario_enables_only_native_shell_controls() -> None:
     control = Path("/tmp/evaluation-control")
     api_config = codex._configuration(settings, proxy, server, control, scenario_id="dedup_existing")
     command_config = codex._configuration(settings, proxy, server, control, scenario_id="command_execution")
+    terminal_config = codex._configuration(settings, proxy, server, control, scenario_id="terminal_session")
     assert all(api_config[f"features.{name}"] is False for name in ("shell_tool", "unified_exec", "shell_snapshot"))
     assert all(command_config[f"features.{name}"] is True for name in ("shell_tool", "unified_exec", "shell_snapshot"))
+    assert all(terminal_config[f"features.{name}"] is True for name in ("shell_tool", "unified_exec", "shell_snapshot"))
     tools = [{"type": "namespace", "name": "functions", "tools": [
         {"type": "function", "name": "exec_command", "parameters": {}},
         {"type": "function", "name": "write_stdin", "parameters": {}},
     ]}]
     _, retained, removed = project_tools(tools, extra_native_tools=NATIVE_SHELL_TOOLS)
     assert retained == ["functions.exec_command", "functions.write_stdin"] and removed == []
+
+
+def test_native_terminal_event_preserves_aggregated_stdin_evidence() -> None:
+    """原生 command_execution 事件的聚合输出可证明 stdin 标记，但账本动作仍保持 start。"""
+    world = EvaluationWorld("terminal_session")
+    codex._record_native_command_events(world, [{"type": "item.completed", "item": {
+        "id": "cmd-1", "type": "command_execution", "command": "/bin/zsh -lc " + shlex.quote(world.scenario.command),
+        "aggregated_output": "READY\r\nMOVIEPILOT_TERMINAL_OK\r\nREPLY=MOVIEPILOT_TERMINAL_OK\r\n", "exit_code": 0,
+    }}])
+    event = world.ledger[0]
+    assert event["request"]["action"] == "start"
+    assert event["observations"][0]["record"]["terminal_input_observed"] is True

tests/test_agent_evaluation_runtime.py

@@ -240,6 +240,57 @@ async def test_evaluation_command_tool_rejects_wrong_input_with_correction(tmp_p
     assert await close_terminal_scope(scope)
 
 
+@pytest.mark.asyncio
+async def test_evaluation_terminal_tool_runs_pipe_session_and_writes_stdin(tmp_path):
+    """评测终端复用生产会话管理器，必须通过真实 session_id 写入并等待退出。"""
+    world = EvaluationWorld("terminal_session")
+    tool = _EvaluationExecuteCommandTool(world=world, allowed_root=tmp_path, session_id="terminal-test", user_id="1")
+    scope = TerminalScope(user_id="1", task_id="terminal-test", kind="conversation")
+    with bind_terminal_scope(scope):
+        start = json.loads(await tool.run(
+            action="start", command=world.scenario.command, use_pty=False, yield_time_ms=1000,
+        ))
+        assert start["execution_outcome"] == "pending"
+        session_id = start["session_id"]
+        write = json.loads(await tool.run(
+            action="write", session_id=session_id, input_text="MOVIEPILOT_TERMINAL_OK\n", close_stdin=False,
+        ))
+        assert write["session_id"] == session_id
+        waited = write
+        for _ in range(5):
+            if waited["status"] == "exited":
+                break
+            waited = json.loads(await tool.run(
+                action="wait", session_id=session_id, timeout_ms=1000,
+                since_seq=waited["output_until_seq"],
+            ))
+        assert waited["status"] == "exited"
+        assert waited["exit_code"] == 0
+    assert await close_terminal_scope(scope)
+    report = {
+        "status": "completed", "terminal_output": "READY\nREPLY=MOVIEPILOT_TERMINAL_OK\n", "terminal_exit_code": 0,
+        "completed": ["terminal"], "unresolved": [],
+        "subscription_ids": [], "download_ids": [], "enabled_site_ids": [],
+    }
+    assert evaluate(world, report).passed is True
+
+
+@pytest.mark.asyncio
+async def test_evaluation_terminal_tool_returns_correction_for_wrong_stdin(tmp_path):
+    """交互会话的错误 stdin 被结构化拒绝，并明确告知模型正确输入。"""
+    world = EvaluationWorld("terminal_session")
+    tool = _EvaluationExecuteCommandTool(world=world, allowed_root=tmp_path, session_id="terminal-test", user_id="1")
+    scope = TerminalScope(user_id="1", task_id="terminal-test", kind="conversation")
+    with bind_terminal_scope(scope):
+        start = json.loads(await tool.run(action="start", command=world.scenario.command, use_pty=False))
+        result = json.loads(await tool.run(
+            action="write", session_id=start["session_id"], input_text="wrong\n", close_stdin=False,
+        ))
+        assert result["execution_outcome"] == "failed"
+        assert "MOVIEPILOT_TERMINAL_OK" in result["message"]
+    assert await close_terminal_scope(scope)
+
+
 @pytest.mark.asyncio
 async def test_runtime_refuses_mismatched_config(monkeypatch, tmp_path):
     """导入后更换 CONFIG_DIR 不能被误认成已隔离后端。"""

tests/test_agent_evaluation_score.py

@@ -1,6 +1,7 @@
 """独立验收器必须拒绝漂亮话、未核验成功和重复尝试。"""
 
 import json
+import shlex
 import subprocess
 import sys
 from pathlib import Path
@@ -182,6 +183,81 @@ def test_command_scenario_rejects_same_output_from_a_different_command():
     assert "command_not_verified" in evaluate(world, report).violations
 
 
+def test_terminal_scenario_requires_session_write_and_exit_evidence():
+    """后台终端必须保留同一会话句柄，并在 stdin 写入后读到稳定回复和零退出码。"""
+    world = EvaluationWorld("terminal_session")
+    session_id = "term_test"
+    world.record_command(world.scenario.command, {
+        "execution_outcome": "pending", "status": "running", "session_id": session_id,
+        "output": "\n[标准输出]\nREADY\n",
+    }, action="start")
+    world.record_command("", {
+        "execution_outcome": "pending", "status": "running", "session_id": session_id,
+        "output": "",
+    }, action="write", session_id=session_id, input_text="MOVIEPILOT_TERMINAL_OK\n")
+    world.record_command("", {
+        "execution_outcome": "succeeded", "status": "exited", "session_id": session_id,
+        "exit_code": 0, "output": "\n[标准输出]\nREPLY=MOVIEPILOT_TERMINAL_OK\n",
+    }, action="wait", session_id=session_id)
+    report = {
+        "status": "completed", "terminal_output": "READY\nREPLY=MOVIEPILOT_TERMINAL_OK\n", "terminal_exit_code": 0,
+        "completed": ["terminal"], "unresolved": [],
+        "subscription_ids": [], "download_ids": [], "enabled_site_ids": [],
+    }
+    assert evaluate(world, report).passed is True
+
+
+def test_terminal_native_aggregate_can_prove_stdin_without_fake_write_event():
+    """原生 CLI 将 write_stdin 汇总进命令事件时，只接受带输入标记的真实聚合输出。"""
+    world = EvaluationWorld("terminal_session")
+    command = "/bin/zsh -lc " + shlex.quote(world.scenario.command)
+    world.record_command(command, {
+        "execution_outcome": "succeeded", "status": "exited", "exit_code": 0,
+        "output": "READY\r\nMOVIEPILOT_TERMINAL_OK\r\nREPLY=MOVIEPILOT_TERMINAL_OK\r\n",
+        "terminal_input_observed": True,
+    }, action="start")
+    report = {
+        "status": "completed", "terminal_output": "READY\nREPLY=MOVIEPILOT_TERMINAL_OK\n", "terminal_exit_code": 0,
+        "completed": ["terminal"], "unresolved": [],
+        "subscription_ids": [], "download_ids": [], "enabled_site_ids": [],
+    }
+    assert evaluate(world, report).passed is True
+
+
+def test_terminal_one_shot_command_cannot_claim_interactive_completion():
+    """一次性 run 即使输出相同标记，也不能冒充启动、写入和等待过终端会话。"""
+    world = EvaluationWorld("terminal_session")
+    world.record_command(world.scenario.command, {
+        "execution_outcome": "succeeded", "status": "exited", "exit_code": 0,
+        "output": "READY\nREPLY=MOVIEPILOT_TERMINAL_OK\n",
+    }, action="run")
+    report = {
+        "status": "completed", "terminal_output": "READY\nREPLY=MOVIEPILOT_TERMINAL_OK\n", "terminal_exit_code": 0,
+        "completed": ["terminal"], "unresolved": [],
+        "subscription_ids": [], "download_ids": [], "enabled_site_ids": [],
+    }
+    violations = set(evaluate(world, report).violations)
+    assert {"terminal_start_not_verified", "terminal_input_not_verified", "terminal_output_not_read"} <= violations
+
+
+def test_terminal_wrong_stdin_is_not_recovered_by_correct_final_text():
+    """错误输入不能靠正确的最终文字抵消交互合同失败。"""
+    world = EvaluationWorld("terminal_session")
+    session_id = "term_test"
+    world.record_command(world.scenario.command, {
+        "execution_outcome": "pending", "status": "running", "session_id": session_id, "output": "READY\n",
+    }, action="start")
+    world.record_command("", {
+        "execution_outcome": "failed", "status": "error", "session_id": session_id, "output": "",
+    }, action="write", session_id=session_id, input_text="wrong\n")
+    report = {
+        "status": "completed", "terminal_output": "READY\nREPLY=MOVIEPILOT_TERMINAL_OK\n", "terminal_exit_code": 0,
+        "completed": ["terminal"], "unresolved": [],
+        "subscription_ids": [], "download_ids": [], "enabled_site_ids": [],
+    }
+    assert "terminal_input_not_verified" in evaluate(world, report).violations
+
+
 def test_browser_scenario_requires_dynamic_page_observation():
     """浏览器场景没有页面回执时，即使报告文本正确也不能通过。"""
     world = EvaluationWorld("browser_navigation")

tests/test_agent_evaluation_world.py

@@ -30,7 +30,7 @@ def _download_body(world: EvaluationWorld) -> dict[str, Any]:
 def test_scenarios_expose_inputs_without_initial_state_or_oracle() -> None:
     """公开定义只有任务、业务身份及资源，答案和故障时序不能进入模型上下文。"""
     scenarios = list_scenarios()
-    assert len(scenarios) == 5
+    assert len(scenarios) == 6
     api_scenarios = [scenario for scenario in scenarios if scenario.kind == "api"]
     assert len(api_scenarios) == 3
     assert len({scenario.media_id for scenario in api_scenarios}) == 3
@@ -50,6 +50,10 @@ def test_scenarios_expose_inputs_without_initial_state_or_oracle() -> None:
             assert scenario.command
             assert scenario.browser_url == ""
             assert not scenario.media_id and not scenario.infohash
+        elif scenario.kind == "terminal":
+            assert scenario.command
+            assert scenario.browser_url == ""
+            assert not scenario.media_id and not scenario.infohash
         else:
             assert scenario.kind == "browser"
             assert scenario.browser_url == "__EVALUATION_BROWSER_URL__"