MoviePilot/tests/test_web_agent_stream.py

import asyncio
import time
from queue import Queue
from threading import Event as ThreadEvent
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock, patch

import pytest

from app import schemas
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
from app.api.endpoints.agent import web_agent_stream
from app.application.messaging import agent as agent_application
from app.application.messaging.agent import (
    _WEB_AGENT_FILE_REGISTRY,
    AgentInteractionOption,
    agent_interaction_manager,
    attach_web_agent_message_queue,
    build_web_agent_message_update_event,
    configure_web_agent_message_runtime,
    detach_web_agent_message_queue,
    dispatch_web_agent_message_event,
    extract_web_agent_message_from_event_data,
    reset_web_agent_message_runtime,
    wait_web_agent_background_tasks,
)
from app.application.messaging.agent import (
    WebAgentEventPublisher as _WebAgentEventPublisher,
)
from app.application.messaging.agent import (
    apply_web_agent_display_event as _apply_web_agent_display_event,
)
from app.application.messaging.agent import (
    build_web_agent_command_items as _build_web_agent_command_items,
)
from app.application.messaging.agent import (
    build_web_agent_display_message_from_events as _build_web_agent_display_message_from_events,
)
from app.application.messaging.agent import (
    build_web_agent_input_attachments as _build_web_agent_input_attachments,
)
from app.application.messaging.agent import (
    build_web_agent_message_events as _build_web_agent_message_events,
)
from app.application.messaging.agent import (
    build_web_agent_session_id as _build_web_agent_session_id,
)
from app.application.messaging.agent import (
    build_web_agent_session_id_async as _build_web_agent_session_id_async,
)
from app.application.messaging.agent import (
    build_web_agent_traditional_callback_payload as _build_web_agent_traditional_callback_payload,
)
from app.application.messaging.agent import (
    collect_web_agent_traditional_events as _collect_web_agent_traditional_events,
)
from app.application.messaging.agent import (
    has_web_agent_traditional_interaction as _has_web_agent_traditional_interaction,
)
from app.application.messaging.agent import (
    prepare_web_agent_audio_attachment_path_async as _prepare_web_agent_audio_attachment_path_async,
)
from app.application.messaging.agent import (
    resolve_web_agent_audio_refs as _resolve_web_agent_audio_refs,
)
from app.application.messaging.agent import (
    resolve_web_agent_choice_payload as _resolve_web_agent_choice_payload,
)
from app.application.messaging.agent import (
    split_web_agent_output as _split_web_agent_output,
)
from app.application.messaging.agent import (
    transcribe_web_agent_audio_files as _transcribe_web_agent_audio_files,
)
from app.application.messaging.chat import AgentChatService, configure_agent_chat_service
from app.application.messaging.skill import skill_interaction_manager
from app.application.messaging.webagentstream import (
    _build_steering_ack_stream,
    _build_web_agent_output_callback,
    _build_web_agent_steering_callback,
    _build_web_agent_tool_event_callback,
)
from app.chain.message import MessageChain
from app.db.oper.agentchat import AgentChatOper
from app.runtime.events import Event
from app.schemas.notification import ChannelCapability, ChannelCapabilityManager
from app.schemas.types import EventType, MessageType, NotificationChannel


@pytest.fixture(autouse=True)
def _running_agent_service():
    """本文件验证运行态 Web Agent 行为，显式提供已启动的 canonical manager。"""
    was_accepting = agent_manager._accepting_tasks
    agent_manager._accepting_tasks = True
    MessageChain._user_sessions.clear()
    configure_web_agent_message_runtime(
        message_handler=lambda **kwargs: MessageChain().handle_message(**kwargs),
        session_binder=lambda user_id, session_id: MessageChain().bind_user_session(user_id, session_id),
    )
    try:
        with (
            patch(
                "app.application.agent.get_running_agent_manager",
                return_value=agent_manager,
            ),
            patch(
                "app.application.agent.get_web_agent_type",
                side_effect=_get_web_agent_type,
            ),
            patch(
                "app.chain.message.get_running_agent_manager",
                return_value=agent_manager,
            ),
        ):
            yield
    finally:
        reset_web_agent_message_runtime()
        MessageChain._user_sessions.clear()
        agent_manager._accepting_tasks = was_accepting


def test_web_agent_non_transport_helpers_are_application_owned():
    """附件、音频、事件投影和会话写入不得回流到 FastAPI 端点。"""
    ownership = {
        "WebAgentEventPublisher",
        "apply_web_agent_display_event",
        "build_web_agent_input_attachments",
        "build_web_agent_message_events_async",
        "build_web_agent_session_id_async",
        "build_web_agent_stream",
        "collect_web_agent_traditional_events",
        "get_web_agent_registered_file",
        "prepare_web_agent_audio_attachment_path_async",
        "save_web_agent_display_snapshot",
        "transcribe_web_agent_audio_input",
    }

    assert all(hasattr(agent_application, name) for name in ownership)
    assert all(not hasattr(agent_endpoint, name) for name in ownership)


@pytest.mark.asyncio
async def test_steering_ack_stream_reports_a_queued_message_without_an_assistant_bubble():
    """运行中补充消息只返回短确认流，主 Agent 的展示气泡由原流继续持有。"""
    message = SteeringMessage.create(session_id="session", user_id="user", text="补充要求")
    events = [event async for event in _build_steering_ack_stream(session_id="session", message=message)]

    assert events == [
        {"type": "start", "session_id": "session"},
        {
            "type": "steering",
            "status": "queued",
            "message_id": message.message_id,
            "content": "补充要求",
        },
        {"type": "done"},
    ]


def test_web_agent_steering_uses_current_assistant_identity_when_segments_are_equal():
    """展示内容相同的助手段也必须把 steering 插入到当前对象之后。"""
    assistant_before = {
        "id": "assistant-same",
        "role": "assistant",
        "content": "",
        "status": "streaming",
    }
    assistant_current = dict(assistant_before)
    display_messages = [assistant_before, assistant_current]
    published = []

    def build_display_message(*, role, status="done", **_kwargs):
        """构造最小展示消息，保持测试只验证时间线定位。"""
        return {"id": f"{role}-{len(published)}", "role": role, "status": status, "content": ""}

    callback = _build_web_agent_steering_callback(
        display_messages=display_messages,
        assistant_message_ref={"message": assistant_current},
        event_publisher=type("Publisher", (), {"publish": published.append})(),
        build_display_message=build_display_message,
        build_input_attachments=lambda **_kwargs: [],
    )
    message = SteeringMessage.create(session_id="session", user_id="user", text="插入当前段之后")

    callback(message, "applied")

    assert display_messages[0] is assistant_before
    assert display_messages[1] is assistant_current
    assert display_messages[1]["status"] == "done"
    assert display_messages[2]["role"] == "user"
    assert display_messages[3]["role"] == "assistant"
    assert published and published[0]["assistant_message_id"] == "assistant-same"


def test_web_agent_stream_events_keep_their_assistant_segment_identity():
    """迟到的文本和工具事件必须携带产生它们的助手段 ID。"""
    published = []
    applied = []
    assistant_message_ref = {"message": {"id": "assistant-before"}}

    output_callback = _build_web_agent_output_callback(
        assistant_message_ref=assistant_message_ref,
        event_publisher=type("Publisher", (), {"publish": published.append})(),
        split_output=lambda delta: [{"type": "delta", "content": delta}],
        apply_display_event=lambda event, _message: applied.append(event),
    )
    tool_event_callback = _build_web_agent_tool_event_callback(
        assistant_message_ref=assistant_message_ref,
        event_publisher=type("Publisher", (), {"publish": published.append})(),
        apply_display_event=lambda event, _message: applied.append(event),
    )

    output_callback("前段")
    tool_event_callback({"type": "tool", "tool_id": "tool-before", "status": "running"})
    assistant_message_ref["message"] = {"id": "assistant-after"}
    output_callback("后段")
    tool_event_callback({"type": "tool", "tool_id": "tool-after", "status": "running"})

    assert [event["assistant_message_id"] for event in applied] == [
        "assistant-before",
        "assistant-before",
        "assistant-after",
        "assistant-after",
    ]
    assert [event["assistant_message_id"] for event in published] == [
        "assistant-before",
        "assistant-before",
        "assistant-after",
        "assistant-after",
    ]


def test_split_web_agent_output_extracts_verbose_tool_message():
    """应将啰嗦模式工具提示拆成独立工具事件，并保留渠道展示文案。"""
    events = _split_web_agent_output("准备查询。\n\n⚙️ => 查询站点\n\n已完成")

    assert events == [
        {"type": "delta", "content": "准备查询。\n\n"},
        {"type": "tool", "message": "⚙️ => 查询站点"},
        {"type": "delta", "content": "已完成"},
    ]


def test_split_web_agent_output_extracts_summary_tool_message():
    """应将非啰嗦模式工具汇总行拆成独立工具事件，并保留渠道展示文案。"""
    events = _split_web_agent_output("（查询了 2 次数据）\n\n这里是结果")

    assert events == [
        {"type": "tool", "message": "（查询了 2 次数据）"},
        {"type": "delta", "content": "\n这里是结果"},
    ]


