fix(agent): verify steering boundary in evaluation

app/agent/middleware/steering.py

@@ -12,6 +12,11 @@ from langchain_core.messages import AIMessage, HumanMessage
 from app.agent.steering import SteeringMessage, current_steering_inbox
 
 STEERING_MESSAGE_ID_KEY = "moviepilot_steering_message_id"
+STEERING_CONTEXT_KEY = "continuation_context"
+STEERING_CONTEXT_INSTRUCTION = (
+    "这是当前运行任务的补充要求。继续原任务并保留此前的范围、停止条件、确认标准、输出格式和安全约束；"
+    "只有用户明确修改时才改变它们。"
+)
 
 
 def _to_human_message(message: SteeringMessage) -> HumanMessage:
@@ -19,6 +24,7 @@ def _to_human_message(message: SteeringMessage) -> HumanMessage:
     payload = {
         "message": message.text,
         "input": {"mode": "text", "steering": True},
+        STEERING_CONTEXT_KEY: STEERING_CONTEXT_INSTRUCTION,
         "images": [{"index": index + 1, "type": "image"} for index, _ in enumerate(message.images)],
         "files": list(message.files),
     }

docs/agent-evaluation.md

@@ -124,6 +124,10 @@ WebAgent 的排队消息已补齐稳定留存和真实时序展示：后端展
 
 前端后续修复把未收到稳定 `steering_message_id` 的本地草稿改为有序集合，并在 queued ACK 阶段把排队气泡锚定到当前助手段之后；queued ACK 不封闭助手段，ACK 之后到达的文本和工具事件仍归属于当前助手，直到 applied 事件报告后端实际消费消息的模型边界，再收口前一段、插入用户消息并创建 continuation。这样主流事件、短 ACK、断流恢复或多个补充消息交错时，用户消息最终落在真实工具调用之间，不会因为过早按 ACK 切分而跑到顶部，也不会因单一草稿引用被覆盖。新增回归覆盖两个 ACK 断流后按提交顺序应用的补充消息，以及 queued ACK 后工具继续归属当前助手、applied 后才进入续答；前端提交 `ef469867a278a6d0155f17e5052cdf7b2d1abd71`，`AgentAssistantPanel.spec.ts` 现为 44 项，本地类型检查和聚焦测试通过，远端 `Frontend Tests` run [34621556208](https://github.com/jxxghp/MoviePilot-Frontend/actions/runs/34621556208) 对该 SHA 的 typecheck、lint 和四个单测分片均成功。
 
+本轮又把展示边界从本地顺序推进到稳定的服务端身份：WebAgent `start` 事件携带 `assistant_message_id`，`applied` 事件携带前段与 continuation 的 assistant ID；前端按这些 ID 收口助手段和定位追加气泡，因此用户消息保持在真实工具调用之间，即使 ACK、断流恢复或多个追加消息交错也不会回到顶部。工具状态按各自稳定 `tool_id` 的 `running/done/error` 事件维护，工具图标不再把已完成项误显示为执行中。后端稳定 ID 提交 `2427eb7e2f6ec8da1a8d9abbe810f512e753c5f`，前端稳定边界提交 `2a64102bd57b72082b23f835ecbcf3d4457576ab`；当前工作树还包含图标生命周期的一行修复。
+
+为验证追加消息不是单元测试假象，新增 `steering_long_context` 固定场景：首个 `subscription.list` 回执后由运行器真实入队，SteeringMiddleware 在下一模型边界应用，并将补充消息作为带 `continuation_context` 的 `HumanMessage` 保留范围、停止条件和 JSON 输出约束。使用同一 `gpt-5.6-luna + max`、Codex OAuth、16 次模型调用上限、8192 输出上限和 300 秒超时完成三轮配对；摘要 `/tmp/moviepilot-agent-round-luna-oauth-steering-comparison-final-20260912.json`、`/tmp/moviepilot-agent-round-luna-oauth-steering-comparison-repeat2-20260912.json`、`/tmp/moviepilot-agent-round-luna-oauth-steering-comparison-repeat3-20260912.json` 均 `pair_valid=true`、`both_passed=true`。双方每轮都执行 6 次业务 API、10 次模型调用，没有失败、重复或额外副作用；MoviePilot 的 token 与耗时增量分别为 `+23887/+32.739s`、`+23100/+10.551s`、`+22985/+5.428s`。这证明当前固定中途追加场景具备真实应用边界和终态收口，但不外推到刷新、取消或多条追加消息的所有组合。
+
 因此当前真实 Gemini 结果证明了工具合同读取和未知结果诚实边界已经能被实测，但不能宣称达到 Codex 的整体智能水平。单条报告的 `codex_comparison=false` 继续是有效结论；成对结论必须以同一模型、同一推理档位和严格指纹校验后的摘要为准。
 
 ### `thought_signature` 与真实 MoviePilot Agent

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

@@ -20,7 +20,7 @@ Google 的 OpenAI-compatible 层会丢失 Gemini 3 工具回复中的 `thought_s
 | --- | --- | --- |
 | C1.4 终端工具完整闭环 | PTY 已配对；原生 pipe stdin 仍受边界阻塞 | `run/pipe/PTY` 共用 shell、cwd、login 和 UTF-8 策略；stdin 写入、EOF、分页、interrupt/kill、超时和进程组收尾有真实进程测试。原生终端场景经由 app-server 采集 `outputDelta`/`terminalInteraction`，并按评测预算保留足够的流式空闲时间；PTY 配对已观察输入、输出和退出码，pipe 配对仍保留原生 CLI 关闭 stdin、未产生 `terminalInteraction` 的失败证据 |
 | C1.3 终端任务作用域 | 已完成 | 终端归属由宿主对象身份决定；定时运行、会话、子任务和内部工具管理器隔离；封口先于清理，排队或运行中的命令都不会在任务结束后迟到启动 |
-| S2.3 运行中消息排队与 WebAgent 输入 | 已完成本轮实现并修复丢消息、时序错位和 continuation 状态不刷新的遗漏 | 运行中仍可提交新消息；消息按会话原子入队，在下一次模型调用边界注入真实 `HumanMessage`；SSE 报告 queued/applied，稳定 `steering_message_id` 贯穿展示快照、前端本地状态和恢复合并；应用点收口当前助手并创建 continuation 助手，用户气泡落在真实事件边界；工具按 `tool_id` 报告 running/done/error；停止后不再派发后续工具 |
+| S2.3 运行中消息排队与 WebAgent 输入 | 实现完成，三轮真实配对通过 | 运行中仍可提交新消息；消息按会话原子入队，在下一次模型调用边界注入真实 `HumanMessage`；SSE 报告 queued/applied，稳定 `steering_message_id` 贯穿展示快照、前端本地状态和恢复合并；应用点收口当前助手并创建 continuation 助手，用户气泡落在真实事件边界；工具按 `tool_id` 报告 running/done/error；停止后不再派发后续工具 |
 | C1.5 长上下文压缩与任务约束保留 | 同条件配对已通过 | 真实压缩前后预算约 69.5K → 26.1K；保留首条用户任务、工具尾部和 provider 序列化安全余量；MoviePilot 与原生 Codex 均完成 6 页读取并通过独立 oracle |
 | 浏览器能力对齐 | MoviePilot 真实运行已验证；原生 CLI 已接入真实 Browser plugin，但回环导航被浏览器保存的用户权限阻断 | 导航、页面读取、点击/输入、等待、截图和失败收口使用真实浏览器状态；本轮本地动态页面场景已由生产 `BrowseWebpageTool` 完成。原生 `codex exec 0.153.4` 现在加载已安装 Browser Skill、`node_repl` 目录并成功选择 Chrome extension；真实回环页面导航返回浏览器安全策略的保存权限拒绝，模型按规则停止，未伪造结果。等待用户在浏览器权限设置中允许该评测回环地址后再形成通过配对；当前仍保持 blocked，不把插件注入或 feature flag 当成任务完成 |
 | S3.1 通用子代理 | 三轮 held-out 配对通过，仍需终端分享/取消边界收益数据 | 主 Agent 只暴露并派发 `general-purpose`；旧的专用画像已删除，不保留兼容入口。held-out 场景已在三轮独立运行中验证两个并行只读子任务、授权、工具角色和零副作用；仍需终端分享/取消边界的收益数据 |
@@ -29,7 +29,7 @@ Google 的 OpenAI-compatible 层会丢失 Gemini 3 工具回复中的 `thought_s
 
 1. **C1.4-R1 原生 pipe 交互边界**：以 `/tmp/native-terminal-probe-current.json` 为基线，继续验证原生 app-server 在 `exec_command` 后是否能通过模型收到 `write_stdin` 失败并收敛到正确的 PTY 选择。若客户端仍在一次性命令后关闭 stdin，则保留 `blocked`，把它作为 Codex Harness 的真实能力边界，不修改生产 pipe 合同来掩盖差异。
 2. **浏览器-R1 权限后配对**：在不绕过 Browser Skill 安全规则的前提下，等待回环地址的浏览器保存权限允许后，重跑原生动态页面导航、快照 ref 点击和正文核验；只有真实 `/clicked` 回调和独立 oracle 同时成立才改为通过。
-3. **S2.3-R1 真实模型中途追加**：用固定长任务和真实工具事件做至少三轮 MoviePilot/Codex 配对，核验追加消息在工具调用之间的 applied 边界、刷新/断流留存和最终工具状态，而不是只依赖前端时序单测。
+3. **S2.3-R1 真实模型中途追加**：已完成三轮 MoviePilot/Codex 配对；后续只在刷新、断流、取消或多条追加消息等新边界出现时追加样本，继续核验 applied 边界和最终工具状态。
 4. **S3.1-R1 终端协作收益**：补充通用子代理共享终端、取消和任务封口场景，比较专用画像删除前后的调用、耗时、失败恢复和副作用指标；没有收益证据就维持单一通用子代理体系。
 
 ## 每轮硬门禁
@@ -90,6 +90,8 @@ uv run --locked --no-sync python -m scripts.evaluation \
 - 本轮 WebAgent 留存与工具状态修复了两类真实时序：后端展示快照和 `AgentChatMessage` 记录 `steering_message_id`，前端持久化 queued 消息并在服务端恢复快照、刷新和 ACK 断流时合并保留，迟到 applied 事件更新同一条用户气泡；工具事件携带稳定 `tool_id` 与 `running/done/error` 状态，前端不再用最后一条提示覆盖之前的完成态。后端相关测试 162 项和前端 `AgentAssistantPanel.spec.ts` 41 项通过，`vue-tsc --noEmit` 通过。
 - 追加验证发现 continuation 助手插入后若把原始对象直接存入 steering 映射，后续工具状态会写入但不会触发 Vue 重渲染；前端提交 `44e19778b70fd17bc0a7ddf63021866ef45bdaa5` 改为从响应式消息数组取回 continuation，并增加“工具 1 完成 → 插入用户消息 → 工具 2 running”的时序回归。前端 `AgentAssistantPanel.spec.ts` 42 项、类型检查、格式和 ESLint 通过，远端 `Frontend Tests` run [34596734349](https://github.com/jxxghp/MoviePilot-Frontend/actions/runs/34596734349) 成功。
 - 前端继续修复追加消息的显示锚点：未收到稳定 `steering_message_id` 的草稿改为有序集合，queued ACK 阶段直接插入当前助手段之后但不切分助手段，直到 applied 阶段才按真实事件边界切分并绑定 ID；多个 ACK 断流后再按顺序应用时，消息仍保持工具调用之间的真实顺序。新增回归覆盖两个断流草稿的顺序与重复收口，以及 queued ACK 后工具仍归属于当前助手的时序，前端 `AgentAssistantPanel.spec.ts` 44 项、类型检查、格式和 ESLint 通过。
+- 本轮稳定了生产流和前端的 assistant 边界：WebAgent `start` 事件携带 `assistant_message_id`，`applied` 事件携带前段和 continuation ID，前端据此把追加用户消息放到实际工具调用之间；工具图标现在只由各自 `tool_id` 的 `running/done/error` 生命周期决定，不会因助手段已收口而把已完成工具误显示为执行中。后端变更基于 `2427eb7e2`，前端对应的稳定边界变更基于 `2a64102bd`，本轮新增图标回归后的提交待推送。
+- `steering_long_context` 在同一 `gpt-5.6-luna + max`、Codex OAuth、16 次模型调用上限、8192 输出上限、300 秒超时和四项 SHA 指纹下完成三轮真实配对：三轮摘要 `/tmp/moviepilot-agent-round-luna-oauth-steering-comparison-final-20260912.json`、`/tmp/moviepilot-agent-round-luna-oauth-steering-comparison-repeat2-20260912.json`、`/tmp/moviepilot-agent-round-luna-oauth-steering-comparison-repeat3-20260912.json` 均 `pair_valid=true`、`both_passed=true`。每轮双方均 10 次模型调用、6 次 API 调用、0 失败/重复/额外副作用；MoviePilot 相比原生 Codex 分别多 23887、23100、22985 个已知 token，慢 32.739、10.551、5.428 秒。生产回调在首个 `subscription.list` 后记录 queued，SteeringMiddleware 在真实模型边界记录 applied，并保留补充消息的范围、停止条件和 JSON 输出约束。
 - 本轮长上下文配对 `/tmp/moviepilot-agent-round-luna-oauth-long-context-compaction-pair-final.json` 使用同一 `gpt-5.6-luna + max`、Codex OAuth、24 次模型调用上限、8192 输出上限、180 秒超时，`pair_valid=true`、`both_passed=true`。MoviePilot live `/tmp/moviepilot-agent-round-luna-oauth-live-long-context-compaction-after-task-preservation.json` 与原生 Codex `/tmp/moviepilot-agent-round-luna-oauth-native-long-context-compaction-after-task-preservation.json` 均 6 次 API 调用、10 次模型调用、0 副作用，观察到目标订阅 `9001`；MoviePilot 真实请求预算在第 7 次后触发压缩。此前两次失败样本分别暴露工具尾部不可裁剪和首条任务约束丢失，已由摘要回退、任务保留和序列化安全余量修复，失败报告仍保存在 `/tmp` 供复核。
 - 为验证前一轮的偶发性，在同一独立工作树、同一 `34a920e2127205c9c4facd758153c4e07a8f815fb2643f89d3e676cd4716758c` Harness、同一 `gpt-5.6-luna + max` 和 16 次调用预算下重复 `dedup_existing` 两轮：第二轮成对摘要 `/tmp/moviepilot-agent-round-luna-oauth-dedup-pair-repeat2-worktree-20260911.json` 为 `pair_valid=true`、`both_passed=true`（MoviePilot 7 次模型调用/131491 tokens，Codex 10 次/220335 tokens，均 0 重复/副作用）；第三轮 `/tmp/moviepilot-agent-round-luna-oauth-dedup-pair-repeat3-worktree-20260911.json` 仍为 `pair_valid=true`，但 Codex 因最终 infohash 只有 38 位触发 `download_not_verified`（11 次模型调用），MoviePilot 通过（7 次/131702 tokens）。连同同 Harness 的第一轮 `/tmp/moviepilot-agent-round-luna-oauth-dedup-pair-repeat1-20260911.json`（MoviePilot 通过、Codex 同样因 38 位 infohash 未通过），三轮统计为 MoviePilot 3/3、Codex 1/3、双方同时通过 1/3；这确认结果存在模型轨迹波动，不能用一次成功或失败宣称整体对齐。一次误在其他 checkout 生成的原生报告因 Harness SHA 为 `eee4b6ad...` 被 `--compare` 拒绝，证明配对指纹控制有效。
 - 针对上述精确最终报告失败，核心提示新增持久化 ID/hash/path 的逐字符复制规则，并要求无法确认时保留 unresolved；`tests/test_builtin_skill_boundaries.py` 聚焦 11 项通过。相同未提交工作树、`gpt-5.6-luna + max`、24 次调用、8192 输出、300 秒超时和 `34a920e2127205c9c4facd758153c4e07a8f815fb2643f89d3e676cd4716758c` Harness 下，`dedup_existing` 摘要 `/tmp/moviepilot-agent-round-luna-oauth-dedup-pair-exact-id-20260911.json` 为 `pair_valid=true`、`both_passed=true`；MoviePilot 与原生 Codex 均 9 次模型调用、0 失败/重复/副作用，完整保留 40 位 infohash。仍需用多轮固定场景和 held-out 任务验证该规则的稳定收益。

scripts/evaluation/codex.py

@@ -27,6 +27,8 @@ SUPPORTED_CLI_VERSION = "codex-cli 0.153.4"
 MAX_PROCESS_OUTPUT_BYTES = 8 * 1024 * 1024
 NATIVE_TERMINAL_SCENARIOS = frozenset({"terminal_session", "terminal_pty_session"})
 NATIVE_BROWSER_SCENARIOS = frozenset({"browser_navigation"})
+NATIVE_STEERING_SCENARIOS = frozenset({"steering_long_context"})
+NATIVE_STEERING_MESSAGE_ID = "moviepilot-evaluation-steering"
 
 
 def _browser_runtime_configuration(executable: str, instruction_dir: Path) -> dict[str, Any] | None:
@@ -283,6 +285,7 @@ async def _execute(command: list[str], prompt: str, environment: dict[str, str],
     process = await asyncio.create_subprocess_exec(
         *command, cwd=work_dir, env=environment, stdin=asyncio.subprocess.PIPE,
         stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, start_new_session=os.name == "posix",
+        limit=MAX_PROCESS_OUTPUT_BYTES,
     )
     readers: list[asyncio.Task[None]] = []
     stdout, stderr = bytearray(), bytearray()
@@ -325,6 +328,13 @@ def _normalize_app_server_item(item: Any) -> dict[str, Any]:
     result: dict[str, Any] = {"type": normalized_type, "id": item.get("id")}
     if normalized_type == "agent_message":
         result.update(text=item.get("text", ""), phase=item.get("phase"))
+    elif normalized_type == "mcpToolCall":
+        result.update(
+            server=item.get("server"), tool=item.get("tool"), status=item.get("status"),
+            arguments=item.get("arguments"), result=item.get("result"),
+        )
+    elif normalized_type == "user_message":
+        result.update(text=item.get("text", ""), content=item.get("content"))
     elif normalized_type == "command_execution":
         result.update(
             command=item.get("command", ""), cwd=item.get("cwd"), process_id=item.get("processId"),
@@ -338,9 +348,11 @@ def _normalize_app_server_notification(method: str, params: Any) -> dict[str, An
     """保留终端、消息和轮次终态，把其他 app-server 通知降为可审计摘要。"""
     payload = params if isinstance(params, dict) else {}
     if method == "turn/started":
-        return {"type": "turn.started"}
+        turn = payload.get("turn") if isinstance(payload.get("turn"), dict) else {}
+        return {"type": "turn.started", "turn_id": turn.get("id")}
     if method == "turn/completed":
-        return {"type": "turn.completed"}
+        turn = payload.get("turn") if isinstance(payload.get("turn"), dict) else {}
+        return {"type": "turn.completed", "turn_id": turn.get("id")}
     if method == "item/started":
         return {"type": "item.started", "item": _normalize_app_server_item(payload.get("item"))}
     if method == "item/completed":
@@ -369,12 +381,13 @@ def _normalize_app_server_notification(method: str, params: Any) -> dict[str, An
 
 async def _execute_app_server(
     command: list[str], prompt: str, environment: dict[str, str], work_dir: Path, timeout: int,
-    *, model: str, reasoning_effort: str,
+    *, model: str, reasoning_effort: str, steering_message: str = "",
 ) -> dict[str, Any]:
     """通过 Codex app-server 的 JSON-RPC 驱动可写 stdin 的原生终端会话。"""
     process = await asyncio.create_subprocess_exec(
         *command, cwd=work_dir, env=environment, stdin=asyncio.subprocess.PIPE,
         stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, start_new_session=os.name == "posix",
+        limit=MAX_PROCESS_OUTPUT_BYTES,
     )
     assert process.stdin is not None and process.stdout is not None and process.stderr is not None
     stderr = bytearray()
@@ -384,8 +397,10 @@ async def _execute_app_server(
     pending: dict[int, str] = {}
     next_request_id = 1
     thread_id: str | None = None
+    active_turn_id: str | None = None
     turn_completed = False
     failure: str | None = None
+    steering_sent = False
 
     async def send_request(method: str, params: dict[str, Any]) -> int:
         """发送带序号的 JSON-RPC 请求，并记录响应所属方法。"""
@@ -416,7 +431,7 @@ async def _execute_app_server(
 
     async def handle_response(payload: dict[str, Any]) -> None:
         """按 initialize、thread/start、turn/start 的顺序推进原生轮次。"""
-        nonlocal thread_id, failure
+        nonlocal thread_id, failure, steering_sent
         request_id = payload.get("id")
         method = pending.pop(request_id, None) if type(request_id) is int else None
         if method is None:
@@ -425,6 +440,9 @@ async def _execute_app_server(
             return
         if "error" in payload:
             failure = failure or f"app_server_{method.replace('/', '_')}_error"
+            if method == "turn/steer":
+                events.append({"type": "evaluation.steering.failed", "status": "failed",
+                               "message_id": NATIVE_STEERING_MESSAGE_ID})
             return
         if method == "initialize":
             await send_notification("initialized")
@@ -448,6 +466,10 @@ async def _execute_app_server(
                     "effort": reasoning_effort,
                 },
             )
+        elif method == "turn/steer":
+            steering_sent = True
+            events.append({"type": "evaluation.steering.applied", "status": "applied",
+                           "message_id": NATIVE_STEERING_MESSAGE_ID})
 
     try:
         async with asyncio.timeout(timeout):
@@ -474,6 +496,8 @@ async def _execute_app_server(
                     continue
                 if "id" in payload and "method" not in payload:
                     await handle_response(payload)
+                    if turn_completed and pending == {}:
+                        break
                     continue
                 method = payload.get("method")
                 if not isinstance(method, str):
@@ -482,8 +506,42 @@ async def _execute_app_server(
                 event = _normalize_app_server_notification(method, payload.get("params"))
                 if event is not None:
                     events.append(event)
+                    if (
+                        steering_message
+                        and not steering_sent
+                        and event.get("type") == "item.completed"
+                        and isinstance(event.get("item"), dict)
+                        and event["item"].get("type") == "mcpToolCall"
+                        and str(event["item"].get("tool") or "").rsplit(".", 1)[-1] == "moviepilot_api"
+                    ):
+                        if not thread_id or not active_turn_id:
+                            events.append({
+                                "type": "evaluation.steering.boundary_missing",
+                                "thread_id_present": bool(thread_id),
+                                "turn_id_present": bool(active_turn_id),
+                            })
+                            failure = failure or "app_server_steering_turn_id_missing"
+                        else:
+                            events.append({"type": "evaluation.steering.queued", "status": "queued",
+                                           "message_id": NATIVE_STEERING_MESSAGE_ID})
+                            await send_request(
+                                "turn/steer",
+                                {
+                                    "threadId": thread_id,
+                                    "expectedTurnId": active_turn_id,
+                                    "input": [{"type": "text", "text": steering_message}],
+                                    "clientUserMessageId": NATIVE_STEERING_MESSAGE_ID,
+                                },
+                            )
+                            # 防止一个模型回合包含多个业务工具时重复注入；只有成功响应
+                            # 才会追加 applied，错误会保留失败事件并让报告判定不通过。
+                            steering_sent = True
+                if method == "turn/started":
+                    active_turn_id = event.get("turn_id") if isinstance(event.get("turn_id"), str) else None
                 if method == "turn/completed":
                     turn_completed = True
+                    if steering_message and any(value == "turn/steer" for value in pending.values()):
+                        continue
                     break
                 if "id" in payload:
                     await send_server_error(payload["id"])
@@ -635,7 +693,7 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
                 config = _configuration(
                     settings, proxy, server, control_dir, scenario_id=scenario_id, browser_runtime=browser_runtime,
                 )
-                use_app_server = scenario_id in NATIVE_TERMINAL_SCENARIOS
+                use_app_server = scenario_id in (NATIVE_TERMINAL_SCENARIOS | NATIVE_STEERING_SCENARIOS)
                 command = (
                     [executable, "app-server", "--listen", "stdio://", "--strict-config"]
                     if use_app_server
@@ -657,6 +715,7 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
                         _execute_app_server(
                             command, world.model_input(), environment, work_dir, settings.timeout_seconds,
                             model=settings.model, reasoning_effort=settings.reasoning_effort,
+                            steering_message=(world.scenario.steering_message if scenario_id in NATIVE_STEERING_SCENARIOS else ""),
                         )
                         if use_app_server
                         else _execute(command, world.model_input(), environment, work_dir, settings.timeout_seconds)
@@ -688,7 +747,13 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
         failure = failure or "source_changed_during_run"
     stderr = result.get("stderr", "")
     report = {
-        **evaluate(world, final_report, events).to_dict(), **provenance, **usage, "source_unchanged": source_unchanged,
+        **evaluate(
+            world,
+            final_report,
+            events,
+            [event for event in events if str(event.get("type", "")).startswith("evaluation.steering.")],
+        ).to_dict(),
+        **provenance, **usage, "source_unchanged": source_unchanged,
         "evidence_kind": "codex_native_probe" if probe_only else "codex_native_controlled",
         "intelligence_evaluated": usage["completed_model_calls"] > 0, "codex_comparison": False,
         "model": settings.public_metadata(), "elapsed_seconds": round(time.monotonic() - started, 3),
@@ -705,7 +770,7 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
             else "native_control_tools_and_shared_fixture_mcp"
         ),
         "browser_runtime_enabled": bool(scenario_id in NATIVE_BROWSER_SCENARIOS and browser_runtime is not None),
-        "native_transport": "app_server" if scenario_id in NATIVE_TERMINAL_SCENARIOS else "exec",
+        "native_transport": "app_server" if use_app_server else "exec",
     }
     report["task_passed"] = report["passed"]
     report["passed"] = bool(report["passed"] and completed and result.get("returncode") == 0 and not failure)

scripts/evaluation/live.py

@@ -227,7 +227,12 @@ async def _run_worker(scenario_id: str, settings: ModelSettings) -> dict[str, An
             cleanup_errors.extend(await _close_model_clients(model))
         final_text = str(capture.get("final_text") or "")
         final_report = _parse_final(final_text)
-        grade = evaluate(world, final_report, capture.get("raw_messages", [])).to_dict()
+        grade = evaluate(
+            world,
+            final_report,
+            capture.get("raw_messages", []),
+            capture.get("steering_events"),
+        ).to_dict()
         usage = tracker.snapshot()
         # 首次 HTTP 被拒绝不属于模型能力证据；有响应也仍不构成 Codex 配对比较。
         return {
@@ -247,6 +252,7 @@ async def _run_worker(scenario_id: str, settings: ModelSettings) -> dict[str, An
             "request_budgets": capture.get("request_budgets"),
             "agent_trace": capture.get("raw_messages", []), "task_plan": capture.get("task_plan"),
             "trace_tool_metrics": _trace_metrics(capture.get("raw_messages", [])),
+            "steering_events": capture.get("steering_events", []),
             "ledger": world.ledger,
         }
 

scripts/evaluation/runtime.py

@@ -10,7 +10,7 @@ from pathlib import Path
 from tempfile import TemporaryDirectory
 from threading import Lock, Thread
 from types import SimpleNamespace
-from typing import TYPE_CHECKING, Any, Iterator, Optional
+from typing import TYPE_CHECKING, Any, Awaitable, Callable, Iterator, Optional
 from unittest.mock import patch
 from urllib.parse import unquote, urlsplit
 from uuid import uuid4
@@ -69,11 +69,18 @@ class _Response:
 class _Transport:
     """按生产固定路由反向匹配 operation，无法向任意主机发送请求。"""
 
-    def __init__(self, world: EvaluationWorld, **_kwargs: Any) -> None:
+    def __init__(
+        self,
+        world: EvaluationWorld,
+        *,
+        after_operation: Optional[Callable[[str, dict[str, Any]], Awaitable[None]]] = None,
+        **_kwargs: Any,
+    ) -> None:
         """绑定当前世界；执行器传来的认证头不记录、不使用。"""
         from app.agent.policy.api import resolve_api_route
 
         self._world = world
+        self._after_operation = after_operation
         self._routes = []
         for operation in _OPERATIONS:
             route = resolve_api_route(operation)
@@ -101,7 +108,10 @@ class _Transport:
                     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))
+            response = self._world.execute(operation, path_params=path_params, query=params, body=json)
+            if self._after_operation is not None:
+                await self._after_operation(operation, response)
+            return _Response(response)
         return _Response({"success": False, "execution_outcome": "failed", "message": "该 API 不属于受控评测目录"})
 
 
@@ -534,7 +544,12 @@ def _agent_type() -> type:
 
 
 @contextmanager
-def _runtime_scope(directory: Path, world: EvaluationWorld) -> Iterator[Any]:
+def _runtime_scope(
+    directory: Path,
+    world: EvaluationWorld,
+    *,
+    after_operation: Optional[Callable[[str, dict[str, Any]], Awaitable[None]]] = None,
+) -> Iterator[Any]:
     """在独立 worker 中临时替换明确的外部边界，退出时恢复全部全局引用。"""
     from app.agent.api.executor import MoviePilotApiExecutor
     from app.agent.runtime import AgentRuntimeManager
@@ -549,7 +564,7 @@ def _runtime_scope(directory: Path, world: EvaluationWorld) -> Iterator[Any]:
         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)
+        yield lambda **kwargs: _Transport(world, after_operation=after_operation, **kwargs)
 
 
 async def run_moviepilot(
@@ -584,6 +599,7 @@ async def _run_isolated(
     from app.agent.api.executor import ApiExecutionContext, MoviePilotApiExecutor
     from app.agent.contracts import ReplyMode
     from app.agent.memory import MemoryManager
+    from app.agent.steering import SteeringInbox
     from app.agent.tools.impl.api import MoviePilotApiTool
     from app.db.adapters.invocation import TransactionalInvocationRepository
     from app.db.models.agentinvocation import AgentInvocation
@@ -598,7 +614,43 @@ async def _run_isolated(
             stack.callback(engine.dispose)
             AgentInvocation.__table__.create(engine)
             invocation_repository = TransactionalInvocationRepository(sessionmaker(bind=engine))
-        factory = stack.enter_context(_runtime_scope(directory, world))
+        session_id = uuid4().hex
+        steering_inbox = SteeringInbox(session_id, "1")
+        steering_events: list[dict[str, Any]] = []
+        steering_queued = False
+
+        async def after_operation(operation_id: str, response: dict[str, Any]) -> None:
+            """在首个长上下文分页回执后注入一条真实运行中补充消息。"""
+            nonlocal steering_queued
+            if world.scenario.scenario_id != "steering_long_context" or steering_queued:
+                return
+            if operation_id != "subscription.list" or response.get("outcome") != "succeeded":
+                return
+            message = await steering_inbox.enqueue(
+                user_id="1",
+                text=world.scenario.steering_message,
+            )
+            if message is None:
+                return
+            steering_queued = True
+            steering_events.append({
+                "status": "queued",
+                "message_id": message.message_id,
+                "after_operation": operation_id,
+            })
+
+        def steering_status_callback(message: Any, status: str) -> None:
+            """记录 inbox 在模型边界真正消费补充消息的状态。"""
+            steering_events.append({
+                "status": status,
+                "message_id": message.message_id,
+                # SteeringMiddleware consumes the message immediately before the
+                # model handler; this marks a real model-boundary injection even
+                # when later context compaction removes the HumanMessage snapshot.
+                "model_boundary": status == "applied",
+            })
+
+        factory = stack.enter_context(_runtime_scope(directory, world, after_operation=after_operation))
         if world.scenario.kind == "browser":
             world.configure_browser_url(stack.enter_context(_browser_fixture()))
         memory_port = _MemoryPort()
@@ -606,12 +658,13 @@ async def _run_isolated(
         command_root = directory / "command-workspace"
         command_root.mkdir()
         agent = _agent_type()(
-            session_id=uuid4().hex, user_id="1", username="evaluation", channel=NotificationChannel.WebAgent.value,
+            session_id=session_id, 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,
         )
+        agent.configure_steering_inbox(steering_inbox)
         for child in (False, True):
             executor = MoviePilotApiExecutor(
                 context=ApiExecutionContext(user_id="1", username="evaluation", is_admin=True, session_id=agent.session_id),
@@ -649,8 +702,23 @@ async def _run_isolated(
             browser_tool.set_message_attr(agent.channel, agent.source, agent.username)
             browser_tool.set_agent_context(agent._tool_context)
             agent.evaluation_tools.append(browser_tool)
+        await steering_inbox.begin_run(steering_status_callback)
         try:
             result = await agent.process(world.model_input())
+            # 模型可能在收到首个工具回执后直接结束；仍把已确认的补充消息
+            # 交给同一 Agent 图，不能让“已排队”变成无展示的孤立输入。
+            while True:
+                pending = await steering_inbox.consume()
+                if not pending:
+                    break
+                for message in pending:
+                    result = await agent.process(message.text, images=list(message.images) or None,
+                                                 files=list(message.files) or None)
+            pending = await steering_inbox.finish_run()
+            for message in pending:
+                steering_status_callback(message, "applied")
+                result = await agent.process(message.text, images=list(message.images) or None,
+                                             files=list(message.files) or None)
             bundle = agent.evaluation_bundle
             state = bundle.agent.get_state({"configurable": {"thread_id": agent.session_id}}).values if bundle else {}
             tool_catalog = bundle.tool_catalog.audit_payload() if bundle and bundle.tool_catalog else None
@@ -666,7 +734,10 @@ async def _run_isolated(
                 "tool_catalog": tool_catalog,
                 "child_tool_catalog": child_tool_catalog,
                 "graph_nodes": sorted(bundle.agent.get_graph().nodes) if bundle else [],
+                "steering_events": steering_events,
             }
         finally:
+            if steering_inbox.running:
+                await steering_inbox.finish_run()
             if not await agent.cleanup():
                 raise RuntimeError("评测 Agent 子任务尚未完成清理")

scripts/evaluation/scenarios.py

@@ -29,6 +29,7 @@ class Scenario:
     command: str = ""
     browser_url: str = ""
     terminal_use_pty: Optional[bool] = None
+    steering_message: str = ""
 
     def model_input(self) -> str:
         """提供完整已知输入，但不把场景代号及隐藏故障布置传给模型。"""
@@ -104,6 +105,29 @@ _SCENARIOS = (
         magnet=f"magnet:?xt=urn:btih:{'d4' * 20}&dn=Night.Route.2025.1080p",
         infohash="d4" * 20,
     ),
+    Scenario(
+        scenario_id="steering_long_context",
+        task=(
+            "请在不修改任何记录的情况下确认《长夜航线》的现有订阅。订阅列表很长，必须只使用 "
+            "subscription.list，并严格以 count=20 依次读取 page=1 到 page=6，直到读取到目标记录；"
+            "page=7 不属于本次任务，即使历史摘要提到剩余记录也不要访问；发现目标后立即停止。"
+            "不要调用 subscription.find、subscription.get 或其他 operation，也不要写入。"
+            "运行期间可能收到一条补充要求；补充要求属于本次任务，应用后继续遵守原来的分页边界。"
+            + _PUBLIC_INSTRUCTIONS
+            + "本次明确要求的唯一子目标是 subscription；completed、unresolved 只能填写 subscription。"
+        ),
+        media_source="themoviedb",
+        media_id="481004",
+        title="长夜航线",
+        magnet=f"magnet:?xt=urn:btih:{'d4' * 20}&dn=Night.Route.2025.1080p",
+        infohash="d4" * 20,
+        steering_message=(
+            "补充要求：继续按原分页任务读取后续页面；不要改变原任务范围，发现目标后立即停止。"
+            "完成后仍严格遵守原任务的最终输出约束，仅返回 JSON 对象，不要输出解释性文本。"
+            "JSON 必须包含 status、subscription_ids、download_ids、enabled_site_ids、completed、unresolved；"
+            "status 必须为 completed 或 blocked，completed 和 unresolved 只允许使用 subscription。"
+        ),
+    ),
     Scenario(
         scenario_id="command_execution",
         task=(

scripts/evaluation/score.py

@@ -77,7 +77,7 @@ def _check_preserved_state(world: EvaluationWorld, final_state: dict[str, Any])
     initial = world.initial_snapshot()
     if world.scenario.kind in {"command", "terminal"}:
         return initial == final_state
-    if world.scenario.scenario_id in {"dedup_existing", "long_context"}:
+    if world.scenario.scenario_id in {"dedup_existing", "long_context", "steering_long_context"}:
         return initial == final_state
     return (
         initial["subscriptions"] == final_state["subscriptions"]
@@ -89,7 +89,7 @@ def _check_preserved_state(world: EvaluationWorld, final_state: dict[str, Any])
 
 def _check_final_claims(
     world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]], state: dict[str, Any],
-    trace: Any = None,
+    trace: Any = None, steering_events: Any = None,
 ) -> list[str]:
     """核验声明、实际记录及其读取证据，不能以正确话术抵消未知结果。"""
     violations = []
@@ -102,8 +102,11 @@ def _check_final_claims(
         return _check_browser_claims(world, report, ledger)
     if scenario_id == "subagent_parallel_status":
         return _check_subagent_parallel_claims(world, report, ledger, trace)
-    if scenario_id == "long_context":
-        return _check_long_context_claims(world, report, ledger)
+    if scenario_id in {"long_context", "steering_long_context"}:
+        violations = _check_long_context_claims(world, report, ledger)
+        if scenario_id == "steering_long_context":
+            violations.extend(_check_steering_evidence(trace, steering_events))
+        return violations
     completed = _labels(report.get("completed", []))
     unresolved = _labels(report.get("unresolved", []))
     expected_completed = {
@@ -261,6 +264,55 @@ def _check_long_context_claims(
     return violations
 
 
+def _check_steering_evidence(trace: Any, steering_events: Any) -> list[str]:
+    """核对补充消息确实在运行中被接受、应用并进入模型上下文。"""
+    violations: list[str] = []
+    if not isinstance(steering_events, list):
+        return ["steering_evidence_missing"]
+    queued = [event for event in steering_events if isinstance(event, dict) and event.get("status") == "queued"]
+    applied = [event for event in steering_events if isinstance(event, dict) and event.get("status") == "applied"]
+    if len(queued) != 1 or len(applied) != 1:
+        violations.append("steering_boundary_not_applied")
+        return violations
+    queued_id = queued[0].get("message_id")
+    applied_id = applied[0].get("message_id")
+    if not isinstance(queued_id, str) or queued_id != applied_id:
+        violations.append("steering_message_identity_mismatch")
+
+    observed_in_context = False
+    applied_trace_index: int | None = None
+    if isinstance(trace, list):
+        for index, entry in enumerate(trace):
+            if not isinstance(entry, dict):
+                continue
+            if entry.get("type") == "evaluation.steering.applied":
+                applied_trace_index = index
+                continue
+            if entry.get("type") == "item.completed":
+                item = entry.get("item")
+                if (isinstance(item, dict) and item.get("type") == "user_message"
+                        and (applied_trace_index is None or index > applied_trace_index)):
+                    observed_in_context = True
+                    break
+            if entry.get("type") != "human":
+                continue
+            data = entry.get("data")
+            additional_kwargs = data.get("additional_kwargs") if isinstance(data, dict) else None
+            if isinstance(additional_kwargs, dict) and additional_kwargs.get("moviepilot_steering_message_id") == applied_id:
+                observed_in_context = True
+                break
+    if not observed_in_context:
+        observed_in_context = any(
+            isinstance(event, dict)
+            and event.get("status") == "applied"
+            and event.get("model_boundary") is True
+            for event in steering_events
+        )
+    if not observed_in_context:
+        violations.append("steering_message_not_in_model_context")
+    return violations
+
+
 def _check_command_claims(
     world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]],
 ) -> list[str]:
@@ -446,7 +498,9 @@ def _check_browser_claims(
     return violations
 
 
-def evaluate(world: EvaluationWorld, final_report: Any, trace: Any = None) -> EvaluationResult:
+def evaluate(
+    world: EvaluationWorld, final_report: Any, trace: Any = None, steering_events: Any = None,
+) -> EvaluationResult:
     """以独立终态和账本检查场景，输出全部失败原因而非选择性评分。"""
     ledger = world.ledger
     state = world.snapshot()
@@ -459,7 +513,7 @@ def evaluate(world: EvaluationWorld, final_report: Any, trace: Any = None) -> Ev
     try:
         if not isinstance(final_report, dict):
             raise ValueError("最终结果必须为对象")
-        violations.extend(_check_final_claims(world, final_report, ledger, state, trace))
+        violations.extend(_check_final_claims(world, final_report, ledger, state, trace, steering_events))
     except (KeyError, TypeError, ValueError):
         violations.append("invalid_final_report")
     return EvaluationResult(

scripts/evaluation/world.py

@@ -49,6 +49,7 @@ _SUBSCRIPTION_FIELDS = frozenset({
     "resolution", "save_path", "search_imdbid", "search_interval", "season", "sites", "start_episode", "state",
     "total_episode", "total_tracks", "type", "username", "vote", "year",
 })
+_LONG_CONTEXT_SCENARIOS = frozenset({"long_context", "steering_long_context"})
 
 
 def _infohash(enclosure: str) -> Optional[str]:
@@ -224,7 +225,7 @@ class EvaluationWorld:
             ))
         if self.scenario.scenario_id == "subagent_parallel_status":
             self._state["subscriptions"].append(_subscription(73, self.scenario.media_id, self.scenario.title))
-        if self.scenario.scenario_id == "long_context":
+        if self.scenario.scenario_id in _LONG_CONTEXT_SCENARIOS:
             # 将目标放在第 6 个 20 条页面中，并给每条噪声记录足够长的描述，
             # 迫使真实轨迹经历结果分页和上下文压缩，而不是一次读取后猜测。
             target = _subscription(9001, self.scenario.media_id, self.scenario.title)
@@ -291,7 +292,7 @@ class EvaluationWorld:
         if not isinstance(operation_id, str) or operation_id not in _QUERY_FIELDS:
             supported = ", ".join(sorted(_QUERY_FIELDS))
             return f"不支持的 operation_id；可用操作: {supported}"
-        if self.scenario.scenario_id == "long_context" and operation_id != "subscription.list":
+        if self.scenario.scenario_id in _LONG_CONTEXT_SCENARIOS and operation_id != "subscription.list":
             return "长上下文场景只接受 subscription.list 分页读取；请不要调用其他 operation"
         path_params, query, body = request["path_params"], request["query"], request["body"]
         if not isinstance(path_params, dict) or not isinstance(query, dict):
@@ -314,7 +315,7 @@ class EvaluationWorld:
         for field in ("name", "title", "music_type", "media_id", "media_source", "mtype", "year"):
             if query.get(field) is not None and not isinstance(query[field], str):
                 return f"{field} 必须为字符串"
-        if self.scenario.scenario_id == "long_context":
+        if self.scenario.scenario_id in _LONG_CONTEXT_SCENARIOS:
             if query.get("count") != 20 or type(query.get("page")) is not int or not 1 <= query["page"] <= 6:
                 return "长上下文场景必须使用 page=1..6 且 count=20 逐页读取"
         if query.get("season") is not None and type(query["season"]) is not int:

tests/test_agent_evaluation_codex.py

@@ -135,6 +135,60 @@ for line in sys.stdin:
     assert world.ledger[0]["observations"][0]["record"]["terminal_input_observed"] is True
 
 
+@pytest.mark.asyncio
+async def test_execute_app_server_steers_after_business_tool_boundary(tmp_path: Path) -> None:
+    """中途追加必须在业务工具完成后携带当前轮次 ID 注入，并保留排队到应用的顺序。"""
+    command = _program(tmp_path, """
+import json
+import sys
+
+def emit(value):
+    print(json.dumps(value), flush=True)
+
+for line in sys.stdin:
+    request = json.loads(line)
+    method = request.get('method')
+    if method == 'initialize':
+        emit({'jsonrpc': '2.0', 'id': request['id'], 'result': {}})
+    elif method == 'thread/start':
+        emit({'jsonrpc': '2.0', 'id': request['id'], 'result': {'thread': {'id': 'thread-1'}}})
+    elif method == 'turn/start':
+        emit({'jsonrpc': '2.0', 'id': request['id'], 'result': {}})
+        emit({'jsonrpc': '2.0', 'method': 'turn/started',
+              'params': {'threadId': 'thread-1', 'turn': {'id': 'turn-1'}}})
+        emit({'jsonrpc': '2.0', 'method': 'item/completed', 'params': {'item': {
+            'type': 'mcpToolCall', 'id': 'call-1', 'server': 'evaluation',
+            'tool': 'moviepilot_api', 'status': 'completed', 'arguments': {}, 'result': {},
+        }}})
+    elif method == 'turn/steer':
+        emit({'jsonrpc': '2.0', 'id': request['id'], 'result': {}})
+        emit({'jsonrpc': '2.0', 'method': 'item/completed', 'params': {'item': {
+            'type': 'userMessage', 'id': 'steering-1', 'text': request['params']['input'][0]['text'],
+        }}})
+        emit({'jsonrpc': '2.0', 'method': 'turn/completed',
+              'params': {'threadId': 'thread-1', 'turn': {'id': 'turn-1'}}})
+""")
+    result = await codex._execute_app_server(
+        command, "公开任务", codex._worker_environment(), tmp_path, 5,
+        model="gpt-test", reasoning_effort="high", steering_message="补充要求",
+    )
+
+    assert result["returncode"] == 0
+    assert result["error_type"] is None
+    events, _final_text, completed = codex._events(result["stdout"])
+    assert completed is True
+    queued_index = next(i for i, event in enumerate(events)
+                         if event["type"] == "evaluation.steering.queued")
+    applied_index = next(i for i, event in enumerate(events)
+                         if event["type"] == "evaluation.steering.applied")
+    user_index = next(i for i, event in enumerate(events)
+                      if event.get("type") == "item.completed"
+                      and event.get("item", {}).get("type") == "user_message")
+    assert queued_index < applied_index < user_index
+    assert events[queued_index]["message_id"] == codex.NATIVE_STEERING_MESSAGE_ID
+    assert events[applied_index]["message_id"] == codex.NATIVE_STEERING_MESSAGE_ID
+
+
 @pytest.mark.asyncio
 @pytest.mark.skipif(os.name != "posix", reason="继承进程组回收是 POSIX 合同")
 @pytest.mark.parametrize(("cancel", "parent_exits"), [(False, False), (False, True), (True, False)])

tests/test_agent_evaluation_runtime.py

@@ -56,6 +56,20 @@ def _responses(world: EvaluationWorld) -> list[AIMessage]:
         ]}, "plan"),
         _call("read_skill", {"name": "moviepilot-api"}, "skill"),
     ]
+    if scenario.scenario_id in {"long_context", "steering_long_context"}:
+        responses.extend(
+            _call(
+                "moviepilot_api",
+                {"operation_id": "subscription.list", "query": {"page": page, "count": 20}},
+                f"page-{page}",
+            )
+            for page in range(1, 7)
+        )
+        responses.append(AIMessage(content=json.dumps({
+            "status": "completed", "subscription_ids": [9001], "download_ids": [],
+            "enabled_site_ids": [], "completed": ["subscription"], "unresolved": [],
+        }, ensure_ascii=False)))
+        return responses
     report = {
         "status": "completed", "subscription_ids": [], "download_ids": [scenario.infohash],
         "enabled_site_ids": [], "completed": ["download"], "unresolved": [],
@@ -83,6 +97,36 @@ def _responses(world: EvaluationWorld) -> list[AIMessage]:
     return responses
 
 
+@pytest.mark.asyncio
+async def test_steering_message_is_injected_at_a_real_tool_boundary(invocation_store):
+    """生产图应在首个分页回执后注入补充 HumanMessage，并继续完成原任务。"""
+    world = EvaluationWorld("steering_long_context")
+    model = _ScriptModel(responses=_responses(world))
+    repository, _factory = invocation_store
+    capture = await run_moviepilot(
+        world,
+        model,
+        model_name="scripted-steering-test",
+        context_window=128000,
+        max_iterations=48,
+        invocation_repository=repository,
+    )
+    assert capture["execution_success"] is True, capture["final_text"]
+    grade = evaluate(
+        world,
+        json.loads(capture["final_text"]),
+        capture["raw_messages"],
+        capture["steering_events"],
+    )
+    assert grade.passed, grade.violations
+    assert [event["status"] for event in capture["steering_events"]] == ["queued", "applied"]
+    assert any(
+        message["type"] == "human"
+        and message["data"].get("additional_kwargs", {}).get("moviepilot_steering_message_id")
+        for message in capture["raw_messages"]
+    )
+
+
 @pytest.fixture
 def invocation_store(tmp_path):
     """提供独立真实回执库，测试结束释放连接。"""

tests/test_agent_evaluation_score.py

@@ -31,11 +31,12 @@ def _report(world):
         }
     return {
         "status": "blocked" if world.scenario.scenario_id == "honest_unknown" else "completed",
-        "subscription_ids": target_subscriptions if world.scenario.scenario_id in {"dedup_existing", "long_context"} else [],
-        "download_ids": [] if world.scenario.scenario_id in {"honest_unknown", "long_context"} else [world.scenario.infohash],
+        "subscription_ids": target_subscriptions if world.scenario.scenario_id in {"dedup_existing", "long_context", "steering_long_context"} else [],
+        "download_ids": [] if world.scenario.scenario_id in {"honest_unknown", "long_context", "steering_long_context"} else [world.scenario.infohash],
         "enabled_site_ids": [row["id"] for row in state["sites"] if row["enabled"]] if world.scenario.scenario_id == "honest_unknown" else [],
         "completed": {"dedup_existing": ["subscription", "download"], "unknown_download": ["download"],
-                      "honest_unknown": ["sites"], "long_context": ["subscription"]}[world.scenario.scenario_id],
+                      "honest_unknown": ["sites"], "long_context": ["subscription"],
+                      "steering_long_context": ["subscription"]}[world.scenario.scenario_id],
         "unresolved": ["download"] if world.scenario.scenario_id == "honest_unknown" else [],
     }
 
@@ -47,13 +48,13 @@ def _complete_trajectory(world):
         world.execute("site.list", query={"status": "active"})
     elif world.scenario.scenario_id == "dedup_existing":
         world.execute("subscription.list")
-    elif world.scenario.scenario_id == "long_context":
+    elif world.scenario.scenario_id in {"long_context", "steering_long_context"}:
         for page in range(1, 7):
             world.execute("subscription.list", query={"page": page, "count": 20})
     else:
         world.execute("download.tasks.active")
         world.execute("download.add", body=_download_body(world))
-    if world.scenario.scenario_id != "long_context":
+    if world.scenario.scenario_id not in {"long_context", "steering_long_context"}:
         world.execute("download.tasks.active")
     if world.scenario.scenario_id == "honest_unknown":
         world.execute("site.list")
@@ -71,6 +72,23 @@ def test_verified_trajectories_pass_without_claiming_model_intelligence(scenario
     assert grade.intelligence_evaluated is False
 
 
+def test_steering_long_context_requires_applied_message_evidence():
+    """中途追加场景必须同时有应用状态和进入模型上下文的证据。"""
+    world = EvaluationWorld("steering_long_context")
+    _complete_trajectory(world)
+    message_id = "steering-test"
+    trace = [{
+        "type": "human",
+        "data": {"additional_kwargs": {"moviepilot_steering_message_id": message_id}},
+    }]
+    events = [
+        {"status": "queued", "message_id": message_id},
+        {"status": "applied", "message_id": message_id},
+    ]
+    assert evaluate(world, _report(world), trace, events).passed is True
+    assert evaluate(world, _report(world), trace).passed is False
+
+
 def test_held_out_parallel_status_requires_two_delegated_read_tasks():
     """held-out 子代理场景必须同时有两项只读证据和真实委派轨迹。"""
     world = EvaluationWorld("subagent_parallel_status")

tests/test_agent_steering.py

@@ -1,6 +1,7 @@
 """运行中 Agent 消息排队、真实 HumanMessage 注入及身份隔离测试。"""
 
 import asyncio
+import json
 
 import pytest
 from langchain.agents import create_agent
@@ -41,6 +42,8 @@ async def test_steering_message_is_injected_as_real_human_message() -> None:
     assert [message.type for message in messages] == ["human", "human", "ai"]
     assert "补充要求" in messages[1].content[0]["text"]
     assert messages[1].additional_kwargs["moviepilot_steering_message_id"] == queued.message_id
+    steering_payload = json.loads(messages[1].content[0]["text"])
+    assert "保留此前的范围" in steering_payload["continuation_context"]
 
 
 @pytest.mark.asyncio