diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index d7654ff6c147a..de60322718bb4 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -5545,11 +5545,16 @@ class BasePlatformAdapter(ABC): coerce_plaintext_gateway_command(event) - # Rewrite ``event.source.thread_id`` via the installed recovery hook - # (Telegram DM topic mode) so the session key, guard checks, and - # downstream delivery all agree on the same lane. - # Offloaded: the sync hook must not block the loop. - await asyncio.to_thread(self._apply_topic_recovery, event) + # Telegram topic recovery only applies to private DM topic lanes. Do + # not submit a no-op check for group/forum/channel traffic to the + # shared default executor: a busy pool would delay message dispatch. + needs_topic_recovery = ( + getattr(self, "_topic_recovery_fn", None) is not None + and event.source.platform == Platform.TELEGRAM + and event.source.chat_type == "dm" + ) + if needs_topic_recovery: + await asyncio.to_thread(self._apply_topic_recovery, event) session_key = build_session_key( event.source, diff --git a/tests/gateway/test_active_session_text_merge.py b/tests/gateway/test_active_session_text_merge.py index e3dd819b2884a..ac8df747c6504 100644 --- a/tests/gateway/test_active_session_text_merge.py +++ b/tests/gateway/test_active_session_text_merge.py @@ -115,6 +115,63 @@ def _debounced_event(adapter: BasePlatformAdapter, session_key: str) -> MessageE return adapter._text_debounce[session_key].event +@pytest.mark.asyncio +async def test_non_dm_message_does_not_wait_for_topic_recovery_executor(monkeypatch): + """Group messages must not queue behind the shared thread pool. + + Topic recovery only applies to Telegram DM topic mode. Offloading that + no-op check for every group message makes ingress wait behind unrelated + blocking jobs when the default executor is saturated. + """ + adapter = _make_adapter() + recovery = MagicMock(return_value=None) + adapter.set_topic_recovery_fn(recovery) + executor_called = False + never_release = asyncio.Event() + + async def _blocked_to_thread(*args, **kwargs): + nonlocal executor_called + executor_called = True + await never_release.wait() + + monkeypatch.setattr(asyncio, "to_thread", _blocked_to_thread) + + await asyncio.wait_for( + adapter.handle_message(_make_event("/status", chat_type="group")), + timeout=1.0, + ) + await asyncio.sleep(0) + + assert executor_called is False + recovery.assert_not_called() + + +@pytest.mark.asyncio +async def test_dm_topic_recovery_stays_offloaded(monkeypatch): + """Real Telegram DM topic recovery must still run outside the event loop.""" + adapter = _make_adapter() + recovery = MagicMock(return_value="topic-222") + adapter.set_topic_recovery_fn(recovery) + offloaded = False + + async def _inline_to_thread(func, *args, **kwargs): + nonlocal offloaded + offloaded = True + return func(*args, **kwargs) + + monkeypatch.setattr(asyncio, "to_thread", _inline_to_thread) + event = _make_event("hello", chat_type="dm", thread_id="1") + original_source = event.source + + await adapter.handle_message(event) + await asyncio.sleep(0) + + assert offloaded is True + assert recovery.call_count == 1 + assert recovery.call_args.args[0] is original_source + assert event.source.thread_id == "topic-222" + + @pytest.mark.asyncio async def test_rapid_text_followups_accumulate_instead_of_replacing(): """Rapid TEXT follow-ups must all survive in the pending event."""