Simplify agent message handling and streaming cleanup

tests/test_agent_background_output.py

@@ -56,6 +56,13 @@ class _FakeStreamingFailingAgent(_FakeFailingAgent):
         yield None
 
 
+class _FakeStreamingAgent(_FakeAgent):
+    async def astream(self, _messages, **_kwargs):
+        return
+        # 保持 async generator 形态，当前用例不需要实际 token。
+        yield None
+
+
 class StreamChunkTimeoutError(RuntimeError):
     """模拟 langchain_openai 的流式分块超时异常。"""
 
@@ -191,6 +198,81 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
         self.assertNotIn("Tune or disable", sent_message)
         self.assertEqual(expected, agent._streamed_output)
 
+    async def test_streaming_success_stops_streaming_once(self):
+        """流式正常完成时不应在 finally 中重复停止流式输出。"""
+        agent = MoviePilotAgent(session_id="stream-ok", user_id="user-1")
+        agent.channel = "Telegram"
+        agent.source = "telegram-test"
+        agent._tool_context = {"user_reply_sent": False}
+        agent._streamed_output = ""
+        agent.stream_handler = SimpleNamespace(
+            set_dispatch_policy=lambda allow_dispatch_without_context=False: None,
+            start_streaming=AsyncMock(),
+            flush_pending_tool_summary=lambda: "",
+            stop_streaming=AsyncMock(return_value=(True, "已发送")),
+        )
+        agent._should_stream = lambda: True
+        agent._create_agent = AsyncMock(
+            return_value=_FakeStreamingAgent([AIMessage(content="已发送")])
+        )
+        agent.send_agent_message = AsyncMock()
+
+        await agent._execute_agent([HumanMessage(content="测试")])
+
+        agent.stream_handler.stop_streaming.assert_awaited_once()
+
+    async def test_tool_sent_reply_does_not_persist_raw_agent_messages(self):
+        """工具已发送用户回复时不应把工具调用状态写入下一轮记忆。"""
+        agent = MoviePilotAgent(session_id="tool-reply", user_id="user-1")
+        agent.channel = "Telegram"
+        agent.source = "telegram-test"
+        agent._tool_context = {"user_reply_sent": True}
+        agent._streamed_output = ""
+        agent.stream_handler = SimpleNamespace(
+            stop_streaming=AsyncMock(return_value=(False, ""))
+        )
+        agent._should_stream = lambda: False
+        agent._create_agent = AsyncMock(
+            return_value=_FakeAgent([AIMessage(content="消息已发送")])
+        )
+        agent.send_agent_message = AsyncMock()
+
+        with patch.object(memory_manager, "save_agent_messages") as save_messages:
+            await agent._execute_agent([HumanMessage(content="测试")])
+
+        save_messages.assert_not_called()
+
+    async def test_process_does_not_mutate_cached_agent_messages(self):
+        """处理新消息时不应直接修改记忆缓存中的历史消息列表。"""
+        agent = MoviePilotAgent(
+            session_id="cached-memory",
+            user_id="user-1",
+            channel="Telegram",
+            source="telegram-test",
+        )
+        cached_messages = [HumanMessage(content="上一轮")]
+        captured = {}
+
+        async def _execute_agent(messages):
+            captured["messages"] = messages
+            return "消息已发送", {}
+
+        agent._execute_agent = AsyncMock(side_effect=_execute_agent)
+
+        with (
+            patch.object(
+                memory_manager, "get_agent_messages", return_value=cached_messages
+            ),
+            patch.object(agent, "prepare_chat_title", new=AsyncMock()),
+            patch.object(agent, "_save_display_history_messages"),
+        ):
+            result = await agent.process("继续")
+
+        self.assertEqual("消息已发送", result)
+        self.assertEqual(1, len(cached_messages))
+        self.assertIsNot(cached_messages, captured["messages"])
+        self.assertEqual(2, len(captured["messages"]))
+
     async def test_background_non_streaming_sends_when_reply_mode_dispatch(self):
         agent = MoviePilotAgent(session_id="bg-test", user_id="system")
         agent.channel = None

tests/test_agent_image_support.py

@@ -569,37 +569,6 @@ class AgentImageSupportTest(unittest.TestCase):
 
         self.assertEqual(payload.image_url, "https://example.com/poster.png")
 
-    def test_send_message_input_normalizes_html_parse_mode(self):
-        payload = SendMessageInput(
-            explanation="send html notice",
-            message="<b>处理完成</b>",
-            parse_mode="html",
-        )
-
-        self.assertEqual(payload.parse_mode, "HTML")
-
-    def test_send_message_input_normalizes_common_html_tags(self):
-        payload = SendMessageInput(
-            explanation="send html notice",
-            message="<h1>标题</h1><p>第一行<br>第二行</p><ul><li>A</li></ul>",
-            parse_mode="HTML",
-        )
-
-        self.assertEqual(
-            payload.message,
-            "<b>标题</b>\n第一行\n第二行\n• A\n",
-        )
-
-    def test_send_message_input_rejects_unsupported_html_tags(self):
-        with self.assertRaises(ValueError) as error:
-            SendMessageInput(
-                explanation="send html notice",
-                message="<table><tr><td>A</td></tr></table>",
-                parse_mode="HTML",
-            )
-
-        self.assertIn("HTML 标签 <table> 不受 Telegram 支持", str(error.exception))
-
     def test_send_message_tool_uses_regular_notification_type(self):
         """发送消息工具应按普通通知消息登记。"""
 
