diff --git a/astrbot/core/config/default.py b/astrbot/core/config/default.py
index bdabcd933e..85764a635a 100644
--- a/astrbot/core/config/default.py
+++ b/astrbot/core/config/default.py
@@ -214,10 +214,13 @@
     "platform_specific": {
         # 平台特异配置：按平台分类，平台下按功能分组
         "lark": {
-            "pre_ack_emoji": {"enable": False, "emojis": ["Typing"]},
+            "pre_ack_emoji": {"enable": False, "emojis": ["Typing"], "auto_remove": True},
         },
         "telegram": {
-            "pre_ack_emoji": {"enable": False, "emojis": ["✍️"]},
+            "pre_ack_emoji": {"enable": False, "emojis": ["✍️"], "auto_remove": True},
+        },
+        "discord": {
+            "pre_ack_emoji": {"enable": False, "emojis": ["✍️"], "auto_remove": True},
         },
     },
     "wake_prefix": ["/"],
@@ -3555,6 +3558,13 @@ class ChatProviderTemplate(TypedDict):
                             "platform_specific.lark.pre_ack_emoji.enable": True,
                         },
                     },
+                    "platform_specific.lark.pre_ack_emoji.auto_remove": {
+                        "description": "[飞书] 处理完毕后自动撤回表情",
+                        "type": "bool",
+                        "condition": {
+                            "platform_specific.lark.pre_ack_emoji.enable": True,
+                        },
+                    },
                     "platform_specific.telegram.pre_ack_emoji.enable": {
                         "description": "[Telegram] 启用预回应表情",
                         "type": "bool",
@@ -3568,6 +3578,13 @@ class ChatProviderTemplate(TypedDict):
                             "platform_specific.telegram.pre_ack_emoji.enable": True,
                         },
                     },
+                    "platform_specific.telegram.pre_ack_emoji.auto_remove": {
+                        "description": "[Telegram] 处理完毕后自动撤回表情",
+                        "type": "bool",
+                        "condition": {
+                            "platform_specific.telegram.pre_ack_emoji.enable": True,
+                        },
+                    },
                     "platform_specific.discord.pre_ack_emoji.enable": {
                         "description": "[Discord] 启用预回应表情",
                         "type": "bool",
@@ -3581,6 +3598,13 @@ class ChatProviderTemplate(TypedDict):
                             "platform_specific.discord.pre_ack_emoji.enable": True,
                         },
                     },
+                    "platform_specific.discord.pre_ack_emoji.auto_remove": {
+                        "description": "[Discord] 处理完毕后自动撤回表情",
+                        "type": "bool",
+                        "condition": {
+                            "platform_specific.discord.pre_ack_emoji.enable": True,
+                        },
+                    },
                 },
             },
         },
