diff --git a/docs/refactor/agent-codex-parity-plan.md b/docs/refactor/agent-codex-parity-plan.md index 2e13f148a..6aa851c0e 100644 --- a/docs/refactor/agent-codex-parity-plan.md +++ b/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 进入原流，最终只保存一个助手展示气泡。该回归验证了宿主排队和展示边界；真实模型长任务配对仍待补跑。 diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py index 3a19f3a35..061ed28b9 100644 --- a/tests/test_web_agent_stream.py +++ b/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** "