fix(channels): synchronize ChannelStore reads (#5083)

Signed-off-by: cuishuang <imcusg@gmail.com>

backend/app/channels/AGENTS.md

@@ -6,7 +6,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
 
 **Components**:
 - `message_bus.py` - Async pub/sub hub (`InboundMessage` → queue → dispatcher; `OutboundMessage` → callbacks → channels)
-- `store.py` - JSON-file persistence mapping `channel_name:chat_id[:topic_id]` → `thread_id` (keys are `channel:chat` for root conversations and `channel:chat:topic` for threaded conversations)
+- `store.py` - JSON-file persistence mapping `channel_name:chat_id[:topic_id]` → `thread_id` (keys are `channel:chat` for root conversations and `channel:chat:topic` for threaded conversations). Every access to `_data` must be protected by `_lock`; `list_entries()` snapshots keys and copied entries under the lock, then formats the result after releasing it so concurrent channel threads cannot resize the dictionary during iteration without extending the critical section.
 - `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout`
   A swallowed streaming failure publishes its final outbound before releasing the inbound dedupe key, so a provider redelivery can retry without overtaking the terminal reply.
   **What may be published from the stream is an allowlist, not a denylist** (`_accumulate_stream_text`): only assistant message types — LangChain serializes `AIMessage.type` as `"ai"` and `AIMessageChunk.type` as `"AIMessageChunk"`, plus the OpenAI-style `"assistant"` spelling for foreign runtimes — become displayable text. The previous rule rejected only payloads whose `type` contained `"tool"` and therefore published everything else, which leaked DeerFlow's hidden model context to every streaming IM channel: `DynamicContextMiddleware` injects recalled memory as a hidden `HumanMessage` (`type == "human"`) and rewrites the user's own turn into a new `HumanMessage`, `DurableContextMiddleware` injects a hidden `<durable_context_data>` `HumanMessage`, and LangGraph fans those state writes out on the `messages-tuple` stream. Proved live on a Buzz relay, which published a `<memory>` fact block and, in another run, a verbatim echo of the user's own message as the assistant's reply. Matching is by prefix (`ai` / `assistant`), never substring, because ordinary words contain `"ai"` (`chain`, `domain`). The message type is resolved by `_stream_payload_type`, which handles both the `model_dump()` shape DeerFlow's own gateway emits and LangChain's `to_json()` constructor shape (whose top-level `type` is the literal `"constructor"`, with the class name at the tail of the `id` path). A bare `str` payload is no longer accepted at all: it carries no type information, so it cannot be attributed to the assistant, and nothing in DeerFlow produces one (`runtime/serialization.py::serialize_messages_tuple` always emits `[message_dict, metadata]`).

backend/app/channels/store.py

@@ -81,8 +81,9 @@ class ChannelStore:
 
     def get_thread_id(self, channel_name: str, chat_id: str, topic_id: str | None = None) -> str | None:
         """Look up the DeerFlow thread_id for a given IM conversation/topic."""
-        entry = self._data.get(self._key(channel_name, chat_id, topic_id))
+        with self._lock:
-        return entry["thread_id"] if entry else None
+            entry = self._data.get(self._key(channel_name, chat_id, topic_id))
+            return entry["thread_id"] if entry else None
 
     def set_thread_id(
         self,
@@ -138,8 +139,11 @@ class ChannelStore:
 
     def list_entries(self, channel_name: str | None = None) -> list[dict[str, Any]]:
         """List all stored mappings, optionally filtered by channel."""
+        with self._lock:
+            entries = [(key, entry.copy()) for key, entry in self._data.items()]
+
         results = []
-        for key, entry in self._data.items():
+        for key, entry in entries:
             parts = key.split(":", 2)
             ch = parts[0]
             chat = parts[1] if len(parts) > 1 else ""

backend/tests/test_channels.py

@@ -7,7 +7,7 @@ import json
 import logging
 import tempfile
 import threading
-from concurrent.futures import Future
+from concurrent.futures import Future, ThreadPoolExecutor
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -264,6 +264,47 @@ class TestChannelStore:
         assert len(entries) == 1
         assert entries[0]["channel_name"] == "slack"
 
+    def test_channel_store_concurrent_list_and_mutation(self, store, monkeypatch):
+        iteration_started = threading.Event()
+        mutation_requested = threading.Event()
+        mutation_finished = threading.Event()
+
+        class CoordinatedData(dict):
+            def items(self):
+                iterator = iter(super().items())
+                first = next(iterator)
+                iteration_started.set()
+
+                if store._lock.locked():
+                    assert mutation_requested.wait(timeout=5), "mutation thread never requested the store lock"
+                else:
+                    assert mutation_finished.wait(timeout=5), "mutation thread never changed the unlocked store"
+
+                yield first
+                yield from iterator
+
+        store._data = CoordinatedData(
+            {
+                "slack:ch1": {"thread_id": "t1", "user_id": "u1", "created_at": 1.0, "updated_at": 1.0},
+                "feishu:ch2": {"thread_id": "t2", "user_id": "u2", "created_at": 2.0, "updated_at": 2.0},
+            }
+        )
+        monkeypatch.setattr(store, "_save", lambda: None)
+
+        def mutate():
+            assert iteration_started.wait(timeout=5), "list_entries never started iterating"
+            mutation_requested.set()
+            store.set_thread_id("test", "new", "t3")
+            mutation_finished.set()
+
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            list_future = executor.submit(store.list_entries)
+            mutation_future = executor.submit(mutate)
+            mutation_future.result(timeout=5)
+            entries = list_future.result(timeout=5)
+
+        assert {(entry["channel_name"], entry["chat_id"]) for entry in entries} == {("slack", "ch1"), ("feishu", "ch2")}
+
     def test_persistence(self, tmp_path):
         path = tmp_path / "store.json"
         store1 = ChannelStore(path=path)