diff --git a/astrbot/core/pipeline/pre_ack_emoji.py b/astrbot/core/pipeline/pre_ack_emoji.py
new file mode 100644
index 0000000000..c1b3ff96a4
--- /dev/null
+++ b/astrbot/core/pipeline/pre_ack_emoji.py
@@ -0,0 +1,69 @@
+import random
+from dataclasses import dataclass
+
+from astrbot.core import logger
+from astrbot.core.platform import AstrMessageEvent
+
+
+@dataclass
+class EmojiRef:
+    """贴出的表情引用，包含撤回所需的全部信息。"""
+
+    emoji: str
+    reaction_id: str | None = None  # 飞书需要 reaction_id 来撤回
+
+
+class PreAckEmojiManager:
+    """预回应表情管理器。
+
+    在 pipeline 执行前贴表情，执行后根据配置撤回。
+    运行在洋葱模型外层，不参与 stage 调度。
+    """
+
+    SUPPORTED_PLATFORMS = ("telegram", "lark", "discord")
+
+    def __init__(self, config: dict) -> None:
+        self.config = config
+
+    def _get_cfg(self, platform: str) -> dict:
+        return (
+            self.config.get("platform_specific", {})
+            .get(platform, {})
+            .get("pre_ack_emoji", {})
+        ) or {}
+
+    async def add_emoji(self, event: AstrMessageEvent) -> EmojiRef | None:
+        """贴表情。返回 EmojiRef，或 None（未贴）。"""
+        platform = event.get_platform_name()
+        if platform not in self.SUPPORTED_PLATFORMS:
+            return None
+
+        cfg = self._get_cfg(platform)
+        emojis = cfg.get("emojis") or []
+
+        if not cfg.get("enable", False) or not emojis:
+            return None
+
+        emoji = random.choice(emojis)
+        try:
+            reaction_id = await event.react(emoji)
+            return EmojiRef(emoji=emoji, reaction_id=reaction_id)
+        except Exception as e:
+            logger.warning(f"{platform} 预回应表情发送失败: {e}")
+            return None
+
+    async def remove_emoji(self, event: AstrMessageEvent, ref: EmojiRef | None) -> None:
+        """根据配置撤回表情。"""
+        if ref is None:
+            return
+
+        platform = event.get_platform_name()
+        cfg = self._get_cfg(platform)
+
+        if not cfg.get("auto_remove", True):
+            return
+
+        try:
+            await event.remove_react(ref.emoji, reaction_id=ref.reaction_id)
+        except Exception as e:
+            logger.warning(f"{platform} 预回应表情撤回失败: {e}")
diff --git a/astrbot/core/pipeline/preprocess_stage/stage.py b/astrbot/core/pipeline/preprocess_stage/stage.py
index 464f584f8e..4b3d67585e 100644
--- a/astrbot/core/pipeline/preprocess_stage/stage.py
+++ b/astrbot/core/pipeline/preprocess_stage/stage.py
@@ -1,5 +1,4 @@
 import asyncio
-import random
 import traceback
 from collections.abc import AsyncGenerator
 
@@ -26,26 +25,6 @@ async def process(
         event: AstrMessageEvent,
     ) -> None | AsyncGenerator[None, None]:
         """在处理事件之前的预处理"""
-        # 平台特异配置：platform_specific.<platform>.pre_ack_emoji
-        supported = {"telegram", "lark", "discord"}
-        platform = event.get_platform_name()
-        cfg = (
-            self.config.get("platform_specific", {})
-            .get(platform, {})
-            .get("pre_ack_emoji", {})
-        ) or {}
-        emojis = cfg.get("emojis") or []
-        if (
-            cfg.get("enable", False)
-            and platform in supported
-            and emojis
-            and event.is_at_or_wake_command
-        ):
-            try:
-                await event.react(random.choice(emojis))
-            except Exception as e:
-                logger.warning(f"{platform} 预回应表情发送失败: {e}")
-
         # 路径映射
         if mappings := self.platform_settings.get("path_mapping", []):
             # 支持 Record，Image 消息段的路径映射。
diff --git a/astrbot/core/pipeline/scheduler.py b/astrbot/core/pipeline/scheduler.py
index ffb9c5c99c..5705a66c04 100644
--- a/astrbot/core/pipeline/scheduler.py
+++ b/astrbot/core/pipeline/scheduler.py
@@ -9,6 +9,7 @@
 from astrbot.core.utils.active_event_registry import active_event_registry
 
 from .bootstrap import ensure_builtin_stages_registered
+from .pre_ack_emoji import PreAckEmojiManager
 from .context import PipelineContext
 from .stage import registered_stages
 from .stage_order import STAGES_ORDER
@@ -24,6 +25,7 @@ def __init__(self, context: PipelineContext) -> None:
         )  # 按照顺序排序
         self.ctx = context  # 上下文对象
         self.stages = []  # 存储阶段实例
+        self.pre_ack_emoji_mgr = PreAckEmojiManager(context.astrbot_config)
 
     async def initialize(self) -> None:
         """初始化管道调度器时, 初始化所有阶段"""
