Merge pull request #2978 from AaronZ345/agent/feishu-message-recall

feat(feishu): cancel recalled messages

channel/chat_channel.py

@@ -562,6 +562,42 @@ class ChatChannel(Channel):
                         semaphore.release()
             time.sleep(0.2)
 
+    def cancel_message(self, session_id: str, message_id: str):
+        """Cancel one channel message without disturbing later queued work.
+
+        Queued contexts are matched by their original channel message ID. An
+        in-flight agent run is cancelled through the per-request token that the
+        channel placed on the context before dispatch.
+        """
+        removed = 0
+        with self.lock:
+            session = self.sessions.get(session_id)
+            if session is not None:
+                context_queue = session[0]
+                kept = []
+                for _ in range(context_queue.qsize()):
+                    context = context_queue.get_nowait()
+                    context_queue.task_done()
+                    message = context.get("msg") if context is not None else None
+                    if getattr(message, "msg_id", None) == message_id:
+                        removed += 1
+                    else:
+                        kept.append(context)
+                for context in kept:
+                    context_queue.put(context)
+
+        from agent.protocol import get_cancel_registry
+
+        active = get_cancel_registry().cancel_request(message_id)
+        logger.info(
+            "[chat_channel] message recall: session=%s, message=%s, queued=%s, active=%s",
+            session_id,
+            message_id,
+            removed,
+            active,
+        )
+        return removed, active
+
     # 取消session_id对应的所有任务，只能取消排队的消息和已提交线程池但未执行的任务
     def cancel_session(self, session_id):
         with self.lock:

channel/feishu/README.md

@@ -63,7 +63,7 @@ python3 app.py
 2. 进入应用详情 -> 事件订阅
 3. 选择 **将事件发送至开发者服务器**
 4. 填写请求地址: `http://your-domain:9891/`
-5. 添加事件: `im.message.receive_v1` (接收消息v2.0)
+5. 添加事件: `im.message.receive_v1` (接收消息v2.0) 和 `im.message.recalled_v1` (消息撤回)
 6. 保存配置
 
 ### 4. 注意事项
@@ -101,7 +101,7 @@ python3 app.py
 1. 登录[飞书开放平台](https://open.feishu.cn/)
 2. 进入应用详情 -> 事件订阅
 3. 选择 **使用长连接接收事件**
-4. 添加事件: `im.message.receive_v1` (接收消息v2.0)
+4. 添加事件: `im.message.receive_v1` (接收消息v2.0) 和 `im.message.recalled_v1` (消息撤回)
 5. 保存配置
 
 ### 5. 注意事项
@@ -168,7 +168,7 @@ Address already in use
 ### 收不到消息
 
 1. 检查飞书应用的事件订阅配置
-2. 确认已添加 `im.message.receive_v1` 事件
+2. 确认已添加 `im.message.receive_v1` 和 `im.message.recalled_v1` 事件
 3. 检查应用权限: 需要 `im:message` 权限
 4. 查看日志中的错误信息
 

channel/feishu/feishu_channel.py

@@ -251,6 +251,8 @@ class FeiShuChanel(ChatChannel):
         super().__init__()
         # 历史消息id暂存，用于幂等控制
         self.receivedMsgs = ExpiredDict(60 * 60 * 7.1)
+        # Route recall events back to the session that accepted the message.
+        self._message_sessions = ExpiredDict(60 * 60 * 7.1)
         self._http_server = None
         self._ws_client = None
         self._ws_thread = None
@@ -387,6 +389,19 @@ class FeiShuChanel(ChatChannel):
             except Exception as e:
                 logger.error(f"[FeiShu] websocket handle message error: {e}", exc_info=True)
 
+        def handle_message_recalled_event(
+            data: lark.im.v1.P2ImMessageRecalledV1,
+        ) -> None:
+            """Cancel only the task created by the recalled Feishu message."""
+            try:
+                event_dict = json.loads(lark.JSON.marshal(data))
+                self._handle_message_recalled_event(event_dict.get("event", {}))
+            except Exception as e:
+                logger.error(
+                    f"[FeiShu] websocket handle message recall error: {e}",
+                    exc_info=True,
+                )
+
         def handle_card_action(data):
             """Handle Card 2.0 button callbacks and update the card in place."""
             try:
@@ -403,6 +418,7 @@ class FeiShuChanel(ChatChannel):
         event_handler = (
             lark.EventDispatcherHandler.builder("", "")
             .register_p2_im_message_receive_v1(handle_message_event)
+            .register_p2_im_message_recalled_v1(handle_message_recalled_event)
             .register_p2_card_action_trigger(handle_card_action)
             .build()
         )
@@ -591,6 +607,29 @@ class FeiShuChanel(ChatChannel):
         )
         return response
 
