feat: add plugin-scoped text input sessions (#6068)

2026-09-04 23:17:20 +08:00 · 2026-07-07 05:15:18 +08:00
parent 136c1baed3
commit 827ed8330c
8 changed files with 1380 additions and 6 deletions
@@ -29,7 +29,12 @@ from app.db.models import TransferHistory
 from app.db.transferhistory_oper import TransferHistoryOper
 from app.db.user_oper import UserOper
 from app.helper.directory import DirectoryHelper
-from app.helper.interaction import agent_interaction_manager, media_interaction_manager, PendingMediaInteraction
+from app.helper.interaction import (
+    agent_interaction_manager,
+    media_interaction_manager,
+    plugin_input_interaction_manager,
+    PendingMediaInteraction,
+)
 from app.helper.torrent import TorrentHelper
 from app.log import logger
 from app.schemas import CommingMessage, DownloadDirectory, FileURI, NotExistMediaInfo, Notification
@@ -160,7 +165,7 @@ class MessageChain(ChainBase):
            source: str,
            userid: Union[str, int],
            username: str,
-            text: str,
+            text: Optional[str],
            original_message_id: Optional[Union[str, int]] = None,
            original_chat_id: Optional[str] = None,
            images: Optional[List[CommingMessage.MessageImage]] = None,
@@ -201,6 +206,20 @@ class MessageChain(ChainBase):
                    )
                    return

+            if self._handle_plugin_input_interaction(
+                    channel=channel,
+                    source=source,
+                    userid=userid,
+                    username=username,
+                    text=text,
+                    original_chat_id=original_chat_id,
+                    images=images,
+                    audio_refs=audio_refs,
+                    files=files,
+                    has_audio_input=has_audio_input,
+            ):
+                return
+
            is_agent_message = self._is_agent_message(
                userid=userid,
                text=text,
@@ -259,7 +278,7 @@ class MessageChain(ChainBase):
            source: str,
            userid: Union[str, int],
            username: str,
-            text: str,
+            text: Optional[str],
            original_message_id: Optional[Union[str, int]] = None,
            original_chat_id: Optional[str] = None,
            images: Optional[List[CommingMessage.MessageImage]] = None,
@@ -290,6 +309,20 @@ class MessageChain(ChainBase):
                )
            return False

+        if self._handle_plugin_input_interaction(
+                channel=channel,
+                source=source,
+                userid=userid,
+                username=username,
+                text=text,
+                original_chat_id=original_chat_id,
+                images=images,
+                audio_refs=audio_refs,
+                files=files,
+                has_audio_input=has_audio_input,
+        ):
+            return False
+
        no_ai_requested, no_ai_text = self._strip_no_ai_prefix(text)
        if no_ai_requested:
            text = no_ai_text
@@ -415,6 +448,112 @@ class MessageChain(ChainBase):
        )
        return False

+    def _handle_plugin_input_interaction(
+            self,
+            channel: MessageChannel,
+            source: str,
+            userid: Union[str, int],
+            username: str,
+            text: str,
+            original_chat_id: Optional[Union[str, int]] = None,
+            images: Optional[List[CommingMessage.MessageImage]] = None,
+            audio_refs: Optional[List[str]] = None,
+            files: Optional[List[CommingMessage.MessageAttachment]] = None,
+            has_audio_input: bool = False,
+    ) -> bool:
+        """
+        将插件输入会话中的下一条普通文本派发给指定插件。
+        """
+        if not text or not text.strip() or images or audio_refs or files or has_audio_input:
+            return False
+        if text.startswith("CALLBACK:"):
+            return False
+
+        request, status = plugin_input_interaction_manager.consume_by_user(
+            userid, channel, source, original_chat_id
+        )
+        if not request:
+            return False
+
+        if status == "expired":
+            self.eventmanager.send_event(
+                EventType.MessageAction,
+                {
+                    "plugin_id": request.plugin_id,
+                    "__mp_target_plugin_id": request.plugin_id,
+                    "text": f"plugin_input_expired|{request.request_id}",
+                    "userid": userid,
+                    "channel": channel,
+                    "source": source,
+                    "username": username,
+                    "chat_id": original_chat_id,
+                    "prompt_id": request.prompt_id,
+                    "input_session_id": request.request_id,
+                    "expired": True,
+                    "payload": request.payload,
+                },
+            )
+            self.post_message(
+                Notification(
+                    channel=channel,
+                    source=source,
+                    userid=userid,
+                    username=username,
+                    title="插件输入已超时，请重新发起操作。",
+                    save_history=False,
+                )
+            )
+            return not text.strip().startswith("/")
+
+        if text.strip().lower() in {"取消", "退出", "q", "quit", "exit"}:
+            self.eventmanager.send_event(
+                EventType.MessageAction,
+                {
+                    "plugin_id": request.plugin_id,
+                    "__mp_target_plugin_id": request.plugin_id,
+                    "text": f"plugin_input_cancel|{request.request_id}",
+                    "userid": userid,
+                    "channel": channel,
+                    "source": source,
+                    "username": username,
+                    "chat_id": original_chat_id,
+                    "prompt_id": request.prompt_id,
+                    "input_session_id": request.request_id,
+                    "cancelled": True,
+                    "payload": request.payload,
+                },
+            )
+            self.post_message(
+                Notification(
+                    channel=channel,
+                    source=source,
+                    userid=userid,
+                    username=username,
+                    title="已取消插件输入",
+                    save_history=False,
+                )
+            )
+            return True
+
+        self.eventmanager.send_event(
+            EventType.MessageAction,
+            {
+                "plugin_id": request.plugin_id,
+                "__mp_target_plugin_id": request.plugin_id,
+                "text": f"plugin_input|{request.request_id}",
+                "input_text": text,
+                "userid": userid,
+                "channel": channel,
+                "source": source,
+                "username": username,
+                "chat_id": original_chat_id,
+                "prompt_id": request.prompt_id,
+                "input_session_id": request.request_id,
+                "payload": request.payload,
+            },
+        )
+        return True
+
    @classmethod
    def _strip_no_ai_prefix(cls, text: str) -> Tuple[bool, str]:
        """
@@ -439,11 +439,19 @@ class EventManager(metaclass=Singleton):
        if not handlers:
            logger.debug(f"No handlers found for broadcast event: {event}")
            return
+        target_plugin_id = None
+        if event.event_type == EventType.MessageAction and isinstance(event.event_data, dict):
+            target_plugin_id = event.event_data.get("__mp_target_plugin_id")
        # 为每个处理器提供独立的事件实例，防止某个处理器对 event_data 的修改影响其他处理器
        for handler_id, handler in handlers.items():
+            if target_plugin_id and not self.__should_dispatch_to_target_plugin(
+                    handler, handler_id, str(target_plugin_id)
+            ):
+                continue
            # 仅浅拷贝顶层字典，避免不必要的深拷贝开销；这样可以隔离键级别的替换/赋值
            if isinstance(event.event_data, dict):
                event_data_copy = event.event_data.copy()
+                event_data_copy.pop("__mp_target_plugin_id", None)
            else:
                event_data_copy = event.event_data
            isolated_event = Event(event_type=event.event_type,
@@ -459,6 +467,34 @@ class EventManager(metaclass=Singleton):
                # 对于同步函数，在线程池中运行
                self.__executor.submit(self.__safe_invoke_handler, handler, isolated_event)

+    @classmethod
+    def __should_dispatch_to_target_plugin(
+            cls,
+            handler: Callable,
+            handler_identifier: str,
+            target_plugin_id: str,
+    ) -> bool:
+        """
+        限定插件输入事件只投递给目标插件，避免自由文本被其他插件观察到。
+        """
+        class_name, method_name = cls.__parse_handler_names(handler)
+        if class_name != target_plugin_id:
+            return False
+        identifier_parts = (handler_identifier or "").split(".")
+        if len(identifier_parts) < 2:
+            logger.debug(
+                "Target plugin dispatch skipped because handler identifier is invalid: "
+                f"target={target_plugin_id}, handler={handler_identifier}"
+            )
+            return False
+        if identifier_parts[-2:] != [class_name, method_name]:
+            logger.debug(
+                "Target plugin dispatch skipped because handler identifier does not match handler: "
+                f"target={target_plugin_id}, handler={handler_identifier}, parsed={class_name}.{method_name}"
+            )
+            return False
+        return True
+
    def __safe_invoke_handler(self, handler: Callable, event: Event):
        """
        调用处理器，处理链式或广播事件
