From 7a6c4a994a86583d2a3c056ee9d0f157d4f030c2 Mon Sep 17 00:00:00 2001 From: Yufeng He <40085740+he-yufeng@users.noreply.github.com> Date: Fri, 26 Jun 2026 15:39:57 +0800 Subject: [PATCH] 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 | 48 +++++++++++++++++++++++++------ backend/tests/test_channels.py | 51 +++++++++++++++++++++++++++++++++ 2 files changed, 90 insertions(+), 9 deletions(-) diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index d030c7db7..235ecce52 100644 --- a/backend/app/channels/manager.py +++ b/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 diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index a92a6f34d..7cddb4551 100644 --- a/backend/tests/test_channels.py +++ b/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