def test_split_web_agent_output_preserves_standalone_newline_delta():
    """独立换行增量应保留，避免流式 Markdown 列表被拼成同一行。"""
    chunks = [
        "可以这样操作：",
        "\n",
        "- **搜索资源**：搜索电影",
        "\n",
        "- **下载管理**：添加任务",
    ]
    content = ""

    for chunk in chunks:
        for event in _split_web_agent_output(chunk):
            if event["type"] == "delta":
                content += event["content"]

    assert content == "可以这样操作：\n- **搜索资源**：搜索电影\n- **下载管理**：添加任务"


def test_web_agent_event_publisher_coalesces_text_before_semantic_events():
    """连续文本应合并，且工具事件前的文本顺序不能改变。"""

    async def scenario():
        publisher = _WebAgentEventPublisher()
        try:
            for index in range(100):
                publisher.publish({"type": "delta", "content": str(index % 10)})
            publisher.publish({"type": "tool", "message": "查询完成"})

            first = await asyncio.wait_for(publisher.get(), timeout=1)
            second = await asyncio.wait_for(publisher.get(), timeout=1)
            return first, second, publisher.max_depth
        finally:
            await publisher.aclose()

    first, second, max_depth = asyncio.run(scenario())

    assert first == {
        "type": "delta",
        "content": "".join(str(index % 10) for index in range(100)),
    }
    assert second == {"type": "tool", "message": "查询完成"}
    assert max_depth == 2


def test_web_agent_event_publisher_rejects_events_after_close():
    """连接关闭后必须显式拒绝事件，避免把敏感结果误报为已交付。"""

    async def scenario():
        publisher = _WebAgentEventPublisher()
        await publisher.aclose()
        return publisher.publish({"type": "interaction-protected", "content": "secret"})

    assert asyncio.run(scenario()) is False


def test_build_web_agent_session_id_is_stable_per_user_and_seed():
    """同一用户和前端会话标识应生成稳定的服务端会话 ID。"""
    user = SimpleNamespace(id=1, name="admin")

    first = _build_web_agent_session_id(user, "browser-session")
    second = _build_web_agent_session_id(user, "browser-session")
    other = _build_web_agent_session_id(user, "other-session")

    assert first == second
    assert first != other
    assert first.startswith("web-agent:")


def test_build_web_agent_session_id_reuses_accessible_history():
    """传入已有历史会话 ID 时应直接复用，避免跨渠道继续对话丢上下文。"""
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)
    AgentChatOper().save_display_messages(
        session_id="telegram-session",
        user_id="telegram-user",
        username="tester",
        channel=NotificationChannel.Telegram.value,
        source="telegram-main",
        messages=[],
        title="Telegram 会话",
    )
    configure_agent_chat_service(AgentChatService(repository=AgentChatOper()))

    assert _build_web_agent_session_id(user, "telegram-session") == "telegram-session"


def test_build_web_agent_session_id_async_uses_native_async_persistence():
    """异步 Web 会话解析应通过 native async 会话服务读取历史。"""
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)
    service = SimpleNamespace(
        get=AsyncMock(
            return_value=SimpleNamespace(
                user_id="telegram-user",
                username="tester",
                agent_messages=[],
            )
        )
    )

    with patch(
        "app.application.messaging.agent.get_configured_agent_chat_service",
        return_value=service,
    ):
        session_id = asyncio.run(_build_web_agent_session_id_async(user, "telegram-session"))

    assert session_id == "telegram-session"
    service.get.assert_awaited_once_with("telegram-session")


def test_apply_web_agent_display_event_updates_snapshot():
    """WebAgent SSE 事件应按到达顺序聚合为服务端展示快照。"""
    message = {
        "id": "assistant-1",
        "role": "assistant",
        "content": "",
        "createdAt": 1,
        "status": "streaming",
        "tools": [],
        "segments": [],
        "attachments": [],
        "choices": [],
    }

    _apply_web_agent_display_event({"type": "delta", "content": "你好"}, message)
    _apply_web_agent_display_event({"type": "tool", "message": "查询订阅"}, message)
    _apply_web_agent_display_event({"type": "delta", "content": "，查询完成"}, message)
    _apply_web_agent_display_event(
        {
            "type": "attachment",
            "attachment": {"kind": "file", "url": "message/agent/file/a"},
        },
        message,
    )
    _apply_web_agent_display_event({"type": "done"}, message)

    assert message["content"] == "你好，查询完成"
    assert message["status"] == "done"
    assert len(message["tools"]) == 1
    assert message["tools"][0]["message"] == "查询订阅"
    assert message["tools"][0]["status"] == "done"
    assert message["segments"] == [
        {"type": "text", "content": "你好"},
        {"type": "tool", "toolIndex": 0},
        {"type": "text", "content": "，查询完成"},
    ]
    assert message["attachments"] == [{"kind": "file", "url": "message/agent/file/a"}]


def test_apply_web_agent_display_event_tracks_parallel_tool_lifecycle_by_id():
    """并行工具必须按稳定调用 ID 独立收口，不能由后一条开始事件覆盖前一条状态。"""
    message = {
        "id": "assistant-1",
        "role": "assistant",
        "content": "",
        "createdAt": 1,
        "status": "streaming",
        "tools": [],
        "segments": [],
        "attachments": [],
        "choices": [],
    }

    _apply_web_agent_display_event(
        {
            "type": "tool",
            "tool_id": "tool-a",
            "tool_name": "search",
            "message": "搜索媒体",
            "status": "running",
        },
        message,
    )
    _apply_web_agent_display_event(
        {
            "type": "tool",
            "tool_id": "tool-b",
            "tool_name": "download",
            "message": "检查下载器",
            "status": "running",
        },
        message,
    )
    _apply_web_agent_display_event(
        {"type": "tool", "tool_id": "tool-a", "status": "done"},
        message,
    )

    assert [(tool["id"], tool["status"]) for tool in message["tools"]] == [
        ("tool-a", "done"),
        ("tool-b", "running"),
    ]
    assert message["segments"] == [
        {"type": "tool", "toolIndex": 0},
        {"type": "tool", "toolIndex": 1},
    ]

    _apply_web_agent_display_event(
        {"type": "tool", "tool_id": "tool-b", "status": "error"},
        message,
    )
    _apply_web_agent_display_event({"type": "done"}, message)

    assert [tool["status"] for tool in message["tools"]] == ["done", "error"]