@@ -398,6 +398,286 @@ class MediaInteractionManager:
 media_interaction_manager = MediaInteractionManager()


+@dataclass
+class PendingPluginInputInteraction:
+    """
+    记录插件临时接管用户下一条文本输入的会话。
+    """
+
+    request_id: str
+    user_id: str
+    plugin_id: str
+    channel: Optional[MessageChannel]
+    source: Optional[str]
+    username: Optional[str]
+    chat_id: Optional[str] = None
+    prompt_id: Optional[str] = None
+    payload: Optional[Any] = None
+    timeout_seconds: int = 120
+    created_at: datetime = field(default_factory=datetime.now)
+
+    @property
+    def expires_at(self) -> datetime:
+        return self.created_at + timedelta(seconds=max(1, self.timeout_seconds))
+
+
+class PluginInputInteractionManager:
+    """
+    管理插件输入会话。
+
+    会话按用户和渠道绑定；同一用户在同一渠道只保留一个待输入会话。
+    """
+
+    EXPIRED_GRACE_SECONDS = 300
+
+    def __init__(self):
+        self._by_id: Dict[str, PendingPluginInputInteraction] = {}
+        self._by_user_channel: Dict[Tuple[str, Optional[MessageChannel], Optional[str], Optional[str]], str] = {}
+        self._expired_by_user_channel: Dict[
+            Tuple[str, Optional[MessageChannel], Optional[str], Optional[str]],
+            PendingPluginInputInteraction,
+        ] = {}
+        self._lock = Lock()
+
+    @staticmethod
+    def _user_channel_source_key(
+            user_id: Union[str, int],
+            channel: Optional[MessageChannel],
+            source: Optional[str] = None,
+            chat_id: Optional[Union[str, int]] = None,
+    ) -> Tuple[str, Optional[MessageChannel], Optional[str], Optional[str]]:
+        return str(user_id), channel, source, str(chat_id) if chat_id not in (None, "") else None
+
+    @classmethod
+    def _keys_overlap(
+            cls,
+            left: Tuple[str, Optional[MessageChannel], Optional[str], Optional[str]],
+            right: Tuple[str, Optional[MessageChannel], Optional[str], Optional[str]],
+    ) -> bool:
+        left_user, left_channel, left_source, left_chat_id = left
+        right_user, right_channel, right_source, right_chat_id = right
+        if left_user != right_user:
+            return False
+        if left_chat_id and right_chat_id and left_chat_id != right_chat_id:
+            return False
+        if (left_channel is None and left_source is None) or (right_channel is None and right_source is None):
+            return left_channel == right_channel and left_source == right_source
+        channel_overlap = left_channel == right_channel or left_channel is None or right_channel is None
+        source_overlap = left_source == right_source or left_source is None or right_source is None
+        return channel_overlap and source_overlap
+
+    def _cleanup_locked(self) -> None:
+        now = datetime.now()
+        expired_tombstones = [
+            key
+            for key, request in self._expired_by_user_channel.items()
+            if request.expires_at + timedelta(seconds=self.EXPIRED_GRACE_SECONDS) < now
+        ]
+        for key in expired_tombstones:
+            self._expired_by_user_channel.pop(key, None)
+
+        expired = [
+            request_id
+            for request_id, request in self._by_id.items()
+            if request.expires_at < now
+        ]
+        for request_id in expired:
+            request = self._by_id.pop(request_id, None)
+            if request:
+                key = self._user_channel_source_key(
+                    request.user_id,
+                    request.channel,
+                    request.source,
+                    request.chat_id,
+                )
+                self._by_user_channel.pop(key, None)
+                self._expired_by_user_channel[key] = request
+
+    def create_or_replace(
+            self,
+            user_id: Union[str, int],
+            plugin_id: str,
+            channel: Optional[MessageChannel],
+            source: Optional[str],
+            username: Optional[str],
+            chat_id: Optional[Union[str, int]] = None,
+            prompt_id: Optional[str] = None,
+            timeout_seconds: int = 120,
+            payload: Optional[Any] = None,
+    ) -> PendingPluginInputInteraction:
+        with self._lock:
+            self._cleanup_locked()
+            key = self._user_channel_source_key(user_id, channel, source, chat_id)
+            old_request_ids = [
+                request_id
+                for stored_key, request_id in self._by_user_channel.items()
+                if self._keys_overlap(stored_key, key)
+            ]
+            for old_request_id in old_request_ids:
+                self._by_id.pop(old_request_id, None)
+            self._by_user_channel = {
+                stored_key: request_id
+                for stored_key, request_id in self._by_user_channel.items()
+                if request_id not in old_request_ids
+            }
+            self._expired_by_user_channel = {
+                stored_key: request
+                for stored_key, request in self._expired_by_user_channel.items()
+                if not self._keys_overlap(stored_key, key)
+            }
+
+            request = PendingPluginInputInteraction(
+                request_id=uuid.uuid4().hex[:12],
+                user_id=str(user_id),
+                plugin_id=plugin_id,
+                channel=channel,
+                source=source,
+                username=username,
+                chat_id=str(chat_id) if chat_id not in (None, "") else None,
+                prompt_id=prompt_id,
+                timeout_seconds=timeout_seconds,
+                payload=payload,
+            )
+            self._by_id[request.request_id] = request
+            self._by_user_channel[key] = request.request_id
+            return request
+
+    def get_by_user(
+            self,
+            user_id: Union[str, int],
+            channel: Optional[MessageChannel] = None,
+            source: Optional[str] = None,
+            chat_id: Optional[Union[str, int]] = None,
+    ) -> Optional[PendingPluginInputInteraction]:
+        with self._lock:
+            self._cleanup_locked()
+            request_id = self._find_request_id_locked(user_id, channel, source, chat_id)
+            if request_id:
+                return self._by_id.get(request_id)
+            return None
+
+    def pop_by_user(
+            self,
+            user_id: Union[str, int],
+            channel: Optional[MessageChannel] = None,
+            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
+
+    def consume_by_user(
+            self,
+            user_id: Union[str, int],
+            channel: Optional[MessageChannel] = None,
+            source: Optional[str] = None,
+            chat_id: Optional[Union[str, int]] = None,
+    ) -> 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
+            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)
+                return request, "expired"
+            self._cleanup_locked()
+            return None, None
+
+    def _find_request_id_locked(
+            self,
+            user_id: Union[str, int],
+            channel: Optional[MessageChannel],
+            source: Optional[str],
+            chat_id: Optional[Union[str, int]] = None,
+    ) -> Optional[str]:
+        _, request_id = self._find_key_and_request_id_locked(user_id, channel, source, chat_id)
+        return request_id
+
+    def _find_key_and_request_id_locked(
+            self,
+            user_id: Union[str, int],
+            channel: Optional[MessageChannel],
+            source: Optional[str],
+            chat_id: Optional[Union[str, int]] = None,
+    ) -> Tuple[Optional[Tuple[str, Optional[MessageChannel], Optional[str], Optional[str]]], Optional[str]]:
+        for key in self._candidate_keys(user_id, channel, source, chat_id):
+            request_id = self._by_user_channel.get(key)
+            if request_id:
+                return key, request_id
+        return None, None
+
+    def _find_expired_key_and_request_locked(
+            self,
+            user_id: Union[str, int],
+            channel: Optional[MessageChannel],
+            source: Optional[str],
+            chat_id: Optional[Union[str, int]] = None,
+    ) -> Tuple[Optional[Tuple[str, Optional[MessageChannel], Optional[str], Optional[str]]],
+               Optional[PendingPluginInputInteraction]]:
+        now = datetime.now()
+        for key in self._candidate_keys(user_id, channel, source, chat_id):
+            request = self._expired_by_user_channel.get(key)
+            if not request:
+                continue
+            if request.expires_at + timedelta(seconds=self.EXPIRED_GRACE_SECONDS) < now:
+                self._expired_by_user_channel.pop(key, None)
+                continue
+            return key, request
+        return None, None
+
+    def _candidate_keys(
+            self,
+            user_id: Union[str, int],
+            channel: Optional[MessageChannel],
+            source: Optional[str],
+            chat_id: Optional[Union[str, int]] = None,
+    ) -> List[Tuple[str, Optional[MessageChannel], Optional[str], Optional[str]]]:
+        chat_key = str(chat_id) if chat_id not in (None, "") else None
+        candidates = [
+            self._user_channel_source_key(user_id, channel, source, chat_key),
+        ]
+        if source is not None:
+            candidates.append(self._user_channel_source_key(user_id, channel, None, chat_key))
+        if channel is not None and source is not None:
+            candidates.append(self._user_channel_source_key(user_id, None, source, chat_key))
+        if channel is None and source is None:
+            wildcard_key = self._user_channel_source_key(user_id, None, None, chat_key)
+            candidates.append(wildcard_key)
+        if chat_key is not None:
+            candidates.append(self._user_channel_source_key(user_id, channel, source, None))
+            if source is not None:
+                candidates.append(self._user_channel_source_key(user_id, channel, None, None))
+            if channel is not None and source is not None:
+                candidates.append(self._user_channel_source_key(user_id, None, source, None))
+            if channel is None and source is None:
+                candidates.append(self._user_channel_source_key(user_id, None, None, None))
+        return candidates
+
+    def remove(self, request_id: str) -> None:
+        with self._lock:
+            request = self._by_id.pop(request_id, None)
+            if request:
+                self._by_user_channel.pop(
+                    self._user_channel_source_key(request.user_id, request.channel, request.source, request.chat_id),
+                    None,
+                )
+
+    def clear(self) -> None:
+        with self._lock:
+            self._by_id.clear()
+            self._by_user_channel.clear()
+            self._expired_by_user_channel.clear()
+
+
+plugin_input_interaction_manager = PluginInputInteractionManager()
+
+
@dataclass(frozen=True)
 class AgentInteractionOption:
    """
@@ -521,6 +521,7 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]):
                        userid=userid,
                        link=message.link,
                        buttons=message.buttons,
+                        force_reply=message.force_reply,
                        original_message_id=message.original_message_id,
                        original_chat_id=message.original_chat_id,
                        disable_web_page_preview=message.disable_web_page_preview,
@@ -15,6 +15,10 @@ from telebot.types import (
    InlineKeyboardButton,
    InputMediaPhoto,
 )
+try:
+    from telebot.types import ForceReply
+except ImportError:
+    ForceReply = None
 from telegramify_markdown import standardize, telegramify  # noqa
 try:
    from telegramify_markdown import entities_to_markdownv2  # noqa
@@ -584,6 +588,7 @@ class Telegram:
            userid: Optional[str] = None,
            link: Optional[str] = None,
            buttons: Optional[List[List[dict]]] = None,
+            force_reply: bool = False,
            original_message_id: Optional[int] = None,
            original_chat_id: Optional[str] = None,
            disable_web_page_preview: Optional[bool] = None,
@@ -598,6 +603,7 @@ class Telegram:
        :param userid: 用户ID，如有则只发消息给该用户
        :param link: 跳转链接
        :param buttons: 按钮列表，格式：[[{"text": "按钮文本", "callback_data": "回调数据"}]]
+        :param force_reply: 是否请求 Telegram 客户端强制回复
        :param original_message_id: 原消息ID，如果提供则编辑原消息
        :param original_chat_id: 原消息的聊天ID，编辑消息时需要
        :param disable_web_page_preview: 是否禁用链接预览
@@ -634,9 +640,31 @@ class Telegram:
            reply_markup = None
            if buttons:
                reply_markup = self._create_inline_keyboard(buttons)
+            elif force_reply and ForceReply:
+                reply_markup = self._create_force_reply_markup()

            # 判断是编辑消息还是发送新消息
            if original_message_id and original_chat_id:
+                if force_reply and reply_markup and not buttons:
+                    sent = self.__send_request(
+                        userid=original_chat_id,
+                        image=image,
+                        caption=caption,
+                        reply_markup=reply_markup,
+                        disable_web_page_preview=disable_web_page_preview,
+                        parse_mode=parse_mode,
+                        reply_to_message_id=original_message_id,
+                    )
+                    self._stop_typing_if_needed(chat_id, stop_typing)
+                    if sent and hasattr(sent, "message_id"):
+                        return {
+                            "success": True,
+                            "message_id": sent.message_id,
+                            "chat_id": sent.chat.id if hasattr(sent, "chat") else chat_id,
+                        }
+                    elif sent:
+                        return {"success": True}
+                    return {"success": False}
                # 编辑消息
                result = self.__edit_message(
                    original_chat_id,
@@ -679,6 +707,18 @@ class Telegram:
            self._stop_typing_if_needed(chat_id, stop_typing)
            return {"success": False}

+    @staticmethod
+    def _create_force_reply_markup():
+        if not ForceReply:
+            return None
+        try:
+            return ForceReply(selective=True, input_field_placeholder="请输入内容")
+        except TypeError:
+            try:
+                return ForceReply(selective=True)
+            except TypeError:
+                return ForceReply()
+
    def send_voice(
            self,
            voice_path: str,
@@ -1285,12 +1325,14 @@ class Telegram:
            reply_markup: Optional[InlineKeyboardMarkup] = None,
            disable_web_page_preview: Optional[bool] = None,
            parse_mode: Optional[str] = None,
+            reply_to_message_id: Optional[int] = None,
    ):
        """
        向Telegram发送报文，返回发送的消息对象
        :param reply_markup: 内联键盘
        :param disable_web_page_preview: 是否禁用链接预览
        :param parse_mode: Telegram 消息格式类型，默认 MarkdownV2，可传 HTML
