test(agent): cover web steering stream race

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

@@ -76,3 +76,4 @@ uv run --locked --no-sync python -m scripts.evaluation \
 - 命令行与浏览器能力阶段：`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 对齐。
 - `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 进入原流，最终只保存一个助手展示气泡。该回归验证了宿主排队和展示边界；真实模型长任务配对仍待补跑。

tests/test_web_agent_stream.py

@@ -12,6 +12,7 @@ from app.agent.contracts import ReplyMode
 
 # pylint: disable=no-name-in-module  # 旧公开入口由 runtime compat 惰性注入，Pylint 无法静态解析。
 from app.agent.orchestrator import agent_manager
+from app.agent.session import AgentSessionOwner
 from app.agent.steering import SteeringMessage
 from app.agent.web import _get_web_agent_type
 from app.api.endpoints import agent as agent_endpoint
@@ -1233,6 +1234,145 @@ def test_web_agent_stream_binds_session_to_agent_manager():
             worker.cancel()
 
 
+def test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream():
+    """第一条 WebAgent 流运行时的第二条请求应 ACK 后注入同一助手回合。"""
+    first_payload = schemas.AgentWebChatRequest(
+        text="开始长任务",
+        session_id="mid-run-steering",
+        echo_user=True,
+    )
+    second_payload = schemas.AgentWebChatRequest(
+        text="补充：只保留最终结果",
+        session_id="mid-run-steering",
+        echo_user=True,
+    )
+    request = SimpleNamespace(is_disconnected=AsyncMock(return_value=False))
+    user = SimpleNamespace(id=1, name="admin", is_superuser=True)
+    session_id = "web-agent:mid-run-steering"
+
+    async def scenario():
+        """并发消费两条 HTTP 流，检查排队和应用的先后顺序。"""
+        owner = AgentSessionOwner()
+        owner._accepting_tasks = True
+        started = asyncio.Event()
+        release = asyncio.Event()
+        instances = []
+
+        class BlockingWebAgent:
+            """阻塞首轮推理并记录后续 steering 消息的测试 Agent。"""
+
+            def __init__(self, **kwargs):
+                """保存会话回调并登记实例数量。"""
+                self.__dict__.update(kwargs)
+                self.processed = []
+                instances.append(self)
+
+            def set_output_callback(self, output_callback):
+                """更新当前助手流的文本回调。"""
+                self.output_callback = output_callback
+
+            def set_protected_output_callback(self, protected_output_callback):
+                """更新当前助手流的敏感结果回调。"""
+                self.protected_output_callback = protected_output_callback
+
+            def set_message_callback(self, message_callback):
+                """更新当前助手流的主动消息回调。"""
+                self.message_callback = message_callback
+
+            async def process(self, message, **_kwargs):
+                """首轮保持运行，后续回合输出补充消息已生效。"""
+                self.processed.append(message)
+                if message == "开始长任务":
+                    self.output_callback("首轮处理中")
+                    started.set()
+                    await release.wait()
+                else:
+                    self.output_callback("补充已应用")
+                return message
+
+            async def cleanup(self):
+                """模拟 Agent 资源清理。"""
+                return True
+
+        async def collect(iterator, initial=None):
+            """收集 SSE 文本，支持保留已读取的首个事件。"""
+            chunks = list(initial or [])
+            async for chunk in iterator:
+                chunks.append(chunk.decode("utf-8") if isinstance(chunk, bytes) else chunk)
+            return chunks
+
+        try:
+            with (
+                patch(
+                    "app.application.messaging.agent.get_api_runtime_config_snapshot",
+                    return_value=SimpleNamespace(ai_agent_enable=True),
+                ),
+                patch(
+                    "app.application.messaging.agent.is_web_agent_traditional_message",
+                    return_value=False,
+                ),
+                patch(
+                    "app.application.messaging.agent.has_web_agent_traditional_interaction",
+                    return_value=False,
+                ),
+                patch(
+                    "app.application.messaging.agent.build_web_agent_session_id_async",
+                    return_value=session_id,
+                ),
+                patch(
+                    "app.application.agent.get_running_agent_manager",
+                    return_value=owner,
+                ),
+                patch(
+                    "app.application.agent.get_web_agent_type",
+                    return_value=BlockingWebAgent,
+                ),
+                patch(
+                    "app.application.messaging.agent.save_web_agent_display_snapshot",
+                    new_callable=AsyncMock,
+                ) as save_snapshot,
+            ):
+                first_response = await web_agent_stream(first_payload, request, user)
+                first_iterator = first_response.body_iterator
+                first_start = await first_iterator.__anext__()
+                await asyncio.wait_for(started.wait(), timeout=1)
+
+                second_response = await web_agent_stream(second_payload, request, user)
+                second_body = "".join(await collect(second_response.body_iterator))
+                assert '"status": "queued"' in second_body
+
+                release.set()
+                first_body = "".join(await collect(first_iterator, [first_start]))
+                await wait_web_agent_background_tasks()
+
+                return (
+                    first_body,
+                    second_body,
+                    instances,
+                    save_snapshot,
+                )
+        finally:
+            worker = owner._session_workers.pop(session_id, None)
+            if worker:
+                worker.cancel()
+                await asyncio.gather(worker, return_exceptions=True)
+            owner._session_queues.pop(session_id, None)
+            owner._session_active_tasks.pop(session_id, None)
+            owner._session_steering_inboxes.pop(session_id, None)
+            owner.active_agents.clear()
+
+    first_body, second_body, instances, save_snapshot = asyncio.run(scenario())
+
+    assert '"status": "queued"' in second_body
+    assert '"status": "applied"' in first_body
+    assert first_body.count('data: {"type": "start"') == 1
+    assert first_body.count('data: {"type": "done"') == 1
+    assert len(instances) == 1
+    assert instances[0].processed == ["开始长任务", "补充：只保留最终结果"]
+    messages = save_snapshot.await_args.kwargs["messages"]
+    assert [message["role"] for message in messages] == ["user", "user", "assistant"]
+
+
 def test_web_agent_stream_emits_secret_result_only_as_protected_event():
     """敏感结果只能进入命名 protected SSE，不能进入普通快照。"""
     secret_marker = "WEB_SECRET_MARKER **literal** <img src=x>"