fix(gateway): offload channel directory session scans

gateway/channel_directory.py

@@ -6,6 +6,7 @@ Built on gateway startup, refreshed periodically (every 5 min), and saved to
 action="list" and for resolving human-friendly channel names to numeric IDs.
 """
 
+import asyncio
 import json
 import logging
 from datetime import datetime
@@ -121,7 +122,7 @@ async def build_channel_directory(adapters: Dict[Any, Any]) -> Dict[str, Any]:
     for platform, adapter in adapters.items():
         try:
             if platform == Platform.DISCORD:
-                platforms["discord"] = _build_discord(adapter)
+                platforms["discord"] = await asyncio.to_thread(_build_discord, adapter)
             elif platform == Platform.SLACK:
                 platforms["slack"] = await _build_slack(adapter)
         except Exception as e:
@@ -142,7 +143,7 @@ async def build_channel_directory(adapters: Dict[Any, Any]) -> Dict[str, Any]:
             or plat_name not in adapter_platform_names
         ):
             continue
-        platforms[plat_name] = _build_from_sessions(plat_name)
+        platforms[plat_name] = await asyncio.to_thread(_build_from_sessions, plat_name)
 
     # Include plugin-registered platforms (dynamic enum members aren't in
     # Platform.__members__, so the loop above misses them). Same
@@ -156,7 +157,7 @@ async def build_channel_directory(adapters: Dict[Any, Any]) -> Dict[str, Any]:
                 and entry.name not in platforms
                 and entry.name in adapter_platform_names
             ):
-                platforms[entry.name] = _build_from_sessions(entry.name)
+                platforms[entry.name] = await asyncio.to_thread(_build_from_sessions, entry.name)
     except Exception:
         pass
 
@@ -223,7 +224,7 @@ async def _build_slack(adapter) -> List[Dict[str, Any]]:
     """
     team_clients = getattr(adapter, "_team_clients", None) or {}
     if not team_clients:
-        return _build_from_sessions("slack")
+        return await asyncio.to_thread(_build_from_sessions, "slack")
 
     channels: List[Dict[str, Any]] = []
     seen_ids: set = set()
@@ -267,7 +268,7 @@ async def _build_slack(adapter) -> List[Dict[str, Any]]:
             continue
 
     # Merge in DM/group entries discovered from session history.
-    for entry in _build_from_sessions("slack"):
+    for entry in await asyncio.to_thread(_build_from_sessions, "slack"):
         if entry.get("id") not in seen_ids:
             channels.append(entry)
             seen_ids.add(entry.get("id"))

tests/gateway/test_channel_directory.py

@@ -3,6 +3,7 @@
 import asyncio
 import json
 import os
+import threading
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -84,6 +85,85 @@ class TestBuildChannelDirectoryWrites:
         assert result == previous
 
 
+class TestBuildChannelDirectoryOffload:
+    def test_discord_builder_runs_off_event_loop_thread(self, tmp_path):
+        from gateway.config import Platform
+
+        cache_file = tmp_path / "channel_directory.json"
+        loop_thread = threading.get_ident()
+        builder_threads = []
+
+        def fake_build_discord(_adapter):
+            builder_threads.append(threading.get_ident())
+            return []
+
+        with patch("gateway.channel_directory._build_discord", side_effect=fake_build_discord), \
+             patch("gateway.channel_directory.DIRECTORY_PATH", cache_file):
+            asyncio.run(build_channel_directory({Platform.DISCORD: object()}))
+
+        assert builder_threads
+        assert all(tid != loop_thread for tid in builder_threads)
+
+    def test_session_discovery_runs_off_event_loop_thread(self, tmp_path):
+        from gateway.config import Platform
+
+        cache_file = tmp_path / "channel_directory.json"
+        loop_thread = threading.get_ident()
+        calls = []
+
+        def fake_build_from_sessions(platform_name):
+            calls.append((platform_name, threading.get_ident()))
+            return []
+
+        with patch("gateway.channel_directory._build_from_sessions", side_effect=fake_build_from_sessions), \
+             patch("gateway.channel_directory.DIRECTORY_PATH", cache_file):
+            asyncio.run(build_channel_directory({Platform.TELEGRAM: object()}))
+
+        assert [name for name, _ in calls] == ["telegram"]
+        assert calls[0][1] != loop_thread
+
+    def test_plugin_session_discovery_runs_off_event_loop_thread(self, tmp_path):
+        cache_file = tmp_path / "channel_directory.json"
+        loop_thread = threading.get_ident()
+        calls = []
+        plugin_entry = SimpleNamespace(name="irc")
+
+        def fake_build_from_sessions(platform_name):
+            calls.append((platform_name, threading.get_ident()))
+            return []
+
+        with patch("gateway.channel_directory._build_from_sessions", side_effect=fake_build_from_sessions), \
+             patch("gateway.channel_directory.DIRECTORY_PATH", cache_file), \
+             patch(
+                 "gateway.platform_registry.platform_registry.plugin_entries",
+                 return_value=[plugin_entry],
+             ):
+            asyncio.run(build_channel_directory({"irc": object()}))
+
+        assert [name for name, _ in calls] == ["irc"]
+        assert calls[0][1] != loop_thread
+
+    def test_slack_session_merge_runs_off_event_loop_thread(self):
+        loop_thread = threading.get_ident()
+        calls = []
+
+        class FakeSlackClient:
+            async def users_conversations(self, **_kwargs):
+                return {"ok": True, "channels": []}
+
+        def fake_build_from_sessions(platform_name):
+            calls.append((platform_name, threading.get_ident()))
+            return [{"id": "D1", "name": "Alice", "type": "dm"}]
+
+        adapter = SimpleNamespace(_team_clients={"T1": FakeSlackClient()})
+        with patch("gateway.channel_directory._build_from_sessions", side_effect=fake_build_from_sessions):
+            channels = asyncio.run(_build_slack(adapter))
+
+        assert channels == [{"id": "D1", "name": "Alice", "type": "dm"}]
+        assert [name for name, _ in calls] == ["slack"]
+        assert calls[0][1] != loop_thread
+
+
 class TestResolveChannelName:
     def _setup(self, tmp_path, platforms):
         cache_file = _write_directory(tmp_path, platforms)