From adfc307677935b44915cad301e673ef79f589c76 Mon Sep 17 00:00:00 2001 From: cui fliter Date: Sat, 29 Aug 2026 08:01:38 +0800 Subject: [PATCH] fix(channels): synchronize ChannelStore reads (#5083) Signed-off-by: cuishuang --- backend/app/channels/AGENTS.md | 2 +- backend/app/channels/store.py | 10 +++++--- backend/tests/test_channels.py | 43 +++++++++++++++++++++++++++++++++- 3 files changed, 50 insertions(+), 5 deletions(-) diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index f1b233c23..b64158811 100644 --- a/backend/app/channels/AGENTS.md +++ b/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 `` `HumanMessage`, and LangGraph fans those state writes out on the `messages-tuple` stream. Proved live on a Buzz relay, which published a `` 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]`). diff --git a/backend/app/channels/store.py b/backend/app/channels/store.py index 81f5d61e4..83e0d22ee 100644 --- a/backend/app/channels/store.py +++ b/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)) - return entry["thread_id"] if entry else None + with self._lock: + 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 "" diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index 5a21ff491..99b444f01 100644 --- a/backend/tests/test_channels.py +++ b/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)