fix(channels): serialize per-chat thread creation to avoid duplicate threads (#3799)

Signed-off-by: Yufeng He <40085740+he-yufeng@users.noreply.github.com>

backend/app/channels/manager.py

@@ -806,6 +806,9 @@ class ChannelManager:
         self._require_bound_identity = require_bound_identity
         self._client = None  # lazy init — langgraph_sdk async client
         self._channel_metadata_synced: set[str] = set()
+        # Per-conversation locks so concurrent inbound messages for the same
+        # chat don't race to create duplicate threads (see _get_or_create_thread).
+        self._thread_create_locks: dict[tuple[str, str, str | None], asyncio.Lock] = {}
         self._skill_storage: SkillStorage | None = None
         self._csrf_token = generate_csrf_token()
         self._semaphore: asyncio.Semaphore | None = None
@@ -1211,6 +1214,36 @@ class ChannelManager:
         logger.info("[Manager] new thread created through Gateway: thread_id=%s for chat_id=%s topic_id=%s", thread_id, msg.chat_id, msg.topic_id)
         return thread_id
 
+    async def _get_or_create_thread(self, client, msg: InboundMessage) -> tuple[str, bool]:
+        """Return ``(thread_id, created)``, creating a thread only if needed.
+
+        Each inbound message is dispatched on its own task, so two messages that
+        arrive close together for the same chat would both look up a missing
+        thread and then both create one — the second store silently overwrites
+        the first, orphaning a Gateway thread and splitting the conversation.
+        Serialize the create path per conversation and re-check inside the lock
+        so only the first message creates a thread and the rest reuse it.
+        """
+        thread_id = await self._lookup_thread_id(msg)
+        if thread_id:
+            return thread_id, False
+
+        key = (msg.channel_name, msg.chat_id, msg.topic_id)
+        lock = self._thread_create_locks.setdefault(key, asyncio.Lock())
+        try:
+            async with lock:
+                # A concurrent message for the same chat may have created the
+                # thread while we were waiting on the lock.
+                thread_id = await self._lookup_thread_id(msg)
+                if thread_id:
+                    return thread_id, False
+                return await self._create_thread(client, msg), True
+        finally:
+            # Once the thread is stored, later messages short-circuit on the
+            # lookup above and never reach this lock, so it's safe to drop the
+            # entry and keep the registry bounded to in-flight conversations.
+            self._thread_create_locks.pop(key, None)
+
     async def _update_thread_channel_metadata(self, client, msg: InboundMessage, thread_id: str) -> None:
         """Best-effort source metadata backfill for existing IM-created threads."""
         # The metadata (provider/chat/topic) is constant for a thread, so one
@@ -1248,18 +1281,15 @@ class ChannelManager:
         client = self._get_client()
         storage_user_id = _channel_storage_user_id(msg)
 
-        # Look up existing DeerFlow thread.
-        # topic_id may be None (e.g. Telegram private chats) — the store
-        # handles this by using the "channel:chat_id" key without a topic suffix.
-        thread_id = await self._lookup_thread_id(msg)
-        if thread_id:
+        # Look up the existing DeerFlow thread, creating one if this is the
+        # first message for the chat. topic_id may be None (e.g. Telegram
+        # private chats) — the store handles this by using the "channel:chat_id"
+        # key without a topic suffix.
+        thread_id, created = await self._get_or_create_thread(client, msg)
+        if not created:
             logger.info("[Manager] reusing thread: thread_id=%s for topic_id=%s", thread_id, msg.topic_id)
             await self._update_thread_channel_metadata(client, msg, thread_id)
 
-        # No existing thread found — create a new one
-        if thread_id is None:
-            thread_id = await self._create_thread(client, msg)
-
         assistant_id, run_config, run_context = self._resolve_run_params(msg, thread_id)
 
         # If the inbound message contains file attachments, let the channel

backend/tests/test_channels.py

@@ -638,6 +638,57 @@ class TestChannelManager:
         assert headers["Cookie"] == f"csrf_token={csrf_token}"
         assert headers["X-DeerFlow-Internal-Token"]
 
+    def test_concurrent_inbound_for_same_chat_reuses_single_thread(self):
+        # Each inbound message is dispatched on its own task, so two messages
+        # arriving close together for the same chat can both look up a missing
+        # thread before either stores one. Without per-conversation locking they
+        # each create a thread and the second store overwrites the first,
+        # orphaning a Gateway thread and splitting the conversation. The create
+        # path must be serialized so only one thread is created and reused.
+        from app.channels.manager import ChannelManager
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            created_ids: list[str] = []
+            first_create_started = asyncio.Event()
+            release_create = asyncio.Event()
+
+            async def blocking_create(*, metadata=None, headers=None):
+                thread_id = f"thread-{len(created_ids) + 1}"
+                created_ids.append(thread_id)
+                first_create_started.set()
+                # Hold the create open so a second concurrent message has a
+                # chance to race in before this one stores its thread_id.
+                await release_create.wait()
+                return {"thread_id": thread_id}
+
+            mock_client = MagicMock()
+            mock_client.threads.create = blocking_create
+            manager._client = mock_client
+
+            msg = InboundMessage(channel_name="slack", chat_id="C1", user_id="U1", text="hi")
+
+            task1 = asyncio.create_task(manager._get_or_create_thread(mock_client, msg))
+            await first_create_started.wait()
+            # task2 should block on the per-conversation lock rather than enter
+            # threads.create a second time.
+            task2 = asyncio.create_task(manager._get_or_create_thread(mock_client, msg))
+            await asyncio.sleep(0)
+            release_create.set()
+
+            (tid1, created1), (tid2, created2) = await asyncio.gather(task1, task2)
+
+            assert len(created_ids) == 1
+            assert tid1 == tid2 == "thread-1"
+            assert created1 is True
+            assert created2 is False
+            assert store.get_thread_id("slack", "C1") == "thread-1"
+
+        _run(go())
+
     def test_fetch_gateway_includes_internal_auth_headers(self, monkeypatch):
         from app.channels.manager import ChannelManager