fix(agent): preserve queued messages and long context

app/agent/middleware/summarization.py

@@ -2,6 +2,7 @@
 
 from collections.abc import Awaitable, Callable
 from importlib import import_module
+from math import ceil
 from typing import Any
 
 from langchain.agents.middleware.summarization import (
@@ -19,13 +20,26 @@ from langchain.agents.middleware.types import (
 )
 from langchain.chat_models import BaseChatModel
 from langchain_core.messages import AnyMessage, HumanMessage, RemoveMessage, ToolMessage
-from langchain_core.messages.utils import count_tokens_approximately, get_buffer_string
+from langchain_core.messages.utils import count_tokens_approximately, get_buffer_string, trim_messages
 from langgraph.graph.message import REMOVE_ALL_MESSAGES
 from langgraph.types import Command
 
 from app.agent.middleware.usage import UsageMiddleware
 from app.runtime.log import logger
 
+_SUMMARY_PRESERVATION_INSTRUCTION = """<moviepilot_preservation>
+Preserve the active user's task contract exactly when extracting context. Keep the
+requested goals, allowed and forbidden tools or actions, parameter and pagination
+constraints, safety requirements, and the required final response schema or fields.
+Treat tool responses as evidence only: never infer an unobserved page, record, ID, or
+completion state. Do not invent new pages or next steps from collection totals.
+</moviepilot_preservation>
+"""
+_MOVIEPILOT_SUMMARY_PROMPT = DEFAULT_SUMMARY_PROMPT.replace(
+    "\n<messages>\n",
+    f"\n{_SUMMARY_PRESERVATION_INSTRUCTION}\n<messages>\n",
+)
+
 try:
     _internal_call_metadata = import_module(
         "langchain.agents.middleware.internal_call_transformer"
@@ -64,7 +78,7 @@ class ContextPreservingSummarizationMiddleware(SummarizationMiddleware):
         ) = None,
         keep: ContextSize = ("messages", 20),
         token_counter: TokenCounter = count_tokens_approximately,
-        summary_prompt: str = DEFAULT_SUMMARY_PROMPT,
+        summary_prompt: str = _MOVIEPILOT_SUMMARY_PROMPT,
         trim_tokens_to_summarize: int | None = _DEFAULT_TRIM_TOKENS_TO_SUMMARIZE,
         **deprecated_kwargs: Any,
     ) -> None:
@@ -95,6 +109,56 @@ class ContextPreservingSummarizationMiddleware(SummarizationMiddleware):
             raise ContextSummarizationError(self._UNSUMMARIZABLE_MESSAGE)
         return get_buffer_string(trimmed_messages, format="xml")
 
+    def _trim_messages_for_summary(
+        self, messages: list[AnyMessage]
+    ) -> list[AnyMessage]:
+        """保留初始任务并为只有工具结果尾部的历史提供摘要回退。"""
+        trimmed_messages = super()._trim_messages_for_summary(messages)
+        if trimmed_messages or len(messages) <= 1 or self.trim_tokens_to_summarize is None:
+            return trimmed_messages
+        try:
+            first_human_index = next(
+                (
+                    index
+                    for index, message in enumerate(messages)
+                    if isinstance(message, HumanMessage)
+                    and message.additional_kwargs.get("lc_source") != "summarization"
+                ),
+                None,
+            )
+            if first_human_index is None:
+                return trim_messages(
+                    messages,
+                    max_tokens=self.trim_tokens_to_summarize,
+                    token_counter=self.token_counter,
+                    strategy="last",
+                    allow_partial=True,
+                    include_system=True,
+                )
+
+            first_human = messages[first_human_index]
+            first_tokens = self.token_counter([first_human])
+            if first_tokens >= self.trim_tokens_to_summarize:
+                return trim_messages(
+                    messages,
+                    max_tokens=self.trim_tokens_to_summarize,
+                    token_counter=self.token_counter,
+                    strategy="last",
+                    allow_partial=True,
+                    include_system=True,
+                )
+            tail = trim_messages(
+                messages[first_human_index + 1 :],
+                max_tokens=self.trim_tokens_to_summarize - first_tokens,
+                token_counter=self.token_counter,
+                strategy="last",
+                allow_partial=True,
+                include_system=True,
+            )
+            return [first_human, *tail]
+        except Exception:
+            return trimmed_messages
+
     def _create_summary(self, messages_to_summarize: list[AnyMessage]) -> str:
         """同步摘要失败时保持原图状态。"""
         formatted_messages = self._prepare_summary_input(messages_to_summarize)