@@ -49,7 +51,9 @@ async def _process_stages(self, event: AstrMessageEvent, from_stage=0) -> None:
 
             if isinstance(coroutine, AsyncGenerator):
                 # 如果返回的是异步生成器, 实现洋葱模型的核心
+                did_yield = False
                 async for _ in coroutine:
+                    did_yield = True
                     # 此处是前置处理完成后的暂停点(yield), 下面开始执行后续阶段
                     if event.is_stopped():
                         logger.debug(
@@ -66,6 +70,10 @@ async def _process_stages(self, event: AstrMessageEvent, from_stage=0) -> None:
                             f"阶段 {stage.__class__.__name__} 已终止事件传播。",
                         )
                         break
+
+                # 洋葱阶段已通过递归处理了后续所有阶段，跳出外层循环避免重复执行
+                if did_yield:
+                    break
             else:
                 # 如果返回的是普通协程(不含yield的async函数), 则不进入下一层(基线条件)
                 # 简单地等待它执行完成, 然后继续执行下一个阶段
@@ -83,6 +91,7 @@ async def execute(self, event: AstrMessageEvent) -> None:
 
         """
         active_event_registry.register(event)
+        emoji = await self.pre_ack_emoji_mgr.add_emoji(event)
         try:
             await self._process_stages(event)
 
@@ -92,4 +101,5 @@ async def execute(self, event: AstrMessageEvent) -> None:
 
             logger.debug("pipeline 执行完毕。")
         finally:
+            await self.pre_ack_emoji_mgr.remove_emoji(event, emoji)
             active_event_registry.unregister(event)
diff --git a/astrbot/core/platform/astr_message_event.py b/astrbot/core/platform/astr_message_event.py
index 021a4bff7c..b68780634c 100644
--- a/astrbot/core/platform/astr_message_event.py
+++ b/astrbot/core/platform/astr_message_event.py
@@ -451,15 +451,31 @@ async def send(self, message: MessageChain) -> None:
         )
         self._has_send_oper = True
 
-    async def react(self, emoji: str) -> None:
+    async def react(self, emoji: str) -> str | None:
         """对消息添加表情回应。
 
         默认实现为发送一条包含该表情的消息。
-        注意：此实现并不一定符合所有平台的原生“表情回应”行为。
+        注意：此实现并不一定符合所有平台的原生"表情回应"行为。
         如需支持平台原生的消息反应功能，请在对应平台的子类中重写本方法。
+
+        Returns:
+            平台特定的 reaction 标识符(如飞书的 reaction_id)，用于后续撤回。
+            大多数平台返回 None。
         """
         await self.send(MessageChain([Plain(emoji)]))
 
+    async def remove_react(self, emoji: str, reaction_id: str | None = None) -> None:
+        """移除消息上的表情回应。
+
+        默认实现为空操作。
+        如需支持平台原生的撤回表情功能，请在对应平台的子类中重写本方法。
+
+        Args:
+            emoji: 要移除的表情
+            reaction_id: 平台特定的 reaction 标识符（如飞书的 reaction_id）
+        """
+        pass
+
     async def get_group(self, group_id: str | None = None, **kwargs) -> Group | None:
         """获取一个群聊的数据, 如果不填写 group_id: 如果是私聊消息，返回 None。如果是群聊消息，返回当前群聊的数据。
 
diff --git a/astrbot/core/platform/sources/discord/discord_platform_event.py b/astrbot/core/platform/sources/discord/discord_platform_event.py
index 02d4dae868..6da5a9af83 100644
--- a/astrbot/core/platform/sources/discord/discord_platform_event.py
+++ b/astrbot/core/platform/sources/discord/discord_platform_event.py
@@ -280,6 +280,17 @@ async def react(self, emoji: str) -> None:
         except Exception as e:
             logger.error(f"[Discord] 添加反应失败: {e}")
 
