From 8c0afac5d1e6cad0ade3f5e1ae7e26d603f4527b Mon Sep 17 00:00:00 2001 From: qqcomeup Date: Thu, 9 Jul 2026 12:52:22 +0800 Subject: [PATCH] feat: support prompt-bound plugin input replies (#6087) --- app/chain/message.py | 24 +- app/helper/interaction.py | 77 ++++- app/modules/telegram/__init__.py | 19 +- app/schemas/message.py | 2 + tests/test_media_interaction.py | 551 +++++++++++++++++++++++++++++++ tests/test_telegram.py | 244 +++++++++++++- 6 files changed, 903 insertions(+), 14 deletions(-) diff --git a/app/chain/message.py b/app/chain/message.py index e9b278bdd..fc5aa46c3 100644 --- a/app/chain/message.py +++ b/app/chain/message.py @@ -141,9 +141,9 @@ class MessageChain(ChainBase): logger.debug(f"未识别到消息内容：：{body}{form}{args}") return - # 获取原消息ID信息 original_message_id = info.message_id original_chat_id = info.chat_id + reply_to_message_id = info.reply_to_message_id # 处理消息 self.handle_message( @@ -154,6 +154,7 @@ class MessageChain(ChainBase): text=text, original_message_id=original_message_id, original_chat_id=original_chat_id, + reply_to_message_id=reply_to_message_id, images=images, audio_refs=audio_refs, files=files, @@ -171,6 +172,7 @@ class MessageChain(ChainBase): images: Optional[List[CommingMessage.MessageImage]] = None, audio_refs: Optional[List[str]] = None, files: Optional[List[CommingMessage.MessageAttachment]] = None, + reply_to_message_id: Optional[Union[str, int]] = None, ) -> None: """ 识别消息内容，执行操作 @@ -213,6 +215,7 @@ class MessageChain(ChainBase): username=username, text=text, original_chat_id=original_chat_id, + reply_to_message_id=reply_to_message_id, images=images, audio_refs=audio_refs, files=files, @@ -255,6 +258,7 @@ class MessageChain(ChainBase): text=text, original_message_id=original_message_id, original_chat_id=original_chat_id, + reply_to_message_id=reply_to_message_id, images=images, audio_refs=audio_refs, files=files, @@ -286,6 +290,7 @@ class MessageChain(ChainBase): files: Optional[List[CommingMessage.MessageAttachment]] = None, has_audio_input: bool = False, processing_status: Optional[_ProcessingStatus] = None, + reply_to_message_id: Optional[Union[str, int]] = None, ) -> bool: """执行实际消息路由，便于统一包裹处理中状态。""" @@ -316,6 +321,7 @@ class MessageChain(ChainBase): username=username, text=text, original_chat_id=original_chat_id, + reply_to_message_id=reply_to_message_id, images=images, audio_refs=audio_refs, files=files, @@ -444,6 +450,8 @@ class MessageChain(ChainBase): "userid": userid, "channel": channel, "source": source, + "chat_id": original_chat_id, + "reply_to_message_id": reply_to_message_id, }, ) return False @@ -460,6 +468,7 @@ class MessageChain(ChainBase): audio_refs: Optional[List[str]] = None, files: Optional[List[CommingMessage.MessageAttachment]] = None, has_audio_input: bool = False, + reply_to_message_id: Optional[Union[str, int]] = None, ) -> bool: """ 将插件输入会话中的下一条普通文本派发给指定插件。 @@ -469,8 +478,14 @@ class MessageChain(ChainBase): if text.startswith("CALLBACK:"): return False + is_cancel_text = text.strip().lower() in {"取消", "退出", "q", "quit", "exit"} request, status = plugin_input_interaction_manager.consume_by_user( - userid, channel, source, original_chat_id + userid, + channel, + source, + original_chat_id, + reply_to_message_id=reply_to_message_id, + bypass_reply_check=is_cancel_text, ) if not request: return False @@ -487,6 +502,7 @@ class MessageChain(ChainBase): "source": source, "username": username, "chat_id": original_chat_id, + "reply_to_message_id": reply_to_message_id, "prompt_id": request.prompt_id, "input_session_id": request.request_id, "expired": True, @@ -505,7 +521,7 @@ class MessageChain(ChainBase): ) return not text.strip().startswith("/") - if text.strip().lower() in {"取消", "退出", "q", "quit", "exit"}: + if is_cancel_text: self.eventmanager.send_event( EventType.MessageAction, { @@ -517,6 +533,7 @@ class MessageChain(ChainBase): "source": source, "username": username, "chat_id": original_chat_id, + "reply_to_message_id": reply_to_message_id, "prompt_id": request.prompt_id, "input_session_id": request.request_id, "cancelled": True, @@ -547,6 +564,7 @@ class MessageChain(ChainBase): "source": source, "username": username, "chat_id": original_chat_id, + "reply_to_message_id": reply_to_message_id, "prompt_id": request.prompt_id, "input_session_id": request.request_id, "payload": request.payload, diff --git a/app/helper/interaction.py b/app/helper/interaction.py index 27088d3ad..0bb079915 100644 --- a/app/helper/interaction.py +++ b/app/helper/interaction.py @@ -415,6 +415,8 @@ class PendingPluginInputInteraction: payload: Optional[Any] = None timeout_seconds: int = 120 created_at: datetime = field(default_factory=datetime.now) + # Optional reply binding for channels that can report reply_to_message_id. + prompt_message_id: Optional[str] = None @property def expires_at(self) -> datetime: @@ -504,6 +506,8 @@ class PluginInputInteractionManager: prompt_id: Optional[str] = None, timeout_seconds: int = 120, payload: Optional[Any] = None, + *, + prompt_message_id: Optional[Union[str, int]] = None, ) -> PendingPluginInputInteraction: with self._lock: self._cleanup_locked() @@ -526,6 +530,13 @@ class PluginInputInteractionManager: if not self._keys_overlap(stored_key, key) } + normalized_chat_id = str(chat_id) if chat_id not in (None, "") else None + normalized_prompt_message_id = ( + str(prompt_message_id) + if channel == MessageChannel.Telegram and normalized_chat_id and prompt_message_id not in (None, "") + else None + ) + request = PendingPluginInputInteraction( request_id=uuid.uuid4().hex[:12], user_id=str(user_id), @@ -533,8 +544,9 @@ class PluginInputInteractionManager: channel=channel, source=source, username=username, - chat_id=str(chat_id) if chat_id not in (None, "") else None, + chat_id=normalized_chat_id, prompt_id=prompt_id, + prompt_message_id=normalized_prompt_message_id, timeout_seconds=timeout_seconds, payload=payload, ) @@ -563,8 +575,16 @@ class PluginInputInteractionManager: source: Optional[str] = None, chat_id: Optional[Union[str, int]] = None, ) -> Optional[PendingPluginInputInteraction]: - request, _ = self.consume_by_user(user_id, channel, source, chat_id) - return request + with self._lock: + self._cleanup_locked() + key, request_id = self._find_key_and_request_id_locked(user_id, channel, source, chat_id) + if request_id: + self._by_user_channel.pop(key, None) + return self._by_id.pop(request_id, None) + expired_key, request = self._find_expired_key_and_request_locked(user_id, channel, source, chat_id) + if expired_key: + self._expired_by_user_channel.pop(expired_key, None) + return request def consume_by_user( self, @@ -572,23 +592,64 @@ class PluginInputInteractionManager: channel: Optional[MessageChannel] = None, source: Optional[str] = None, chat_id: Optional[Union[str, int]] = None, + *, + reply_to_message_id: Optional[Union[str, int]] = None, + bypass_reply_check: bool = False, ) -> Tuple[Optional[PendingPluginInputInteraction], Optional[str]]: with self._lock: key, request_id = self._find_key_and_request_id_locked(user_id, channel, source, chat_id) if request_id: - self._by_user_channel.pop(key, None) - request = self._by_id.pop(request_id, None) - if request: - status = "expired" if request.expires_at < datetime.now() else "active" - return request, status + request = self._by_id.get(request_id) + if not request: + self._by_user_channel.pop(key, None) + elif request.expires_at < datetime.now(): + self._by_user_channel.pop(key, None) + self._by_id.pop(request_id, None) + if request.prompt_message_id: + return None, None + return request, "expired" + elif not self._reply_matches_prompt( + request, + chat_id, + reply_to_message_id, + ignore_reply_to_message_id=bypass_reply_check, + ): + return None, None + else: + self._by_user_channel.pop(key, None) + self._by_id.pop(request_id, None) + return request, "active" + self._cleanup_locked() key, request = self._find_expired_key_and_request_locked(user_id, channel, source, chat_id) if request: self._expired_by_user_channel.pop(key, None) + if request.prompt_message_id: + return None, None return request, "expired" self._cleanup_locked() return None, None + @staticmethod + def _reply_matches_prompt( + request: PendingPluginInputInteraction, + chat_id: Optional[Union[str, int]], + reply_to_message_id: Optional[Union[str, int]], + *, + ignore_reply_to_message_id: bool = False, + ) -> bool: + if not request.prompt_message_id: + return True + if not request.chat_id or chat_id in (None, ""): + return False + if str(chat_id) != str(request.chat_id): + return False + if ignore_reply_to_message_id: + return True + if reply_to_message_id in (None, ""): + return False + return str(reply_to_message_id) == str(request.prompt_message_id) + def _find_request_id_locked( self, user_id: Union[str, int], diff --git a/app/modules/telegram/__init__.py b/app/modules/telegram/__init__.py index 2af5be926..912e174ea 100644 --- a/app/modules/telegram/__init__.py +++ b/app/modules/telegram/__init__.py @@ -252,9 +252,11 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]): 处理普通文本消息 """ text = msg.get("text") or msg.get("caption") + message_id = msg.get("message_id") user_id = msg.get("from", {}).get("id") user_name = msg.get("from", {}).get("username") chat_id = msg.get("chat", {}).get("id") + reply_to_message_id = (msg.get("reply_to_message") or {}).get("message_id") # 将 text_link 实体中的 URL 嵌入到文本中 if text: @@ -309,7 +311,9 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]): userid=user_id, username=user_name, text=cleaned_text, + message_id=message_id, chat_id=str(chat_id) if chat_id else None, + reply_to_message_id=reply_to_message_id, images=images if images else None, audio_refs=audio_refs if audio_refs else None, files=files if files else None, @@ -514,6 +518,12 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]): parse_mode=message.parse_mode, ) else: + # Telegram 的 reply_markup 不能同时承载 InlineKeyboard 和 ForceReply。 + # 普通通知只清空可编辑消息 ID，仍保留原会话作为新消息目标。 + has_interaction_context = bool(message.buttons or message.force_reply) + original_message_id = ( + message.original_message_id if has_interaction_context else None + ) client.send_msg( title=message.title, text=message.text, @@ -522,7 +532,7 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]): link=message.link, buttons=message.buttons, force_reply=message.force_reply, - original_message_id=message.original_message_id, + original_message_id=original_message_id, original_chat_id=message.original_chat_id, disable_web_page_preview=message.disable_web_page_preview, parse_mode=message.parse_mode, @@ -735,12 +745,19 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]): parse_mode=message.parse_mode, ) else: + # direct message 只禁用编辑旧消息；仅 ForceReply 使用 original_chat_id + # 发回原会话，并保留 original_message_id 让 client reply_to 原消息。 + original_chat_id = message.original_chat_id if message.force_reply else None + original_message_id = message.original_message_id if message.force_reply else None result = client.send_msg( title=message.title, text=message.text, image=message.image, userid=userid, link=message.link, + force_reply=message.force_reply, + original_message_id=original_message_id, + original_chat_id=original_chat_id, disable_web_page_preview=message.disable_web_page_preview, parse_mode=message.parse_mode, ) diff --git a/app/schemas/message.py b/app/schemas/message.py index 95c5585c8..66bcb57cb 100644 --- a/app/schemas/message.py +++ b/app/schemas/message.py @@ -175,6 +175,8 @@ class CommingMessage(BaseModel): message_id: Optional[Union[str, int]] = None # 聊天ID（用于回调时定位聊天） chat_id: Optional[str] = None + # 回复目标消息ID（用于 ForceReply 等回复场景） + reply_to_message_id: Optional[Union[str, int]] = None # 完整的回调查询信息（原始数据） callback_query: Optional[Dict] = None # 图片列表（图片URL或file_id） diff --git a/tests/test_media_interaction.py b/tests/test_media_interaction.py index acf3433b0..c777ce669 100644 --- a/tests/test_media_interaction.py +++ b/tests/test_media_interaction.py @@ -160,6 +160,110 @@ def test_message_routes_text_reply_to_media_interaction_before_ai(): handle_ai.assert_not_called() +def test_message_process_preserves_parser_message_id_context(): + """消息链不按渠道解释 message_id，只透传解析器给出的原消息上下文。""" + chain = MessageChain() + incoming = CommingMessage( + channel=MessageChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + text="东张西望", + message_id=101, + chat_id="chat-a", + reply_to_message_id=99, + ) + + with patch.object(chain, "message_parser", return_value=incoming), patch.object( + chain, "handle_message" + ) as handle_message: + chain.process(body=None, form=None, args={"source": "telegram-test"}) + + handle_message.assert_called_once() + kwargs = handle_message.call_args.kwargs + assert kwargs["original_message_id"] == 101 + assert kwargs["original_chat_id"] == "chat-a" + assert kwargs["reply_to_message_id"] == 99 + + +def test_message_process_keeps_callback_message_id_as_edit_context(): + """按钮回调的 message_id 仍应作为机器人原消息 ID 传递，供编辑原消息使用。""" + chain = MessageChain() + incoming = CommingMessage( + channel=MessageChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + text="CALLBACK:demo", + is_callback=True, + message_id=101, + chat_id="chat-a", + ) + + with patch.object(chain, "message_parser", return_value=incoming), patch.object( + chain, "handle_message" + ) as handle_message: + chain.process(body=None, form=None, args={"source": "telegram-test"}) + + handle_message.assert_called_once() + kwargs = handle_message.call_args.kwargs + assert kwargs["original_message_id"] == 101 + assert kwargs["original_chat_id"] == "chat-a" + + +def test_message_process_preserves_non_telegram_plain_message_id(): + """非 Telegram 渠道保持旧行为，普通消息 ID 仍向下传递给渠道实现自行解释。""" + chain = MessageChain() + incoming = CommingMessage( + channel=MessageChannel.Slack, + source="slack-test", + userid="10001", + username="tester", + text="hello", + message_id="slack-message-ts", + chat_id="slack-channel", + ) + + with patch.object(chain, "message_parser", return_value=incoming), patch.object( + chain, "handle_message" + ) as handle_message: + chain.process(body=None, form=None, args={"source": "slack-test"}) + + handle_message.assert_called_once() + kwargs = handle_message.call_args.kwargs + assert kwargs["original_message_id"] == "slack-message-ts" + assert kwargs["original_chat_id"] == "slack-channel" + + +def test_handle_message_keeps_legacy_positional_images_argument(): + """新增 reply_to_message_id 不应改变旧位置参数 images/audio/files 的含义。""" + chain = MessageChain() + images = [CommingMessage.MessageImage(ref="tg://file_id/photo-1")] + + with patch.object( + chain, "_handle_plugin_input_interaction", return_value=False + ), patch.object( + chain, "_mark_message_processing_started", return_value=None + ), patch.object( + chain, "_mark_message_processing_finished" + ), patch.object(chain, "_handle_message_core", return_value=False) as handle_core: + chain.handle_message( + MessageChannel.Telegram, + "telegram-test", + "10001", + "tester", + "带图消息", + None, + "chat-a", + images, + ) + + handle_core.assert_called_once() + kwargs = handle_core.call_args.kwargs + assert kwargs["images"] == images + assert kwargs["reply_to_message_id"] is None + + def test_plugin_input_session_captures_plain_text_before_media_interaction(): """插件输入会话存在时，普通文本应派发给插件而不是媒体交互。""" chain = MessageChain() @@ -209,6 +313,7 @@ def test_plugin_input_session_captures_plain_text_before_media_interaction(): "source": "wechat-test", "username": "tester", "chat_id": None, + "reply_to_message_id": None, "prompt_id": "prompt-1", "input_session_id": request.request_id, "payload": {"step": "name"}, @@ -528,6 +633,326 @@ def test_plugin_input_session_does_not_capture_other_chat_text(): assert payload["chat_id"] == "chat-a" +def test_plugin_input_prompt_message_requires_matching_reply(): + """绑定提示消息 ID 的插件输入只应消费当前 ForceReply 回复。""" + chain = MessageChain() + request = plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Telegram, + source="telegram-test", + username="tester", + chat_id="chat-a", + prompt_message_id="prompt-current", + payload={"step": "keyword"}, + ) + + with patch.object(chain, "_record_user_message") as record_message, patch.object( + chain.eventmanager, "send_event" + ) as send_event: + chain.handle_message( + channel=MessageChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + text="旧回复框文本", + original_chat_id="chat-a", + reply_to_message_id="prompt-old", + ) + + record_message.assert_called_once() + assert plugin_input_interaction_manager.get_by_user( + "10001", MessageChannel.Telegram, "telegram-test", "chat-a" + ) == request + assert not any( + call.args and call.args[0] == EventType.MessageAction + for call in send_event.call_args_list + ) + + with patch.object(chain, "_record_user_message") as record_message, patch.object( + chain.eventmanager, "send_event" + ) as send_event: + chain.handle_message( + channel=MessageChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + text="当前回复框文本", + original_chat_id="chat-a", + reply_to_message_id="prompt-current", + ) + + record_message.assert_not_called() + send_event.assert_called_once() + event_type, payload = send_event.call_args.args + assert event_type == EventType.MessageAction + assert payload["input_session_id"] == request.request_id + assert payload["input_text"] == "当前回复框文本" + assert payload["reply_to_message_id"] == "prompt-current" + assert plugin_input_interaction_manager.get_by_user( + "10001", MessageChannel.Telegram, "telegram-test", "chat-a" + ) is None + + +def test_plugin_input_prompt_message_matches_integer_reply_ids(): + """真实 Telegram message_id 为 int，应与内部 str 归一化后的 prompt_message_id 匹配。""" + chain = MessageChain() + request = plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Telegram, + source="telegram-test", + username="tester", + chat_id=10001, + prompt_message_id=99, + payload={"step": "keyword"}, + ) + + with patch.object(chain.eventmanager, "send_event") as send_event: + chain.handle_message( + channel=MessageChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + text="翡翠台", + original_chat_id=10001, + reply_to_message_id=99, + ) + + send_event.assert_called_once() + event_type, payload = send_event.call_args.args + assert event_type == EventType.MessageAction + assert payload["input_session_id"] == request.request_id + assert payload["input_text"] == "翡翠台" + + +def test_plugin_input_prompt_message_ignores_plain_text_without_reply(): + """用户未使用 ForceReply 回复框直接发文本时，绑定会话不应消费该文本。""" + chain = MessageChain() + request = plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Telegram, + source="telegram-test", + username="tester", + chat_id="chat-a", + prompt_message_id="prompt-current", + payload={"step": "keyword"}, + ) + + with patch.object(chain, "_record_user_message") as record_message, patch.object( + chain.eventmanager, "send_event" + ) as send_event: + chain.handle_message( + channel=MessageChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + text="直接输入文本", + original_chat_id="chat-a", + ) + + record_message.assert_called_once() + assert plugin_input_interaction_manager.get_by_user( + "10001", MessageChannel.Telegram, "telegram-test", "chat-a" + ) == request + assert not any( + call.args and call.args[0] == EventType.MessageAction + for call in send_event.call_args_list + ) + + +def test_plugin_input_prompt_message_allows_direct_cancel_without_reply(): + """绑定 ForceReply 时，取消词应能直接结束会话，避免用户被残留回复框卡住。""" + chain = MessageChain() + request = plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Telegram, + source="telegram-test", + username="tester", + chat_id="chat-a", + prompt_message_id="prompt-current", + payload={"step": "keyword"}, + ) + + with patch.object(chain, "_record_user_message") as record_message, patch.object( + chain.eventmanager, "send_event" + ) as send_event, patch.object(chain, "post_message") as post_message: + chain.handle_message( + channel=MessageChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + text="取消", + original_chat_id="chat-a", + ) + + record_message.assert_not_called() + send_event.assert_called_once() + event_type, payload = send_event.call_args.args + assert event_type == EventType.MessageAction + assert payload["input_session_id"] == request.request_id + assert payload["cancelled"] is True + post_message.assert_called_once() + assert plugin_input_interaction_manager.get_by_user( + "10001", MessageChannel.Telegram, "telegram-test", "chat-a" + ) is None + + +def test_expired_prompt_message_cancel_text_falls_back_to_normal_search_without_notice(): + """绑定 ForceReply 过期后，即使输入取消词也应静默放行给普通文本链路。""" + chain = MessageChain() + plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Telegram, + source="telegram-test", + username="tester", + chat_id="chat-a", + prompt_message_id="prompt-expired", + timeout_seconds=60, + ).created_at = datetime.now() - timedelta(seconds=61) + + with patch.object(chain, "_record_user_message") as record_message, patch.object( + chain.eventmanager, "send_event" + ) as send_event, patch.object( + chain, "_handle_message_core", return_value=False + ) as handle_core: + chain.handle_message( + channel=MessageChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + text="取消", + original_chat_id="chat-a", + reply_to_message_id="prompt-expired", + ) + + record_message.assert_called_once() + handle_core.assert_called_once() + assert not any( + call.args and call.args[0] == EventType.MessageAction + for call in send_event.call_args_list + ) + assert plugin_input_interaction_manager.get_by_user( + "10001", MessageChannel.Telegram, "telegram-test", "chat-a" + ) is None + + +def test_plugin_input_prompt_message_requires_matching_chat_id(): + """绑定提示消息 ID 时还必须匹配 chat_id，避免跨聊天同号消息误消费。""" + chain = MessageChain() + request = plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Telegram, + source="telegram-test", + username="tester", + chat_id="chat-a", + prompt_message_id="prompt-current", + payload={"step": "keyword"}, + ) + + with patch.object(chain, "_record_user_message") as record_message, patch.object( + chain.eventmanager, "send_event" + ) as send_event: + chain.handle_message( + channel=MessageChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + text="其他聊天同号回复", + original_chat_id="chat-b", + reply_to_message_id="prompt-current", + ) + + record_message.assert_called_once() + assert plugin_input_interaction_manager.get_by_user( + "10001", MessageChannel.Telegram, "telegram-test", "chat-a" + ) == request + assert not any( + call.args and call.args[0] == EventType.MessageAction + for call in send_event.call_args_list + ) + + +def test_expired_prompt_message_input_falls_back_to_normal_search_without_notice(): + """回复过期 ForceReply 时不提示插件输入超时，交回普通文本搜索。""" + chain = MessageChain() + request = plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Telegram, + source="telegram-test", + username="tester", + chat_id="chat-a", + prompt_message_id="prompt-expired", + timeout_seconds=60, + ) + request.created_at = datetime.now() - timedelta(seconds=61) + + with patch.object(chain, "_record_user_message") as record_message, patch.object( + chain.eventmanager, "send_event" + ) as send_event, patch.object(chain, "post_message") as post_message: + chain.handle_message( + channel=MessageChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + text="过期回复框文本", + original_chat_id="chat-a", + reply_to_message_id="prompt-expired", + ) + + record_message.assert_called_once() + post_message.assert_not_called() + assert not any( + call.args and call.args[0] == EventType.MessageAction + for call in send_event.call_args_list + ) + assert plugin_input_interaction_manager.get_by_user( + "10001", MessageChannel.Telegram, "telegram-test", "chat-a" + ) is None + + +def test_expired_prompt_message_without_reply_falls_back_and_clears_state(): + """绑定会话过期后，未命中回复框的文本也应放行并清理过期状态。""" + chain = MessageChain() + request = plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Telegram, + source="telegram-test", + username="tester", + chat_id="chat-a", + prompt_message_id="prompt-expired", + timeout_seconds=60, + ) + request.created_at = datetime.now() - timedelta(seconds=61) + + with patch.object(chain, "_record_user_message") as record_message, patch.object( + chain.eventmanager, "send_event" + ) as send_event: + chain.handle_message( + channel=MessageChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + text="过期后直接输入", + original_chat_id="chat-a", + ) + + record_message.assert_called_once() + assert not any( + call.args and call.args[0] == EventType.MessageAction + for call in send_event.call_args_list + ) + assert plugin_input_interaction_manager.get_by_user( + "10001", MessageChannel.Telegram, "telegram-test", "chat-a" + ) is None + + def test_plugin_input_chatless_session_keeps_legacy_chat_fallback(): """旧插件未绑定 chat_id 时，同 source 消息仍可兼容消费。""" chain = MessageChain() @@ -875,6 +1300,92 @@ def test_plugin_input_session_with_no_channel_and_no_source_does_not_match_speci ) +def test_plugin_input_create_or_replace_keeps_legacy_positional_timeout_and_payload(): + """新增 prompt_message_id 不应改变旧位置参数 timeout_seconds/payload 的含义。""" + request = plugin_input_interaction_manager.create_or_replace( + "10001", + "demo_plugin", + MessageChannel.Telegram, + "telegram-test", + "tester", + "chat-a", + "prompt-id", + 30, + {"step": "legacy"}, + ) + + assert request.timeout_seconds == 30 + assert request.payload == {"step": "legacy"} + assert request.prompt_message_id is None + + +def test_plugin_input_create_or_replace_ignores_prompt_message_without_chat_id(): + """缺少 chat_id 时不启用 prompt_message_id 绑定，避免创建永远无法消费的会话。""" + request = plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Telegram, + source="telegram-test", + username="tester", + prompt_message_id="prompt-current", + ) + + assert request.chat_id is None + assert request.prompt_message_id is None + + +def test_plugin_input_create_or_replace_ignores_prompt_message_for_non_telegram_channel(): + """非 Telegram 渠道不启用 prompt_message_id 绑定，避免渠道无法上报回复 ID 时卡死。""" + request = plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Slack, + source="slack-test", + username="tester", + chat_id="slack-channel", + prompt_message_id="prompt-current", + ) + + assert request.chat_id == "slack-channel" + assert request.prompt_message_id is None + + consumed, status = plugin_input_interaction_manager.consume_by_user( + "10001", + MessageChannel.Slack, + "slack-test", + "slack-channel", + ) + assert consumed == request + assert status == "active" + + +def test_plugin_input_bypass_reply_check_still_requires_matching_chat_id(): + """取消词绕过 reply_id 校验时，仍必须匹配绑定会话的 chat_id。""" + request = plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Telegram, + source="telegram-test", + username="tester", + chat_id="chat-a", + prompt_message_id="prompt-current", + ) + + consumed, status = plugin_input_interaction_manager.consume_by_user( + "10001", + MessageChannel.Telegram, + "telegram-test", + "chat-b", + bypass_reply_check=True, + ) + + assert consumed is None + assert status is None + assert plugin_input_interaction_manager.get_by_user( + "10001", MessageChannel.Telegram, "telegram-test", "chat-a" + ) == request + + def test_plugin_input_specific_session_replaces_overlapping_no_channel_session(): """同用户创建具体渠道会话时，应替换重叠的无渠道会话，避免下一条消息被连环接管。""" old_request = plugin_input_interaction_manager.create_or_replace( @@ -918,6 +1429,46 @@ def test_plugin_input_session_pop_by_user_consumes_once(): ) is None +def test_plugin_input_session_pop_by_user_ignores_prompt_message_binding(): + """主动清理会话时不应要求提供 ForceReply 的 reply_to_message_id。""" + request = plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Telegram, + source="telegram-test", + username="tester", + prompt_message_id="prompt-current", + ) + + assert plugin_input_interaction_manager.pop_by_user( + "10001", MessageChannel.Telegram, "telegram-test" + ) == request + assert plugin_input_interaction_manager.get_by_user( + "10001", MessageChannel.Telegram, "telegram-test" + ) is None + + +def test_plugin_input_session_pop_by_user_removes_expired_prompt_session(): + """主动清理已过期会话时，也应移除过期表中的绑定 ForceReply 会话。""" + request = plugin_input_interaction_manager.create_or_replace( + user_id="10001", + plugin_id="demo_plugin", + channel=MessageChannel.Telegram, + source="telegram-test", + username="tester", + prompt_message_id="prompt-current", + timeout_seconds=60, + ) + request.created_at = datetime.now() - timedelta(seconds=61) + + assert plugin_input_interaction_manager.pop_by_user( + "10001", MessageChannel.Telegram, "telegram-test" + ) == request + assert plugin_input_interaction_manager.pop_by_user( + "10001", MessageChannel.Telegram, "telegram-test" + ) is None + + def test_target_plugin_filter_only_allows_target_plugin_handler(): """带目标插件的输入事件不应投递给其他插件或模块级处理器。""" diff --git a/tests/test_telegram.py b/tests/test_telegram.py index 81b5c70dc..1d142d0ef 100644 --- a/tests/test_telegram.py +++ b/tests/test_telegram.py @@ -2,6 +2,7 @@ """ Telegram 模块单元测试（pytest 原生）。 """ +import json from types import SimpleNamespace from unittest.mock import MagicMock, Mock, patch @@ -51,6 +52,38 @@ def test_send_msg_success(telegram): # 验证返回值：send_msg 失败时返回 {"success": False}（非空字典，仅 truthy 检查会漏判），故显式断言 success assert result and result.get("success") + +def test_telegram_parser_preserves_reply_to_message_id(): + """Telegram ForceReply 回复应保留来源消息和被回复消息的 message_id。""" + module = TelegramModule() + client_config = SimpleNamespace(name="telegram-test", config={}) + client = SimpleNamespace(bot_username="mp_bot") + payload = { + "update_id": 1, + "message": { + "message_id": 101, + "from": {"id": 10001, "username": "tester"}, + "chat": {"id": 10001, "type": "private"}, + "text": "东张西望", + "reply_to_message": {"message_id": 99, "text": "请输入节目关键词"}, + }, + } + + with patch.object(module, "get_config", return_value=client_config), patch.object( + module, "get_instance", return_value=client + ): + message = module.message_parser( + source="telegram-test", + body=json.dumps(payload), + form=None, + args={}, + ) + + assert message.text == "东张西望" + assert message.message_id == 101 + assert message.chat_id == "10001" + assert message.reply_to_message_id == 99 + def test_send_msg_with_longtext(telegram): """测试发送长消息""" result = telegram.send_msg( @@ -309,8 +342,75 @@ def test_telegram_module_passes_parse_mode_to_client(): assert client.send_msg.call_args.kwargs["parse_mode"] == "HTML" +def test_telegram_module_plain_post_message_keeps_chat_without_editing_source_message(): + """普通通知应保留原会话目标，同时避免把来源消息 ID 当成编辑目标。""" + module = TelegramModule() + client = Mock() + + with patch.object( + module, + "get_configs", + return_value={"telegram-test": SimpleNamespace(name="telegram-test")}, + ), patch.object( + module, "check_message", return_value=True + ), patch.object( + module, "get_instance", return_value=client + ): + module.post_message( + Notification( + channel=MessageChannel.Telegram, + source="telegram-test", + title="Agent 回复", + text="处理完成", + original_message_id=123, + original_chat_id="chat-a", + ) + ) + + client.send_msg.assert_called_once() + kwargs = client.send_msg.call_args.kwargs + assert kwargs["original_message_id"] is None + assert kwargs["original_chat_id"] == "chat-a" + + def test_telegram_module_passes_force_reply_to_client(): - """模块发送通知时应透传消息指定的force_reply""" + """模块发送通知时应透传交互消息参数""" + module = TelegramModule() + client = Mock() + buttons = [[{"text": "取消", "callback_data": "cancel"}]] + + with patch.object( + module, + "get_configs", + return_value={"telegram-test": SimpleNamespace(name="telegram-test")}, + ), patch.object( + module, "check_message", return_value=True + ), patch.object( + module, "get_instance", return_value=client + ): + module.post_message( + Notification( + channel=MessageChannel.Telegram, + source="telegram-test", + title="请输入目录", + text="回复目录路径", + force_reply=True, + buttons=buttons, + original_message_id=123, + original_chat_id="chat-a", + ) + ) + + client.send_msg.assert_called_once() + kwargs = client.send_msg.call_args.kwargs + assert kwargs["force_reply"] is True + assert kwargs["buttons"] == buttons + assert kwargs["original_message_id"] == 123 + assert kwargs["original_chat_id"] == "chat-a" + + +def test_telegram_module_force_reply_sends_new_prompt_message(): + """无按钮 ForceReply 应保留原消息 ID，让 client 发新提示并 reply_to 原消息。""" module = TelegramModule() client = Mock() @@ -330,11 +430,135 @@ def test_telegram_module_passes_force_reply_to_client(): title="请输入目录", text="回复目录路径", force_reply=True, + original_message_id=123, + original_chat_id="chat-a", ) ) client.send_msg.assert_called_once() - assert client.send_msg.call_args.kwargs["force_reply"] is True + kwargs = client.send_msg.call_args.kwargs + assert kwargs["force_reply"] is True + assert kwargs["buttons"] is None + assert kwargs["original_message_id"] == 123 + assert kwargs["original_chat_id"] == "chat-a" + + +def test_telegram_module_direct_force_reply_sends_new_prompt_message(): + """direct message 的无按钮 ForceReply 同样保留原消息 ID，交给 client 发送新提示。""" + module = TelegramModule() + client = Mock() + client.send_msg.return_value = { + "success": True, + "message_id": 456, + "chat_id": "chat-a", + } + + with patch.object( + module, + "get_configs", + return_value={"telegram-test": SimpleNamespace(name="telegram-test")}, + ), patch.object( + module, "check_message", return_value=True + ), patch.object( + module, "get_instance", return_value=client + ): + response = module.send_direct_message( + Notification( + channel=MessageChannel.Telegram, + source="telegram-test", + title="请输入目录", + text="回复目录路径", + force_reply=True, + original_message_id=123, + original_chat_id="chat-a", + ) + ) + + client.send_msg.assert_called_once() + kwargs = client.send_msg.call_args.kwargs + assert kwargs["force_reply"] is True + assert "buttons" not in kwargs + assert kwargs["original_message_id"] == 123 + assert kwargs["original_chat_id"] == "chat-a" + assert response.message_id == 456 + + +def test_telegram_module_direct_buttons_keep_new_message_behavior(): + """direct message 不透传原消息上下文，避免从发新消息变成编辑旧消息。""" + module = TelegramModule() + client = Mock() + buttons = [[{"text": "确认", "callback_data": "confirm"}]] + client.send_msg.return_value = { + "success": True, + "message_id": 456, + "chat_id": "chat-a", + } + + with patch.object( + module, + "get_configs", + return_value={"telegram-test": SimpleNamespace(name="telegram-test")}, + ), patch.object( + module, "check_message", return_value=True + ), patch.object( + module, "get_instance", return_value=client + ): + response = module.send_direct_message( + Notification( + channel=MessageChannel.Telegram, + source="telegram-test", + title="请选择", + text="请选择一个操作", + buttons=buttons, + original_message_id=123, + original_chat_id="chat-a", + ) + ) + + client.send_msg.assert_called_once() + kwargs = client.send_msg.call_args.kwargs + assert "buttons" not in kwargs + assert kwargs["original_message_id"] is None + assert kwargs["original_chat_id"] is None + assert response.message_id == 456 + + +def test_telegram_module_plain_direct_message_keeps_userid_target(): + """普通 direct message 不使用 original_chat_id，避免把私聊消息发回原群聊。""" + module = TelegramModule() + client = Mock() + client.send_msg.return_value = { + "success": True, + "message_id": 456, + "chat_id": "10001", + } + + with patch.object( + module, + "get_configs", + return_value={"telegram-test": SimpleNamespace(name="telegram-test")}, + ), patch.object( + module, "check_message", return_value=True + ), patch.object( + module, "get_instance", return_value=client + ): + response = module.send_direct_message( + Notification( + channel=MessageChannel.Telegram, + source="telegram-test", + userid="10001", + title="普通通知", + text="只发给用户", + original_chat_id="group-1", + ) + ) + + client.send_msg.assert_called_once() + kwargs = client.send_msg.call_args.kwargs + assert kwargs["userid"] == "10001" + assert kwargs["original_message_id"] is None + assert kwargs["original_chat_id"] is None + assert response.message_id == 456 def test_send_msg_with_force_reply_uses_force_reply_when_no_buttons(telegram): @@ -389,6 +613,22 @@ def test_send_msg_with_force_reply_and_original_message_sends_new_prompt(telegra assert send_kwargs["reply_markup"].__class__.__name__ == "ForceReply" +def test_send_msg_new_direct_context_message_prefers_original_chat(telegram): + """不编辑旧消息时，original_chat_id 仍用于把新消息发回原交互会话。""" + result = telegram.send_msg( + title="请输入关键词", + text="回复节目关键词", + userid="10001", + original_chat_id="group-1", + ) + + assert result and result.get("success") + telegram.bot.edit_message_text.assert_not_called() + send_kwargs = telegram.bot.send_message.call_args.kwargs + assert send_kwargs["chat_id"] == "group-1" + assert "reply_to_message_id" not in send_kwargs + + def test_edit_msg_falls_back_to_caption_when_original_message_has_no_text(telegram): """编辑图片消息时应在文本编辑失败后回退为 caption 编辑。""" telegram.bot.edit_message_text.side_effect = Exception(