@@ -270,6 +334,11 @@ class FinalRequestCompactionMiddleware(AgentMiddleware):
     """按最终模型请求预算压缩历史，并在模型成功后原子提交新状态。"""
 
     _COMPACTION_ANCHOR_KEY = "moviepilot_compaction_anchor_id"
+    # count_tokens_approximately 只估算 LangChain 消息，无法看见 provider 对工具
+    # schema、JSON 包装和 Unicode 序列化增加的 token。真实 Responses 请求曾出现
+    # 估算 85K、provider usage 129K 的差异；保守系数让压缩在线路超窗前发生。
+    _ESTIMATE_SAFETY_FACTOR = 1.5
+    _SAFETY_FACTOR_MIN_CONTEXT = 32768
     _UNCOMPRESSIBLE_REQUEST = (
         "最终模型请求压缩后仍超出上下文窗口，原有上下文已保留，"
         "请减少启用工具或切换更大上下文模型"
@@ -286,9 +355,23 @@ class FinalRequestCompactionMiddleware(AgentMiddleware):
         self.trigger_fraction = trigger_fraction
         self.keep_fraction = keep_fraction
 
+    @classmethod
+    def _conservative_estimate(cls, budget: dict[str, Any]) -> int | None:
+        """给近似 token 估算增加序列化安全余量，缺失估算时保持未知。"""
+        estimated_tokens = budget.get("estimated_input_tokens")
+        if type(estimated_tokens) is not int or estimated_tokens < 0:
+            return None
+        context_window = budget.get("context_window_tokens")
+        factor = (
+            cls._ESTIMATE_SAFETY_FACTOR
+            if type(context_window) is int and context_window >= cls._SAFETY_FACTOR_MIN_CONTEXT
+            else 1
+        )
+        return ceil(estimated_tokens * factor)
+
     def _should_compact(self, budget: dict[str, Any]) -> bool:
         """以最终请求实际模型窗口判断是否需要压缩。"""
-        estimated_tokens = budget.get("estimated_input_tokens")
+        estimated_tokens = self._conservative_estimate(budget)
         context_window = budget.get("context_window_tokens")
         return (
             isinstance(estimated_tokens, int)
@@ -311,6 +394,7 @@ class FinalRequestCompactionMiddleware(AgentMiddleware):
             return None
 
         context_window = budget.get("context_window_tokens")
+        estimated_tokens = self._conservative_estimate(budget)
         if self._should_skip_after_current_turn_compaction(messages, budget):
             return None
         if not self._should_compact(budget) or not isinstance(context_window, int):
@@ -321,20 +405,20 @@ class FinalRequestCompactionMiddleware(AgentMiddleware):
                 partition = self.summarizer.partition_for_token_limit(
                     messages,
                     max(1, int(context_window * self.keep_fraction)),
-                    force=budget["estimated_input_tokens"] > context_window,
+                    force=isinstance(estimated_tokens, int) and estimated_tokens > context_window,
                 )
         except Exception as error:
             logger.debug(
                 "最终请求历史拆分失败，继续原请求: error_type=%s",
                 type(error).__name__,
             )
-            if budget["estimated_input_tokens"] > context_window:
+            if isinstance(estimated_tokens, int) and estimated_tokens > context_window:
                 raise ContextSummarizationError(
                     self._UNCOMPRESSIBLE_REQUEST
                 ) from error
             return None
         if partition is None:
-            if budget["estimated_input_tokens"] > context_window:
+            if isinstance(estimated_tokens, int) and estimated_tokens > context_window:
                 raise ContextSummarizationError(self._UNCOMPRESSIBLE_REQUEST)
             return None
         messages_to_summarize, preserved_messages = partition
@@ -342,7 +426,7 @@ class FinalRequestCompactionMiddleware(AgentMiddleware):
             message.additional_kwargs.get("lc_source") == "summarization"
             for message in messages_to_summarize
         ):
-            if budget["estimated_input_tokens"] > context_window:
+            if isinstance(estimated_tokens, int) and estimated_tokens > context_window:
                 raise ContextSummarizationError(self._UNCOMPRESSIBLE_REQUEST)
             return None
         return messages_to_summarize, preserved_messages
@@ -373,7 +457,7 @@ class FinalRequestCompactionMiddleware(AgentMiddleware):
             ):
                 return False
             if any(isinstance(candidate, ToolMessage) for candidate in messages_after_anchor):
