diff --git a/app/agent/callback/__init__.py b/app/agent/callback/__init__.py index 878315113..cb99e7e50 100644 --- a/app/agent/callback/__init__.py +++ b/app/agent/callback/__init__.py @@ -514,8 +514,8 @@ class StreamingHandler: # 将未落地的工具统计补入缓冲区,避免流式结束时丢失这段执行信息 self.flush_pending_tool_summary() - # 执行最后一次刷新 - await self._flush_messages() + # 执行最后一次刷新;渠道若有消息级编辑间隔,应等待收尾正文落地。 + await self._flush_messages(final=True) await self._finalize_current_message() @@ -998,13 +998,15 @@ class StreamingHandler: pass self._flush_task = None - async def _flush_messages(self) -> None: + async def _flush_messages(self, final: bool = False) -> None: """ 按补充输入边界依次收口旧消息,再把后续内容发送到新的渠道消息。 + + :param final: 是否执行等待消息级编辑间隔的收尾刷新 """ while self._message_boundaries: boundary, original_message_id = self._message_boundaries[0] - await self._flush(end_offset=boundary) + await self._flush(end_offset=boundary, final=final) with self._lock: current_text = self._buffer[self._msg_start_offset : boundary] if current_text and current_text != self._sent_text: @@ -1017,7 +1019,7 @@ class StreamingHandler: self._sent_text = "" self._sent_thinking_status = "" self._original_message_id = original_message_id - await self._flush() + await self._flush(final=final) async def _finalize_current_message(self) -> None: """ @@ -1026,12 +1028,19 @@ class StreamingHandler: if self._message_response is not None: await run_in_threadpool(_StreamChain().finalize_message, self._message_response) - async def _flush(self, end_offset: Optional[int] = None) -> None: + async def _flush( + self, + end_offset: Optional[int] = None, + final: bool = False, + ) -> None: """ 将当前缓冲区内容刷新到用户消息 - 如果还没有发送过消息,先发送一条新消息并记录message_id - 如果已经发送过消息,编辑该消息为最新的完整内容 - 如果当前消息内容超过长度限制,冻结当前消息并发送新消息继续输出 + + :param end_offset: 当前消息允许读取的缓冲区结束位置 + :param final: 是否执行等待消息级编辑间隔的最终编辑 """ with self._lock: # 当前消息的文本 = buffer 中从 _msg_start_offset 开始的部分 @@ -1123,6 +1132,7 @@ class StreamingHandler: if rich_message: # 通用编辑接口不增加渠道专属参数,通过元数据交给 Telegram 模块消费。 metadata["telegram_rich_message"] = rich_message + metadata["telegram_wait_for_edit_interval"] = final success = await run_in_threadpool( chain.edit_message, channel=channel_enum, diff --git a/app/modules/telegram/module.py b/app/modules/telegram/module.py index ba5b1f155..2f0b04e9b 100644 --- a/app/modules/telegram/module.py +++ b/app/modules/telegram/module.py @@ -627,7 +627,7 @@ class TelegramModule(_MessageChannelModuleBase[Telegram]): :param text: 新的消息内容 :param title: 消息标题 :param buttons: 新的按钮列表 - :param metadata: 其他元信息;telegram_rich_message 用于流式富文本编辑 + :param metadata: telegram_rich_message 为流式富文本;telegram_wait_for_edit_interval 表示等待编辑间隔收尾 :param parse_mode: Telegram 消息格式类型,默认 MarkdownV2,可传 HTML :return: 编辑是否成功 """ @@ -646,6 +646,7 @@ class TelegramModule(_MessageChannelModuleBase[Telegram]): buttons=buttons, parse_mode=parse_mode, rich_message=(metadata or {}).get("telegram_rich_message"), + wait_for_interval=bool((metadata or {}).get("telegram_wait_for_edit_interval")), ) if result: return True diff --git a/app/modules/telegram/telegram.py b/app/modules/telegram/telegram.py index 72be309ee..9499c1827 100644 --- a/app/modules/telegram/telegram.py +++ b/app/modules/telegram/telegram.py @@ -312,15 +312,33 @@ class Telegram: ) return True - def _reserve_edit(self, chat_id: Union[str, int], message_id: Union[str, int]) -> bool: + def _reserve_edit( + self, + chat_id: Union[str, int], + message_id: Union[str, int], + wait_for_interval: bool = False, + ) -> bool: """ 原子预留一次消息编辑,执行聊天冷却和消息级最小间隔检查。 + + :param chat_id: 编辑目标会话标识 + :param message_id: 编辑目标消息标识 + :param wait_for_interval: 是否等待一次消息级间隔,最长两秒;聊天 429 冷却不会等待 + :return: 成功预留编辑时返回 True,否则返回 False """ self._ensure_edit_state() if self._is_chat_rate_limited(chat_id): return False - now = time.monotonic() key = (str(chat_id), int(message_id)) + if wait_for_interval: + with self._edit_state_lock: + remaining = self._edit_next_allowed.get(key, 0.0) - time.monotonic() + if remaining > 0: + # 调用方在线程池内收尾;不占用事件循环,也不持锁等待或无限追赶并发编辑。 + time.sleep(min(remaining, TELEGRAM_EDIT_MIN_INTERVAL_SECONDS)) + if self._is_chat_rate_limited(chat_id): + return False + now = time.monotonic() with self._edit_state_lock: if self._edit_next_allowed.get(key, 0.0) > now: return False @@ -1547,6 +1565,7 @@ class Telegram: stop_typing: bool = False, parse_mode: Optional[str] = None, rich_message: Optional[str] = None, + wait_for_interval: bool = False, ) -> Optional[bool]: """ 限频更新原消息,富文本失败时尝试纯文本回退。 @@ -1559,12 +1578,13 @@ class Telegram: :param stop_typing: 是否在操作收尾时请求停止输入状态,默认 False :param parse_mode: 消息格式,默认 MarkdownV2;支持 HTML 和 plain 纯文本 :param rich_message: 可选富文本正文;非空时覆盖普通标题和正文 + :param wait_for_interval: 是否等待消息级编辑间隔后再发送;429 冷却不会等待 :return: 更新成功或内容未变化返回 True;限流、空内容或失败返回 False;客户端缺失返回 None """ if not self._bot: return None - if not self._reserve_edit(chat_id, message_id): + if not self._reserve_edit(chat_id, message_id, wait_for_interval=wait_for_interval): return False parse_mode = self._normalize_parse_mode(parse_mode) diff --git a/tests/test_agent_streaming_lifecycle.py b/tests/test_agent_streaming_lifecycle.py index 1dd3aa8ce..3def168bd 100644 --- a/tests/test_agent_streaming_lifecycle.py +++ b/tests/test_agent_streaming_lifecycle.py @@ -256,10 +256,13 @@ async def test_streaming_start_waits_for_concurrent_stop_final_flush() -> None: flush_started = asyncio.Event() release_flush = asyncio.Event() - async def blocking_flush() -> None: + async def blocking_flush(*, final: bool = False) -> None: """ 把最终刷新停留在生命周期锁内,验证新启动等待。 + + :param final: 是否执行停止阶段的最终刷新 """ + assert final flush_started.set() await release_flush.wait() diff --git a/tests/test_telegram.py b/tests/test_telegram.py index dc08d105c..02f0a92b7 100644 --- a/tests/test_telegram.py +++ b/tests/test_telegram.py @@ -15,7 +15,7 @@ from app.domain.metainfo import MetaInfo from app.modules.telegram.botapi import TelegramApiError, TelegramNetworkError from app.modules.telegram.module import TelegramModule from app.modules.telegram.telegram import Telegram -from app.schemas.message import Message, MessageResponse +from app.schemas.message import Message from app.schemas.types import MediaType, NotificationChannel @@ -45,11 +45,27 @@ def telegram(): @pytest.mark.asyncio -async def test_agent_fallback_skips_progress_during_telegram_edit_interval(telegram, monkeypatch): +@pytest.mark.parametrize("partial_answer, trailing_tool", [(False, False), (True, False), (True, True)]) +async def test_agent_final_edit_waits_during_telegram_edit_interval( + telegram, monkeypatch, partial_answer, trailing_tool, +): """ - 真实 Telegram 两秒编辑间隔触发回退时,已发送统计不能随最终答案再发一次。 + 真实 Telegram 两秒编辑间隔内收尾时,应等待并把完整答案编辑回原消息。 """ - monkeypatch.setattr("app.modules.telegram.telegram.time.monotonic", lambda: 100.0) + now = [100.0] + waits = [] + monkeypatch.setattr("app.modules.telegram.telegram.time.monotonic", lambda: now[0]) + + def advance(seconds): + """ + 记录并推进 Telegram 编辑等待,保持用例离线且无需真实延时。 + + :param seconds: 等待秒数 + """ + waits.append(seconds) + now[0] += seconds + + monkeypatch.setattr("app.modules.telegram.telegram.time.sleep", advance) telegram.bot.send_rich_message.return_value = { "message_id": 101, "chat": {"id": "fake_chat_id"}, } @@ -57,35 +73,53 @@ async def test_agent_fallback_skips_progress_during_telegram_edit_interval(teleg handler._channel = NotificationChannel.Telegram.value handler._source = "bot" handler._streaming_enabled = True + handler._original_message_id = "question" + module = TelegramModule() + module._channel = NotificationChannel.Telegram + monkeypatch.setattr(module, "get_configs", lambda: {"bot": SimpleNamespace(name="bot")}) + monkeypatch.setattr(module, "get_instance", lambda _name: telegram) + monkeypatch.setattr(module, "check_message", lambda *_args: True) + monkeypatch.setattr(module, "get_notification_targets", lambda _message: None) async def dispatch(func, *args, **kwargs): """ 保留真实 Telegram 发送及限频逻辑,只隔离宿主模块路由和远端 Bot API。 """ if func.__name__ == "send_direct_message": - message = args[0] - result = telegram.send_msg(title="", text=message.text, rich_message=message.rich_message) - return MessageResponse(source="bot", **result) + return module.send_direct_message(args[0]) if func.__name__ == "edit_message": - return telegram.edit_msg( - chat_id=kwargs["chat_id"], message_id=kwargs["message_id"], - text=kwargs["text"], - rich_message=kwargs["metadata"]["telegram_rich_message"], - ) + return module.edit_message(**kwargs) return True monkeypatch.setattr("app.agent.callback.run_in_threadpool", dispatch) handler.record_tool_call("execute_command", tool_kwargs={"command": "capture"}) handler.thinking_started() await handler._flush_messages() - handler.emit("截图已发送。") + handler.emit("| 剧名 | 简介 |\n| --- | --- |\n" + "| 示例 | 推荐理由 |\n" * 8) + if partial_answer: + now[0] += 2.0 + await handler._flush_messages() + handler.emit("\n其中这些剧集评价最稳。\n\n要继续搜索吗?") + if trailing_tool: + handler.record_tool_call("read_file", tool_kwargs={"file_path": "result.txt"}) + handler.flush_pending_tool_summary() + expected_text = handler._buffer - all_sent, _ = await handler.stop_streaming() + all_sent, final_text = await handler.stop_streaming() - assert not all_sent - assert await handler.take() == "截图已发送。" + assert all_sent + assert waits == [2.0] + assert await handler.take() == "" telegram.bot.send_rich_message.assert_called_once() - telegram.bot.edit_message_text.assert_not_called() + assert telegram.bot.edit_message_text.call_count == (2 if partial_answer else 1) + assert telegram.bot.edit_message_text.call_args.kwargs["message_id"] == 101 + rich_html = telegram.bot.edit_message_text.call_args.kwargs["rich_message"]["html"] + assert "" in rich_html + assert rich_html.count("") == 8 + assert "要继续搜索吗?" in rich_html + assert final_text == expected_text + assert "要继续搜索吗?" in final_text + assert "思考中" not in final_text @pytest.mark.parametrize("url", [None, "", " ", 123, False]) diff --git a/tests/test_telegram_rich_message_fallback.py b/tests/test_telegram_rich_message_fallback.py index f2d636f5d..737e25767 100644 --- a/tests/test_telegram_rich_message_fallback.py +++ b/tests/test_telegram_rich_message_fallback.py @@ -279,7 +279,8 @@ class TestEditMsgFallsBackToPlainText: assert result is False client._api.edit_message_text.assert_called_once() - def test_429_cooldown_skips_follow_up_edits(self, monkeypatch): + @pytest.mark.parametrize("wait_for_interval", [False, True]) + def test_429_cooldown_skips_follow_up_edits(self, monkeypatch, wait_for_interval): """ 同一聊天的冷却期内后续编辑不应再触碰 Bot API。 """ @@ -293,7 +294,12 @@ class TestEditMsgFallsBackToPlainText: assert client.edit_msg(chat_id="chat", message_id=5, text="正文", rich_message="# 结果") is False now[0] += 10 - assert client.edit_msg(chat_id="chat", message_id=5, text="更新", rich_message="# 更新") is False + with patch("app.modules.telegram.telegram.time.sleep") as sleep_mock: + assert client.edit_msg( + chat_id="chat", message_id=5, text="更新", rich_message="# 更新", + wait_for_interval=wait_for_interval, + ) is False + sleep_mock.assert_not_called() assert client._api.edit_message_text.call_count == 1 now[0] += 8 client._api.edit_message_text.side_effect = None @@ -301,6 +307,41 @@ class TestEditMsgFallsBackToPlainText: assert client.edit_msg(chat_id="chat", message_id=5, text="更新", rich_message="# 更新") is True assert client._api.edit_message_text.call_count == 2 + @pytest.mark.parametrize("interruption", ["429", "edit"]) + def test_final_edit_rechecks_cooldown_and_interval_after_one_wait(self, monkeypatch, interruption): + """ + 等待期间出现聊天冷却或同消息并发编辑时,不追加请求或无限延长等待。 + """ + client = _make_client() + now = [100.0] + waits = [] + monkeypatch.setattr("app.modules.telegram.telegram.time.monotonic", lambda: now[0]) + client._api.edit_message_text.return_value = {"message_id": 5} + assert client.edit_msg(chat_id="chat", message_id=5, text="首帧", rich_message="# 首帧") is True + + def interrupt_wait(seconds): + """ + 在虚拟等待结束时模拟其他请求带来的状态变化。 + + :param seconds: 本次等待秒数 + """ + waits.append(seconds) + now[0] += seconds + if interruption == "429": + client._mark_rate_limit( + "chat", TelegramApiError("editMessageText", "Too Many Requests", 429, {"retry_after": 17}), + ) + else: + assert client.edit_msg(chat_id="chat", message_id=5, text="并发编辑") is True + + monkeypatch.setattr("app.modules.telegram.telegram.time.sleep", interrupt_wait) + assert client.edit_msg( + chat_id="chat", message_id=5, text="完整正文", rich_message="# 完整正文", + wait_for_interval=True, + ) is False + assert waits == [TELEGRAM_EDIT_MIN_INTERVAL_SECONDS] + assert client._api.edit_message_text.call_count == (1 if interruption == "429" else 2) + def test_successful_edits_are_throttled_and_latest_text_wins(self, monkeypatch): """ 最小编辑间隔内只保留最新一次刷新,解除后发送当前文本。 @@ -318,6 +359,34 @@ class TestEditMsgFallsBackToPlainText: assert client.edit_msg(chat_id="chat", message_id=5, text="最新", rich_message="# 最新") is True assert client._api.edit_message_text.call_count == 2 + def test_final_edit_waits_for_message_interval(self, monkeypatch): + """ + 收尾编辑应等待消息级间隔,确保完整正文覆盖原消息。 + """ + client = _make_client() + now = [100.0] + monkeypatch.setattr("app.modules.telegram.telegram.time.monotonic", lambda: now[0]) + + def advance(seconds): + """ + 用虚拟时钟推进消息级编辑间隔。 + + :param seconds: 等待秒数 + """ + now[0] += seconds + + monkeypatch.setattr("app.modules.telegram.telegram.time.sleep", advance) + client._api.edit_message_text.return_value = {"message_id": 5} + + assert client.edit_msg(chat_id="chat", message_id=5, text="首帧", rich_message="# 首帧") is True + assert client.edit_msg( + chat_id="chat", message_id=5, text="完整正文", rich_message="# 完整正文", + wait_for_interval=True, + ) is True + assert client._api.edit_message_text.call_count == 2 + assert now[0] == 100.0 + TELEGRAM_EDIT_MIN_INTERVAL_SECONDS + assert "完整正文" in client._api.edit_message_text.call_args.kwargs["rich_message"]["html"] + class TestFallbackDisablesMarkdownV2: """
推荐理由