+        :param reply_to_message_id: 回复的原消息ID
        :return: 发送成功返回消息对象，失败返回None
        """
        parse_mode = self._normalize_parse_mode(parse_mode)
@@ -1299,6 +1341,8 @@ class Telegram:
            "parse_mode": parse_mode,
            "reply_markup": reply_markup,
        }
+        if reply_to_message_id:
+            kwargs["reply_to_message_id"] = reply_to_message_id
        # 处理图片
        image = self.__process_image(image)

@@ -243,6 +243,8 @@ class Notification(BaseModel):
    targets: Optional[dict] = None
    # 按钮列表，格式：[[{"text": "按钮文本", "callback_data": "回调数据", "url": "链接"}]]
    buttons: Optional[List[List[dict]]] = None
+    # Telegram ForceReply 回复标记
+    force_reply: bool = False
    # 原消息ID，用于编辑消息
    original_message_id: Optional[Union[str, int]] = None
    # 原消息的聊天ID，用于编辑消息
@@ -1,13 +1,15 @@
+from datetime import datetime, timedelta
 from unittest.mock import patch

 import pytest

 from app.chain.message import MediaInteractionChain, MessageChain
+from app.core.event import EventManager
 from app.core.context import Context, MediaInfo, TorrentInfo
 from app.core.meta import MetaBase
-from app.helper.interaction import media_interaction_manager
-from app.schemas import TransferDirectoryConf
-from app.schemas.types import MediaType, MessageChannel
+from app.helper.interaction import media_interaction_manager, plugin_input_interaction_manager
+from app.schemas import CommingMessage, TransferDirectoryConf
+from app.schemas.types import EventType, MediaType, MessageChannel


@pytest.fixture(autouse=True)
@@ -15,6 +17,7 @@ def clear_media_interactions():
    """清理媒体交互状态，避免用例之间共享内存会话。"""
    yield
    media_interaction_manager.clear()
+    plugin_input_interaction_manager.clear()


 def _build_meta(name: str) -> MetaBase:
@@ -157,6 +160,795 @@ def test_message_routes_text_reply_to_media_interaction_before_ai():
    handle_ai.assert_not_called()


+def test_plugin_input_session_captures_plain_text_before_media_interaction():
+    """插件输入会话存在时，普通文本应派发给插件而不是媒体交互。"""
+    chain = MessageChain()
+    request = plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=MessageChannel.Wechat,
+        source="wechat-test",
+        username="tester",
+        prompt_id="prompt-1",
+        payload={"step": "name"},
+    )
+    media_interaction_manager.create_or_replace(
+        user_id="10001",
+        channel=MessageChannel.Wechat,
+        source="wechat-test",
+        username="tester",
+        action="Search",
+        keyword="星际穿越",
+        title="星际穿越",
+        meta=_build_meta("星际穿越"),
+        items=[MediaInfo(title="星际穿越", year="2014")],
+    )
+
+    with patch.object(chain, "_record_user_message"), patch(
+        "app.chain.message.MediaInteractionChain.handle_text_interaction",
+        return_value=True,
+    ) as handle_media, patch.object(chain.eventmanager, "send_event") as send_event:
+        chain.handle_message(
+            channel=MessageChannel.Wechat,
+            source="wechat-test",
+            userid="10001",
+            username="tester",
+            text="用户输入内容",
+        )
+
+    handle_media.assert_not_called()
+    send_event.assert_called_once_with(
+        EventType.MessageAction,
+        {
+            "plugin_id": "demo_plugin",
+            "__mp_target_plugin_id": "demo_plugin",
+            "text": f"plugin_input|{request.request_id}",
+            "input_text": "用户输入内容",
+            "userid": "10001",
+            "channel": MessageChannel.Wechat,
+            "source": "wechat-test",
+            "username": "tester",
+            "chat_id": None,
+            "prompt_id": "prompt-1",
+            "input_session_id": request.request_id,
+            "payload": {"step": "name"},
+        },
+    )
+    assert plugin_input_interaction_manager.get_by_user("10001", MessageChannel.Wechat) is None
+
+
+def test_plugin_input_session_does_not_record_sensitive_text_history():
+    """插件输入命中时不应先写入普通用户消息历史。"""
+    chain = MessageChain()
+    plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=MessageChannel.Telegram,
+        source="telegram-test",
+        username="tester",
+    )
+
+    with patch.object(chain, "_record_user_message") as record_message, patch.object(
+        chain.eventmanager, "send_event"
+    ):
+        chain.handle_message(
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            userid="10001",
+            username="tester",
+            text="secret-value",
+        )
+
+    record_message.assert_not_called()
+
+
+def test_plugin_input_session_captures_slash_like_text_before_commands():
+    """插件输入会话中的 /path 文本不应被当成 slash 命令。"""
+    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",
+        prompt_id="path",
+    )
+
+    with patch.object(chain, "_record_user_message"), patch.object(
+        chain.eventmanager, "send_event"
+    ) as send_event:
+        chain.handle_message(
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            userid="10001",
+            username="tester",
+            text="/downloads/tv",
+        )
+
+    send_event.assert_called_once()
+    event_type, payload = send_event.call_args.args
+    assert event_type == EventType.MessageAction
+    assert payload["text"] == f"plugin_input|{request.request_id}"
+    assert payload["input_text"] == "/downloads/tv"
+    assert payload["__mp_target_plugin_id"] == "demo_plugin"
+
+
+def test_plugin_input_session_cancel_notifies_plugin_and_clears():
+    """取消词应清理插件输入会话并通知目标插件。"""
+    chain = MessageChain()
+    plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=MessageChannel.Telegram,
+        source="telegram-test",
+        username="tester",
+    )
+
+    with patch.object(chain, "_record_user_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="取消",
+        )
+
+    send_event.assert_called_once()
+    event_type, payload = send_event.call_args.args
+    assert event_type == EventType.MessageAction
+    assert payload["cancelled"] is True
+    assert payload["__mp_target_plugin_id"] == "demo_plugin"
+    post_message.assert_called_once()
+    assert plugin_input_interaction_manager.get_by_user("10001", MessageChannel.Telegram) is None
+
+
+def test_plugin_input_cancel_does_not_block_next_command():
+    """取消输入后下一条 slash 命令应按正常命令路由处理。"""
+    chain = MessageChain()
+    plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=MessageChannel.Telegram,
+        source="telegram-test",
+        username="tester",
+    )
+
+    with patch.object(chain, "_record_user_message"), patch.object(
+        chain.eventmanager, "send_event"
+    ) as send_event, patch.object(chain, "post_message"):
+        chain.handle_message(
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            userid="10001",
+            username="tester",
+            text="取消",
+        )
+        chain.handle_message(
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            userid="10001",
+            username="tester",
+            text="/tvh",
+        )
+
+    assert send_event.call_args_list[0].args[0] == EventType.MessageAction
+    assert send_event.call_args_list[1].args[0] == EventType.CommandExcute
+    assert send_event.call_args_list[1].args[1]["cmd"] == "/tvh"
+
+
+def test_plugin_input_session_ignores_non_text_messages():
+    """图片/文件等非纯文本消息不应消费待输入会话。"""
+    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",
+    )
+    image = CommingMessage.MessageImage(ref="https://example.invalid/image.jpg")
+
+    with patch.object(chain, "_record_user_message"), patch.object(
+        chain.eventmanager, "send_event"
+    ) as send_event:
+        chain.handle_message(
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            userid="10001",
+            username="tester",
+            text="图片说明",
+            images=[image],
+        )
+
+    assert plugin_input_interaction_manager.get_by_user(
+        "10001", MessageChannel.Telegram, "telegram-test"
+    ) == request
+    assert not any(
+        call.args and call.args[0] == EventType.MessageAction
+        for call in send_event.call_args_list
+    )
+
+
+def test_plugin_input_session_ignores_none_text_messages():
+    """文本为空时不应因 CALLBACK 检查崩溃或消费待输入会话。"""
+    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",
+    )
+    image = CommingMessage.MessageImage(ref="https://example.invalid/image.jpg")
+
+    handled = chain._handle_plugin_input_interaction(
+        channel=MessageChannel.Telegram,
+        source="telegram-test",
+        userid="10001",
+        username="tester",
+        text=None,
+        images=[image],
+    )
+
+    assert handled is False
+    assert plugin_input_interaction_manager.get_by_user(
+        "10001", MessageChannel.Telegram, "telegram-test"
+    ) == request
+
+
+def test_plugin_input_session_is_bound_to_user_and_channel():
+    """同一用户不同渠道的插件输入会话互不匹配。"""
+    request = plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=MessageChannel.Telegram,
+        source="telegram-test",
+        username="tester",
+    )
+
+    assert plugin_input_interaction_manager.get_by_user("10001", MessageChannel.Wechat) is None
+    assert plugin_input_interaction_manager.get_by_user(
+        "10001", MessageChannel.Telegram, "telegram-test"
+    ) == request
+    assert plugin_input_interaction_manager.get_by_user("10002", MessageChannel.Telegram) is None
+
+
+def test_plugin_input_session_does_not_capture_other_channel_text():
+    """生产消费路径也必须保持用户+渠道绑定。"""
+    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",
+    )
+
+    with patch.object(chain, "_record_user_message"), patch.object(
+        chain.eventmanager, "send_event"
+    ) as send_event:
+        chain.handle_message(
+            channel=MessageChannel.Wechat,
+            source="wechat-test",
+            userid="10001",
+            username="tester",
+            text="普通搜索",
+        )
+
+    assert plugin_input_interaction_manager.get_by_user(
+        "10001", MessageChannel.Telegram, "telegram-test"
+    ) == request
+    assert not any(
+        call.args and call.args[0] == EventType.MessageAction
+        for call in send_event.call_args_list
+    )
+
+
+def test_plugin_input_session_does_not_capture_other_source_text():
+    """同一渠道不同来源的插件输入会话互不匹配。"""
+    chain = MessageChain()
+    request = plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=MessageChannel.Telegram,
+        source="telegram-bot-a",
+        username="tester",
+    )
+
+    with patch.object(chain, "_record_user_message"), patch.object(
+        chain.eventmanager, "send_event"
+    ) as send_event:
+        chain.handle_message(
+            channel=MessageChannel.Telegram,
+            source="telegram-bot-b",
+            userid="10001",
+            username="tester",
+            text="普通搜索",
+        )
+
+    assert plugin_input_interaction_manager.get_by_user(
+        "10001", MessageChannel.Telegram, "telegram-bot-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_session_does_not_capture_other_chat_text():
+    """同一用户同一 bot 的不同 chat 不应串扰插件输入会话。"""
+    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",
+    )
+
+    with patch.object(chain, "_record_user_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",
+        )
+
+    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"), 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",
+        )
+
+    send_event.assert_called_once()
+    event_type, payload = send_event.call_args.args
+    assert event_type == EventType.MessageAction
+    assert payload["input_text"] == "真正输入"
+    assert payload["chat_id"] == "chat-a"
+
+
+def test_plugin_input_chatless_session_keeps_legacy_chat_fallback():
+    """旧插件未绑定 chat_id 时，同 source 消息仍可兼容消费。"""
+    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",
+    )
+
+    with patch.object(chain, "_record_user_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",
+        )
+
+    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_wildcard_session_does_not_match_missing_source_with_chat():
+    """chat fallback 不应把完全 wildcard 会话扩大到有渠道但缺来源的消息。"""
+    request = plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=None,
+        source=None,
+        username="tester",
+    )
+
+    assert plugin_input_interaction_manager.consume_by_user(
+        "10001", MessageChannel.Telegram, None, "chat-a"
+    ) == (None, None)
+    assert plugin_input_interaction_manager.get_by_user("10001", None, None) == request
+
+
+def test_plugin_input_chat_bound_session_does_not_match_missing_chat():
+    """绑定 chat_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",
+    )
+
+    assert plugin_input_interaction_manager.consume_by_user(
+        "10001", MessageChannel.Telegram, "telegram-test"
+    ) == (None, None)
+    assert plugin_input_interaction_manager.get_by_user(
+        "10001", MessageChannel.Telegram, "telegram-test", "chat-a"
+    ) == request
+
+
+def test_plugin_input_core_path_preserves_original_chat_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",
+    )
+
+    with patch.object(chain.eventmanager, "send_event") as send_event:
+        chain._handle_message_core(
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            userid="10001",
+            username="tester",
+            text="核心路径输入",
+            original_chat_id="chat-a",
+        )
+
+    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["chat_id"] == "chat-a"
+
+
+def test_plugin_input_session_does_not_capture_missing_source_text():
+    """来源缺失的入站消息不应捕获绑定到具体来源的输入会话。"""
+    chain = MessageChain()
+    request = plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=MessageChannel.Telegram,
+        source="telegram-bot-a",
+        username="tester",
+    )
+
+    with patch.object(chain, "_record_user_message"), patch.object(
+        chain.eventmanager, "send_event"
+    ) as send_event:
+        chain.handle_message(
+            channel=MessageChannel.Telegram,
+            source=None,
+            userid="10001",
+            username="tester",
+            text="普通搜索",
+        )
+
+    assert plugin_input_interaction_manager.get_by_user(
+        "10001", MessageChannel.Telegram, "telegram-bot-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_session_does_not_capture_callback_payload():
+    """待输入会话存在时，按钮回调仍应优先进入回调链而不是投递给插件。"""
+    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",
+    )
+
+    with patch.object(chain, "_record_user_message") as record_message, patch.object(
+        chain.eventmanager, "send_event"
+    ) as send_event, patch.object(chain, "_handle_callback", return_value=True) as handle_callback:
+        chain.handle_message(
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            userid="10001",
+            username="tester",
+            text="CALLBACK:media:req:page-next",
+            original_message_id="msg-1",
+            original_chat_id="chat-a",
+        )
+
+    record_message.assert_not_called()
+    handle_callback.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_session_expires_after_timeout():
+    """插件输入会话超过 TTL 后不再匹配。"""
+    request = plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=MessageChannel.Wechat,
+        source="wechat-test",
+        username="tester",
+        timeout_seconds=120,
+    )
+    request.created_at = datetime.now() - timedelta(seconds=121)
+
+    assert plugin_input_interaction_manager.get_by_user("10001", MessageChannel.Wechat) is None
+
+
+def test_plugin_input_session_expired_text_notifies_plugin_and_continues_routing():
+    """过期插件输入遇到命令时应提示超时，并让命令继续正常路由。"""
+    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",
+        timeout_seconds=120,
+    )
+    request.created_at = datetime.now() - timedelta(seconds=121)
+
+    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="/tvh",
+        )
+
+    record_message.assert_called_once()
+    post_message.assert_called_once()
+    assert send_event.call_count == 2
+    event_type, payload = send_event.call_args_list[0].args
+    assert event_type == EventType.MessageAction
+    assert payload["expired"] is True
+    assert payload["input_session_id"] == request.request_id
+    command_type, command_payload = send_event.call_args_list[1].args
+    assert command_type == EventType.CommandExcute
+    assert command_payload["cmd"] == "/tvh"
+
+
+def test_plugin_input_session_expired_sensitive_text_is_not_recorded_or_routed():
+    """过期插件输入中的普通文本仍可能是敏感信息，应提示后吞掉。"""
+    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",
+        timeout_seconds=120,
+    )
+    request.created_at = datetime.now() - timedelta(seconds=121)
+
+    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="late-secret",
+        )
+
+    record_message.assert_not_called()
+    post_message.assert_called_once()
+    send_event.assert_called_once()
+    event_type, payload = send_event.call_args.args
+    assert event_type == EventType.MessageAction
+    assert payload["expired"] is True
+    assert payload["input_session_id"] == request.request_id
+
+
+def test_plugin_input_expired_text_after_cleanup_is_not_recorded_or_routed():
+    """其他用户触发过期清理后，迟到的普通文本仍应按过期插件输入吞掉。"""
+    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",
+        timeout_seconds=120,
+    )
+    request.created_at = datetime.now() - timedelta(seconds=121)
+
+    plugin_input_interaction_manager.create_or_replace(
+        user_id="10002",
+        plugin_id="demo_plugin",
+        channel=MessageChannel.Telegram,
+        source="telegram-test",
+        username="other",
+        timeout_seconds=120,
+    )
+    assert request.request_id not in plugin_input_interaction_manager._by_id
+
+    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="late-secret",
+        )
+
+    record_message.assert_not_called()
+    post_message.assert_called_once()
+    send_event.assert_called_once()
+    event_type, payload = send_event.call_args.args
+    assert event_type == EventType.MessageAction
+    assert payload["expired"] is True
+    assert payload["input_session_id"] == request.request_id
+
+
+def test_plugin_input_session_with_no_channel_matches_specific_channel():
+    """插件拿不到渠道时创建的会话应能被该用户下一条具体渠道消息消费。"""
+    chain = MessageChain()
+    request = plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=None,
+        source="telegram-test",
+        username="tester",
+    )
+
+    with patch.object(chain, "_record_user_message"), patch.object(
+        chain.eventmanager, "send_event"
+    ) as send_event:
+        chain.handle_message(
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            userid="10001",
+            username="tester",
+            text="keyword",
+        )
+
+    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"] == "keyword"
+
+
+def test_plugin_input_session_with_no_channel_and_no_source_does_not_match_specific_message():
+    """完全缺少渠道来源的会话不应跨渠道捕获具体消息。"""
+    chain = MessageChain()
+    request = plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=None,
+        source=None,
+        username="tester",
+    )
+
+    with patch.object(chain, "_record_user_message"), patch.object(
+        chain.eventmanager, "send_event"
+    ) as send_event:
+        chain.handle_message(
+            channel=MessageChannel.Telegram,
+            source="telegram-test",
+            userid="10001",
+            username="tester",
+            text="keyword",
+        )
+
+    assert plugin_input_interaction_manager.get_by_user("10001", None, None) == request
+    assert not any(
+        call.args and call.args[0] == EventType.MessageAction
+        for call in send_event.call_args_list
+    )
+
+
+def test_plugin_input_specific_session_replaces_overlapping_no_channel_session():
+    """同用户创建具体渠道会话时，应替换重叠的无渠道会话，避免下一条消息被连环接管。"""
+    old_request = plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="old_plugin",
+        channel=None,
+        source="telegram-test",
+        username="tester",
+    )
+    new_request = plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=MessageChannel.Telegram,
+        source="telegram-test",
+        username="tester",
+    )
+
+    assert plugin_input_interaction_manager.pop_by_user(
+        "10001", MessageChannel.Telegram, "telegram-test"
+    ) == new_request
+    assert plugin_input_interaction_manager.pop_by_user(
+        "10001", MessageChannel.Telegram, "telegram-test"
+    ) is None
+
+
+def test_plugin_input_session_pop_by_user_consumes_once():
+    """原子消费应保证同一个会话只返回一次。"""
+    request = plugin_input_interaction_manager.create_or_replace(
+        user_id="10001",
+        plugin_id="demo_plugin",
+        channel=MessageChannel.Telegram,
+        source="telegram-test",
+        username="tester",
+    )
+
+    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():
+    """带目标插件的输入事件不应投递给其他插件或模块级处理器。"""
+
+    def demo_plugin_handler(_event):
+        return None
+
+    def other_plugin_handler(_event):
+        return None
+
+    demo_plugin_handler.__qualname__ = "demo_plugin.handle"
+    other_plugin_handler.__qualname__ = "other_plugin.handle"
+
+    def module_handler(_event):
+        return None
+
+    should_dispatch = EventManager._EventManager__should_dispatch_to_target_plugin
+    handler_id = EventManager._EventManager__get_handler_identifier(demo_plugin_handler)
+
+    assert should_dispatch(
+        demo_plugin_handler, "tests.plugins.demo_plugin.handle", "demo_plugin"
+    ) is True
+    assert should_dispatch(demo_plugin_handler, handler_id, "demo_plugin") is True
+    assert should_dispatch(
+        demo_plugin_handler, "tests.plugins.other_plugin.handle", "demo_plugin"
+    ) is False
+    assert should_dispatch(
+        other_plugin_handler, "tests.plugins.other_plugin.handle", "demo_plugin"
+    ) is False
+    assert should_dispatch(module_handler, "tests.module_handler", "demo_plugin") is False
+
+
 def test_noai_prefix_starts_traditional_search_when_global_ai_enabled():
    """全局 AI 开启时，/noai 前缀应让本条消息进入传统搜索交互。"""
    chain = MessageChain()