+    async def remove_react(self, emoji: str, reaction_id: str | None = None) -> None:
+        """移除 bot 在原消息上的表情回应"""
+        try:
+            if not hasattr(self.message_obj, "raw_message"):
+                return
+            raw = self.message_obj.raw_message
+            if hasattr(raw, "remove_reaction") and self.client.user:
+                await cast(discord.Message, raw).remove_reaction(emoji, self.client.user)
+        except Exception as e:
+            logger.warning(f"[Discord] 移除反应失败: {e}")
+
     def is_slash_command(self) -> bool:
         """判断是否为斜杠命令"""
         return (
diff --git a/astrbot/core/platform/sources/lark/lark_event.py b/astrbot/core/platform/sources/lark/lark_event.py
index 92e3a32b9e..1cb2db7087 100644
--- a/astrbot/core/platform/sources/lark/lark_event.py
+++ b/astrbot/core/platform/sources/lark/lark_event.py
@@ -12,6 +12,7 @@
     CreateImageRequestBody,
     CreateMessageReactionRequest,
     CreateMessageReactionRequestBody,
+    DeleteMessageReactionRequest,
     Emoji,
     ReplyMessageRequest,
     ReplyMessageRequestBody,
@@ -534,7 +535,7 @@ async def _send_media_message(
             receive_id_type=receive_id_type,
         )
 
-    async def react(self, emoji: str) -> None:
+    async def react(self, emoji: str) -> str | None:
         if self.bot.im is None:
             logger.error("[Lark] API Client im 模块未初始化，无法发送表情")
             return
@@ -553,8 +554,32 @@ async def react(self, emoji: str) -> None:
         response = await self.bot.im.v1.message_reaction.acreate(request)
         if not response.success():
             logger.error(f"发送飞书表情回应失败({response.code}): {response.msg}")
+            return None
+
+        # 返回 reaction_id 供调用方保存（如 PreAckEmojiManager 用于后续撤回）
+        if response.data and response.data.reaction_id:
+            return response.data.reaction_id
+        return None
+
+    async def remove_react(self, emoji: str, reaction_id: str | None = None) -> None:
+        if not reaction_id:
             return
 
+        if self.bot.im is None:
+            logger.warning("[Lark] API Client im 模块未初始化，无法撤回表情")
+            return
+
+        request = (
+            DeleteMessageReactionRequest.builder()
+            .message_id(self.message_obj.message_id)
+            .reaction_id(reaction_id)
+            .build()
+        )
+
+        response = await self.bot.im.v1.message_reaction.adelete(request)
+        if not response.success():
+            logger.warning(f"撤回飞书表情回应失败({response.code}): {response.msg}")
+
     async def send_streaming(self, generator, use_fallback: bool = False):
         buffer = None
         async for chain in generator:
diff --git a/astrbot/core/platform/sources/telegram/tg_event.py b/astrbot/core/platform/sources/telegram/tg_event.py
index 43e58960ee..ab040a9e94 100644
--- a/astrbot/core/platform/sources/telegram/tg_event.py
+++ b/astrbot/core/platform/sources/telegram/tg_event.py
@@ -355,6 +355,10 @@ async def react(self, emoji: str | None, big: bool = False) -> None:
         except Exception as e:
             logger.error(f"[Telegram] 添加反应失败: {e}")
 
+    async def remove_react(self, emoji: str, reaction_id: str | None = None) -> None:
+        """移除 bot 在原消息上的表情回应（Telegram 通过设置空 reaction 列表实现）"""
+        await self.react(None)
+
     async def _send_message_draft(
         self,
         chat_id: str,
diff --git a/dashboard/src/i18n/locales/en-US/features/config-metadata.json b/dashboard/src/i18n/locales/en-US/features/config-metadata.json
index 089aca7ad4..a2c898b3da 100644
--- a/dashboard/src/i18n/locales/en-US/features/config-metadata.json
+++ b/dashboard/src/i18n/locales/en-US/features/config-metadata.json
@@ -798,6 +798,9 @@
             "emojis": {
               "description": "Emoji List (Lark Emoji Enum Names)",
               "hint": "Emoji enum names reference: [https://open.feishu.cn/document/server-docs/im-v1/message-reaction/emojis-introduce](https://open.feishu.cn/document/server-docs/im-v1/message-reaction/emojis-introduce)"
+            },
+            "auto_remove": {
+              "description": "[Lark] Auto-remove emoji after processing"
             }
           }
         },