def test_agent_chat_display_schema_preserves_ordered_segments():
    """前端回传会话快照时应保留文字和工具的有序片段。"""
    payload = schemas.AgentChatDisplaySaveRequest(
        messages=[
            {
                "id": "assistant-1",
                "role": "assistant",
                "content": "先检查检查完成",
                "createdAt": 1,
                "status": "done",
                "tools": [{"id": "tool-1", "message": "执行检查", "status": "done"}],
                "segments": [
                    {"type": "text", "content": "先检查"},
                    {"type": "tool", "toolIndex": 0},
                    {"type": "text", "content": "检查完成"},
                ],
                "steering_message_id": "steering-1",
            }
        ]
    )

    assert payload.messages[0].model_dump()["segments"] == [
        {"type": "text", "content": "先检查", "toolIndex": None},
        {"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():
    """WebAgent 用户输入附件应转换为可展示的附件记录。"""
    attachments = _build_web_agent_input_attachments(
        images=["data:image/png;base64,abc"],
        files=[
            {
                "ref": "message/agent/file/file-1",
                "name": "report.txt",
                "mime_type": "text/plain",
                "size": 5,
            }
        ],
        audio_refs=["message/agent/file/audio-1"],
    )

    assert [item["kind"] for item in attachments] == ["image", "file", "audio"]
    assert attachments[1]["name"] == "report.txt"


def test_build_web_agent_command_items_returns_slash_commands():
    """WebAgent 命令建议应返回可展示的斜杠命令。"""
    with patch(
        "app.application.messaging.agent.get_commands",
        return_value={
            "/sites": {"description": "管理站点", "category": "站点"},
            "hidden": {"description": "忽略", "category": "其他"},
            "/hidden": {"description": "隐藏", "category": "其他", "show": False},
        },
    ):
        commands = _build_web_agent_command_items()

    assert commands == [
        {
            "command": "/sites",
            "description": "管理站点",
            "category": "站点",
            "type": "",
            "pid": None,
        }
    ]


def test_build_web_agent_command_items_includes_sites_command():
    """WebAgent 命令建议应包含内建站点管理命令。"""
    with patch(
        "app.application.messaging.agent.get_commands",
        return_value={
            "/sites": {"description": "管理站点", "category": "站点"},
        },
    ):
        commands = _build_web_agent_command_items()

    assert any(command["command"] == "/sites" for command in commands)


def test_build_web_agent_traditional_callback_payload_wraps_callback():
    """传统按钮回调应包装为可继续提交给 MessageChain 的消息。"""
    payload = _build_web_agent_traditional_callback_payload(
        "skills:req-1:root",
        original_message_id="assistant-1",
        original_chat_id="web-session",
    )

    assert payload["message"] == "CALLBACK:skills:req-1:root"
    assert payload["traditional"] is True
    assert payload["original_message_id"] == "assistant-1"
    assert payload["original_chat_id"] == "web-session"


def test_web_agent_stream_returns_error_for_unknown_command():
    """不存在的 WebAgent 斜杠命令应立即返回错误，不进入等待队列。"""
    payload = schemas.AgentWebChatRequest(
        text="/missing_command 参数",
        session_id="browser-session",
    )
    request = SimpleNamespace()
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)

    with (
        patch(
            "app.application.messaging.agent.get_command",
            return_value=None,
        ),
        patch("app.chain.message.MessageChain.handle_message") as handle_message,
    ):
        response = asyncio.run(web_agent_stream(payload, request, user))
        body = "".join(asyncio.run(_collect_streaming_response(response)))

    assert "error" in body
    assert "命令不存在：/missing_command" in body
    handle_message.assert_not_called()


def test_web_agent_stream_does_not_bind_request_scoped_chat_service():
    """流式路由不能把请求级 Agent 会话服务捕获到后台任务。"""
    from app.api.dependencies.agent import get_agent_chat_service
    from app.api.endpoints import agent as agent_endpoint

    route = next(route for route in agent_endpoint.router.routes if getattr(route, "name", None) == "web_agent_stream")

    assert all(dependency.call is not get_agent_chat_service for dependency in route.dependant.dependencies)


def test_build_web_agent_message_update_event_converts_buttons():
    """WebAgent 编辑消息应转换为可原地更新卡片的事件。"""
    event = build_web_agent_message_update_event(
        message_id="assistant-1",
        title="技能管理",
        text="请选择操作",
        buttons=[[{"text": "返回", "callback_data": "skills:req-1:root"}]],
    )

    assert event["type"] == "message_update"
    assert event["target_message"]["id"] == "assistant-1"
    assert event["target_message"]["choices"][0]["title"] == "技能管理"
    assert event["target_message"]["choices"][0]["prompt"] == "请选择操作"
    assert event["target_message"]["choices"][0]["buttons"][0]["label"] == "返回"


def test_build_web_agent_display_message_from_events_marks_done():
    """传统消息事件应聚合为完成态助手展示消息。"""
    message = _build_web_agent_display_message_from_events(
        [
            {"type": "delta", "content": "菜单"},
            {
                "type": "choice",
                "choice": {
                    "id": "choice-1",
                    "prompt": "请选择",
                    "buttons": [{"label": "返回", "callback_data": "back"}],
                },
            },
        ]
    )

    assert message["content"] == "菜单"
    assert message["status"] == "done"
    assert message["choices"][0]["prompt"] == "请选择"


def test_has_web_agent_traditional_interaction_detects_pending_skills():
    """WebAgent 应能识别命令后的传统交互上下文。"""
    skill_interaction_manager.clear()
    try:
        skill_interaction_manager.create_or_replace(
            user_id="1",
            channel=NotificationChannel.WebAgent,
            source="web-agent",
            username="admin",
        )

        assert _has_web_agent_traditional_interaction("1") is True
        assert _has_web_agent_traditional_interaction("2") is False
    finally:
        skill_interaction_manager.clear()


def test_web_agent_admin_context_uses_current_user_id():
    """Web Agent 工具权限应按当前登录用户 ID 判断管理员身份。"""
    agent = _get_web_agent_type()(
        session_id="web-agent:session",
        user_id="7",
        channel=NotificationChannel.WebAgent.value,
        source="web-agent",
        username="normal-user",
        replay_mode=ReplyMode.CAPTURE_ONLY,
    )

    lookup_fn = Mock(return_value=SimpleNamespace(is_superuser=True))
    with patch(
        "app.agent.web.get_configured_user_id_lookup",
        return_value=lookup_fn,
    ) as lookup:
        assert asyncio.run(agent._is_system_admin_context()) is True
        lookup.assert_called_once_with()
        lookup_fn.assert_called_once_with(7)


def test_web_agent_reused_for_background_task_disables_streaming():
    """Web Agent 被后台任务复用且渠道已清空时应改用非流式广播。"""
    agent = _get_web_agent_type()(
        session_id="web-agent:scheduled-session",
        user_id="7",
        channel=None,
        source=None,
        username="admin",
        replay_mode=ReplyMode.DISPATCH,
    )

    assert agent.is_background is True
    assert agent._should_stream() is False


def test_web_agent_output_callback_receives_only_new_text():
    """WebAgent 外部回调应接收增量，同时内部仍保留完整输出。"""
    outputs = []
    agent = _get_web_agent_type()(
        session_id="web-agent:incremental-output",
        user_id="7",
        channel=NotificationChannel.WebAgent.value,
        source="web-agent",
        username="admin",
        replay_mode=ReplyMode.CAPTURE_ONLY,
        output_callback=outputs.append,
    )

    agent._handle_stream_text("你")
    agent._handle_stream_text("好")

    assert outputs == ["你", "好"]
    assert agent._streamed_output == "你好"


def test_web_agent_tool_summary_is_emitted_before_following_text():
    """Web 工具状态应在调用发生时输出，不能拖到正文结束后。"""
    outputs = []
    agent = _get_web_agent_type()(
        session_id="web-agent:tool-order",
        user_id="7",
        channel=NotificationChannel.WebAgent.value,
        source="web-agent",
        username="admin",
        replay_mode=ReplyMode.CAPTURE_ONLY,
        output_callback=outputs.append,
    )

    agent.stream_handler.record_tool_call(
        "moviepilot_api",
        tool_kwargs={"operation_id": "scheduler.list"},
    )
    agent._handle_stream_text("查询完成。")

    assert outputs == ["（查询了 1 次数据）\n\n", "查询完成。"]


def test_web_agent_middleware_tool_emits_detail_in_verbose_mode():
    """中间件私有工具在啰嗦模式下应立即输出逐条工具明细。"""
    outputs = []
    agent = _get_web_agent_type()(
        session_id="web-agent:middleware-tool-detail",
        user_id="7",
        channel=NotificationChannel.WebAgent.value,
        source="web-agent",
        username="admin",
        replay_mode=ReplyMode.CAPTURE_ONLY,
        output_callback=outputs.append,
    )

    with patch("app.agent.callback.get_runtime_setting", return_value=True):
        agent.stream_handler.report_tool_call(
            "read_skill",
            tool_message="读取技能说明：moviepilot-api",
            tool_kwargs={"name": "moviepilot-api"},
        )

    assert outputs == ["\n\n⚙️ => 读取技能说明：moviepilot-api\n\n"]


def test_web_agent_channel_supports_streaming_and_attachments():
    """WebAgent 渠道应声明流式、多媒体和文件发送能力。"""
    assert ChannelCapabilityManager.supports_capability(NotificationChannel.WebAgent, ChannelCapability.INLINE_BUTTONS)
    assert ChannelCapabilityManager.supports_capability(
        NotificationChannel.WebAgent, ChannelCapability.CALLBACK_QUERIES
    )
    assert ChannelCapabilityManager.supports_capability(NotificationChannel.WebAgent, ChannelCapability.MESSAGE_EDITING)
    assert ChannelCapabilityManager.supports_capability(NotificationChannel.WebAgent, ChannelCapability.IMAGES)
    assert ChannelCapabilityManager.supports_capability(NotificationChannel.WebAgent, ChannelCapability.AUDIO_OUTPUT)
    assert ChannelCapabilityManager.supports_capability(NotificationChannel.WebAgent, ChannelCapability.FILE_SENDING)


def test_build_web_agent_message_events_extracts_image():
    """Agent 工具发送图片消息时应转换为图片附件事件。"""
    events = _build_web_agent_message_events(
        schemas.Message(
            channel=NotificationChannel.WebAgent,
            mtype=MessageType.Agent,
            title="海报",
            text="已找到图片",
            image="https://example.com/poster.jpg",
        )
    )

    assert events == [
        {"type": "delta", "content": "海报\n\n已找到图片"},
        {
            "type": "attachment",
            "attachment": {
                "kind": "image",
                "url": "https://example.com/poster.jpg",
                "download_url": "https://example.com/poster.jpg",
                "name": "海报",
                "mime_type": None,
            },
        },
    ]


def test_extract_web_agent_message_supports_wrapped_message_event():
    """NoticeMessage 包装 Message 时应仍能解析为 WebAgent 通知。"""
    message = schemas.Message(
        channel=NotificationChannel.WebAgent,
        source="web-agent",
        title="会话状态",
        userid="1",
    )

    extracted = extract_web_agent_message_from_event_data({"message": message, "current_time": "2026-06-26 09:18:38"})

    assert extracted == message


def test_dispatch_web_agent_message_event_accepts_wrapped_message_event():
    """WebAgent 等待队列应接收 message 包装格式的 NoticeMessage 事件。"""
    notice_queue = Queue()
    attach_web_agent_message_queue("1", notice_queue)
    message = schemas.Message(
        channel=NotificationChannel.WebAgent,
        source="web-agent",
        title="会话状态",
        userid="1",
    )

    try:
        dispatch_web_agent_message_event(
            Event(
                EventType.NoticeMessage,
                {"message": message, "current_time": "2026-06-26 09:18:38"},
            )
        )
    finally:
        detach_web_agent_message_queue("1", notice_queue)

    assert notice_queue.get_nowait() == message


def test_collect_web_agent_traditional_events_does_not_emit_submit_hint():
    """传统命令未产生通知时不应返回“命令已提交”的兜底提示。"""
    user = SimpleNamespace(id=1, name="admin")

    with (
        patch(
            "app.chain.message.MessageChain.handle_message",
        ),
        patch(
            "app.application.messaging.agent.WEB_AGENT_TRADITIONAL_IDLE_TIMEOUT_SECONDS",
            0.01,
        ),
        patch(
            "app.application.messaging.agent.WEB_AGENT_TRADITIONAL_MAX_WAIT_SECONDS",
            0.05,
        ),
    ):
        events = asyncio.run(
            _collect_web_agent_traditional_events(
                text="/session_status",
                current_user=user,
            )
        )

    assert events == []


def test_build_web_agent_message_events_registers_local_file(tmp_path):
    """Agent 工具发送本地文件时应生成可下载附件事件。"""
    file_path = tmp_path / "report.txt"
    file_path.write_text("hello", encoding="utf-8")

    events = _build_web_agent_message_events(
        schemas.Message(
            channel=NotificationChannel.WebAgent,
            mtype=MessageType.Agent,
            file_path=str(file_path),
            file_name="report.txt",
        )
    )

    assert len(events) == 1
    attachment = events[0]["attachment"]
    assert events[0]["type"] == "attachment"
    assert attachment["kind"] == "file"
    assert attachment["name"] == "report.txt"
    assert attachment["mime_type"] == "text/plain"
    assert attachment["size"] == 5
    assert attachment["url"].startswith("message/agent/file/")


def test_build_web_agent_message_events_registers_voice_attachment(tmp_path):
    """Agent 工具发送语音时应转换为可播放的音频附件事件。"""
    voice_path = tmp_path / "reply.wav"
    voice_path.write_bytes(b"wav-bytes")

    events = _build_web_agent_message_events(
        schemas.Message(
            channel=NotificationChannel.WebAgent,
            mtype=MessageType.Agent,
            text="你好",
            voice_path=str(voice_path),
        )
    )

    assert len(events) == 2
    assert events[0] == {"type": "delta", "content": "你好"}
    attachment = events[1]["attachment"]
    assert events[1]["type"] == "attachment"
    assert attachment["kind"] == "audio"
    assert attachment["name"] == "reply.wav"
    assert attachment["mime_type"] == "audio/wav"
    assert attachment["size"] == len(b"wav-bytes")
    assert attachment["url"].startswith("message/agent/file/")


def test_prepare_web_agent_audio_attachment_async_keeps_loop_responsive(tmp_path):
    """异步转码等待期间事件循环仍应可调度其它任务。"""
    source_path = tmp_path / "reply.opus"
    source_path.write_bytes(b"opus-bytes")
    converted_path = tmp_path / "voice" / "reply_web_abcdef12.wav"
    started = asyncio.Event()
    release = asyncio.Event()

    class FakeProcess:
        returncode = 0

        async def communicate(self):
            started.set()
            await release.wait()
            converted_path.write_bytes(b"wav-bytes")
            return b"", b""

    async def fake_create_subprocess_exec(*args, **kwargs):
        assert args[0] == "/usr/bin/ffmpeg"
        return FakeProcess()

    async def scenario():
        with (
            patch(
                "app.application.messaging.agent.shutil.which",
                return_value="/usr/bin/ffmpeg",
            ),
            patch(
                "app.application.messaging.agent.uuid.uuid4",
                return_value=SimpleNamespace(hex="abcdef1234567890"),
            ),
            patch(
                "app.application.messaging.agent.asyncio.create_subprocess_exec",
                side_effect=fake_create_subprocess_exec,
            ),
            patch(
                "app.application.messaging.agent.get_api_runtime_config_snapshot",
                return_value=SimpleNamespace(temp_path=tmp_path),
            ),
        ):
            conversion_task = asyncio.create_task(_prepare_web_agent_audio_attachment_path_async(str(source_path)))
            await asyncio.wait_for(started.wait(), timeout=1)
            heartbeat = asyncio.create_task(asyncio.sleep(0))
            await heartbeat
            assert not conversion_task.done()
            release.set()
            return await conversion_task

    output_path = asyncio.run(scenario())

    assert output_path == converted_path
    assert output_path.read_bytes() == b"wav-bytes"


def test_prepare_web_agent_audio_attachment_async_cancellation_reaps_process(tmp_path):
    """取消 WebAgent 转码时应终止并回收 ffmpeg，不能留下半成品。"""
    source_path = tmp_path / "reply.opus"
    source_path.write_bytes(b"opus-bytes")
    output_path = tmp_path / "voice" / "reply_web_abcdef12.wav"
    started = asyncio.Event()
    killed = False

    class FakeProcess:
        returncode = None
        _release = asyncio.Event()

        def kill(self):
            nonlocal killed
            killed = True
            self.returncode = -9
            self._release.set()

        async def communicate(self):
            started.set()
            if self.returncode is None:
                await self._release.wait()
            return b"", b""

    async def fake_create_subprocess_exec(*args, **kwargs):
        return FakeProcess()

    async def scenario():
        with (
            patch(
                "app.application.messaging.agent.shutil.which",
                return_value="/usr/bin/ffmpeg",
            ),
            patch(
                "app.application.messaging.agent.uuid.uuid4",
                return_value=SimpleNamespace(hex="abcdef1234567890"),
            ),
            patch(
                "app.application.messaging.agent.asyncio.create_subprocess_exec",
                side_effect=fake_create_subprocess_exec,
            ),
            patch(
                "app.application.messaging.agent.get_api_runtime_config_snapshot",
                return_value=SimpleNamespace(temp_path=tmp_path),
            ),
        ):
            conversion_task = asyncio.create_task(_prepare_web_agent_audio_attachment_path_async(str(source_path)))
            await asyncio.wait_for(started.wait(), timeout=5)
            conversion_task.cancel()
            with pytest.raises(asyncio.CancelledError):
                await conversion_task

    asyncio.run(scenario())

    assert killed is True
    assert not output_path.exists()


def test_prepare_web_agent_audio_attachment_async_communicate_error_reaps_process(tmp_path):
    """ffmpeg 通信异常时应终止仍运行的进程并回退原文件。"""
    source_path = tmp_path / "reply.opus"
    source_path.write_bytes(b"opus-bytes")
    started = asyncio.Event()
    killed = False
    communicate_calls = 0

    class FakeProcess:
        returncode = None

        def kill(self):
            nonlocal killed
            killed = True
            self.returncode = -9

        async def communicate(self):
            nonlocal communicate_calls
            communicate_calls += 1
            started.set()
            if self.returncode is None:
                raise OSError("pipe closed")
            return b"", b""

    async def fake_create_subprocess_exec(*args, **kwargs):
        return FakeProcess()

    async def scenario():
        with (
            patch(
                "app.application.messaging.agent.shutil.which",
                return_value="/usr/bin/ffmpeg",
            ),
            patch(
                "app.application.messaging.agent.asyncio.create_subprocess_exec",
                side_effect=fake_create_subprocess_exec,
            ),
            patch(
                "app.application.messaging.agent.get_api_runtime_config_snapshot",
                return_value=SimpleNamespace(temp_path=tmp_path),
            ),
        ):
            output_path = await _prepare_web_agent_audio_attachment_path_async(str(source_path))
            return output_path

    output_path = asyncio.run(scenario())

    assert output_path == source_path
    assert killed is True
    assert communicate_calls == 2


def test_prepare_web_agent_audio_attachment_async_cancellation_cleans_completed_output(
    tmp_path,
):
    """转码完成后检查产物期间取消，也应清理未登记的 WAV。"""
    source_path = tmp_path / "reply.opus"
    source_path.write_bytes(b"opus-bytes")
    output_path = tmp_path / "voice" / "reply_web_abcdef12.wav"
    exists_started = asyncio.Event()

    class FakeProcess:
        returncode = 0

        async def communicate(self):
            output_path.parent.mkdir(parents=True, exist_ok=True)
            output_path.write_bytes(b"wav-bytes")
            return b"", b""

    async def fake_create_subprocess_exec(*args, **kwargs):
        return FakeProcess()

    async def fake_run_in_threadpool(func, *args, **kwargs):
        if getattr(func, "__name__", "") == "exists":
            exists_started.set()
            await asyncio.Event().wait()
        return await asyncio.to_thread(func, *args, **kwargs)

    async def scenario():
        with (
            patch(
                "app.application.messaging.agent.shutil.which",
                return_value="/usr/bin/ffmpeg",
            ),
            patch(
                "app.application.messaging.agent.uuid.uuid4",
                return_value=SimpleNamespace(hex="abcdef1234567890"),
            ),
            patch(
                "app.application.messaging.agent.asyncio.create_subprocess_exec",
                side_effect=fake_create_subprocess_exec,
            ),
            patch(
                "app.application.messaging.agent.get_api_runtime_config_snapshot",
                return_value=SimpleNamespace(temp_path=tmp_path),
            ),
            patch(
                "app.application.messaging.agent.run_in_threadpool",
                side_effect=fake_run_in_threadpool,
            ),
        ):
            conversion_task = asyncio.create_task(_prepare_web_agent_audio_attachment_path_async(str(source_path)))
            await asyncio.wait_for(exists_started.wait(), timeout=1)
            assert output_path.exists()
            conversion_task.cancel()
            with pytest.raises(asyncio.CancelledError):
                await conversion_task

    asyncio.run(scenario())

    assert not output_path.exists()


def test_prepare_web_agent_audio_attachment_async_timeout_falls_back(tmp_path):
    """转码超时应回退原文件并回收 ffmpeg。"""
    source_path = tmp_path / "reply.opus"
    source_path.write_bytes(b"opus-bytes")
    started = asyncio.Event()
    killed = False

    class FakeProcess:
        returncode = None
        _release = asyncio.Event()

        def kill(self):
            nonlocal killed
            killed = True
            self.returncode = -9
            self._release.set()

        async def communicate(self):
            started.set()
            if self.returncode is None:
                await self._release.wait()
            return b"", b""

    async def fake_create_subprocess_exec(*args, **kwargs):
        return FakeProcess()

    async def scenario():
        with (
            patch(
                "app.application.messaging.agent.shutil.which",
                return_value="/usr/bin/ffmpeg",
            ),
            patch(
                "app.application.messaging.agent.asyncio.create_subprocess_exec",
                side_effect=fake_create_subprocess_exec,
            ),
            patch(
                "app.application.messaging.agent.get_api_runtime_config_snapshot",
                return_value=SimpleNamespace(temp_path=tmp_path),
            ),
            patch(
                "app.application.messaging.agent.WEB_AGENT_AUDIO_CONVERSION_TIMEOUT_SECONDS",
                0.01,
            ),
        ):
            conversion_task = asyncio.create_task(_prepare_web_agent_audio_attachment_path_async(str(source_path)))
            await asyncio.wait_for(started.wait(), timeout=1)
            return await conversion_task

    output_path = asyncio.run(scenario())

    assert output_path == source_path
    assert killed is True


def test_transcribe_web_agent_audio_files_reads_registered_upload(tmp_path):
    """WebAgent 上传录音应从临时附件登记表读取并转写为文本。"""
    voice_path = tmp_path / "recording.webm"
    voice_path.write_bytes(b"webm-bytes")
    _WEB_AGENT_FILE_REGISTRY["audio-test"] = {
        "path": voice_path,
        "name": "recording.webm",
        "mime_type": "audio/webm",
        "created_at": time.time(),
    }

    try:
        with (
            patch(
                "app.application.messaging.agent.is_audio_input_available",
                return_value=True,
            ),
            patch(
                "app.application.messaging.agent.transcribe_audio",
                return_value="帮我推荐一部电影",
            ) as transcribe_audio,
        ):
            audio_files = _resolve_web_agent_audio_refs(["message/agent/file/audio-test"])
            _WEB_AGENT_FILE_REGISTRY.pop("audio-test", None)
            transcript = _transcribe_web_agent_audio_files(audio_files)
    finally:
        _WEB_AGENT_FILE_REGISTRY.pop("audio-test", None)

    assert transcript == "帮我推荐一部电影"
    transcribe_audio.assert_called_once_with(
        content=b"webm-bytes",
        filename="recording.webm",
    )


def test_web_agent_stream_returns_error_when_voice_transcription_fails():
    """仅发送语音且转写失败时应直接返回错误事件。"""
    payload = schemas.AgentWebChatRequest(
        text="",
        session_id="browser-session",
        audio_refs=["message/agent/file/missing"],
    )
    request = SimpleNamespace()
    user = SimpleNamespace(id=1, name="admin")

    with (
        patch(
            "app.application.messaging.agent.get_api_runtime_config_snapshot",
            return_value=SimpleNamespace(ai_agent_enable=True),
        ),
        patch(
            "app.application.messaging.agent.transcribe_web_agent_audio_files",
            return_value=None,
        ) as transcribe_audio,
    ):
        response = asyncio.run(web_agent_stream(payload, request, user))
        body = "".join(asyncio.run(_collect_streaming_response(response)))

    transcribe_audio.assert_not_called()
    assert "error" in body
    assert "语音识别失败" in body


def test_web_agent_stream_does_not_block_event_loop_during_transcription():
    """同步音频 provider 等待时，事件循环仍应让其他任务获得执行机会。"""
    payload = schemas.AgentWebChatRequest(
        text="",
        session_id="browser-session",
        audio_refs=["message/agent/file/audio-test"],
    )
    request = SimpleNamespace(headers={})
    user = SimpleNamespace(id=1, name="admin")

    transcription_started = ThreadEvent()
    transcription_release = ThreadEvent()

    def blocking_transcription(_audio_refs):
        transcription_started.set()
        assert transcription_release.wait(timeout=2)
        return None

    async def scenario():
        stream_task = asyncio.create_task(web_agent_stream(payload, request, user))
        assert await asyncio.to_thread(transcription_started.wait, 1)
        assert stream_task.done() is False
        transcription_release.set()
        return await stream_task

    try:
        with (
            patch(
                "app.application.messaging.agent.get_api_runtime_config_snapshot",
                return_value=SimpleNamespace(ai_agent_enable=True),
            ),
            patch(
                "app.application.messaging.agent.resolve_web_agent_audio_refs",
                return_value=[Mock()],
            ),
            patch(
                "app.application.messaging.agent.transcribe_web_agent_audio_files",
                side_effect=blocking_transcription,
            ),
        ):
            response = asyncio.run(scenario())
    finally:
        transcription_release.set()

    body = "".join(asyncio.run(_collect_streaming_response(response)))
    assert "语音识别失败" in body


def test_web_agent_stream_binds_session_to_agent_manager():
    """WebAgent 普通对话应统一进入 AgentManager 并绑定远程命令会话。"""
    payload = schemas.AgentWebChatRequest(
        text="查看会话",
        session_id="browser-session",
    )
    request = SimpleNamespace(is_disconnected=AsyncMock(return_value=False))
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)

    class FakeWebAgent:
        """测试用 WebAgent，模拟 AgentManager 内部的持久实例。"""

        def __init__(self, **kwargs):
            self.__dict__.update(kwargs)
            self.processed = []

        def set_output_callback(self, output_callback):
            """更新当前 SSE 输出回调。"""
            self.output_callback = output_callback

        def set_protected_output_callback(self, protected_output_callback):
            """更新当前 SSE 受保护输出回调。"""
            self.protected_output_callback = protected_output_callback

        def set_message_callback(self, message_callback):
            """更新当前 SSE 通知回调。"""
            self.message_callback = message_callback

        async def process(self, message, **kwargs):
            """模拟一次 WebAgent 推理输出。"""
            self.processed.append((message, kwargs))
            self.output_callback("状态正常")
            return "状态正常"

        async def cleanup(self):
            """模拟 Agent 资源清理。"""
            return None

    session_id = _build_web_agent_session_id(user, payload.session_id)
    MessageChain._user_sessions.clear()
    agent_manager.active_agents.pop(session_id, None)
    agent_manager._session_queues.pop(session_id, None)
    worker = agent_manager._session_workers.pop(session_id, None)
    if worker:
        worker.cancel()

    async def scenario():
        response = await web_agent_stream(payload, request, user)
        return "".join(await _collect_streaming_response(response))

    try:
        with (
            patch(
                "app.application.messaging.agent.get_api_runtime_config_snapshot",
                return_value=SimpleNamespace(ai_agent_enable=True),
            ),
            patch(
                "app.application.agent.get_web_agent_type",
                return_value=FakeWebAgent,
            ),
        ):
            body = asyncio.run(scenario())

        assert "状态正常" in body
        assert MessageChain._user_sessions["1"][0] == session_id
        assert isinstance(agent_manager.active_agents[session_id], FakeWebAgent)
    finally:
        MessageChain._user_sessions.clear()
        agent = agent_manager.active_agents.pop(session_id, None)
        if agent:
            asyncio.run(agent.cleanup())
        agent_manager._session_queues.pop(session_id, None)
        worker = agent_manager._session_workers.pop(session_id, None)
        if worker:
            worker.cancel()