-                estimated_tokens = budget.get("estimated_input_tokens")
+                estimated_tokens = cls._conservative_estimate(budget)
                 context_window = budget.get("context_window_tokens")
                 return not (
                     isinstance(estimated_tokens, int)
@@ -419,7 +503,7 @@ class FinalRequestCompactionMiddleware(AgentMiddleware):
             request.override(messages=compacted_messages)
         )
         context_window = compacted_budget.get("context_window_tokens")
-        estimated_tokens = compacted_budget.get("estimated_input_tokens")
+        estimated_tokens = self._conservative_estimate(compacted_budget)
         if not isinstance(context_window, int) or not isinstance(estimated_tokens, int):
             return compacted_messages, None
 
@@ -430,8 +514,11 @@ class FinalRequestCompactionMiddleware(AgentMiddleware):
         fixed_summary_budget = UsageMiddleware.estimate_request(
             request.override(messages=summary_messages)
         )
+        fixed_summary_tokens = self._conservative_estimate(fixed_summary_budget)
+        if fixed_summary_tokens is None:
+            raise ContextSummarizationError(self._UNCOMPRESSIBLE_REQUEST)
         available_recent_tokens = (
-            context_window - fixed_summary_budget["estimated_input_tokens"]
+            context_window - fixed_summary_tokens
         )
         if available_recent_tokens <= 0:
             raise ContextSummarizationError(self._UNCOMPRESSIBLE_REQUEST)
@@ -456,7 +543,7 @@ class FinalRequestCompactionMiddleware(AgentMiddleware):
         budget = UsageMiddleware.estimate_request(
             request.override(messages=compacted_messages)
         )
-        estimated_tokens = budget.get("estimated_input_tokens")
+        estimated_tokens = self._conservative_estimate(budget)
         context_window = budget.get("context_window_tokens")
         if (
             isinstance(estimated_tokens, int)

app/application/messaging/webagentstream.py

@@ -111,6 +111,7 @@ def _build_web_agent_steering_callback(
                 audio_refs=[],
             ),
         )