@@ -809,6 +812,9 @@
             "emojis": {
               "description": "Emoji List (Unicode)",
               "hint": "Telegram only supports a fixed reaction set, reference: [https://gist.github.com/Soulter/3f22c8e5f9c7e152e967e8bc28c97fc9](https://gist.github.com/Soulter/3f22c8e5f9c7e152e967e8bc28c97fc9)"
+            },
+            "auto_remove": {
+              "description": "[Telegram] Auto-remove emoji after processing"
             }
           }
         },
@@ -820,6 +826,9 @@
             "emojis": {
               "description": "Emoji List (Unicode or Custom Emoji Name)",
               "hint": "Enter Unicode emoji symbols, e.g., 👍, 🤔, ⏳"
+            },
+            "auto_remove": {
+              "description": "[Discord] Auto-remove emoji after processing"
             }
           }
         }
diff --git a/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json b/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json
index 158dbf3806..8675b714dc 100644
--- a/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json
+++ b/dashboard/src/i18n/locales/zh-CN/features/config-metadata.json
@@ -801,6 +801,9 @@
             "emojis": {
               "description": "表情列表(飞书表情枚举名)",
               "hint": "表情枚举名参考:[https://open.feishu.cn/document/server-docs/im-v1/message-reaction/emojis-introduce](https://open.feishu.cn/document/server-docs/im-v1/message-reaction/emojis-introduce)"
+            },
+            "auto_remove": {
+              "description": "[飞书] 处理完毕后自动撤回表情"
             }
           }
         },
@@ -812,6 +815,9 @@
             "emojis": {
               "description": "表情列表(Unicode)",
               "hint": "Telegram 仅支持固定反应集合,参考:[https://gist.github.com/Soulter/3f22c8e5f9c7e152e967e8bc28c97fc9](https://gist.github.com/Soulter/3f22c8e5f9c7e152e967e8bc28c97fc9)"
+            },
+            "auto_remove": {
+              "description": "[Telegram] 处理完毕后自动撤回表情"
             }
           }
         },
@@ -823,6 +829,9 @@
             "emojis": {
               "description": "表情列表（Unicode 或自定义表情名）",
               "hint": "填写 Unicode 表情符号，例如：👍、🤔、⏳"
+            },
+            "auto_remove": {
+              "description": "[Discord] 处理完毕后自动撤回表情"
             }
           }
         }