def test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream():
    """第一条 WebAgent 流运行时的第二条请求应在真实应用点切分助手回合。"""
    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 == "开始长任务":
                    if callable(self.tool_event_callback):
                        self.tool_event_callback(
                            {
                                "type": "tool",
                                "status": "running",
                                "tool_id": "tool-steering-boundary",
                                "tool_name": "search",
                                "message": "检查任务状态",
                            }
                        )
                    self.output_callback("首轮处理中")
                    if callable(self.tool_event_callback):
                        self.tool_event_callback(
                            {
                                "type": "tool",
                                "status": "done",
                                "tool_id": "tool-steering-boundary",
                            }
                        )
                    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 '"assistant_message_id":' 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", "assistant", "user", "assistant"]
    assert messages[2]["steering_message_id"]
    assert messages[1]["status"] == "done"
    assert messages[3]["status"] == "done"
    assert messages[1]["content"] == "首轮处理中"
    assert messages[3]["content"] == "补充已应用"
    assert messages[1]["tools"] == [
        {
            "id": "tool-steering-boundary",
            "tool_name": "search",
            "message": "检查任务状态",
            "status": "done",
        }
    ]


def test_web_agent_stream_emits_secret_result_only_as_protected_event():
    """敏感结果只能进入命名 protected SSE，不能进入普通快照。"""
    secret_marker = "WEB_SECRET_MARKER **literal** <img src=x>"
    payload = schemas.AgentWebChatRequest(
        text="确认",
        session_id="browser-secret",
        echo_user=True,
    )
    request = SimpleNamespace(
        headers={"X-MoviePilot-Agent-Interaction": "1"},
        is_disconnected=AsyncMock(return_value=False),
    )
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)

    class FakeProtectedAgent:
        """直接触发受保护输出的 WebAgent 测试替身。"""

        def __init__(self, **kwargs):
            self.__dict__.update(kwargs)
            self._pending_secret_confirmation = SimpleNamespace(
                channel=NotificationChannel.WebAgent.value,
                source="web-agent",
                original_chat_id="",
            )

        def has_pending_secret_confirmation(self):
            """模拟当前会话存在有效的敏感读取确认。"""
            return self._pending_secret_confirmation is not None

        def set_output_callback(self, output_callback):
            self.output_callback = output_callback

        def set_message_callback(self, message_callback):
            self.message_callback = message_callback

        def set_protected_output_callback(self, protected_output_callback):
            self.protected_output_callback = protected_output_callback

        async def process(self, _message, **_kwargs):
            self.protected_output_callback(secret_marker)
            return "敏感设置确认已处理。"

        async def cleanup(self):
            return None

    session_id = _build_web_agent_session_id(user, payload.session_id)
    existing_messages = [
        {"role": "user", "content": "此前的问题", "status": "done"},
        {"role": "assistant", "content": "此前的回答", "status": "done"},
    ]
    existing_chat = AgentChatOper().save_display_messages(
        session_id=session_id,
        user_id="1",
        username="admin",
        channel=NotificationChannel.WebAgent.value,
        source="web-agent",
        messages=existing_messages,
        client_session_id=payload.session_id,
    )
    agent_manager.active_agents[session_id] = FakeProtectedAgent(
        session_id=session_id,
        user_id="1",
        channel=NotificationChannel.WebAgent.value,
        source="web-agent",
        username="admin",
    )
    agent_manager._session_queues.pop(session_id, None)
    worker = agent_manager._session_workers.pop(session_id, None)
    if worker:
        worker.cancel()

    async def scenario():
        response = await web_agent_stream(payload, request, user)
        return "".join(await _collect_streaming_response(response))

    try:
        with (
            patch(
                "app.application.messaging.agent.get_api_runtime_config_snapshot",
                return_value=SimpleNamespace(ai_agent_enable=True),
            ),
            patch(
                "app.application.agent.get_web_agent_type",
                return_value=FakeProtectedAgent,
            ),
            patch(
                "app.application.messaging.agent.save_web_agent_display_snapshot",
                new_callable=AsyncMock,
            ) as save_snapshot,
        ):
            body = asyncio.run(scenario())

        assert "event: interaction-protected\n" in body
        assert secret_marker in body
        save_snapshot.assert_not_called()
        preserved_chat = AgentChatOper().get(session_id=session_id, user_id="1")
        assert preserved_chat.display_messages == existing_messages
        assert preserved_chat.message_count == 2
        assert preserved_chat.preview == "此前的回答"
    finally:
        agent = agent_manager.active_agents.pop(session_id, None)
        if agent:
            asyncio.run(agent.cleanup())
        agent_manager._session_queues.pop(session_id, None)
        worker = agent_manager._session_workers.pop(session_id, None)
        if worker:
            worker.cancel()
        AgentChatOper().delete_by_id(existing_chat.id)