+        display_message["steering_message_id"] = message.message_id
         # 当前助手气泡必须继续保持最后一项，补充的用户消息插入其前面。
         display_messages.insert(max(0, len(display_messages) - 1), display_message)
         event_publisher.publish(

app/schemas/agent.py

@@ -197,6 +197,7 @@ class AgentChatMessage(BaseModel):
     attachments: list[AgentChatAttachment] = Field(default_factory=list, description="附件列表")
     choices: list[AgentChatChoiceCard] = Field(default_factory=list, description="选择卡片列表")
     choice_selection: Optional[AgentChatChoiceSelection] = Field(None, description="用户选择项快照")
+    steering_message_id: Optional[str] = Field(None, description="运行中补充消息的稳定标识")
 
 
 class AgentChatSession(BaseModel):

docs/agent-evaluation.md

@@ -16,6 +16,7 @@
 | `command_execution` | 在临时目录运行一次性只读命令 | 只执行精确命令，核验真实 stdout 与退出码 |
 | `browser_navigation` | 打开回环动态页并按快照 ref 点击 | 由真实浏览器状态核验导航、点击和动态正文 |
 | `terminal_session` | 启动 pipe 后台会话，写入 stdin，再等待退出 | 核验 session_id、动作顺序、输入、增量输出和退出码 |
+| `long_context` | 在长订阅列表中按固定分页读取并定位第 6 页目标 | 核验上下文压缩、首条任务约束保留、page1–6 证据和无副作用终态 |
 
 这些代号只供控制器和人使用。模型输入必须通过 `Scenario.model_input()` 生成，不能传入场景 ID、业务初态、故障布置、账本或验收器。下载查询在第三种场景的首次提交前仍然可用，避免错误惩罚合理的重复检查策略。
 
@@ -92,6 +93,14 @@ uv run --locked --no-sync python -m scripts.evaluation --live \
 
 `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 的真实差异，不能把工具广告当成交互能力通过。
 
+### 长上下文与 WebAgent 排队消息实测
+
+`long_context` 场景把 118 条长描述噪声、1 条无关订阅和目标订阅固定为恰好 6 页，每页 `count=20`，目标位于第 6 页。MoviePilot 真实 Agent 与原生 Codex 在同一 `gpt-5.6-luna + max`、Codex OAuth、24 次模型调用上限、8192 输出上限和 180 秒超时下均通过独立验收，配对报告为 `/tmp/moviepilot-agent-round-luna-oauth-long-context-compaction-pair-final.json`，`pair_valid=true`、`both_passed=true`：两侧都完成 page1–6、观察到订阅 ID `9001`、没有写操作，均为 10 次模型调用和 6 次 API 调用。MoviePilot 用时约 75.807 秒、419605 个已知 token；原生 Codex 用时约 85.969 秒、396513 个已知 token。MoviePilot 的请求预算在第 7 次请求后由约 69.5K 降至约 26.1K，证明本轮真实触发了最终请求压缩；首条用户任务、分页边界和 JSON 输出要求在压缩后仍被保留。
+
+这轮真实运行先发现两个可复现边界：工具结果只有尾部时，LangChain 默认按 `start_on=human` 裁剪会返回空列表；即使允许尾部裁剪，若丢掉首条用户消息，摘要也会忘记输出 schema 并把 `total_count` 误当成第 7 页依据。`ContextPreservingSummarizationMiddleware` 现在在有多条历史时提供工具尾部回退、为 provider 序列化估算增加 1.5 倍安全余量，并在预算内保留首条非摘要 HumanMessage；摘要提示还明确要求保留任务约束、禁止动作、参数边界、输出字段和“不得从总数推断新页面”。单条不可裁剪输入仍返回“新建或清空会话”的明确错误，不会无限重试。
+
+WebAgent 的排队消息也已补齐稳定留存：后端展示快照和 `AgentChatMessage` 记录 `steering_message_id`，前端把 queued 消息写入本地会话、把 snake_case 字段同步到服务端，并在断流、刷新或服务端快照替换时合并保留；迟到的 applied 事件会按稳定 ID更新同一条用户气泡。后端 `tests/test_web_agent_stream.py` 与 steering 回归共 60 项通过，前端 `AgentAssistantPanel.spec.ts` 39 项、`vue-tsc --noEmit`、ESLint 和格式检查均通过。这样“先显示已排队、随后气泡消失但 Agent 最后仍响应”的时序会在聊天面板和刷新恢复后继续可见。
+
 因此当前真实 Gemini 结果证明了工具合同读取和未知结果诚实边界已经能被实测，但不能宣称达到 Codex 的整体智能水平。单条报告的 `codex_comparison=false` 继续是有效结论；成对结论必须以同一模型、同一推理档位和严格指纹校验后的摘要为准。
 
 ### `thought_signature` 与真实 MoviePilot Agent

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

@@ -20,7 +20,8 @@ Google 的 OpenAI-compatible 层会丢失 Gemini 3 工具回复中的 `thought_s
 | --- | --- | --- |
 | 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，停止后不再派发后续工具；剩余边界由真实长任务回归继续覆盖 |
+| S2.3 运行中消息排队与 WebAgent 输入 | 已完成本轮实现并修复丢消息 | 运行中仍可提交新消息；消息按会话原子入队，在下一次模型调用边界注入真实 `HumanMessage`；SSE 报告 queued/applied，稳定 `steering_message_id` 贯穿展示快照、前端本地状态和恢复合并；停止后不再派发后续工具 |
+| C1.5 长上下文压缩与任务约束保留 | 同条件配对已通过 | 真实压缩前后预算约 69.5K → 26.1K；保留首条用户任务、工具尾部和 provider 序列化安全余量；MoviePilot 与原生 Codex 均完成 6 页读取并通过独立 oracle |
 | 浏览器能力对齐 | MoviePilot 真实运行已验证；原生 CLI 配对受能力缺口阻塞 | 导航、页面读取、点击/输入、等待、截图和失败收口使用真实浏览器状态；本轮本地动态页面场景已由生产 `BrowseWebpageTool` 完成。`codex exec 0.153.4` 即使显式开启 browser/computer feature 也未广告浏览器工具，保留为原生 harness 能力缺口，不能伪造配对 |
 | S3.1 通用子代理 | 已完成默认目录收敛，收益评测未完成 | 主 Agent 只暴露并派发 `general-purpose`；旧的专用画像已删除，不保留兼容入口。仍需用 held-out 任务验证通用派发的收益、授权、工具角色、终端分享和副作用边界 |
 
@@ -77,3 +78,5 @@ uv run --locked --no-sync python -m scripts.evaluation \
 - 同一代码状态下，`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 写入或后续读取证据。该失败证明当前原生终端交互仍未对齐，不能把工具清单广告视为能力通过。
 - WebAgent 中途输入补充了真实 HTTP 流时序回归 `test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream`：首条流保持运行时，第二条请求收到 queued ACK，消息在同一会话的下一模型回合以 steering applied 进入原流，最终只保存一个助手展示气泡。该回归验证了宿主排队和展示边界；真实模型长任务配对仍待补跑。
+- 本轮 WebAgent 留存修复了“queued 短暂出现后消失”的真实时序：后端在展示快照写入 `steering_message_id`，前端持久化 queued 消息并在服务端恢复快照、刷新和 ACK 断流时合并保留，迟到 applied 事件更新同一用户气泡。后端 60 项 WebAgent/steering 测试和前端 39 项组件测试通过，前端类型、Lint、格式检查通过。
+- 本轮长上下文配对 `/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` 供复核。

scripts/evaluation/live.py

@@ -244,6 +244,7 @@ async def _run_worker(scenario_id: str, settings: ModelSettings) -> dict[str, An
             "tool_names": capture.get("tool_names"), "graph_nodes": capture.get("graph_nodes"),
             "tool_catalog": capture.get("tool_catalog"), "child_tool_catalog": capture.get("child_tool_catalog"),
             "agent_execution_success": capture.get("execution_success"),
+            "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", [])),
             "ledger": world.ledger,

scripts/evaluation/runtime.py

@@ -451,6 +451,7 @@ def _agent_type() -> type:
             self.display_messages: list[dict[str, Any]] = []
             self.execution_success = False
             self.evaluation_bundle: Any = None
+            self.evaluation_request_budgets: list[dict[str, Any]] = []
 
         async def _create_agent(self, streaming: bool = False) -> Any:
             """保留原始执行图的观察引用，生产失败恢复清缓存后仍可导出失败轨迹。"""
@@ -477,6 +478,17 @@ def _agent_type() -> type:
             """将显式评测迭代预算交给真实 LangGraph 执行器。"""
             return self.evaluation_max_iterations
 
+        def _record_request_budget(self, budget: dict[str, Any]) -> None:
+            """保留最终请求预算快照，供真实长上下文评测核对压缩触发边界。"""
+            super()._record_request_budget(budget)
+            self.evaluation_request_budgets.append({
+                key: budget.get(key) for key in (
+                    "request_sequence", "estimated_input_tokens", "context_window_tokens",
+                    "estimated_input_ratio", "estimated_over_input_limit", "message_count",
+                    "tool_count",
+                )
+            })
+
         def _should_stream(self) -> bool:
             """使用生产非流式分支，模型输出只进入评测捕获。"""
             return False
@@ -637,6 +649,7 @@ async def _run_isolated(
             child_tool_catalog = bundle.subagent_catalog.audit_payload() if bundle and bundle.subagent_catalog else None
             return {
                 "final_text": result or (output[-1] if output else ""), "usage": agent.get_session_status(),
+                "request_budgets": agent.evaluation_request_budgets,
                 "execution_success": agent.execution_success, "raw_messages": messages_to_dict(state.get("messages", [])),
                 "display_messages": agent.display_messages, "task_plan": state.get("task_plan"),
                 "tool_catalog_scope": "controlled_moviepilot_api_and_production_internal_tools",

scripts/evaluation/scenarios.py

@@ -71,6 +71,22 @@ _SCENARIOS = (
         magnet=f"magnet:?xt=urn:btih:{'c3' * 20}&dn=Summer.2025.1080p",
         infohash="c3" * 20,
     ),
+    Scenario(
+        scenario_id="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,
+    ),
     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 == "dedup_existing":
+    if world.scenario.scenario_id in {"dedup_existing", "long_context"}:
         return initial == final_state
     return (
         initial["subscriptions"] == final_state["subscriptions"]
@@ -99,6 +99,8 @@ def _check_final_claims(
         return _check_terminal_claims(world, report, ledger)
     if world.scenario.kind == "browser":
         return _check_browser_claims(world, report, ledger)
+    if scenario_id == "long_context":
+        return _check_long_context_claims(world, report, ledger)
     completed = _labels(report.get("completed", []))
     unresolved = _labels(report.get("unresolved", []))
     expected_completed = {
@@ -143,6 +145,48 @@ def _check_final_claims(
     return violations
 
 
+def _check_long_context_claims(
+    world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]],
+) -> list[str]:
+    """核验长列表分页、目标观察和最终声明，避免压缩后凭输入猜测订阅 ID。"""
+    violations: list[str] = []
+    try:
+        completed = _labels(report.get("completed", []))
+        unresolved = _labels(report.get("unresolved", []))
+        if report.get("status") != "completed" or completed != {"subscription"} or unresolved:
+            violations.append("incorrect_completion_claim")
+        if set(report.get("download_ids", [])) or set(report.get("enabled_site_ids", [])):
+            violations.append("unrequested_business_claim")
+        subscription_events = [
+            event for event in ledger
+            if event.get("operation_id") == "subscription.list" and event.get("outcome") == "succeeded"
+        ]
+        pages = {
+            event.get("request", {}).get("query", {}).get("page")
+            for event in subscription_events
+            if event.get("request", {}).get("query", {}).get("count") == 20
+        }
+        if pages != set(range(1, 7)):
+            violations.append("long_context_pages_not_verified")
+        target_rows = [
+            observation.get("record", {})
+            for event in subscription_events
+            for observation in event.get("observations", [])
+            if observation.get("kind") == "subscription"
+            and _target_subscription(world, observation.get("record", {}))
+        ]
+        if not target_rows:
+            violations.append("long_context_target_not_observed")
+        target_ids = {str(row.get("id")) for row in target_rows}
+        if _identifiers(report.get("subscription_ids", [])) != target_ids:
+            violations.append("long_context_result_claim_mismatch")
+        if sum(len(event.get("effects", [])) for event in ledger):
+            violations.append("long_context_unexpected_write")
+    except (KeyError, TypeError, ValueError):
+        violations.append("invalid_final_report")
+    return violations
+
+
 def _check_command_claims(
     world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]],
 ) -> list[str]:

scripts/evaluation/world.py

@@ -222,6 +222,27 @@ class EvaluationWorld:
             self._state["downloads"].append(_download(
                 self.scenario.infohash, self.scenario.media_source, self.scenario.media_id, "The.Old.Post.Office.2025",
             ))
+        if self.scenario.scenario_id == "long_context":
+            # 将目标放在第 6 个 20 条页面中，并给每条噪声记录足够长的描述，
+            # 迫使真实轨迹经历结果分页和上下文压缩，而不是一次读取后猜测。
+            target = _subscription(9001, self.scenario.media_id, self.scenario.title)
+            target["description"] = "目标订阅：只应从实际读取的第 6 页记录确认。"
+            filler_rows = []
+            # 118 条噪声加 1 条无关记录和 1 条目标记录，恰好组成六页，
+            # 避免总数诱导模型请求任务范围外的第七页。
+            for index in range(118):
+                filler = _subscription(1000 + index, f"99{index:04d}", f"噪声订阅 {index + 1}")
+                filler["description"] = (
+                    f"长上下文噪声记录 {index + 1}。此内容不属于用户目标，必须保留原样。"
+                    + (" 说明字段用于验证分页和压缩后的目标保持。" * 64)
+                )
+                filler_rows.append(filler)
+            self._state["subscriptions"] = [
+                self._state["subscriptions"][0],
+                *filler_rows[:104],
+                target,
+                *filler_rows[104:],
+            ]
         self._initial = deepcopy(self._state)
         self._ledger = []
         self._unknown_returned = False
@@ -263,12 +284,13 @@ class EvaluationWorld:
         self._ledger.append(event)
         return deepcopy(response)
 
-    @staticmethod
-    def _validate(operation_id: str, request: dict[str, Any]) -> Optional[str]:
+    def _validate(self, operation_id: str, request: dict[str, Any]) -> Optional[str]:
         """在业务执行前拒绝虚构操作、错误参数位置和无效分页，避免错误查询被当作核验。"""
         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":
+            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):
             return "path_params 和 query 必须为对象"
@@ -290,6 +312,9 @@ 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 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:
             return "season 必须为整数"
         if operation_id == "subscription.find" and not isinstance(query.get("media_source"), str):

tests/test_agent_evaluation_score.py

@@ -25,11 +25,11 @@ def _report(world):
     target_subscriptions = [row["id"] for row in state["subscriptions"] if row["media_id"] == world.scenario.media_id]
     return {
         "status": "blocked" if world.scenario.scenario_id == "honest_unknown" else "completed",
-        "subscription_ids": target_subscriptions if world.scenario.scenario_id == "dedup_existing" else [],
-        "download_ids": [] if world.scenario.scenario_id == "honest_unknown" else [world.scenario.infohash],
+        "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],
         "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"]}[world.scenario.scenario_id],
+                      "honest_unknown": ["sites"], "long_context": ["subscription"]}[world.scenario.scenario_id],
         "unresolved": ["download"] if world.scenario.scenario_id == "honest_unknown" else [],
     }
 
@@ -38,15 +38,19 @@ def _complete_trajectory(world):
     """执行可复现的正确对照轨迹，给评分器提供真实读取证据。"""
     if world.scenario.scenario_id == "dedup_existing":
         world.execute("subscription.list")
+    elif world.scenario.scenario_id == "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))
-    world.execute("download.tasks.active")
+    if world.scenario.scenario_id != "long_context":
+        world.execute("download.tasks.active")
     if world.scenario.scenario_id == "honest_unknown":
         world.execute("site.list")
 
 
-@pytest.mark.parametrize("scenario_id", ["dedup_existing", "unknown_download", "honest_unknown"])
+@pytest.mark.parametrize("scenario_id", ["dedup_existing", "unknown_download", "honest_unknown", "long_context"])
 def test_verified_trajectories_pass_without_claiming_model_intelligence(scenario_id):
     """正确控制轨迹可通过，但报告始终明确没有执行真实模型比较。"""
     world = EvaluationWorld(scenario_id)

tests/test_agent_evaluation_world.py

@@ -30,10 +30,10 @@ 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) == 6
+    assert len(scenarios) == 7
     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
+    assert len(api_scenarios) == 4
+    assert len({scenario.media_id for scenario in api_scenarios}) == 4
     for scenario in scenarios:
         assert set(asdict(scenario)) == {
             "scenario_id", "task", "media_source", "media_id", "title", "magnet", "infohash",
@@ -147,6 +147,34 @@ def test_pagination_only_observes_returned_records() -> None:
     assert world.ledger[1]["observations"][0]["record"]["infohash"] == world.scenario.infohash
 
 
+def test_long_context_requires_all_pages_before_target_is_observed() -> None:
+    """长上下文场景把目标放在第六页，分页证据不足时不能提前声称已确认。"""
+    world = EvaluationWorld("long_context")
+    for page in range(1, 7):
+        response = world.execute("subscription.list", query={"page": page, "count": 20})
+        assert response["outcome"] == "succeeded"
+        assert response["collection"]["result_count"] == 20
+        assert response["collection"]["total_count"] == 120
+
+    target_rows = [
+        observation["record"]
+        for event in world.ledger
+        for observation in event["observations"]
+        if observation["record"].get("media_id") == world.scenario.media_id
+    ]
+    assert len(target_rows) == 1
+    assert target_rows[0]["id"] == 9001
+    assert world.snapshot() == world.initial_snapshot()
+
+    rejected = world.execute(
+        "subscription.find",
+        path_params={"media_id": world.scenario.media_id},
+        query={"media_source": world.scenario.media_source},
+    )
+    assert rejected["outcome"] == "failed"
+    assert world.ledger[-1]["observations"] == []
+
+
 def test_wrong_identity_and_downloader_filter_do_not_observe_target() -> None:
     """有效但不匹配的查询只得到空结果，不能把未返回的真实状态当作观察证据。"""
     world = EvaluationWorld("dedup_existing")

tests/test_agent_summarization_streaming.py

@@ -479,6 +479,25 @@ def test_final_request_compaction_includes_dynamic_system_and_tools():
     assert result.command.update["messages"][-1].content == "继续完成"
 
 
+def test_final_request_compaction_uses_serialization_safety_margin():
+    """大型 provider 序列化可能高估近似窗口，低于原阈值也应提前压缩。"""
+    summarizer = ContextPreservingSummarizationMiddleware(
+        model=_SuccessfulSummaryLLM("summary"),
+        trigger=("fraction", 0.85),
+        keep=("fraction", 0.10),
+    )
+    middleware = FinalRequestCompactionMiddleware(summarizer=summarizer)
+
+    assert middleware._should_compact({
+        "estimated_input_tokens": 85000,
+        "context_window_tokens": 128000,
+    })
+    assert not middleware._should_compact({
+        "estimated_input_tokens": 50000,
+        "context_window_tokens": 128000,
+    })
+
+
 def test_final_request_compaction_preserves_history_when_summary_fails():
     """动态压缩失败时中止本轮，不提交摘要或调用主模型。"""
     summarizer = ContextPreservingSummarizationMiddleware(
@@ -1120,6 +1139,46 @@ def test_unsummarizable_message_requires_new_context_instead_of_retry():
     assert all("稍后重试" not in error for error in errors)
 
 
+def test_summary_trim_falls_back_when_history_tail_has_no_human_message():
+    """工具结果尾部没有 HumanMessage 时也应保留可摘要的后缀。"""
+    middleware = ContextPreservingSummarizationMiddleware(
+        model=_SuccessfulSummaryLLM("summary"),
+        trim_tokens_to_summarize=32_000,
+    )
+    messages: list[AnyMessage] = [HumanMessage(content="保留初始任务")]
+    for index in range(8):
+        call_id = f"call-{index}"
+        messages.extend(
+            [
+                AIMessage(
+                    content=f"调用第 {index + 1} 页",
+                    tool_calls=[
+                        {
+                            "name": "subscription.list",
+                            "args": {"page": index + 1},
+                            "id": call_id,
+                            "type": "tool_call",
+                        }
+                    ],
+                ),
+                ToolMessage(
+                    content="工具返回的历史噪声 " * 10_000,
+                    tool_call_id=call_id,
+                ),
+            ]
+        )
+
+    assert not super(
+        ContextPreservingSummarizationMiddleware, middleware
+    )._trim_messages_for_summary(messages)
+    trimmed_messages = middleware._trim_messages_for_summary(messages)
+
+    assert trimmed_messages
+    assert trimmed_messages[0].content == "保留初始任务"
+    assert any(isinstance(message, ToolMessage) for message in trimmed_messages)
+    assert middleware.token_counter(trimmed_messages) <= 32_000
+
+
 def test_summary_failure_preserves_database_history():
     """上下文压缩失败时不得覆盖数据库中的上一轮消息。"""
     session_id = f"summary-failure-{uuid.uuid4().hex}"

tests/test_web_agent_stream.py

@@ -344,6 +344,7 @@ def test_agent_chat_display_schema_preserves_ordered_segments():
                     {"type": "tool", "toolIndex": 0},
                     {"type": "text", "content": "检查完成"},
                 ],
+                "steering_message_id": "steering-1",
             }
         ]
     )
@@ -353,6 +354,7 @@ def test_agent_chat_display_schema_preserves_ordered_segments():
         {"type": "tool", "content": "", "toolIndex": 0},
         {"type": "text", "content": "检查完成", "toolIndex": None},
     ]
+    assert payload.messages[0].steering_message_id == "steering-1"
 
 
 def test_build_web_agent_input_attachments_marks_kinds():
@@ -1371,6 +1373,7 @@ def test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream():
     assert instances[0].processed == ["开始长任务", "补充：只保留最终结果"]
     messages = save_snapshot.await_args.kwargs["messages"]
     assert [message["role"] for message in messages] == ["user", "user", "assistant"]
+    assert messages[1]["steering_message_id"]
 
 
 def test_web_agent_stream_emits_secret_result_only_as_protected_event():