@@ -309,6 +309,86 @@ def test_telegram_module_passes_parse_mode_to_client():
    assert client.send_msg.call_args.kwargs["parse_mode"] == "HTML"


+def test_telegram_module_passes_force_reply_to_client():
+    """模块发送通知时应透传消息指定的force_reply"""
+    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="请输入目录",
+                text="回复目录路径",
+                force_reply=True,
+            )
+        )
+
+    client.send_msg.assert_called_once()
+    assert client.send_msg.call_args.kwargs["force_reply"] is True
+
+
+def test_send_msg_with_force_reply_uses_force_reply_when_no_buttons(telegram):
+    """无按钮时force_reply应生成Telegram ForceReply标记"""
+    result = telegram.send_msg(
+        title="请输入目录",
+        text="回复目录路径",
+        force_reply=True,
+    )
+
+    assert result and result.get("success")
+    send_kwargs = telegram.bot.send_message.call_args.kwargs
+    reply_markup = send_kwargs["reply_markup"]
+    assert reply_markup.__class__.__name__ == "ForceReply"
+    if hasattr(reply_markup, "to_dict"):
+        assert reply_markup.to_dict()["force_reply"] is True
+        assert reply_markup.to_dict().get("selective") is True
+    else:
+        assert getattr(reply_markup, "selective", None) is True
+
+
+def test_send_msg_with_force_reply_keeps_inline_keyboard_when_buttons_exist(telegram):
+    """按钮存在时force_reply不能覆盖InlineKeyboardMarkup"""
+    result = telegram.send_msg(
+        title="请选择目录",
+        text="点击按钮选择",
+        buttons=[[{"text": "默认", "callback_data": "default"}]],
+        force_reply=True,
+    )
+
+    assert result and result.get("success")
+    send_kwargs = telegram.bot.send_message.call_args.kwargs
+    reply_markup = send_kwargs["reply_markup"]
+    assert reply_markup.__class__.__name__ == "InlineKeyboardMarkup"
+
+
+def test_send_msg_with_force_reply_and_original_message_sends_new_prompt(telegram):
+    """编辑消息场景不能带ForceReply，应改为发送新的回复提示。"""
+    result = telegram.send_msg(
+        title="请输入关键词",
+        text="回复节目关键词",
+        force_reply=True,
+        original_message_id=123,
+        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 send_kwargs["reply_to_message_id"] == 123
+    assert send_kwargs["reply_markup"].__class__.__name__ == "ForceReply"
+
+
 def test_edit_msg_falls_back_to_caption_when_original_message_has_no_text(telegram):
    """编辑图片消息时应在文本编辑失败后回退为 caption 编辑。"""
    telegram.bot.edit_message_text.side_effect = Exception(