diff --git a/plugins/platforms/feishu/adapter.py b/plugins/platforms/feishu/adapter.py index 96528646939dc..0b5d25f3c2c3e 100644 --- a/plugins/platforms/feishu/adapter.py +++ b/plugins/platforms/feishu/adapter.py @@ -255,6 +255,10 @@ async def _read_limited_feishu_webhook_body(request: Any, max_bytes: int) -> byt # delete-failures, not a capacity plan. _FEISHU_PROCESSING_REACTION_CACHE_SIZE = 1024 _FEISHU_MESSAGE_TEXT_CACHE_SIZE = 512 # LRU cap for reply-context message text lookups +_FEISHU_MESSAGE_ITEM_CACHE_SIZE = 128 +_FEISHU_REPLY_CHAIN_MAX_DEPTH = 6 +_FEISHU_REPLY_CONTEXT_MAX_CHARS = 500 +_FEISHU_EARLIER_REPLY_SEPARATOR = "\n[Earlier quoted message]\n" # QR onboarding constants _ONBOARD_ACCOUNTS_URLS = { @@ -1481,6 +1485,7 @@ def __init__(self, config: PlatformConfig): self._sent_message_id_order: List[str] = [] # LRU order for _sent_message_ids_to_chat self._chat_info_cache: Dict[str, Dict[str, Any]] = {} self._message_text_cache: "OrderedDict[str, Optional[str]]" = OrderedDict() + self._message_item_cache: "OrderedDict[str, Any]" = OrderedDict() self._app_lock_identity: Optional[str] = None self._text_batch_state = FeishuBatchState() self._pending_text_batches = self._text_batch_state.events @@ -3253,14 +3258,26 @@ async def _process_inbound_message( if hint: text = f"{hint}\n\n{text}" if text else hint + chat_id = getattr(message, "chat_id", "") or "" thread_id = getattr(message, "thread_id", None) or getattr(message, "root_id", None) or None + # Keep the existing session route above unchanged. root_id is also + # quote metadata, so only an explicit topic ID is safe as a lane guard. + explicit_thread_id = getattr(message, "thread_id", None) or None reply_to_message_id = ( getattr(message, "parent_id", None) or getattr(message, "upper_message_id", None) or getattr(message, "root_id", None) or None ) - reply_to_text = await self._fetch_message_text(reply_to_message_id) if reply_to_message_id else None + reply_to_text = ( + await self._fetch_message_context_chain( + reply_to_message_id, + chat_id=chat_id, + thread_id=explicit_thread_id, + ) + if reply_to_message_id + else None + ) sender_primary = ( getattr(sender_id, "open_id", None) @@ -3280,7 +3297,6 @@ async def _process_inbound_message( len(media_urls), ) - chat_id = getattr(message, "chat_id", "") or "" chat_info = await self.get_chat_info(chat_id) sender_profile = await self._resolve_sender_profile(sender_id, is_bot=is_bot) source = self.build_source( @@ -4177,15 +4193,9 @@ async def _fetch_message_text(self, message_id: str) -> Optional[str]: self._message_text_cache.move_to_end(message_id) return self._message_text_cache[message_id] try: - request = self._build_get_message_request(message_id) - response = await self._run_blocking(self._client.im.v1.message.get, request) - if not response or getattr(response, "success", lambda: False)() is False: - code = getattr(response, "code", "unknown") - msg = getattr(response, "msg", "message lookup failed") - logger.warning("[Feishu] Failed to fetch parent message %s: [%s] %s", message_id, code, msg) + parent = await self._fetch_message_item(message_id) + if parent is None: return None - items = getattr(getattr(response, "data", None), "items", None) or [] - parent = items[0] if items else None body = getattr(parent, "body", None) msg_type = getattr(parent, "msg_type", "") or "" raw_content = getattr(body, "content", "") or "" @@ -4203,6 +4213,117 @@ async def _fetch_message_text(self, message_id: str) -> Optional[str]: logger.warning("[Feishu] Failed to fetch parent message %s", message_id, exc_info=True) return None + async def _fetch_message_item(self, message_id: str) -> Optional[Any]: + """Fetch one message through a shared LRU for text/media reply context.""" + if not self._client or not message_id: + return None + if message_id in self._message_item_cache: + self._message_item_cache.move_to_end(message_id) + return self._message_item_cache[message_id] + try: + request = self._build_get_message_request(message_id) + response = await self._run_blocking(self._client.im.v1.message.get, request) + if not response or getattr(response, "success", lambda: False)() is False: + code = getattr(response, "code", "unknown") + msg = getattr(response, "msg", "message lookup failed") + logger.warning("[Feishu] Failed to fetch parent message %s: [%s] %s", message_id, code, msg) + return None + items = getattr(getattr(response, "data", None), "items", None) or [] + item = items[0] if items else None + if item is not None: + self._message_item_cache[message_id] = item + while len(self._message_item_cache) > _FEISHU_MESSAGE_ITEM_CACHE_SIZE: + self._message_item_cache.popitem(last=False) + return item + except Exception: + logger.warning("[Feishu] Failed to fetch parent message %s", message_id, exc_info=True) + return None + + async def _fetch_message_context_chain( + self, + message_id: str, + *, + chat_id: str, + thread_id: Optional[str], + max_depth: int = _FEISHU_REPLY_CHAIN_MAX_DEPTH, + max_chars: int = _FEISHU_REPLY_CONTEXT_MAX_CHARS, + ) -> Optional[str]: + """Fetch bounded quoted-message ancestry with the direct parent first.""" + if not message_id or max_depth <= 0 or max_chars <= 0: + return None + if not self._client: + return await self._fetch_message_text(message_id) + + current_id: Optional[str] = message_id + seen: set[str] = set() + rendered = "" + + for _ in range(max_depth): + if not current_id or current_id in seen: + break + seen.add(current_id) + + item = await self._fetch_message_item(current_id) + if item is None: + cached_text = self._message_text_cache.get(current_id) + if cached_text and not rendered: + rendered = cached_text[:max_chars] + break + + item_chat_id = str(getattr(item, "chat_id", "") or "") + if chat_id and item_chat_id and item_chat_id != chat_id: + logger.warning( + "[Feishu] Stopped reply ancestry at %s: chat mismatch", + current_id, + ) + break + + item_thread_id = str(getattr(item, "thread_id", "") or "") + if thread_id and item_thread_id and item_thread_id != thread_id: + logger.warning( + "[Feishu] Stopped reply ancestry at %s: thread mismatch", + current_id, + ) + break + + body = getattr(item, "body", None) + text = self._extract_text_from_raw_content( + msg_type=getattr(item, "msg_type", "") or "", + raw_content=getattr(body, "content", "") or "", + mentions=getattr(item, "mentions", None), + ) + if text: + self._message_text_cache[current_id] = text + self._message_text_cache.move_to_end(current_id) + while len(self._message_text_cache) > _FEISHU_MESSAGE_TEXT_CACHE_SIZE: + self._message_text_cache.popitem(last=False) + + if not rendered: + rendered = text[:max_chars] + else: + remaining = max_chars - len(rendered) + if remaining <= len(_FEISHU_EARLIER_REPLY_SEPARATOR): + break + rendered += _FEISHU_EARLIER_REPLY_SEPARATOR + remaining = max_chars - len(rendered) + rendered += text[:remaining] + if len(rendered) >= max_chars: + break + + if rendered and ( + max_chars - len(rendered) <= len(_FEISHU_EARLIER_REPLY_SEPARATOR) + ): + break + + current_id = ( + getattr(item, "parent_id", None) + or getattr(item, "upper_message_id", None) + or getattr(item, "root_id", None) + or None + ) + + return rendered or None + def _extract_text_from_raw_content( self, *, diff --git a/tests/gateway/test_feishu.py b/tests/gateway/test_feishu.py index 0d63659bc13e8..320e06fee6504 100644 --- a/tests/gateway/test_feishu.py +++ b/tests/gateway/test_feishu.py @@ -2078,7 +2078,7 @@ def test_process_inbound_message_fetches_reply_to_text(self): adapter._resolve_sender_profile = AsyncMock( return_value={"user_id": "ou_user", "user_name": "张三", "user_id_alt": None} ) - adapter._fetch_message_text = AsyncMock(return_value="父消息内容") + adapter._fetch_message_context_chain = AsyncMock(return_value="父消息内容") message = SimpleNamespace( chat_id="oc_chat", thread_id=None, @@ -2103,6 +2103,11 @@ def test_process_inbound_message_fetches_reply_to_text(self): event = adapter._dispatch_inbound_event.await_args.args[0] self.assertEqual(event.reply_to_message_id, "om_parent") self.assertEqual(event.reply_to_text, "父消息内容") + adapter._fetch_message_context_chain.assert_awaited_once_with( + "om_parent", + chat_id="oc_chat", + thread_id=None, + ) @patch.dict(os.environ, {}, clear=True) def test_send_replies_in_thread_when_thread_metadata_present(self): @@ -4782,10 +4787,45 @@ def _build_adapter(self): adapter._bot_user_id = "" adapter._bot_name = "Hermes" adapter._message_text_cache = OrderedDict() + adapter._message_item_cache = OrderedDict() adapter._client = Mock() adapter._build_get_message_request = Mock(return_value=object()) return adapter + @staticmethod + def _reply_item( + text, + *, + parent_id=None, + root_id=None, + chat_id="oc_chat", + thread_id=None, + ): + return SimpleNamespace( + body=SimpleNamespace(content=json.dumps({"text": text})), + msg_type="text", + mentions=[], + parent_id=parent_id, + upper_message_id=None, + root_id=root_id, + chat_id=chat_id, + thread_id=thread_id, + ) + + @staticmethod + def _install_reply_messages(adapter, messages): + adapter._build_get_message_request = lambda message_id: SimpleNamespace( + message_id=message_id + ) + + def get_message(request): + response = Mock() + response.success = Mock(return_value=True) + response.data = SimpleNamespace(items=[messages[request.message_id]]) + return response + + adapter._client.im.v1.message.get = Mock(side_effect=get_message) + def test_fetch_message_text_renders_mentions_without_hint_prefix(self): adapter = self._build_adapter() @@ -4876,6 +4916,208 @@ def test_build_mentions_map_string_id_shape(self): _build_mentions_map([bot_oid], _FeishuBotIdentity(open_id="ou_bot"))["@_user_3"].is_self ) + def test_fetch_message_context_chain_keeps_direct_parent_first(self): + adapter = self._build_adapter() + messages = { + "m_parent": self._reply_item("直接父消息", parent_id="m_grandparent"), + "m_grandparent": self._reply_item("更早的原始问题"), + } + self._install_reply_messages(adapter, messages) + + result = asyncio.run( + adapter._fetch_message_context_chain( + "m_parent", + chat_id="oc_chat", + thread_id=None, + ) + ) + + self.assertEqual( + result, + "直接父消息\n[Earlier quoted message]\n更早的原始问题", + ) + + def test_fetch_message_context_chain_follows_root_only_ancestry(self): + adapter = self._build_adapter() + messages = { + "m_parent": self._reply_item("直接父消息", root_id="m_root"), + "m_root": self._reply_item("根消息"), + } + self._install_reply_messages(adapter, messages) + + result = asyncio.run( + adapter._fetch_message_context_chain( + "m_parent", + chat_id="oc_chat", + thread_id=None, + ) + ) + + self.assertEqual( + result, + "直接父消息\n[Earlier quoted message]\n根消息", + ) + + def test_fetch_message_context_chain_stops_on_chat_boundary(self): + adapter = self._build_adapter() + messages = { + "m_parent": self._reply_item("直接父消息", parent_id="m_other_chat", chat_id="oc_chat"), + "m_other_chat": self._reply_item("不同聊天的消息", chat_id="oc_other"), + } + self._install_reply_messages(adapter, messages) + + result = asyncio.run( + adapter._fetch_message_context_chain( + "m_parent", + chat_id="oc_chat", + thread_id=None, + ) + ) + + self.assertEqual(result, "直接父消息") + + def test_fetch_message_context_chain_stops_on_thread_mismatch(self): + adapter = self._build_adapter() + messages = { + "m_parent": self._reply_item("直接父消息", parent_id="m_other_thread", thread_id="omt_current"), + "m_other_thread": self._reply_item("不同线程的消息", thread_id="omt_other"), + } + self._install_reply_messages(adapter, messages) + + result = asyncio.run( + adapter._fetch_message_context_chain( + "m_parent", + chat_id="oc_chat", + thread_id="omt_current", + ) + ) + + self.assertEqual(result, "直接父消息") + + def test_fetch_message_context_chain_keeps_direct_parent_on_ancestor_error(self): + adapter = self._build_adapter() + direct = Mock() + direct.success = Mock(return_value=True) + direct.data = SimpleNamespace( + items=[self._reply_item("直接父消息", parent_id="m_unavailable")] + ) + adapter._build_get_message_request = lambda message_id: SimpleNamespace( + message_id=message_id + ) + adapter._client.im.v1.message.get = Mock( + side_effect=[direct, OSError("offline")] + ) + + result = asyncio.run( + adapter._fetch_message_context_chain( + "m_parent", + chat_id="oc_chat", + thread_id=None, + ) + ) + + self.assertEqual(result, "直接父消息") + + def test_fetch_message_context_chain_falls_back_to_cached_direct_text(self): + adapter = self._build_adapter() + adapter._message_text_cache["m_parent"] = "缓存的直接父消息" + adapter._client.im.v1.message.get = Mock(side_effect=OSError("offline")) + + result = asyncio.run( + adapter._fetch_message_context_chain( + "m_parent", chat_id="oc_chat", thread_id=None + ) + ) + + self.assertEqual(result, "缓存的直接父消息") + + def test_fetch_message_context_chain_stops_before_unneeded_ancestor_call(self): + adapter = self._build_adapter() + messages = { + "m_parent": self._reply_item("p" * 500, parent_id="m_unneeded"), + } + self._install_reply_messages(adapter, messages) + + result = asyncio.run( + adapter._fetch_message_context_chain( + "m_parent", chat_id="oc_chat", thread_id=None + ) + ) + + self.assertEqual(result, "p" * 500) + self.assertEqual(adapter._client.im.v1.message.get.call_count, 1) + + def test_fetch_message_context_chain_honors_total_character_budget(self): + adapter = self._build_adapter() + messages = { + "m_parent": self._reply_item( + "p" * 450, + parent_id="m_grandparent", + ), + "m_grandparent": self._reply_item("g" * 200), + } + self._install_reply_messages(adapter, messages) + + result = asyncio.run( + adapter._fetch_message_context_chain( + "m_parent", + chat_id="oc_chat", + thread_id=None, + max_chars=500, + ) + ) + + self.assertEqual(len(result), 500) + self.assertTrue(result.startswith("p" * 450)) + self.assertIn("[Earlier quoted message]", result) + + def test_fetch_message_context_chain_breaks_parent_cycles(self): + adapter = self._build_adapter() + messages = { + "m_parent": self._reply_item( + "直接父消息", + parent_id="m_grandparent", + ), + "m_grandparent": self._reply_item( + "祖先消息", + parent_id="m_parent", + ), + } + self._install_reply_messages(adapter, messages) + + result = asyncio.run( + adapter._fetch_message_context_chain( + "m_parent", + chat_id="oc_chat", + thread_id=None, + ) + ) + + self.assertEqual( + result, + "直接父消息\n[Earlier quoted message]\n祖先消息", + ) + self.assertEqual(adapter._client.im.v1.message.get.call_count, 2) + + def test_fetch_message_context_chain_caps_api_calls_at_max_depth(self): + adapter = self._build_adapter() + messages = { + f"m{index}": self._reply_item( + str(index), parent_id=f"m{index + 1}" if index < 7 else None + ) + for index in range(8) + } + self._install_reply_messages(adapter, messages) + + result = asyncio.run( + adapter._fetch_message_context_chain( + "m0", chat_id="oc_chat", thread_id=None + ) + ) + + self.assertNotIn("6", result or "") + self.assertEqual(adapter._client.im.v1.message.get.call_count, 6) + class TestFeishuMentionEndToEnd(unittest.TestCase): """High-level scenarios from the design spec — verify the full pipeline."""