MoviePilot/tests/test_agent_tool_streaming.py

"""
验证 Agent 工具结果、语音消息及流式输出的交互契约。
"""

import asyncio
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import langchain.agents as langchain_agents
import pytest

if not hasattr(langchain_agents, "create_agent"):
    langchain_agents.create_agent = lambda *args, **kwargs: None

from app.agent.callback import StreamingHandler
from app.agent.middleware.subagents import is_subagent_stream_metadata
from app.agent.orchestrator import MoviePilotAgent, _ThinkTagStripper
from app.agent.tools.base import MoviePilotTool
from app.agent.tools.impl.api import MoviePilotApiTool
from app.agent.tools.impl.send_voice_message import SendVoiceMessageTool
from app.agent.web import _get_web_agent_streaming_handler_type
from app.api.endpoints.openai import _get_openai_streaming_handler_type
from app.runtime.config import settings
from app.schemas.message import Message, MessageResponse
from app.schemas.types import MessageType, NotificationChannel


def test_think_tag_stripper_waits_for_partial_open_tag():
    """
    普通文本后出现不完整 think 开始标签时不应进入死循环。
    """
    stripper = _ThinkTagStripper()
    outputs = []

    emitted = stripper.process("你好<", outputs.append)
    emitted_next = stripper.process("世界", outputs.append)

    assert emitted is True
    assert emitted_next is True
    assert outputs == ["你好", "<世界"]


def test_think_tag_stripper_hides_split_think_tag_content():
    """
    think 标签被拆分到多个 token 时应继续隐藏思考内容。
    """
    stripper = _ThinkTagStripper()
    outputs = []

    stripper.process("回答前<", outputs.append)
    stripper.process("thi", outputs.append)
    stripper.process("nk>隐藏内容</think>回答后", outputs.append)

    assert outputs == ["回答前", "回答后"]


def test_think_tag_stripper_ignores_orphan_closing_tag():
    """
    模型单独输出 think 结束标签时不应把协议标记展示给用户。
    """
    stripper = _ThinkTagStripper()
    outputs = []

    emitted = stripper.process("</think>", outputs.append)
    stripper.process("最终回答", outputs.append)

    assert emitted is False
    assert outputs == ["最终回答"]


def test_think_tag_stripper_hides_split_orphan_closing_tag():
    """
    孤立结束标签跨 token 到达时仍应被过滤并保留前后正文。
    """
    stripper = _ThinkTagStripper()
    outputs = []

    stripper.process("回答前</", outputs.append)
    stripper.process("think>回答后", outputs.append)

    assert outputs == ["回答前", "回答后"]


def test_stream_keeps_message_channel_thinking_during_hidden_think_tags():
    """
    隐藏的 think 标签内容不能提前结束消息渠道的思考状态。
    """

    class FakeAgent:
        """
        按顺序产生思考标签和最终正文的流式 Agent。
        """

        async def astream(self, *_args, **_kwargs):
            """
            依次产生隐藏思考片段和可见答案。
            """
            def token(content):
                """
                构造一个最小的文本流 token。
                """
                return SimpleNamespace(
                    tool_call_chunks=[],
                    additional_kwargs={},
                    content=content,
                )

            yield {"type": "messages", "data": (token("<think>"), {})}
            assert handler._thinking_active is True
            yield {"type": "messages", "data": (token("隐藏推理"), {})}
            assert handler._thinking_active is True
            yield {"type": "messages", "data": (token("</think>"), {})}
            assert handler._thinking_active is True
            yield {"type": "messages", "data": (token("最终答案"), {})}

    handler = StreamingHandler()
    handler._can_stream = lambda: True
    handler._streaming_enabled = True
    tokens: list[str] = []

    asyncio.run(
        MoviePilotAgent._stream_agent_tokens(
            FakeAgent(),
            {"messages": []},
            {"configurable": {"thread_id": "thinking-lifecycle"}},
            tokens.append,
            stream_handler=handler,
        )
    )

    assert tokens == ["最终答案"]
    assert handler._thinking_active is False


def test_stream_does_not_mark_plain_model_output_as_thinking():
    """
    模型没有输出 reasoning 协议时，流式处理不应创建 thinking 状态。
    """

    class FakeAgent:
        """
        只输出普通文本的流式 Agent。
        """

        async def astream(self, *_args, **_kwargs):
            """
            返回不带思考标记的普通答案。
            """
            token = SimpleNamespace(
                tool_call_chunks=[],
                additional_kwargs={},
                content="普通答案",
            )
            yield {"type": "messages", "data": (token, {})}

    handler = StreamingHandler()
    handler._can_stream = lambda: True
    handler._streaming_enabled = True
    tokens: list[str] = []

    asyncio.run(
        MoviePilotAgent._stream_agent_tokens(
            FakeAgent(),
            {"messages": []},
            {"configurable": {"thread_id": "plain-output"}},
            tokens.append,
            stream_handler=handler,
        )
    )

    assert tokens == ["普通答案"]
    assert handler._thinking_active is False
    assert handler._thinking_started_at is None