def test_web_agent_cancel_keeps_existing_display_history():
    """取消敏感读取不得覆盖当前会话已有的普通展示历史。"""
    payload = schemas.AgentWebChatRequest(
        text="取消",
        session_id="browser-secret-cancel",
        echo_user=True,
    )
    request = SimpleNamespace(
        headers={"X-MoviePilot-Agent-Interaction": "1"},
        is_disconnected=AsyncMock(return_value=False),
    )
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)
    session_id = _build_web_agent_session_id(user, payload.session_id)
    existing_messages = [
        {"role": "user", "content": "保留的问题", "status": "done"},
        {"role": "assistant", "content": "保留的回答", "status": "done"},
    ]
    existing_chat = AgentChatOper().save_display_messages(
        session_id=session_id,
        user_id="1",
        username="admin",
        channel=NotificationChannel.WebAgent.value,
        source="web-agent",
        messages=existing_messages,
        client_session_id=payload.session_id,
    )

    async def scenario():
        response = await web_agent_stream(payload, request, user)
        return "".join(await _collect_streaming_response(response))

    try:
        with (
            patch(
                "app.application.messaging.agent.get_api_runtime_config_snapshot",
                return_value=SimpleNamespace(ai_agent_enable=True),
            ),
            patch.object(
                agent_manager,
                "matches_secret_confirmation",
                return_value=True,
            ),
            patch.object(
                agent_manager,
                "process_message",
                new=AsyncMock(return_value="已取消敏感设置读取。"),
            ),
            patch(
                "app.application.messaging.agent.save_web_agent_display_snapshot",
                new_callable=AsyncMock,
            ) as save_snapshot,
        ):
            body = asyncio.run(scenario())

        assert '"type": "done"' in body
        save_snapshot.assert_not_called()
        preserved_chat = AgentChatOper().get(session_id=session_id, user_id="1")
        assert preserved_chat.display_messages == existing_messages
        assert preserved_chat.message_count == 2
        assert preserved_chat.preview == "保留的回答"
    finally:
        AgentChatOper().delete_by_id(existing_chat.id)