diff --git a/tests/test_pipeline_execution.py b/tests/test_pipeline_execution.py
new file mode 100644
index 0000000000..02f34f2bf3
--- /dev/null
+++ b/tests/test_pipeline_execution.py
@@ -0,0 +1,175 @@
+"""流水线执行次数测试。
+
+验证洋葱模型阶段后的普通阶段只执行一次，
+多层洋葱嵌套不重复执行，
+未挂起的异步生成器阶段不阻断后续执行。
+"""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from astrbot.core.pipeline.scheduler import PipelineScheduler
+
+
+def _make_normal_stage(name: str, counter: dict):
+    """创建普通（非洋葱）阶段"""
+    stage = MagicMock()
+    stage.__class__ = type(name, (), {})
+    stage.__class__.__name__ = name
+    counter[name] = 0
+
+    async def process(event):
+        counter[name] += 1
+
+    stage.process = process
+    return stage
+
+
+def _make_onion_stage(name: str, counter: dict):
+    """创建洋葱（generator）阶段"""
+    stage = MagicMock()
+    stage.__class__ = type(name, (), {})
+    stage.__class__.__name__ = name
+    counter[f"{name}_pre"] = 0
+    counter[f"{name}_post"] = 0
+
+    async def process(event):
+        counter[f"{name}_pre"] += 1
+        yield
+        counter[f"{name}_post"] += 1
+
+    stage.process = process
+    return stage
+
+
+def _make_non_suspending_generator_stage(name: str, counter: dict):
+    """创建不 yield 的异步生成器阶段（立即返回）"""
+    stage = MagicMock()
+    stage.__class__ = type(name, (), {})
+    stage.__class__.__name__ = name
+    counter[name] = 0
+
+    async def process(event):
+        counter[name] += 1
+        return
+        yield  # noqa: RET504 — 使 process 成为 async generator 但永不 yield
+
+    stage.process = process
+    return stage
+
+
+def _make_event():
+    event = MagicMock()
+    event.is_stopped.return_value = False
+    return event
+
+
+def _make_scheduler_with_stages(stages):
+    """创建带有指定 stages 的 PipelineScheduler（绕过正常初始化）"""
+    scheduler = object.__new__(PipelineScheduler)
+    scheduler.stages = stages
+    scheduler.ctx = MagicMock()
+    scheduler.pre_ack_emoji_mgr = MagicMock()
+    scheduler.pre_ack_emoji_mgr.add_emoji = AsyncMock(return_value=None)
+    scheduler.pre_ack_emoji_mgr.remove_emoji = AsyncMock()
+    return scheduler
+
+
+class TestOnionThenNormalExecutesOnce:
+    """洋葱阶段后的普通阶段只执行一次"""
+
+    @pytest.mark.asyncio
+    async def test_normal_after_single_onion(self):
+        counter = {}
+        onion = _make_onion_stage("Onion", counter)
+        normal = _make_normal_stage("Normal", counter)
+
+        scheduler = _make_scheduler_with_stages([onion, normal])
+        event = _make_event()
+
+        await scheduler._process_stages(event)
+
+        assert counter["Onion_pre"] == 1
+        assert counter["Normal"] == 1
+        assert counter["Onion_post"] == 1
+
+    @pytest.mark.asyncio
+    async def test_normal_after_two_onions(self):
+        counter = {}
+        onion1 = _make_onion_stage("Onion1", counter)
+        onion2 = _make_onion_stage("Onion2", counter)
+        normal = _make_normal_stage("Normal", counter)
+
+        scheduler = _make_scheduler_with_stages([onion1, onion2, normal])
+        event = _make_event()
+
+        await scheduler._process_stages(event)
+
+        assert counter["Onion1_pre"] == 1
+        assert counter["Onion2_pre"] == 1
+        assert counter["Normal"] == 1
+        assert counter["Onion2_post"] == 1
+        assert counter["Onion1_post"] == 1
+
+
+class TestMultiOnionNesting:
+    """多层洋葱嵌套不重复执行"""
+
+    @pytest.mark.asyncio
+    async def test_three_layer_onion_nesting(self):
+        counter = {}
+        stages = [
+            _make_onion_stage("O1", counter),
+            _make_onion_stage("O2", counter),
+            _make_onion_stage("O3", counter),
+            _make_normal_stage("Final", counter),
+        ]
+
+        scheduler = _make_scheduler_with_stages(stages)
+        event = _make_event()
+
+        await scheduler._process_stages(event)
+
+        assert counter["O1_pre"] == 1
+        assert counter["O2_pre"] == 1
+        assert counter["O3_pre"] == 1
+        assert counter["Final"] == 1
+        assert counter["O3_post"] == 1
+        assert counter["O2_post"] == 1
+        assert counter["O1_post"] == 1
+
+
+class TestNonSuspendingGeneratorStage:
+    """未挂起的异步生成器阶段不阻断后续执行"""
+
+    @pytest.mark.asyncio
+    async def test_non_yielding_generator_does_not_block(self):
+        counter = {}
+        non_suspending = _make_non_suspending_generator_stage("NonSuspending", counter)
+        normal = _make_normal_stage("After", counter)
+
+        scheduler = _make_scheduler_with_stages([non_suspending, normal])
+        event = _make_event()
+
+        await scheduler._process_stages(event)
+
+        assert counter["NonSuspending"] == 1
+        assert counter["After"] == 1
+
+    @pytest.mark.asyncio
+    async def test_non_yielding_generator_between_onions(self):
+        counter = {}
+        onion = _make_onion_stage("Onion", counter)
+        non_suspending = _make_non_suspending_generator_stage("NonSuspending", counter)
+        normal = _make_normal_stage("Final", counter)
+
+        scheduler = _make_scheduler_with_stages([onion, non_suspending, normal])
+        event = _make_event()
+
+        await scheduler._process_stages(event)
+
+        assert counter["Onion_pre"] == 1
+        assert counter["NonSuspending"] == 1
+        assert counter["Final"] == 1
+        assert counter["Onion_post"] == 1
diff --git a/tests/test_pre_ack_emoji.py b/tests/test_pre_ack_emoji.py
new file mode 100644
index 0000000000..21e2e54ab3
--- /dev/null
+++ b/tests/test_pre_ack_emoji.py
@@ -0,0 +1,166 @@
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from astrbot.core.pipeline.pre_ack_emoji import EmojiRef, PreAckEmojiManager
+
+
+def _make_config(platform: str, enable: bool = True, emojis: list | None = None, auto_remove: bool = True) -> dict:
+    """构造包含 platform_specific 的最小配置字典"""
+    if emojis is None:
+        emojis = ["👍"]
+    return {
+        "platform_specific": {
+            platform: {
+                "pre_ack_emoji": {
+                    "enable": enable,
+                    "emojis": emojis,
+                    "auto_remove": auto_remove,
+                }
+            }
+        }
+    }
+
+
+def _make_event(platform: str = "telegram", is_wake: bool = True) -> MagicMock:
+    """构造模拟的 AstrMessageEvent"""
+    event = MagicMock()
+    event.get_platform_name.return_value = platform
+    event.is_at_or_wake_command = is_wake
+    event.react = AsyncMock(return_value=None)
+    event.remove_react = AsyncMock()
+    return event
+
+
+class TestPreAckEmojiAddEmoji:
+    """测试贴表情逻辑"""
+
+    @pytest.mark.asyncio
+    async def test_disabled_does_not_react(self):
+        """功能关闭时不贴表情"""
+        cfg = _make_config("telegram", enable=False)
+        mgr = PreAckEmojiManager(cfg)
+        event = _make_event("telegram")
+        result = await mgr.add_emoji(event)
+        assert result is None
+        event.react.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_empty_emojis_does_not_react(self):
+        """emojis 列表为空时不贴表情"""
+        cfg = _make_config("telegram", emojis=[])
+        mgr = PreAckEmojiManager(cfg)
+        event = _make_event("telegram")
+        result = await mgr.add_emoji(event)
+        assert result is None
+        event.react.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_unsupported_platform_does_not_react(self):
+        """非支持平台不贴表情"""
+        cfg = _make_config("aiocqhttp", enable=True)
+        mgr = PreAckEmojiManager(cfg)
+        event = _make_event("aiocqhttp")
+        result = await mgr.add_emoji(event)
+        assert result is None
+        event.react.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_emoji_chosen_from_config_list(self):
+        """选出的 emoji 在配置列表内"""
+        emojis = ["👍", "✍️", "🤔"]
+        cfg = _make_config("telegram", emojis=emojis)
+        mgr = PreAckEmojiManager(cfg)
+        event = _make_event("telegram")
+        result = await mgr.add_emoji(event)
+        assert result is not None
+        assert result.emoji in emojis
+        event.react.assert_called_once_with(result.emoji)
+
+    @pytest.mark.asyncio
+    async def test_react_failure_returns_none(self):
+        """贴表情失败时返回 None，不影响主流程"""
+        cfg = _make_config("telegram")
+        mgr = PreAckEmojiManager(cfg)
+        event = _make_event("telegram")
+        event.react.side_effect = Exception("API error")
+        result = await mgr.add_emoji(event)
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_all_three_platforms_supported(self):
+        """telegram、lark、discord 三个平台都支持"""
+        for platform in ("telegram", "lark", "discord"):
+            cfg = _make_config(platform)
+            mgr = PreAckEmojiManager(cfg)
+            event = _make_event(platform)
+            result = await mgr.add_emoji(event)
+            assert result is not None
+            assert result.emoji == "👍"
+            event.react.assert_called_once_with("👍")
+
+    @pytest.mark.asyncio
+    async def test_lark_reaction_id_stored_in_ref(self):
+        """飞书平台 react 返回的 reaction_id 保存在 EmojiRef 中"""
+        cfg = _make_config("lark")
+        mgr = PreAckEmojiManager(cfg)
+        event = _make_event("lark")
+        event.react.return_value = "lark_reaction_123"
+        result = await mgr.add_emoji(event)
+        assert result is not None
+        assert result.reaction_id == "lark_reaction_123"
+
+
+class TestPreAckEmojiRemoveEmoji:
+    """测试撤回表情逻辑"""
+
+    @pytest.mark.asyncio
+    async def test_auto_remove_true_calls_remove_react(self):
+        """auto_remove=True 时撤回"""
+        cfg = _make_config("telegram", auto_remove=True)
+        mgr = PreAckEmojiManager(cfg)
+        event = _make_event("telegram")
+        ref = EmojiRef(emoji="👍")
+        await mgr.remove_emoji(event, ref)
+        event.remove_react.assert_called_once_with("👍", reaction_id=None)
+
+    @pytest.mark.asyncio
+    async def test_auto_remove_false_does_not_remove(self):
+        """auto_remove=False 时不撤回"""
+        cfg = _make_config("telegram", auto_remove=False)
+        mgr = PreAckEmojiManager(cfg)
+        event = _make_event("telegram")
+        ref = EmojiRef(emoji="👍")
+        await mgr.remove_emoji(event, ref)
+        event.remove_react.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_none_ref_skips_removal(self):
+        """ref 为 None 时跳过撤回"""
+        cfg = _make_config("telegram")
+        mgr = PreAckEmojiManager(cfg)
+        event = _make_event("telegram")
+        await mgr.remove_emoji(event, None)
+        event.remove_react.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_remove_failure_logs_warning(self):
+        """撤回失败时记录 warning 日志"""
+        cfg = _make_config("telegram")
+        mgr = PreAckEmojiManager(cfg)
+        event = _make_event("telegram")
+        ref = EmojiRef(emoji="👍")
+        event.remove_react.side_effect = Exception("API error")
+        with patch("astrbot.core.pipeline.pre_ack_emoji.logger") as mock_logger:
+            await mgr.remove_emoji(event, ref)
+            mock_logger.warning.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_lark_reaction_id_passed_to_remove_react(self):
+        """飞书平台撤回时传递 reaction_id"""
+        cfg = _make_config("lark", auto_remove=True)
+        mgr = PreAckEmojiManager(cfg)
+        event = _make_event("lark")
+        ref = EmojiRef(emoji="Typing", reaction_id="lark_reaction_123")
+        await mgr.remove_emoji(event, ref)
+        event.remove_react.assert_called_once_with("Typing", reaction_id="lark_reaction_123")