def test_stream_marks_reasoning_content_as_thinking():
    """
    模型通过 reasoning_content 输出推理时，应进入 thinking 生命周期。
    """

    class FakeAgent:
        """
        输出 reasoning_content 后再输出普通答案的流式 Agent。
        """

        async def astream(self, *_args, **_kwargs):
            """
            返回推理 token 和最终答案 token。
            """
            yield {
                "type": "messages",
                "data": (
                    SimpleNamespace(
                        tool_call_chunks=[],
                        additional_kwargs={"reasoning_content": "正在分析"},
                        content="",
                    ),
                    {},
                ),
            }
            yield {
                "type": "messages",
                "data": (
                    SimpleNamespace(
                        tool_call_chunks=[],
                        additional_kwargs={},
                        content="最终答案",
                    ),
                    {},
                ),
            }

    handler = StreamingHandler()
    handler._can_stream = lambda: True
    handler._streaming_enabled = True
    tokens: list[str] = []

    asyncio.run(
        MoviePilotAgent._stream_agent_tokens(
            FakeAgent(),
            {"messages": []},
            {"configurable": {"thread_id": "reasoning-output"}},
            tokens.append,
            stream_handler=handler,
        )
    )

    assert tokens == ["最终答案"]
    assert handler._thinking_active is False
    assert handler._thinking_started_at is not None


def test_stream_marks_reasoning_content_when_tool_chunks_share_the_token():
    """
    工具调用分片携带 reasoning_content 时，思考状态不能被工具过滤提前吞掉。
    """

    class FakeAgent:
        """
        输出带工具调用分片和 reasoning_content 的流式 Agent。
        """

        async def astream(self, *_args, **_kwargs):
            """
            返回同一 token 中的推理协议和工具调用分片。
            """
            yield {
                "type": "messages",
                "data": (
                    SimpleNamespace(
                        tool_call_chunks=[{"name": "search"}],
                        additional_kwargs={"reasoning_content": "准备调用工具"},
                        content="",
                    ),
                    {},
                ),
            }

    handler = StreamingHandler()
    handler._can_stream = lambda: True
    handler._streaming_enabled = True
    tokens: list[str] = []

    asyncio.run(
        MoviePilotAgent._stream_agent_tokens(
            FakeAgent(),
            {"messages": []},
            {"configurable": {"thread_id": "reasoning-tool-token"}},
            tokens.append,
            stream_handler=handler,
        )
    )

    assert tokens == []
    assert handler._thinking_started_at is not None
    assert handler._thinking_active is True


@pytest.mark.parametrize(
    "thinking_content",
    [[{"type": "thinking", "thinking": "分析"}], {"type": "thought", "thought": "分析"}],
)
def test_stream_marks_thinking_content_blocks(thinking_content):
    """
    内容块协议也必须开启 thinking，且不能把隐藏内容转成正文。
    """

    class FakeAgent:
        """
        输出隐藏思考内容块和普通答案的流式 Agent。
        """

        async def astream(self, *_args, **_kwargs):
            """
            返回思考内容块和最终答案。
            """
            for content in (thinking_content, "最终答案"):
                yield {
                    "type": "messages",
                    "data": (
                        SimpleNamespace(
                            tool_call_chunks=[],
                            additional_kwargs={},
                            content=content,
                        ),
                        {},
                    ),
                }

    handler = StreamingHandler()
    handler._can_stream = lambda: True
    handler._streaming_enabled = True
    tokens: list[str] = []

    asyncio.run(
        MoviePilotAgent._stream_agent_tokens(
            FakeAgent(),
            {"messages": []},
            {"configurable": {"thread_id": "thinking-block"}},
            tokens.append,
            stream_handler=handler,
        )
    )

    assert tokens == ["最终答案"]
    assert handler._thinking_started_at is not None
    assert handler._thinking_active is False


class DummyTool(MoviePilotTool):
    """
    用于流式输出测试的固定结果工具。
    """

    name: str = "dummy_tool"
    description: str = "Dummy tool for streaming tests."

    def get_tool_message(self, **kwargs) -> str:
        """
        返回固定工具执行提示。
        """
        return "run test tool"

    async def run(self, **kwargs) -> str:
        """
        返回固定工具执行结果。
        """
        return "ok"


class AdminOnlyDummyTool(MoviePilotTool):
    """
    仅管理员可用的工具，用于权限拦截场景的流式输出测试。
    """

    name: str = "admin_only_tool"
    description: str = "Admin-only tool for streaming tests."
    require_admin: bool = True

    def get_tool_message(self, **kwargs) -> str:
        """
        返回固定工具执行提示。
        """
        return "run admin only tool"

    async def run(self, **kwargs) -> str:
        """
        返回固定工具执行结果。
        """
        return "ok"