def test_web_agent_stream_rejects_confirmation_without_protected_capability():
    """旧客户端未声明 protected 能力时不得把确认交给 Agent。"""
    payload = schemas.AgentWebChatRequest(
        text="确认",
        session_id="browser-secret-legacy",
        echo_user=False,
    )
    request = SimpleNamespace(
        headers={},
        is_disconnected=AsyncMock(return_value=False),
    )
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)

    async def scenario():
        response = await web_agent_stream(payload, request, user)
        return "".join(await _collect_streaming_response(response))

    with (
        patch(
            "app.application.messaging.agent.get_api_runtime_config_snapshot",
            return_value=SimpleNamespace(ai_agent_enable=True),
        ),
        patch.object(
            agent_manager,
            "matches_secret_confirmation",
            return_value=True,
        ),
        patch.object(agent_manager, "process_message", new=AsyncMock()) as process,
    ):
        body = asyncio.run(scenario())

    assert "不支持安全交付" in body
    process.assert_not_awaited()


def test_web_agent_stream_keeps_confirmation_without_pending_on_normal_path():
    """无待确认操作时，纯文本确认仍是普通 Agent 消息。"""
    payload = schemas.AgentWebChatRequest(
        text="确认",
        session_id="browser-ordinary-confirmation",
        echo_user=True,
    )
    request = SimpleNamespace(headers={}, is_disconnected=AsyncMock(return_value=False))
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)

    async def scenario():
        response = await web_agent_stream(payload, request, user)
        body = "".join(await _collect_streaming_response(response))
        return response, body

    with (
        patch(
            "app.application.messaging.agent.get_api_runtime_config_snapshot",
            return_value=SimpleNamespace(ai_agent_enable=True),
        ),
        patch.object(
            agent_manager,
            "process_message",
            new=AsyncMock(return_value="普通回复"),
        ) as process,
    ):
        response, body = asyncio.run(scenario())

    assert "不支持安全交付" not in body
    assert response.headers.get("X-MoviePilot-Agent-Control") is None
    process.assert_awaited_once()