+    def _handle_message_recalled_event(self, event: dict):
+        """Cancel one recalled message while preserving later queued messages."""
+        message_id = event.get("message_id")
+        if not message_id:
+            logger.warning(f"[FeiShu] invalid message recall event: {event}")
+            return 0, False
+
+        session_id = self._message_sessions.get(message_id)
+        if not session_id:
+            logger.info(
+                f"[FeiShu] ignored recall for unknown message, message_id={message_id}"
+            )
+            return 0, False
+
+        result = self.cancel_message(session_id, message_id)
+        self._message_sessions.pop(message_id, None)
+        logger.info(
+            "[FeiShu] recalled message cancelled, "
+            f"message_id={message_id}, session_id={session_id}, "
+            f"queued={result[0]}, active={result[1]}"
+        )
+        return result
+
     def _handle_message_event(self, event: dict):
         """
         处理消息事件的核心逻辑
@@ -724,6 +763,10 @@ class FeiShuChanel(ChatChannel):
             no_need_at=True
         )
         if context:
+            # Feishu recall events only include message_id/chat_id. Keep the
+            # accepted route and use message_id as the agent cancellation key.
+            context["request_id"] = msg_id
+            self._message_sessions[msg_id] = context["session_id"]
             # 流式回复模式：向 context 注入 on_event 回调，agent 每产出一段文字时会调用它。
             # 回调内部先发送一条占位消息获取 message_id，之后通过 PATCH 接口原地更新内容，
             # 实现打字机效果。回调结束时设置 context["feishu_streamed"]=True，
@@ -2095,6 +2138,7 @@ class FeishuController:
     FAILED_MSG = '{"success": false}'
     SUCCESS_MSG = '{"success": true}'
     MESSAGE_RECEIVE_TYPE = "im.message.receive_v1"
+    MESSAGE_RECALLED_TYPE = "im.message.recalled_v1"
     CARD_ACTION_TYPE = "card.action.trigger"
 
     def GET(self):
@@ -2134,6 +2178,8 @@ class FeishuController:
             # 3. Handle message events.
             if event_type == self.MESSAGE_RECEIVE_TYPE and event:
                 channel._handle_message_event(event)
+            elif event_type == self.MESSAGE_RECALLED_TYPE and event:
+                channel._handle_message_recalled_event(event)
 
             return self.SUCCESS_MSG
 

tests/test_feishu_message_recall.py

@@ -0,0 +1,136 @@
+import json
+import threading
+import time
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from bridge.context import Context, ContextType
+from channel.chat_channel import ChatChannel
+from channel.feishu import feishu_channel
+from channel.feishu.feishu_channel import FeishuController, FeiShuChanel
+from common.dequeue import Dequeue
+from common.expired_dict import ExpiredDict
+
+
+def _context(message_id: str) -> Context:
+    return Context(
+        ContextType.TEXT,
+        message_id,
+        {
+            "session_id": "session-1",
+            "msg": SimpleNamespace(msg_id=message_id),
+        },
+    )
+
+
+def _bare_chat_channel(*contexts: Context) -> ChatChannel:
+    channel = ChatChannel.__new__(ChatChannel)
+    channel.lock = threading.RLock()
+    channel.futures = {}
+    queue = Dequeue()
+    for context in contexts:
+        queue.put(context)
+    channel.sessions = {"session-1": [queue, MagicMock()]}
+    return channel
+
+
+def test_cancel_message_removes_only_recalled_queued_context(monkeypatch):
+    channel = _bare_chat_channel(_context("m1"), _context("m2"), _context("m3"))
+    registry = MagicMock()
+    registry.cancel_request.return_value = False
+    monkeypatch.setattr("agent.protocol.get_cancel_registry", lambda: registry)
+
+    queued, active = channel.cancel_message("session-1", "m2")
+
+    assert (queued, active) == (1, False)
+    remaining = channel.sessions["session-1"][0]
+    assert [remaining.get_nowait().get("msg").msg_id for _ in range(2)] == ["m1", "m3"]
+    registry.cancel_request.assert_called_once_with("m2")
+
+
+def test_cancel_message_targets_active_request_without_clearing_queue(monkeypatch):
+    channel = _bare_chat_channel(_context("later"))
+    registry = MagicMock()
+    registry.cancel_request.return_value = True
+    monkeypatch.setattr("agent.protocol.get_cancel_registry", lambda: registry)
+
+    queued, active = channel.cancel_message("session-1", "active")
+
+    assert (queued, active) == (0, True)
+    remaining = channel.sessions["session-1"][0]
+    assert remaining.get_nowait().get("msg").msg_id == "later"
+
+
+def test_feishu_message_uses_message_id_for_precise_recall(monkeypatch):
+    channel = FeiShuChanel()
+    channel.receivedMsgs = ExpiredDict(60)
+    channel._message_sessions = ExpiredDict(60)
+    monkeypatch.setattr(channel, "fetch_access_token", lambda: "tenant-token")
+    monkeypatch.setattr(channel, "_make_feishu_stream_callback", lambda *_: MagicMock())
+    produced = []
+    monkeypatch.setattr(channel, "produce", produced.append)
+
+    channel._handle_message_event(
+        {
+            "app_id": "cli_bot",
+            "sender": {"sender_id": {"open_id": "ou_user"}},
+            "message": {
+                "message_id": "om_recall_me",
+                "chat_id": "oc_chat",
+                "chat_type": "p2p",
+                "message_type": "text",
+                "create_time": str(int(time.time() * 1000)),
+                "content": json.dumps({"text": "long task"}),
+            },
+        }
+    )
+
+    assert len(produced) == 1
+    assert produced[0]["request_id"] == "om_recall_me"
+    assert channel._message_sessions.get("om_recall_me") == "ou_user"
+
+
+def test_feishu_recall_cancels_only_the_original_message(monkeypatch):
+    channel = FeiShuChanel()
+    channel._message_sessions = ExpiredDict(60)
+    channel._message_sessions["om_recalled"] = "session-1"
+    cancel_message = MagicMock(return_value=(0, True))
+    monkeypatch.setattr(channel, "cancel_message", cancel_message)
+
+    result = channel._handle_message_recalled_event(
+        {"message_id": "om_recalled", "chat_id": "oc_chat"}
+    )
+
+    assert result == (0, True)
+    cancel_message.assert_called_once_with("session-1", "om_recalled")
+    assert channel._message_sessions.get("om_recalled") is None
+
+
+def test_feishu_recall_ignores_unknown_message():
+    channel = FeiShuChanel()
+    channel._message_sessions = ExpiredDict(60)
+
+    assert channel._handle_message_recalled_event({"message_id": "unknown"}) == (0, False)
+
+
+def test_feishu_webhook_routes_message_recall(monkeypatch):
+    channel = FeiShuChanel()
+    channel.feishu_token = "verification-token"
+    handle_recall = MagicMock(return_value=(1, False))
+    monkeypatch.setattr(channel, "_handle_message_recalled_event", handle_recall)
+    event = {"message_id": "om_recalled", "chat_id": "oc_chat"}
+    request = {
+        "header": {
+            "event_type": "im.message.recalled_v1",
+            "token": "verification-token",
+        },
+        "event": event,
+    }
+    monkeypatch.setattr(
+        feishu_channel.web,
+        "data",
+        lambda: json.dumps(request).encode("utf-8"),
+    )
+
+    assert json.loads(FeishuController().POST()) == {"success": True}
+    handle_recall.assert_called_once_with(event)