class TestAgentToolStreaming:
    """
    Agent 工具流式输出测试。
    """

    def test_web_message_callback_can_await_async_delivery(self):
        """
        WebAgent 通知回调支持异步附件准备并保持发送顺序。
        """
        received = []

        async def scenario():
            tool = DummyTool(session_id="session-1", user_id="10001")
            tool.set_message_attr("WebAgent", "web-agent", "admin")

            async def callback(message):
                await asyncio.sleep(0)
                received.append(message.text)

            tool.set_agent_context({"message_callback": callback})
            await tool.send_message(
                Message(
                    text="异步通知",
                    channel=NotificationChannel.WebAgent,
                    mtype=MessageType.Agent,
                )
            )

        asyncio.run(scenario())

        assert received == ["异步通知"]

    async def _run_tool(self, initial_buffer: str) -> tuple[str, str]:
        """
        运行测试工具并返回工具结果与缓冲内容。
        """
        tool = DummyTool(session_id="session-1", user_id="10001")
        handler = StreamingHandler()
        await handler.start_streaming()
        if initial_buffer:
            handler.emit(initial_buffer)
        tool.set_stream_handler(handler)

        with patch.object(settings, "AI_AGENT_VERBOSE", False):
            result = await tool._arun()

        buffered_message = await handler.take()
        return result, buffered_message

    def test_non_verbose_tool_call_flushes_summary_on_take(self):
        """
        校验非详细模式在读取时补充工具调用摘要。
        """
        result, buffered_message = asyncio.run(self._run_tool("prefix"))

        assert result == "ok"
        assert buffered_message == "prefix\n\n（调用了 1 次工具）\n\n"

    def test_report_tool_call_switches_between_detail_and_summary(self):
        """
        中间件工具报告入口应按详细模式选择逐条展示或汇总。
        """

        async def _run(verbose: bool):
            handler = StreamingHandler()
            await handler.start_streaming()
            with patch("app.agent.callback.get_runtime_setting", return_value=verbose):
                handler.report_tool_call(
                    tool_name="read_skill",
                    tool_message="读取技能说明：moviepilot-api",
                    tool_kwargs={"name": "moviepilot-api"},
                )
            return await handler.take()

        detailed_message = asyncio.run(_run(True))
        summary_message = asyncio.run(_run(False))

        assert detailed_message == "\n\n⚙️ => 读取技能说明：moviepilot-api\n\n"
        assert summary_message == "（查询了 1 个技能说明）\n\n"

    def test_permission_failure_adds_boundary_newline(self):
        """
        校验权限拦截时在流式缓冲中补齐工具边界换行。
        """

        async def _run():
            tool = AdminOnlyDummyTool(session_id="session-1", user_id="10001")
            handler = StreamingHandler()
            await handler.start_streaming()
            handler.emit("好的，我来帮您执行")
            tool.set_stream_handler(handler)
            tool.set_agent_context({"is_admin": False})

            result = await tool._arun()
            # 模拟模型在工具被拦截后继续输出失败说明
            handler.emit("抱歉，您没有执行此工具的权限")
            buffered_message = await handler.take()
            return result, buffered_message

        result, buffered_message = asyncio.run(_run())

        assert "没有执行此工具" in result
        assert buffered_message == "好的，我来帮您执行\n抱歉，您没有执行此工具的权限"

    def test_permission_failure_skips_newline_when_buffer_empty(self):
        """
        校验缓冲为空时权限拦截不会补多余的换行。
        """

        async def _run():
            tool = AdminOnlyDummyTool(session_id="session-1", user_id="10001")
            handler = StreamingHandler()
            await handler.start_streaming()
            tool.set_stream_handler(handler)
            tool.set_agent_context({"is_admin": False})

            await tool._arun()
            return await handler.take()

        buffered_message = asyncio.run(_run())

        assert buffered_message == ""

    def test_permission_failure_reuses_existing_newline(self):
        """
        校验缓冲已以换行结尾时权限拦截不再补换行。
        """

        async def _run():
            tool = AdminOnlyDummyTool(session_id="session-1", user_id="10001")
            handler = StreamingHandler()
            await handler.start_streaming()
            handler.emit("好的，我来帮您执行\n")
            tool.set_stream_handler(handler)
            tool.set_agent_context({"is_admin": False})

            await tool._arun()
            handler.emit("抱歉，您没有执行此工具的权限")
            return await handler.take()

        buffered_message = asyncio.run(_run())

        assert buffered_message == "好的，我来帮您执行\n抱歉，您没有执行此工具的权限"

    def test_non_verbose_tool_call_completes_blank_line_before_summary(self):
        """
        校验非详细模式在摘要前补足空行，保证工具摘要独立成段。
        """
        result, buffered_message = asyncio.run(self._run_tool("prefix\n"))

        assert result == "ok"
        assert buffered_message == "prefix\n\n（调用了 1 次工具）\n\n"

    def test_non_verbose_tool_call_keeps_existing_blank_line_before_summary(self):
        """
        校验缓冲区已有空行时摘要不再追加多余换行。
        """
        result, buffered_message = asyncio.run(self._run_tool("prefix\n\n"))

        assert result == "ok"
        assert buffered_message == "prefix\n\n（调用了 1 次工具）\n\n"

    def test_non_verbose_tool_call_emits_summary_even_when_buffer_was_empty(self):
        """
        校验空缓冲区仍会输出工具调用摘要。
        """
        result, buffered_message = asyncio.run(self._run_tool(""))

        assert result == "ok"
        assert buffered_message == "（调用了 1 次工具）\n\n"

    def test_non_verbose_tool_summary_is_inserted_before_next_text(self):
        """
        校验工具摘要插入在后续文本之前。
        """

        async def _run():
            tool = DummyTool(session_id="session-1", user_id="10001")
            handler = StreamingHandler()
            await handler.start_streaming()
            handler.emit("让我来检查一下：")
            tool.set_stream_handler(handler)

            with patch.object(settings, "AI_AGENT_VERBOSE", False):
                await tool._arun()

            handler.emit("已经拿到结果")
            return await handler.take()

        buffered_message = asyncio.run(_run())

        assert buffered_message == "让我来检查一下：\n\n（调用了 1 次工具）\n\n已经拿到结果"

    def test_non_verbose_tool_summary_aggregates_multiple_categories(self):
        """
        校验非详细模式按工具类别聚合摘要。
        """

        async def _run():
            handler = StreamingHandler()
            await handler.start_streaming()
            handler.emit("处理中：")
            handler.record_tool_call(
                tool_name="search_web",
                tool_message="搜索网络内容: MoviePilot",
                tool_kwargs={"query": "MoviePilot"},
            )
            handler.record_tool_call(
                tool_name="search_web",
                tool_message="搜索网络内容: agent streaming",
                tool_kwargs={"query": "agent streaming"},
            )
            handler.record_tool_call(
                tool_name="read_file",
                tool_message="读取文件: a.py",
                tool_kwargs={"file_path": "/tmp/a.py"},
            )
            handler.record_tool_call(
                tool_name="read_file",
                tool_message="读取文件: b.py",
                tool_kwargs={"file_path": "/tmp/b.py"},
            )
            handler.emit("继续分析")
            return await handler.take()

        buffered_message = asyncio.run(_run())

        assert buffered_message == "处理中：\n\n（执行了 2 次搜索，读取了 2 个文件）\n\n继续分析"

    def test_tool_summary_does_not_retain_secret_arguments(self):
        """
        流式聚合状态只能保留递归脱敏后的工具参数。
        """
        secret_marker = "stream-secret-marker-1947"
        handler = StreamingHandler()

        handler.record_tool_call(
            tool_name="execute_command",
            tool_message=f"Authorization: Bearer {secret_marker}",
            tool_kwargs={
                "command": f"curl -H 'Authorization: Bearer {secret_marker}'",
                "headers": {"X-API-Key": secret_marker},
            },
        )

        pending_state = str(handler._pending_tool_stats)
        assert secret_marker not in pending_state
        assert "***" in pending_state

    def test_non_verbose_tool_summary_counts_subagents(self):
        """
        校验非详细模式统计子代理调用次数。
        """

        async def _run():
            handler = StreamingHandler()
            await handler.start_streaming()
            handler.emit("处理中：")
            handler.record_tool_call(
                tool_name="task",
                tool_message="Subagent invoked",
                tool_kwargs={"description": "检查媒体库"},
            )
            handler.record_tool_call(
                tool_name="task",
                tool_message="Subagent invoked",
                tool_kwargs={"description": "检查资源"},
            )
            return await handler.take()

        buffered_message = asyncio.run(_run())

        assert buffered_message == "处理中：\n\n（已调用 2 个子代理）\n\n"

    def test_non_verbose_tool_summary_describes_skill_lookup(self):
        """
        校验非详细模式单独描述 Skill 说明查询。
        """

        async def _run():
            handler = StreamingHandler()
            await handler.start_streaming()
            handler.emit("处理中：")
            handler.record_tool_call(
                tool_name="read_skill",
                tool_message="Reads a MoviePilot skill",
                tool_kwargs={"name": "moviepilot-api"},
            )
            handler.record_tool_call(
                tool_name="read_skill",
                tool_message="Reads a MoviePilot skill",
                tool_kwargs={"name": "moviepilot-api"},
            )
            handler.record_tool_call(
                tool_name="search_memory",
                tool_message="Search MoviePilot Agent memory",
                tool_kwargs={"query": "整理", "category": "activity"},
            )
            return await handler.take()

        buffered_message = asyncio.run(_run())

        assert buffered_message == "处理中：\n\n（查询了 1 个技能说明，检索了 1 次记忆）\n\n"

    def test_non_verbose_tool_summary_counts_subagent_batch_tasks(self):
        """
        校验批量子代理控制工具按子任务数统计。
        """

        async def _run():
            handler = StreamingHandler()
            await handler.start_streaming()
            handler.emit("处理中：")
            handler.record_tool_call(
                tool_name="subagent_task",
                tool_message="Start and manage multiple MoviePilot subagent tasks",
                tool_kwargs={
                    "action": "start",
                    "tasks": [
                        {"description": "检查媒体库"},
                        {"description": "检查下载器"},
                    ],
                },
            )
            return await handler.take()

        buffered_message = asyncio.run(_run())

        assert buffered_message == "处理中：\n\n（已调用 2 个子代理）\n\n"

    def test_subagent_stream_metadata_is_suppressed(self):
        """
        校验子代理流式元数据会被识别并抑制。
        """
        assert is_subagent_stream_metadata({"metadata": {"ls_agent_type": "subagent"}})
        assert is_subagent_stream_metadata({"lc_agent_name": "general-purpose"})
        assert not is_subagent_stream_metadata({"lc_agent_name": "main"})

    def test_openai_streaming_handler_flushes_pending_summary_to_queue(self):
        """
        校验 OpenAI 流式处理器将待发送摘要推入队列。
        """

        async def _run():
            handler = _get_openai_streaming_handler_type()()
            queue: asyncio.Queue = asyncio.Queue()
            handler.bind_queue(queue)
            await handler.start_streaming()
            handler.record_tool_call(
                tool_name="read_file",
                tool_message="读取文件: app.py",
                tool_kwargs={"file_path": "/tmp/app.py"},
            )
            emitted = handler.flush_pending_tool_summary()
            queued = await queue.get()
            buffered_message = await handler.take()
            return emitted, queued, buffered_message

        emitted, queued, buffered_message = asyncio.run(_run())

        assert emitted == "（读取了 1 个文件）\n\n"
        assert queued == emitted
        assert buffered_message == emitted

    def test_flush_sends_direct_message_via_threadpool(self):
        """
        校验刷新时通过线程池发送首条直连消息。
        """
        handler = StreamingHandler()
        handler._channel = NotificationChannel.Telegram.value
        handler._source = "telegram"
        handler._user_id = "10001"
        handler._username = "tester"
        handler._streaming_enabled = True
        handler.emit("hello")

        with patch("app.agent.callback.run_in_threadpool", new_callable=AsyncMock) as run_in_threadpool_mock:
            run_in_threadpool_mock.return_value = MessageResponse(
                message_id=1,
                chat_id=2,
                source="telegram",
                success=True,
            )

            asyncio.run(handler._flush())

        assert run_in_threadpool_mock.await_count == 1
        assert run_in_threadpool_mock.await_args.args[0].__name__ == "send_direct_message"
        notification = run_in_threadpool_mock.await_args.args[1]
        assert notification.mtype == MessageType.Agent
        assert notification.text == "hello"
        assert notification.rich_message == "hello"
        assert handler.has_sent_message

    def test_rich_message_quotes_tool_summary_lines_for_telegram(self):
        """
        校验 Telegram 富文本将工具摘要行转换为引用块，与正文视觉分隔。
        """
        handler = StreamingHandler()
        handler._channel = NotificationChannel.Telegram.value
        handler._source = "telegram"
        handler.record_tool_call(
            tool_name="moviepilot_api",
            tool_message="查看目录",
            tool_kwargs={
                "operation_id": "storage.list",
                "body": {"path": "/tmp"},
            },
        )
        handler.flush_pending_tool_summary()
        text = handler._buffer

        rich_message = handler._get_rich_message(text)

        assert text == "（查看了 1 个目录）\n\n"
        assert rich_message == "> （查看了 1 个目录）\n\n"

    def test_verbose_tool_message_is_quoted_for_telegram(self):
        """
        校验 Telegram 啰嗦模式逐条工具提示也使用引用块。
        """

        async def _run():
            tool = DummyTool(session_id="session-1", user_id="10001")
            handler = StreamingHandler()
            handler._channel = NotificationChannel.Telegram.value
            handler._source = "telegram"
            handler._streaming_enabled = True
            handler.emit("前置内容")
            flush_task = asyncio.create_task(asyncio.sleep(60))
            handler._flush_task = flush_task
            tool.set_stream_handler(handler)

            try:
                with patch.object(settings, "AI_AGENT_VERBOSE", True):
                    await tool._arun()
                return handler._buffer, handler._get_rich_message(handler._buffer)
            finally:
                flush_task.cancel()
                await asyncio.gather(flush_task, return_exceptions=True)
                handler._flush_task = None

        text, rich_message = asyncio.run(_run())

        assert text == "前置内容\n\n⚙️ => run test tool\n\n"
        assert rich_message == "前置内容\n\n> ⚙️ => run test tool\n\n"

    def test_verbose_web_api_tool_message_includes_major_parameters(self):
        """
        Web Agent 啰嗦模式应实时输出 API operation 及其主要参数。
        """

        async def _run():
            emitted = []
            handler = _get_web_agent_streaming_handler_type()(emitted.append)
            await handler.start_streaming(
                channel=NotificationChannel.WebAgent.value,
                source="web-agent",
                user_id="1",
                username="admin",
            )
            tool = MoviePilotApiTool(session_id="session-1", user_id="1")
            tool.set_stream_handler(handler)

            with (
                patch.object(settings, "AI_AGENT_VERBOSE", True),
                patch.object(MoviePilotApiTool, "run", new_callable=AsyncMock) as run_mock,
            ):
                run_mock.return_value = '{"success": true}'
                result = await tool._arun(
                    operation_id="subscription.list",
                    query={"page": 1, "count": 20},
                )
            return result, "".join(emitted), handler._buffer

        result, emitted, buffered = asyncio.run(_run())

        expected = (
            '\n\n⚙️ => 调用 MoviePilot API：subscription.list，主要参数：'
            '{"query": {"page": 1, "count": 20}}\n\n'
        )
        assert result == '{"success": true}'
        assert emitted == expected
        assert buffered == expected

    def test_web_streaming_handler_emits_structured_tool_lifecycle(self):
        """
        Web 流式处理器应发布稳定调用 ID 的开始与完成事件，避免重复文本提示。
        """

        async def _run():
            emitted = []
            tool_events = []
            handler = _get_web_agent_streaming_handler_type()(
                emitted.append,
                tool_events.append,
            )
            await handler.start_streaming(
                channel=NotificationChannel.WebAgent.value,
                source="web-agent",
                user_id="1",
                username="admin",
            )
            with patch("app.agent.callback.get_runtime_setting", return_value=True):
                tool_id = handler.tool_call_started("search", "查询媒体")
                handler.emit_tool_message("查询媒体")
                handler.record_tool_call(
                    tool_name="search",
                    tool_message="查询媒体",
                    tool_kwargs={"query": "MoviePilot"},
                )
                handler.tool_call_finished(tool_id, "done")
            return tool_id, emitted, tool_events

        tool_id, emitted, tool_events = asyncio.run(_run())

        assert tool_id.startswith("tool-")
        assert emitted == []
        assert tool_events == [
            {
                "type": "tool",
                "status": "running",
                "tool_id": tool_id,
                "tool_name": "search",
                "message": "查询媒体",
            },
            {"type": "tool", "status": "done", "tool_id": tool_id},
        ]

    def test_web_streaming_handler_updates_tool_summary_when_not_verbose(self):
        """
        WebAgent 关闭啰嗦模式时应逐次推送最新的工具计数摘要。
        """

        async def _run():
            emitted = []
            tool_events = []
            handler = _get_web_agent_streaming_handler_type()(
                emitted.append,
                tool_events.append,
            )
            await handler.start_streaming(
                channel=NotificationChannel.WebAgent.value,
                source="web-agent",
                user_id="1",
                username="admin",
            )
            with patch("app.agent.callback.get_runtime_setting", return_value=False):
                assert handler.tool_call_started("search", "查询媒体") == ""
                handler.report_tool_call(
                    tool_name="search_web",
                    tool_message="搜索网络内容",
                    tool_kwargs={"query": "MoviePilot"},
                )
                handler.report_tool_call(
                    tool_name="read_file",
                    tool_message="读取文件",
                    tool_kwargs={"file_path": "/tmp/app.py"},
                )
                handler.emit("查询完成")
            return emitted, tool_events

        emitted, tool_events = asyncio.run(_run())

        assert tool_events == []
        assert emitted == [
            "（执行了 1 次搜索）\n\n",
            "（读取了 1 个文件）\n\n",
            "查询完成",
        ]

    def test_web_streaming_handler_updates_normal_tool_when_not_verbose(self):
        """
        普通工具在 WebAgent 非啰嗦模式下也应立即推送计数摘要。
        """

        async def _run():
            emitted = []
            tool_events = []
            handler = _get_web_agent_streaming_handler_type()(
                emitted.append,
                tool_events.append,
            )
            await handler.start_streaming(
                channel=NotificationChannel.WebAgent.value,
                source="web-agent",
                user_id="1",
                username="admin",
            )
            tool = DummyTool(session_id="session-1", user_id="1")
            tool.set_stream_handler(handler)
            with patch.object(settings, "AI_AGENT_VERBOSE", False):
                await tool._arun()
                handler.emit("查询完成")
            return emitted, tool_events

        emitted, tool_events = asyncio.run(_run())

        assert tool_events == []
        assert emitted == ["（调用了 1 次工具）\n\n", "查询完成"]

    @pytest.mark.parametrize(
        "channel",
        [
            NotificationChannel.Telegram,
            NotificationChannel.Feishu,
            NotificationChannel.Slack,
            NotificationChannel.Discord,
        ],
    )
    def test_editable_channels_update_tool_summary_in_place(self, channel):
        """
        所有支持消息编辑的通知渠道都应在原消息上更新工具统计。
        """
        handler = StreamingHandler()
        handler._channel = channel.value
        handler._source = f"{channel.value}-source"
        handler._streaming_enabled = True
        handler.emit("正在处理")
        handler.record_tool_call(
            tool_name="search_web",
            tool_message="搜索网络内容",
            tool_kwargs={"query": "MoviePilot"},
        )

        first_text = "正在处理\n\n（执行了 1 次搜索）\n\n"
        assert handler._buffer == first_text
        with patch("app.agent.callback.run_in_threadpool", new_callable=AsyncMock) as run_in_threadpool_mock:
            run_in_threadpool_mock.return_value = MessageResponse(
                message_id=f"{channel.value}-message",
                chat_id=f"{channel.value}-chat",
                channel=channel,
                source=handler._source,
                success=True,
            )
            asyncio.run(handler._flush())

        handler.record_tool_call(
            tool_name="read_file",
            tool_message="读取文件",
            tool_kwargs={"file_path": "/tmp/app.py"},
        )
        updated_text = "正在处理\n\n（执行了 1 次搜索，读取了 1 个文件）\n\n"
        assert handler._buffer == updated_text

        with patch("app.agent.callback.run_in_threadpool", new_callable=AsyncMock) as run_in_threadpool_mock:
            run_in_threadpool_mock.return_value = True
            asyncio.run(handler._flush())

        assert run_in_threadpool_mock.await_count == 1
        assert run_in_threadpool_mock.await_args.args[0].__name__ == "edit_message"
        assert run_in_threadpool_mock.await_args.kwargs["text"] == updated_text
        assert handler._sent_text == updated_text

        handler.emit("第一批完成")
        handler.record_tool_call(
            tool_name="search_web",
            tool_message="搜索下一批内容",
            tool_kwargs={"query": "next"},
        )
        assert handler._buffer == f"{updated_text}第一批完成\n\n（执行了 1 次搜索）\n\n"

    def test_rich_message_keeps_body_text_unquoted_for_telegram(self):
        """
        校验 Telegram 富文本只转换工具摘要行，正文保持原样。
        """
        handler = StreamingHandler()
        handler._channel = NotificationChannel.Telegram.value
        handler.emit("正文内容\n\n")
        handler.record_tool_call(
            tool_name="execute_command",
            tool_message="执行命令",
            tool_kwargs={},
        )
        handler.emit("后续结论")

        rich_message = handler._get_rich_message(handler._buffer)

        assert rich_message == "正文内容\n\n> （执行了 1 条命令）\n\n后续结论"

    def test_rich_message_keeps_previous_tool_summaries_styled_after_edits(self):
        """
        Telegram 后续编辑时，重复文案的历史工具摘要仍应保持引用样式。
        """
        handler = StreamingHandler()
        handler._channel = NotificationChannel.Telegram.value
        handler._streaming_enabled = True
        handler._can_stream = lambda: True

        for _ in range(3):
            handler.record_tool_call(
                tool_name="execute_command",
                tool_message="执行命令",
                tool_kwargs={},
            )
        handler.emit("第一批完成")
        for _ in range(4):
            handler.record_tool_call(
                tool_name="execute_command",
                tool_message="执行命令",
                tool_kwargs={},
            )

        rich_message = handler._get_rich_message(handler._buffer)

        assert "> （执行了 3 条命令）" in rich_message
        assert "> （执行了 4 条命令）" in rich_message

    def test_rich_message_returns_none_for_non_telegram_channels(self):
        """
        校验非 Telegram 渠道不启用富文本，摘要保持原有纯文本格式。
        """
        handler = StreamingHandler()
        handler._channel = NotificationChannel.Feishu.value
        handler._source = "feishu-main"
        handler.record_tool_call(
            tool_name="execute_command",
            tool_message="执行命令",
            tool_kwargs={},
        )
        handler.flush_pending_tool_summary()

        assert handler._get_rich_message(handler._buffer) is None
        assert handler._buffer == "（执行了 1 条命令）\n\n"

    def test_flush_edits_message_via_threadpool(self):
        """
        校验刷新时通过线程池编辑已有消息。
        """
        handler = StreamingHandler()
        handler._channel = NotificationChannel.Telegram.value
        handler._source = "telegram"
        handler._streaming_enabled = True
        handler._message_response = MessageResponse(
            message_id=1,
            chat_id=2,
            source="telegram",
            success=True,
        )
        handler._sent_text = "hello"
        handler.emit("hello world")

        with patch("app.agent.callback.run_in_threadpool", new_callable=AsyncMock) as run_in_threadpool_mock:
            run_in_threadpool_mock.return_value = True

            asyncio.run(handler._flush())

        assert run_in_threadpool_mock.await_count == 1
        assert run_in_threadpool_mock.await_args.args[0].__name__ == "edit_message"
        assert run_in_threadpool_mock.await_args.kwargs["metadata"]["telegram_rich_message"] == "hello world"
        assert handler._sent_text == "hello world"

    def test_stop_streaming_waits_inflight_initial_flush_before_final_edit(self):
        """
        校验停止流式输出会等待首条消息发送完成再编辑。
        """

        async def _run():
            handler = StreamingHandler()
            handler._channel = NotificationChannel.Feishu.value
            handler._source = "feishu-main"
            handler._user_id = "ou_user"
            handler._streaming_enabled = True
            handler.emit("hello")

            send_started = asyncio.Event()
            allow_send_finish = asyncio.Event()
            calls = []

            async def fake_run_in_threadpool(func, *args, **kwargs):
                calls.append((func.__name__, args, kwargs))
                if func.__name__ == "send_direct_message":
                    send_started.set()
                    await allow_send_finish.wait()
                    return MessageResponse(
                        message_id="om_stream",
                        chat_id="oc_stream",
                        channel=NotificationChannel.Feishu,
                        source="feishu-main",
                        success=True,
                    )
                return True

            with patch(
                "app.agent.callback.run_in_threadpool",
                new=fake_run_in_threadpool,
            ):
                # 模拟定时刷新已经开始发送首条消息，但飞书 API 尚未返回。
                handler._flush_task = asyncio.create_task(handler._flush())
                await send_started.wait()
                handler.emit(" world")

                stop_task = asyncio.create_task(handler.stop_streaming())
                await asyncio.sleep(0)
                assert not stop_task.done()

                allow_send_finish.set()
                all_sent, final_text = await stop_task

            return all_sent, final_text, calls

        all_sent, final_text, calls = asyncio.run(_run())

        assert all_sent
        assert final_text == "hello world"
        assert [call[0] for call in calls] == [
            "send_direct_message",
            "edit_message",
            "finalize_message",
        ]
        edit_kwargs = calls[1][2]
        assert edit_kwargs["message_id"] == "om_stream"
        assert edit_kwargs["text"] == "hello world"

    def test_stop_streaming_uses_generic_finalize_message(self):
        """
        校验停止流式输出会调用通用消息完成接口。
        """
        handler = StreamingHandler()
        handler._message_response = MessageResponse(
            message_id="om_stream",
            chat_id="oc_stream",
            channel=NotificationChannel.Feishu,
            source="feishu-main",
            metadata={"feishu_streaming": {"card_id": "card_stream", "sequence": 2}},
            success=True,
        )
        handler._sent_text = "hello"
        handler._buffer = "hello"
        handler._streaming_enabled = True

        with (
            patch("app.agent.callback.run_in_threadpool", new_callable=AsyncMock) as run_in_threadpool_mock,
            patch.object(handler, "_cancel_flush_task", new_callable=AsyncMock),
            patch.object(handler, "_flush", new_callable=AsyncMock),
        ):
            asyncio.run(handler.stop_streaming())

        assert run_in_threadpool_mock.await_count == 1
        assert run_in_threadpool_mock.await_args.args[0].__name__ == "finalize_message"
        assert run_in_threadpool_mock.await_args.args[1].message_id == "om_stream"

    def test_flush_without_channel_context_does_not_send_direct_message(self):
        """
        校验缺少渠道上下文时不会发送直连消息。
        """
        handler = StreamingHandler()
        handler._streaming_enabled = True
        handler.emit("hello")

        with patch("app.agent.callback.run_in_threadpool", new_callable=AsyncMock) as run_in_threadpool_mock:
            asyncio.run(handler._flush())

        run_in_threadpool_mock.assert_not_awaited()
        assert not handler.has_sent_message

    def test_flush_without_channel_context_dispatch_allowed_sends_direct_message(self):
        """
        校验允许后台派发时缺少渠道上下文也能发送消息。
        """
        handler = StreamingHandler()
        handler._user_id = "10001"
        handler._username = "tester"
        handler._streaming_enabled = True
        handler.set_dispatch_policy(allow_dispatch_without_context=True)
        handler.emit("hello")

        with patch("app.agent.callback.run_in_threadpool", new_callable=AsyncMock) as run_in_threadpool_mock:
            run_in_threadpool_mock.return_value = MessageResponse(
                message_id=1,
                chat_id=2,
                source="telegram",
                success=True,
            )

            asyncio.run(handler._flush())

        assert run_in_threadpool_mock.await_count == 1
        assert run_in_threadpool_mock.await_args.args[0].__name__ == "send_direct_message"
        assert handler.has_sent_message

    def test_flush_passes_original_message_context_to_send_direct_message(self):
        """
        校验刷新发送时保留原始消息上下文。
        """
        handler = StreamingHandler()
        handler._channel = NotificationChannel.Feishu.value
        handler._source = "feishu-main"
        handler._user_id = "ou_user"
        handler._username = "tester"
        handler._original_message_id = "om_origin"
        handler._original_chat_id = "oc_origin"
        handler._streaming_enabled = True
        handler.emit("hello")

        with patch("app.agent.callback.run_in_threadpool", new_callable=AsyncMock) as run_in_threadpool_mock:
            run_in_threadpool_mock.return_value = MessageResponse(
                message_id="om_stream",
                chat_id="oc_origin",
                source="feishu-main",
                success=True,
            )

            asyncio.run(handler._flush())

        notification = run_in_threadpool_mock.await_args.args[1]
        assert notification.original_message_id == "om_origin"
        assert notification.original_chat_id == "oc_origin"

    def test_verbose_background_tool_call_does_not_post_message(self):
        """
        校验详细模式后台工具调用不会主动发送工具消息。
        """

        async def _run():
            tool = DummyTool(session_id="session-1", user_id="10001")
            handler = StreamingHandler()
            await handler.start_streaming()
            tool.set_stream_handler(handler)
            tool.set_message_attr(channel=None, source=None, username="tester")

            with (
                patch.object(settings, "AI_AGENT_VERBOSE", True),
                patch.object(DummyTool, "send_tool_message", new_callable=AsyncMock) as send_tool_message,
            ):
                result = await tool._arun()
                buffered_message = await handler.take()
                return result, buffered_message, send_tool_message

        result, buffered_message, send_tool_message = asyncio.run(_run())

        assert result == "ok"
        send_tool_message.assert_not_awaited()
        assert buffered_message == "（调用了 1 次工具）\n\n"

    def test_verbose_background_dispatch_tool_call_can_post_message(self):
        """
        校验允许后台派发时详细模式工具调用可以发送消息。
        """

        async def _run():
            tool = DummyTool(session_id="session-1", user_id="10001")
            handler = StreamingHandler()
            await handler.start_streaming()
            handler.emit("前置内容")
            tool.set_stream_handler(handler)
            tool.set_message_attr(channel=None, source=None, username="tester")
            tool.set_agent_context({"should_dispatch_reply": True})

            with (
                patch.object(settings, "AI_AGENT_VERBOSE", True),
                patch.object(DummyTool, "send_tool_message", new_callable=AsyncMock) as send_tool_message,
            ):
                result = await tool._arun()
                buffered_message = await handler.take()
                return result, buffered_message, send_tool_message

        result, buffered_message, send_tool_message = asyncio.run(_run())

        assert result == "ok"
        send_tool_message.assert_awaited_once_with("前置内容\n\n⚙️ => run test tool")
        assert buffered_message == ""

    def test_send_voice_message_uses_native_voice_for_supported_channels(self):
        """
        校验支持语音输出的渠道会发送原生语音消息。
        """

        async def _run(channel: NotificationChannel):
            """
            运行指定渠道的语音发送工具。
            """
            tool = SendVoiceMessageTool(session_id="session-1", user_id="10001")
            tool.set_message_attr(
                channel=channel.value,
                source=f"{channel.name.lower()}-main",
                username="tester",
            )

            with (
                patch.object(settings, "LLM_SUPPORT_AUDIO_OUTPUT", True),
                patch.object(settings, "AUDIO_OUTPUT_INCLUDE_TEXT", True),
                patch(
                    "app.agent.tools.impl.send_voice_message.AgentCapabilityManager.is_audio_output_available",
                    return_value=True,
                ),
                patch(
                    "app.agent.tools.impl.send_voice_message.AgentCapabilityManager.synthesize_speech",
                    return_value=Path("/tmp/reply.opus"),
                ) as synthesize_speech,
                patch.object(
                    SendVoiceMessageTool,
                    "send_message",
                    new_callable=AsyncMock,
                ) as send_message,
            ):
                result = await tool.run("你好")
            return result, synthesize_speech, send_message

        for channel in (NotificationChannel.Telegram, NotificationChannel.Feishu, NotificationChannel.WebAgent):
            result, synthesize_speech, send_message = asyncio.run(_run(channel))
            notification = send_message.await_args.args[-1]

            assert result == "语音回复已发送"
            synthesize_speech.assert_called_once_with("你好")
            send_message.assert_awaited_once()
            assert notification.channel == channel
            assert notification.voice_path == "/tmp/reply.opus"
            assert notification.voice_caption == "你好"
            voice_tool = SendVoiceMessageTool(session_id="session-1", user_id="10001")
            assert voice_tool.return_direct
            assert "本工具直接完成用户回复" in voice_tool.description

    def test_send_voice_message_falls_back_for_unsupported_channels(self):
        """
        校验不支持语音输出的渠道继续回退为文字消息。
        """

        async def _run():
            """
            运行不支持语音输出渠道的语音发送工具。
            """
            tool = SendVoiceMessageTool(session_id="session-1", user_id="10001")
            tool.set_message_attr(channel=NotificationChannel.Slack.value, source="slack-main", username="tester")

            with (
                patch.object(settings, "LLM_SUPPORT_AUDIO_OUTPUT", True),
                patch(
                    "app.agent.tools.impl.send_voice_message.AgentCapabilityManager.is_audio_output_available",
                    return_value=True,
                ),
                patch(
                    "app.agent.tools.impl.send_voice_message.AgentCapabilityManager.synthesize_speech"
                ) as synthesize_speech,
                patch.object(
                    SendVoiceMessageTool,
                    "send_message",
                    new_callable=AsyncMock,
                ) as send_message,
            ):
                result = await tool.run("你好")
            return result, synthesize_speech, send_message

        result, synthesize_speech, send_message = asyncio.run(_run())
        notification = send_message.await_args.args[-1]

        assert result == "当前渠道不支持语音回复，已自动回退为文字回复"
        synthesize_speech.assert_not_called()
        send_message.assert_awaited_once()
        assert notification.text == "你好"
        assert notification.voice_path is None