def test_web_agent_stream_drops_secret_result_after_disconnect():
    """确认请求断线后不改造通用队列，并拒绝向关闭的连接投递密钥。"""
    payload = schemas.AgentWebChatRequest(
        text="确认",
        session_id="browser-secret-disconnect",
        echo_user=True,
    )

    agent_started = asyncio.Event()
    release_agent = asyncio.Event()
    agent_completed = asyncio.Event()

    async def disconnect_after_agent_starts():
        """等待确认进入处理流程后再模拟浏览器断线。"""
        await agent_started.wait()
        return True

    request = SimpleNamespace(
        headers={"X-MoviePilot-Agent-Interaction": "1"},
        is_disconnected=AsyncMock(side_effect=disconnect_after_agent_starts),
    )
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)
    session_id = _build_web_agent_session_id(user, payload.session_id)
    existing_messages = [
        {"role": "user", "content": "断线前的问题", "status": "done"},
        {"role": "assistant", "content": "断线前的回答", "status": "done"},
    ]
    existing_chat = AgentChatOper().save_display_messages(
        session_id=session_id,
        user_id="1",
        username="admin",
        channel=NotificationChannel.WebAgent.value,
        source="web-agent",
        messages=existing_messages,
        client_session_id=payload.session_id,
    )

    delivery_results = []

    async def finish_after_disconnect(**kwargs):
        """断线后继续完成只读任务，并尝试向已关闭发布器投递。"""
        agent_started.set()
        await release_agent.wait()
        delivery_results.append(kwargs["protected_output_callback"]("DISCONNECTED_SECRET_MARKER"))
        agent_completed.set()

    async def scenario():
        response = await web_agent_stream(payload, request, user)
        body = "".join(
            await _collect_streaming_response(
                response,
                wait_for_background=False,
            )
        )
        release_agent.set()
        await asyncio.wait_for(agent_completed.wait(), timeout=1)
        await wait_web_agent_background_tasks()
        return body

    try:
        with (
            patch(
                "app.application.messaging.agent.get_api_runtime_config_snapshot",
                return_value=SimpleNamespace(ai_agent_enable=True),
            ),
            patch.object(
                agent_manager,
                "matches_secret_confirmation",
                return_value=True,
            ),
            patch.object(
                agent_manager,
                "process_message",
                new=AsyncMock(side_effect=finish_after_disconnect),
            ) as process,
            patch(
                "app.application.messaging.agent.save_web_agent_display_snapshot",
                new_callable=AsyncMock,
            ) as save_snapshot,
        ):
            body = asyncio.run(scenario())

        assert '"type": "start"' in body
        assert delivery_results == [False]
        assert "cancel_on_waiter_cancel" not in process.await_args.kwargs
        save_snapshot.assert_not_called()
        preserved_chat = AgentChatOper().get(session_id=session_id, user_id="1")
        assert preserved_chat.display_messages == existing_messages
        assert preserved_chat.message_count == 2
        assert preserved_chat.preview == "断线前的回答"
    finally:
        AgentChatOper().delete_by_id(existing_chat.id)


def test_web_agent_stream_emits_heartbeat_during_idle_tool_wait():
    """长时间没有 Agent 事件时应发送 SSE heartbeat 保持连接。"""
    payload = schemas.AgentWebChatRequest(text="分析系统状态", session_id="browser-heartbeat")
    request = SimpleNamespace(is_disconnected=AsyncMock(return_value=False))
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)

    async def slow_process_message(**kwargs):
        """模拟工具执行期间暂时没有可见输出。"""
        await asyncio.sleep(0.035)
        kwargs["output_callback"]("状态正常")

    async def scenario():
        response = await web_agent_stream(payload, request, user)
        return "".join(await _collect_streaming_response(response))

    with (
        patch(
            "app.application.messaging.agent.get_api_runtime_config_snapshot",
            return_value=SimpleNamespace(ai_agent_enable=True),
        ),
        patch(
            "app.application.messaging.agent.WEB_AGENT_STREAM_HEARTBEAT_SECONDS",
            0.01,
        ),
        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="web-agent:heartbeat",
        ),
        patch.object(
            MessageChain,
            "bind_user_session",
        ),
        patch.object(
            agent_manager,
            "process_message",
            side_effect=slow_process_message,
        ),
        patch(
            "app.application.messaging.agent.save_web_agent_display_snapshot",
            new_callable=AsyncMock,
        ),
    ):
        body = asyncio.run(scenario())

    assert ": heartbeat\n\n" in body
    assert '"type": "delta"' in body
    assert '"type": "done"' in body


def test_web_agent_stop_finishes_stream_without_error():
    """停止运行中的 Web Agent 后应正常结束 SSE，不能继续等待或报执行错误。"""
    payload = schemas.AgentWebChatRequest(
        text="执行长任务",
        session_id="browser-stop",
    )
    request = SimpleNamespace(is_disconnected=AsyncMock(return_value=False))
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)
    session_id = "web-agent:stop"

    class BlockingWebAgent:
        """阻塞到会话 worker 被停止的 Web Agent 替身。"""

        started = None

        def __init__(self, **kwargs):
            self.__dict__.update(kwargs)

        async def process(self, _message, **_kwargs):
            """等待外层 worker 取消。"""
            self.started.set()
            await asyncio.Event().wait()

        async def cleanup(self):
            """模拟 Agent 资源清理。"""
            return None

    async def scenario():
        BlockingWebAgent.started = asyncio.Event()
        response = await web_agent_stream(payload, request, user)
        iterator = response.body_iterator.__aiter__()
        received = [await asyncio.wait_for(anext(iterator), timeout=1)]
        await asyncio.wait_for(BlockingWebAgent.started.wait(), timeout=1)

        assert await asyncio.wait_for(agent_manager.stop_current_task(session_id), timeout=1) is True
        while '"type": "done"' not in "".join(received):
            received.append(await asyncio.wait_for(anext(iterator), timeout=1))
        await iterator.aclose()
        await wait_web_agent_background_tasks()
        return "".join(received)

    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.object(
                MessageChain,
                "bind_user_session",
            ),
            patch(
                "app.application.agent.get_web_agent_type",
                return_value=BlockingWebAgent,
            ),
            patch(
                "app.application.messaging.agent.save_web_agent_display_snapshot",
                new_callable=AsyncMock,
            ),
        ):
            body = asyncio.run(scenario())
    finally:
        agent_manager._session_queues.pop(session_id, None)
        agent_manager._session_workers.pop(session_id, None)
        agent_manager.active_agents.pop(session_id, None)

    assert '"type": "done"' in body
    assert '"type": "error"' not in body


def test_web_agent_stream_rechecks_running_service_before_enqueue():
    """响应建立后服务若已关闭，生成器必须稳定返回错误且不向旧 manager 入队。"""
    payload = schemas.AgentWebChatRequest(
        text="检查状态",
        session_id="shutdown-race",
    )
    request = SimpleNamespace(is_disconnected=AsyncMock(return_value=False))
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)
    stale_manager = SimpleNamespace(process_message=AsyncMock())

    async def scenario():
        response = await web_agent_stream(payload, request, user)
        return "".join(await _collect_streaming_response(response))

    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.agent.get_running_agent_manager",
            side_effect=[stale_manager, None],
        ),
        patch(
            "app.application.messaging.agent.save_web_agent_display_snapshot",
            new_callable=AsyncMock,
        ),
    ):
        body = asyncio.run(scenario())

    assert '"type": "error"' in body
    assert '"type": "done"' in body
    stale_manager.process_message.assert_not_awaited()


def test_web_agent_traditional_stream_keeps_alive_and_saves_after_done():
    """传统消息等待期间应保活，且展示快照不能阻塞终态。"""
    payload = schemas.AgentWebChatRequest(text="/状态", session_id="traditional-heartbeat")
    request = SimpleNamespace(is_disconnected=AsyncMock(return_value=False))
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)
    snapshot_started = ThreadEvent()
    snapshot_release = ThreadEvent()
    snapshot_finished = ThreadEvent()

    async def slow_collect(**_kwargs):
        """模拟传统消息链路等待外部结果。"""
        await asyncio.sleep(0.035)
        return [{"type": "delta", "content": "状态正常"}]

    async def slow_snapshot(**_kwargs):
        """阻塞快照写入，便于断言 done 不等待落库。"""
        await asyncio.to_thread(snapshot_started.set)
        await asyncio.to_thread(snapshot_release.wait, 2)
        await asyncio.to_thread(snapshot_finished.set)

    async def scenario():
        response = await web_agent_stream(payload, request, user)
        assert response.headers["cache-control"] == "no-cache, no-transform"
        iterator = response.body_iterator.__aiter__()
        received = []
        while True:
            chunk = await asyncio.wait_for(anext(iterator), timeout=1)
            text = chunk.decode("utf-8") if isinstance(chunk, bytes) else chunk
            received.append(text)
            if '"type": "done"' in text:
                break

        for _ in range(100):
            if snapshot_started.is_set():
                break
            await asyncio.sleep(0.001)
        assert snapshot_started.is_set()
        assert not snapshot_finished.is_set()
        await iterator.aclose()
        assert not snapshot_finished.is_set()
        snapshot_release.set()
        await asyncio.to_thread(snapshot_finished.wait, 1)
        await wait_web_agent_background_tasks()
        return "".join(received)

    try:
        with (
            patch(
                "app.application.messaging.agent.WEB_AGENT_STREAM_HEARTBEAT_SECONDS",
                0.01,
            ),
            patch(
                "app.application.messaging.agent.is_web_agent_traditional_message",
                return_value=True,
            ),
            patch(
                "app.application.messaging.agent.ensure_web_agent_command_allowed",
                return_value=None,
            ),
            patch(
                "app.application.messaging.agent.get_web_agent_unknown_command_message",
                return_value=None,
            ),
            patch(
                "app.application.messaging.agent.build_web_agent_session_id_async",
                return_value="web-agent:traditional-heartbeat",
            ),
            patch(
                "app.application.messaging.agent.collect_web_agent_traditional_events",
                side_effect=slow_collect,
            ),
            patch(
                "app.application.messaging.agent.save_web_agent_display_snapshot",
                new_callable=AsyncMock,
                side_effect=slow_snapshot,
            ),
        ):
            body = asyncio.run(scenario())

        assert ": heartbeat\n\n" in body
        assert '"type": "delta"' in body
        assert '"type": "done"' in body
    finally:
        snapshot_release.set()

    assert snapshot_finished.wait(timeout=1)


