feat: support prompt-bound plugin input replies (#6087)

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,

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)
+        with self._lock:
-        return request
+            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.get(request_id)
-                request = self._by_id.pop(request_id, None)
+                if not request:
-                if request:
+                    self._by_user_channel.pop(key, None)
-                    status = "expired" if request.expires_at < datetime.now() else "active"
+                elif request.expires_at < datetime.now():
-                    return request, status
+                    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],

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,
                     )

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）

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():
     """带目标插件的输入事件不应投递给其他插件或模块级处理器。"""
 

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(