diff --git a/backend/app/channels/feishu.py b/backend/app/channels/feishu.py index 48cc8e8ba..1eeed7e7f 100644 --- a/backend/app/channels/feishu.py +++ b/backend/app/channels/feishu.py @@ -28,6 +28,7 @@ from deerflow.sandbox.sandbox_provider import get_sandbox_provider logger = logging.getLogger(__name__) PENDING_CLARIFICATION_TTL_SECONDS = 30 * 60 +FEISHU_INBOUND_BATCH_WINDOW_SECONDS = 0.75 def _is_feishu_command(text: str) -> bool: @@ -66,6 +67,7 @@ class FeishuChannel(Channel): self._running_card_ids: dict[str, str] = {} self._running_card_tasks: dict[str, asyncio.Task] = {} self._pending_clarifications: dict[tuple[str, str], list[dict[str, Any]]] = {} + self._pending_inbound_batches: dict[tuple[str, str], dict[str, Any]] = {} self._CreateFileRequest = None self._CreateFileRequestBody = None self._CreateImageRequest = None @@ -715,6 +717,113 @@ class FeishuChannel(Channel): return root_id or msg_id, False + @staticmethod + def _is_batchable_file_inbound( + *, + msg_type: InboundMessageType, + text: str, + files: list[dict[str, Any]], + root_id: str | None, + parent_id: str | None, + thread_id: str | None, + ) -> bool: + return msg_type == InboundMessageType.CHAT and text in {"[file]", "[image]"} and len(files) == 1 and not (root_id or parent_id or thread_id) + + def _schedule_prepare_inbound( + self, + msg_id: str, + inbound: InboundMessage, + *, + source_message_ids: list[str] | None = None, + ) -> None: + if self._main_loop and self._main_loop.is_running(): + logger.info("[Feishu] publishing inbound message to bus (type=%s, msg_id=%s)", inbound.msg_type.value, msg_id) + fut = asyncio.run_coroutine_threadsafe( + self._prepare_inbound(msg_id, inbound, source_message_ids=source_message_ids), + self._main_loop, + ) + fut.add_done_callback(lambda f, mid=msg_id: self._log_future_error(f, "prepare_inbound", mid)) + else: + logger.warning("[Feishu] main loop not running, cannot publish inbound message") + + def _schedule_batch_flush(self, key: tuple[str, str], source_message_id: str) -> None: + if self._main_loop and self._main_loop.is_running(): + fut = asyncio.run_coroutine_threadsafe(self._flush_pending_inbound_batch_after(key, source_message_id), self._main_loop) + fut.add_done_callback(lambda f, mid=source_message_id: self._log_future_error(f, "flush_inbound_batch", mid)) + else: + logger.warning("[Feishu] main loop not running, cannot flush inbound batch") + + def _queue_file_inbound_batch(self, msg_id: str, inbound: InboundMessage) -> bool: + key = self._pending_key(inbound.chat_id, inbound.user_id) + should_schedule_flush = False + expired_batch: tuple[str, InboundMessage, list[str]] | None = None + + with self._thread_lock: + batch = self._pending_inbound_batches.get(key) + now = time.time() + if batch: + if now - batch["created_at"] <= FEISHU_INBOUND_BATCH_WINDOW_SECONDS: + batched_inbound = batch["inbound"] + batch["message_ids"].append(msg_id) + batch["text_parts"].append(inbound.text) + batched_inbound.text = "\n\n".join(part for part in batch["text_parts"] if part) + batched_inbound.files.extend(inbound.files) + batched_inbound.metadata["batched_message_ids"] = list(batch["message_ids"]) + logger.info( + "[Feishu] batched inbound file message: chat_id=%s user_id=%s anchor=%s msg_id=%s files=%d", + inbound.chat_id, + inbound.user_id, + batch["anchor_message_id"], + msg_id, + len(batched_inbound.files), + ) + return True + + expired_batch = (batch["anchor_message_id"], batch["inbound"], list(batch["message_ids"])) + + self._pending_inbound_batches[key] = { + "anchor_message_id": msg_id, + "created_at": now, + "inbound": inbound, + "message_ids": [msg_id], + "text_parts": [inbound.text], + } + inbound.metadata["batched_message_ids"] = [msg_id] + should_schedule_flush = True + + if should_schedule_flush: + self._schedule_batch_flush(key, msg_id) + if expired_batch: + anchor_message_id, expired_inbound, source_message_ids = expired_batch + self._schedule_prepare_inbound(anchor_message_id, expired_inbound, source_message_ids=source_message_ids) + return True + + def _pop_pending_inbound_batch(self, key: tuple[str, str], *, anchor_message_id: str | None = None) -> tuple[str, InboundMessage, list[str]] | None: + with self._thread_lock: + batch = self._pending_inbound_batches.get(key) + if not batch: + return None + if anchor_message_id is not None and batch["anchor_message_id"] != anchor_message_id: + return None + self._pending_inbound_batches.pop(key, None) + return batch["anchor_message_id"], batch["inbound"], list(batch["message_ids"]) + + async def _flush_pending_inbound_batch_after(self, key: tuple[str, str], anchor_message_id: str) -> None: + await asyncio.sleep(FEISHU_INBOUND_BATCH_WINDOW_SECONDS) + batch = self._pop_pending_inbound_batch(key, anchor_message_id=anchor_message_id) + if not batch: + return + anchor_message_id, inbound, source_message_ids = batch + logger.info( + "[Feishu] flushing inbound file batch: chat_id=%s user_id=%s anchor=%s messages=%d files=%d", + inbound.chat_id, + inbound.user_id, + anchor_message_id, + len(source_message_ids), + len(inbound.files), + ) + await self._prepare_inbound(anchor_message_id, inbound, source_message_ids=source_message_ids) + @staticmethod def _log_task_error(task: asyncio.Task, name: str, msg_id: str) -> None: """Callback for background asyncio tasks to surface errors.""" @@ -727,11 +836,13 @@ class FeishuChannel(Channel): except Exception: pass - async def _prepare_inbound(self, msg_id: str, inbound) -> None: + async def _prepare_inbound(self, msg_id: str, inbound, *, source_message_ids: list[str] | None = None) -> None: """Kick off Feishu side effects without delaying inbound dispatch.""" inbound = await self._attach_connection_identity(inbound) - reaction_task = asyncio.create_task(self._add_reaction(msg_id, "OK")) - self._track_background_task(reaction_task, name="add_reaction", msg_id=msg_id) + reaction_message_ids = source_message_ids or [msg_id] + for reaction_message_id in reaction_message_ids: + reaction_task = asyncio.create_task(self._add_reaction(reaction_message_id, "OK")) + self._track_background_task(reaction_task, name="add_reaction", msg_id=reaction_message_id) self._ensure_running_card_started(msg_id) await self.bus.publish_inbound(inbound) @@ -922,12 +1033,17 @@ class FeishuChannel(Channel): ) inbound.topic_id = topic_id - # Schedule on the async event loop - if self._main_loop and self._main_loop.is_running(): - logger.info("[Feishu] publishing inbound message to bus (type=%s, msg_id=%s)", msg_type.value, msg_id) - fut = asyncio.run_coroutine_threadsafe(self._prepare_inbound(msg_id, inbound), self._main_loop) - fut.add_done_callback(lambda f, mid=msg_id: self._log_future_error(f, "prepare_inbound", mid)) - else: - logger.warning("[Feishu] main loop not running, cannot publish inbound message") + if self._is_batchable_file_inbound( + msg_type=msg_type, + text=text, + files=files_list, + root_id=root_id, + parent_id=parent_id, + thread_id=feishu_thread_id, + ): + self._queue_file_inbound_batch(msg_id, inbound) + return + + self._schedule_prepare_inbound(msg_id, inbound) except Exception: logger.exception("[Feishu] error processing message") diff --git a/backend/tests/test_feishu_parser.py b/backend/tests/test_feishu_parser.py index 41ccd7137..4dbf00b0c 100644 --- a/backend/tests/test_feishu_parser.py +++ b/backend/tests/test_feishu_parser.py @@ -228,6 +228,7 @@ def test_feishu_on_message_extracts_image_and_file_keys(): assert files == [{"image_key": "img_123"}, {"file_key": "file_456"}] assert "[image]" in mock_make_inbound.call_args[1]["text"] assert "[file]" in mock_make_inbound.call_args[1]["text"] + assert channel._pending_inbound_batches == {} def test_feishu_on_message_reuses_stored_parent_topic_for_card_replies(): @@ -285,6 +286,176 @@ def _make_text_event( return event +def _make_file_event( + file_key: str, + *, + chat_id: str = "chat_1", + message_id: str = "msg_1", + user_id: str = "user_1", + root_id: str | None = None, + parent_id: str | None = None, + thread_id: str | None = None, +): + event = MagicMock() + event.event.message.chat_id = chat_id + event.event.message.message_id = message_id + event.event.message.root_id = root_id + event.event.message.parent_id = parent_id + event.event.message.thread_id = thread_id + event.event.sender.sender_id.open_id = user_id + event.event.message.content = json.dumps({"file_key": file_key}) + return event + + +def test_feishu_batches_top_level_file_messages_from_same_user(monkeypatch): + async def go(): + monkeypatch.setattr("app.channels.feishu.FEISHU_INBOUND_BATCH_WINDOW_SECONDS", 0.01) + bus = MessageBus() + channel = FeishuChannel(bus, {"app_id": "test", "app_secret": "test"}) + channel._main_loop = asyncio.get_running_loop() + channel._add_reaction = AsyncMock() + channel._ensure_running_card_started = MagicMock() + + channel._on_message(_make_file_event("file_a", message_id="msg_file_1")) + channel._on_message(_make_file_event("file_b", message_id="msg_file_2")) + + inbound = await asyncio.wait_for(bus.get_inbound(), timeout=0.5) + + assert inbound.thread_ts == "msg_file_1" + assert inbound.topic_id == "msg_file_1" + assert inbound.text == "[file]\n\n[file]" + assert inbound.files == [{"file_key": "file_a"}, {"file_key": "file_b"}] + assert inbound.metadata["message_id"] == "msg_file_1" + assert inbound.metadata["topic_id"] == "msg_file_1" + assert inbound.metadata["batched_message_ids"] == ["msg_file_1", "msg_file_2"] + channel._ensure_running_card_started.assert_called_once_with("msg_file_1") + assert [call.args for call in channel._add_reaction.call_args_list] == [ + ("msg_file_1", "OK"), + ("msg_file_2", "OK"), + ] + + _run(go()) + + +def test_feishu_rich_text_file_message_does_not_enter_batch(): + bus = MessageBus() + channel = FeishuChannel(bus, {"app_id": "test", "app_secret": "test"}) + channel._schedule_prepare_inbound = MagicMock() + + event = MagicMock() + event.event.message.chat_id = "chat_1" + event.event.message.message_id = "msg_rich_file" + event.event.message.root_id = None + event.event.message.parent_id = None + event.event.message.thread_id = None + event.event.sender.sender_id.open_id = "user_1" + event.event.message.content = json.dumps( + { + "content": [ + [ + {"tag": "text", "text": "Review"}, + {"tag": "file", "file_key": "file_a"}, + ] + ] + } + ) + + channel._on_message(event) + + channel._schedule_prepare_inbound.assert_called_once() + inbound = channel._schedule_prepare_inbound.call_args.args[1] + assert inbound.text == "Review [file]" + assert inbound.files == [{"file_key": "file_a"}] + assert channel._pending_inbound_batches == {} + + +def test_feishu_file_batch_window_expiry_starts_new_topic(monkeypatch): + async def go(): + monkeypatch.setattr("app.channels.feishu.FEISHU_INBOUND_BATCH_WINDOW_SECONDS", 0.01) + bus = MessageBus() + channel = FeishuChannel(bus, {"app_id": "test", "app_secret": "test"}) + channel._main_loop = asyncio.get_running_loop() + channel._add_reaction = AsyncMock() + channel._ensure_running_card_started = MagicMock() + + channel._on_message(_make_file_event("file_a", message_id="msg_file_1")) + first = await asyncio.wait_for(bus.get_inbound(), timeout=0.5) + + channel._on_message(_make_file_event("file_b", message_id="msg_file_2")) + second = await asyncio.wait_for(bus.get_inbound(), timeout=0.5) + + assert first.topic_id == "msg_file_1" + assert first.files == [{"file_key": "file_a"}] + assert second.topic_id == "msg_file_2" + assert second.files == [{"file_key": "file_b"}] + assert channel._ensure_running_card_started.call_args_list[0].args == ("msg_file_1",) + assert channel._ensure_running_card_started.call_args_list[1].args == ("msg_file_2",) + + _run(go()) + + +def test_feishu_explicit_file_reply_does_not_enter_batch(monkeypatch): + async def go(): + monkeypatch.setattr("app.channels.feishu.FEISHU_INBOUND_BATCH_WINDOW_SECONDS", 10.0) + bus = MessageBus() + channel = FeishuChannel(bus, {"app_id": "test", "app_secret": "test"}) + channel._main_loop = asyncio.get_running_loop() + channel._add_reaction = AsyncMock() + channel._ensure_running_card_started = MagicMock() + + channel._on_message(_make_file_event("file_a", message_id="msg_reply", root_id="msg_root")) + + inbound = await asyncio.wait_for(bus.get_inbound(), timeout=0.5) + assert inbound.thread_ts == "msg_reply" + assert inbound.topic_id == "msg_root" + assert inbound.files == [{"file_key": "file_a"}] + assert channel._pending_inbound_batches == {} + + _run(go()) + + +def test_feishu_expired_file_batch_does_not_get_overwritten(monkeypatch): + monkeypatch.setattr("app.channels.feishu.FEISHU_INBOUND_BATCH_WINDOW_SECONDS", 0.5) + now = 0.0 + monkeypatch.setattr("app.channels.feishu.time.time", lambda: now) + + bus = MessageBus() + channel = FeishuChannel(bus, {"app_id": "test", "app_secret": "test"}) + channel._schedule_batch_flush = MagicMock() + channel._schedule_prepare_inbound = MagicMock() + + first = InboundMessage( + channel_name="feishu", + chat_id="chat_1", + user_id="user_1", + text="[file]", + thread_ts="msg_file_1", + files=[{"file_key": "file_a"}], + ) + second = InboundMessage( + channel_name="feishu", + chat_id="chat_1", + user_id="user_1", + text="[file]", + thread_ts="msg_file_2", + files=[{"file_key": "file_b"}], + ) + + channel._queue_file_inbound_batch("msg_file_1", first) + now = 1.0 + channel._queue_file_inbound_batch("msg_file_2", second) + + channel._schedule_prepare_inbound.assert_called_once_with( + "msg_file_1", + first, + source_message_ids=["msg_file_1"], + ) + assert channel._pop_pending_inbound_batch(channel._pending_key("chat_1", "user_1"), anchor_message_id="msg_file_1") is None + + current = channel._pop_pending_inbound_batch(channel._pending_key("chat_1", "user_1"), anchor_message_id="msg_file_2") + assert current == ("msg_file_2", second, ["msg_file_2"]) + + def test_feishu_plain_reply_consumes_pending_clarification_topic(): bus = MessageBus() store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")