def test_web_agent_traditional_stream_drains_collection_on_cancellation():
    """传统 SSE 被取消时必须等待请求级 collection 子任务完成清理。"""
    payload = schemas.AgentWebChatRequest(text="/状态", session_id="traditional-cancel")
    request = SimpleNamespace(is_disconnected=AsyncMock(return_value=False))
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)

    async def scenario():
        """取消正在等待的 SSE 读取，并观察 collection 的清理时序。"""
        started = asyncio.Event()
        cancelling = asyncio.Event()
        release_cleanup = asyncio.Event()
        cleanup_finished = asyncio.Event()

        async def blocked_collect(**_kwargs):
            """阻塞传统消息收集，并在取消后等待测试释放清理。"""
            started.set()
            try:
                await asyncio.Event().wait()
            except asyncio.CancelledError:
                cancelling.set()
                await release_cleanup.wait()
                cleanup_finished.set()
                raise

        with (
            patch(
                "app.application.messaging.agent.is_web_agent_traditional_message",
                return_value=True,
            ),
            patch(
                "app.application.messaging.agent.ensure_web_agent_command_allowed",
                return_value=None,
            ),
            patch(
                "app.application.messaging.agent.get_web_agent_unknown_command_message",
                return_value=None,
            ),
            patch(
                "app.application.messaging.agent.build_web_agent_session_id_async",
                return_value="web-agent:traditional-cancel",
            ),
            patch(
                "app.application.messaging.agent.collect_web_agent_traditional_events",
                side_effect=blocked_collect,
            ),
        ):
            response = await web_agent_stream(payload, request, user)
            iterator = response.body_iterator.__aiter__()
            await asyncio.wait_for(anext(iterator), timeout=1)
            pending_chunk = asyncio.create_task(anext(iterator))
            await asyncio.wait_for(started.wait(), timeout=1)
            pending_chunk.cancel()
            await asyncio.wait_for(cancelling.wait(), timeout=1)
            assert pending_chunk.done() is False
            assert cleanup_finished.is_set() is False

            release_cleanup.set()
            result = await asyncio.gather(pending_chunk, return_exceptions=True)
            assert isinstance(result[0], StopAsyncIteration)
            assert cleanup_finished.is_set() is True

    asyncio.run(scenario())


def test_web_agent_stream_sends_done_before_snapshot_persistence_finishes():
    """展示快照落库缓慢时，前端终态不应被数据库操作阻塞。"""
    payload = schemas.AgentWebChatRequest(text="检查系统", session_id="browser-snapshot")
    request = SimpleNamespace(is_disconnected=AsyncMock(return_value=False))
    user = SimpleNamespace(id=1, name="admin", is_superuser=True)
    snapshot_started = ThreadEvent()
    snapshot_release = ThreadEvent()
    snapshot_finished = ThreadEvent()

    async def immediate_process_message(**kwargs):
        """立即生成一段文本，随后进入终态。"""
        kwargs["output_callback"]("检查完成")

    async def slow_snapshot(**_kwargs):
        """阻塞快照写入，便于验证 done 的发送时机。"""
        await asyncio.to_thread(snapshot_started.set)
        await asyncio.to_thread(snapshot_release.wait, 2)
        await asyncio.to_thread(snapshot_finished.set)

    async def scenario():
        response = await web_agent_stream(payload, request, user)
        iterator = response.body_iterator.__aiter__()
        received = []
        while True:
            chunk = await asyncio.wait_for(anext(iterator), timeout=1)
            text = chunk.decode("utf-8") if isinstance(chunk, bytes) else chunk
            received.append(text)
            if '"type": "done"' in text:
                break

        for _ in range(100):
            if snapshot_started.is_set():
                break
            await asyncio.sleep(0.001)
        assert snapshot_started.is_set()
        assert not snapshot_finished.is_set()

        await iterator.aclose()
        assert not snapshot_finished.is_set()
        snapshot_release.set()
        await asyncio.to_thread(snapshot_finished.wait, 1)
        await wait_web_agent_background_tasks()
        return "".join(received)

    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="web-agent:snapshot",
            ),
            patch.object(
                MessageChain,
                "bind_user_session",
            ),
            patch.object(
                agent_manager,
                "process_message",
                side_effect=immediate_process_message,
            ),
            patch(
                "app.application.messaging.agent.save_web_agent_display_snapshot",
                new_callable=AsyncMock,
                side_effect=slow_snapshot,
            ),
        ):
            body = asyncio.run(scenario())

        assert '"type": "done"' in body
    finally:
        snapshot_release.set()

    assert snapshot_finished.wait(timeout=1)


async def _collect_streaming_response(
    response,
    *,
    wait_for_background: bool = True,
):
    """读取 StreamingResponse，并按用例语义等待生产 owner 完成收尾。"""
    chunks = []
    try:
        async for chunk in response.body_iterator:
            chunks.append(chunk.decode("utf-8") if isinstance(chunk, bytes) else chunk)
    finally:
        if wait_for_background:
            await wait_web_agent_background_tasks()
    return chunks


def test_build_web_agent_message_events_extracts_choice_card():
    """Agent 按钮通知应转换为 Web 选择卡片事件而非普通文本。"""
    events = _build_web_agent_message_events(
        schemas.Message(
            channel=NotificationChannel.WebAgent,
            mtype=MessageType.Agent,
            title="需要你的选择",
            text="请选择要执行的操作",
            buttons=[
                [
                    {
                        "text": "继续下载",
                        "callback_data": "agent_interaction:choice:req-1:1",
                        "description": "继续当前下载任务",
                    }
                ],
                [
                    {
                        "text": "查看详情",
                        "callback_data": "agent_interaction:choice:req-1:2",
                    }
                ],
            ],
        )
    )

    assert events == [
        {
            "type": "choice",
            "choice": {
                "id": "req-1",
                "title": "需要你的选择",
                "prompt": "请选择要执行的操作",
                "buttons": [
                    {
                        "label": "继续下载",
                        "callback_data": "agent_interaction:choice:req-1:1",
                        "description": "继续当前下载任务",
                    },
                    {
                        "label": "查看详情",
                        "callback_data": "agent_interaction:choice:req-1:2",
                    },
                ],
                "button_rows": [
                    [
                        {
                            "label": "继续下载",
                            "callback_data": "agent_interaction:choice:req-1:1",
                            "description": "继续当前下载任务",
                        }
                    ],
                    [
                        {
                            "label": "查看详情",
                            "callback_data": "agent_interaction:choice:req-1:2",
                        }
                    ],
                ],
            },
        }
    ]


def test_resolve_web_agent_choice_payload_returns_next_message():
    """Web 按钮回调应解析为下一条用户消息并返回卡片反馈。"""
    agent_interaction_manager.clear()
    request = agent_interaction_manager.create_request(
        session_id="web-agent:session",
        user_id="1",
        channel=NotificationChannel.WebAgent.value,
        source="web-agent",
        username="admin",
        title="需要你的选择",
        prompt="请选择",
        options=[
            AgentInteractionOption(label="电影", value="我选择电影"),
            AgentInteractionOption(label="电视剧", value="我选择电视剧", description="选择电视剧并继续清理日志"),
        ],
    )

    try:
        result = _resolve_web_agent_choice_payload(
            callback_data=f"agent_interaction:choice:{request.request_id}:2",
            user_id="1",
        )
    finally:
        agent_interaction_manager.clear()

    assert result["message"] == "我选择电视剧"
    assert result["display_message"] == "选择电视剧并继续清理日志"
    assert result["session_id"] == "web-agent:session"
    assert result["feedback"]["prompt"] == "请选择"
    assert result["feedback"]["selected_label"] == "电视剧"
    assert result["feedback"]["selected_value"] == "我选择电视剧"
    assert result["feedback"]["selected_description"] == "选择电视剧并继续清理日志"
    assert result["choice_selection"]["prompt"] == "请选择"
    assert result["choice_selection"]["selected_description"] == "选择电视剧并继续清理日志"
    assert result["choice_selection"]["button_rows"][1][0]["description"] == "选择电视剧并继续清理日志"