diff --git a/channel/chat_channel.py b/channel/chat_channel.py index 96c12d05..afe6d829 100644 --- a/channel/chat_channel.py +++ b/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: diff --git a/channel/feishu/README.md b/channel/feishu/README.md index 818d8f60..9ec89c26 100644 --- a/channel/feishu/README.md +++ b/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. 查看日志中的错误信息 diff --git a/channel/feishu/feishu_channel.py b/channel/feishu/feishu_channel.py index 4a735158..4eeaa22d 100644 --- a/channel/feishu/feishu_channel.py +++ b/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 diff --git a/tests/test_feishu_message_recall.py b/tests/test_feishu_message_recall.py new file mode 100644 index 00000000..104de20b --- /dev/null +++ b/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)