@@ -619,7 +588,6 @@ class AgentImageSupportTest(unittest.TestCase):
                     message="处理完成",
                     title="智能体通知",
                     image_url="https://example.com/poster.png",
-                    parse_mode="HTML",
                 )
             return result, async_post_message
 
@@ -633,7 +601,35 @@ class AgentImageSupportTest(unittest.TestCase):
         self.assertEqual(notification.title, "智能体通知")
         self.assertEqual(notification.text, "处理完成")
         self.assertEqual(notification.image, "https://example.com/poster.png")
-        self.assertEqual(notification.parse_mode, "HTML")
+        self.assertIsNone(notification.parse_mode)
+
+    def test_send_message_tool_ignores_parse_mode_argument(self):
+        """发送消息工具不再支持由 Agent 指定 Telegram parse_mode。"""
+
+        async def _run():
+            tool = SendMessageTool(session_id="session-1", user_id="10001")
+            tool.set_message_attr(
+                channel=MessageChannel.Telegram.value,
+                source="telegram-test",
+                username="tester",
+            )
+
+            with patch(
+                "app.agent.tools.base.ToolChain.async_post_message",
+                new_callable=AsyncMock,
+            ) as async_post_message:
+                result = await tool.run(
+                    message="<b>处理完成</b>",
+                    parse_mode="HTML",
+                )
+            return result, async_post_message
+
+        result, async_post_message = asyncio.run(_run())
+        notification = async_post_message.await_args.args[0]
+
+        self.assertEqual(result, "消息已发送")
+        self.assertEqual(notification.text, "<b>处理完成</b>")
+        self.assertIsNone(notification.parse_mode)
 
     def test_send_message_tool_marks_reply_sent_after_dispatch(self):
         """发送消息工具成功发送后应终止本轮回复。"""
@@ -652,7 +648,7 @@ class AgentImageSupportTest(unittest.TestCase):
                 "app.agent.tools.base.ToolChain.async_post_message",
                 new_callable=AsyncMock,
             ):
-                result = await tool.run(message="<b>处理完成</b>", parse_mode="HTML")
+                result = await tool.run(message="处理完成")
             return result, agent_context
 
         result, agent_context = asyncio.run(_run())
@@ -661,58 +657,6 @@ class AgentImageSupportTest(unittest.TestCase):
         self.assertTrue(agent_context["user_reply_sent"])
         self.assertEqual(agent_context["reply_mode"], "send_message")
 
-    def test_send_message_tool_rejects_unsupported_html_before_dispatch(self):
-        """发送消息工具应在进入消息链路前拒绝不支持的 HTML。"""
-
-        async def _run():
-            tool = SendMessageTool(session_id="session-1", user_id="10001")
-            tool.set_message_attr(
-                channel=MessageChannel.Telegram.value,
-                source="telegram-test",
-                username="tester",
-            )
-
-            with patch(
-                "app.agent.tools.base.ToolChain.async_post_message",
-                new_callable=AsyncMock,
-            ) as async_post_message:
-                result = await tool.run(
-                    message="<table><tr><td>A</td></tr></table>",
-                    parse_mode="HTML",
-                )
-            return result, async_post_message
-
-        result, async_post_message = asyncio.run(_run())
-
-        self.assertIn("HTML 标签 <table> 不受 Telegram 支持", result)
-        async_post_message.assert_not_awaited()
-
-    def test_send_message_tool_rejects_invalid_parse_mode(self):
-        """发送消息工具应拒绝不支持的格式类型。"""
-
-        async def _run():
-            tool = SendMessageTool(session_id="session-1", user_id="10001")
-            tool.set_message_attr(
-                channel=MessageChannel.Telegram.value,
-                source="telegram-test",
-                username="tester",
-            )
-
-            with patch(
-                "app.agent.tools.base.ToolChain.async_post_message",
-                new_callable=AsyncMock,
-            ) as async_post_message:
-                result = await tool.run(
-                    message="处理完成",
-                    parse_mode="Markdown",
-                )
-            return result, async_post_message
-
-        result, async_post_message = asyncio.run(_run())
-
-        self.assertIn("parse_mode 仅支持 MarkdownV2 或 HTML", result)
-        async_post_message.assert_not_awaited()
-
     def test_send_local_file_input_accepts_file_payload(self):
         payload = SendLocalFileInput(
             explanation="send generated report",

tests/test_agent_interaction.py

@@ -39,7 +39,7 @@ class TestAgentInteraction(unittest.TestCase):
         self.assertIn("do not write a final text reply after it", telegram_prompt)
         self.assertNotIn("ask_user_choice", wechat_prompt)
 
-    def test_prompt_injects_send_message_html_hint_only_for_telegram(self):
+    def test_prompt_does_not_inject_send_message_html_hint(self):
         telegram_prompt = prompt_manager.get_agent_prompt(
             channel=MessageChannel.Telegram.value
         )
@@ -47,9 +47,8 @@ class TestAgentInteraction(unittest.TestCase):
             channel=MessageChannel.Wechat.value
         )
 
-        self.assertIn("parse_mode=\"HTML\"", telegram_prompt)
-        self.assertIn("Telegram-supported HTML tags", telegram_prompt)
-        self.assertIn("Do not mix Markdown syntax", telegram_prompt)
+        self.assertNotIn("parse_mode=\"HTML\"", telegram_prompt)
+        self.assertNotIn("Telegram-supported HTML tags", telegram_prompt)
         self.assertNotIn("parse_mode=\"HTML\"", wechat_prompt)
 
     def test_factory_injects_choice_tool_only_for_button_channels(self):