furyhawk/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-20 09:45:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(channels): key inbound dedupe on chat-scoped workspaces (#4287)

Browse Source 
* fix(channels): key inbound dedupe on chat-scoped workspaces

* fix(channels): release the inbound dedupe key on swallowed transient failures

_release_inbound_dedupe_key runs from _handle_message's generic exception
handler, but three sites handle their error in place and never re-raise, so
the key recorded on receipt survives the full dedupe TTL and the provider's
redelivery -- the retry that would have recovered the failure -- is dropped:

- _handle_streaming_chat swallows every stream error into its finally block
- the fire-and-forget runs.create busy branch
- the non-streaming runs.wait busy branch

Keying chat-scoped workspaces gives telegram/feishu/wechat/dingtalk a dedupe
key for the first time, which newly exposes them to this. The mechanism is
pre-existing, not introduced here: WeCom already had a key (streaming plus
aibotid) and shows the same black hole on main.

Also parametrize the dispatch dedupe test over all three chat-scoped
providers so the streaming path is covered end-to-end, and restate the
workspace-fallback comment in terms of what the chain actually guarantees --
both fallbacks are appended last and gated on every earlier source being
absent, so they can only turn "no key" into a key, never change one.

* fix(channels): publish final reply before dedupe release

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Aari
2026-07-19 20:01:52 +08:00
committed by GitHub
co-authored by Willem Jiang
parent d2f8f61e3a
commit 5b65d543b1
4 changed files with 390 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+7 -3
View File
@@ -289,9 +289,12 @@ This section accumulates work toward the **2.1.0** milestone
codes and don't treat a bare "connect" as a bind command; stop Feishu from
creating thread topics and throttle card updates; let the UI runtime channel
config win over `config.yaml`; fix `require_mention` gating on
whitespace-only `bot_login` / `mention_login`; and guard null quote fields in
WeCom. ([#4100], [#4104], [#4131], [#4129], [#3753], [#4229], [#4222], [#4251],
[#3810], [#3674], [#4055], [#4069])
whitespace-only `bot_login` / `mention_login`; guard null quote fields in
WeCom; and key inbound dedupe on chat-scoped workspaces so Telegram, Feishu,
WeChat and DingTalk redeliveries stop re-running the agent on a default
(unbound) configuration, releasing the dedupe key on transient failures so a
redelivery can still recover. ([#4100], [#4104], [#4131], [#4129], [#3753],
[#4229], [#4222], [#4251], [#3810], [#3674], [#4055], [#4069], [#4287])
- **frontend:** Preserve messages and durable context across summarization;
preserve artifacts and stabilize artifact paths during streaming; resolve
relative artifact image paths; retain presented artifacts in the header
@@ -1094,4 +1097,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
[#4246]: https://github.com/bytedance/deer-flow/pull/4246
[#4251]: https://github.com/bytedance/deer-flow/pull/4251
[#4264]: https://github.com/bytedance/deer-flow/pull/4264
[#4287]: https://github.com/bytedance/deer-flow/pull/4287
[#4288]: https://github.com/bytedance/deer-flow/pull/4288
backend/AGENTS.md
+1
View File
@@ -511,6 +511,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
- `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)
- `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.
- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle)
- `service.py` - Manages lifecycle of all configured channels from `config.yaml`
- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` registers the "Working on it..." placeholder as the stream target and edits it in place via `editMessageText`; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured)
backend/app/channels/manager.py
+27 -1
View File
@@ -78,6 +78,11 @@ INBOUND_DEDUPE_MAX_ENTRIES = 4096
# client_id) are not guaranteed identical across a provider's own redelivery, so
# keying dedupe on them would miss exactly the retries we want to absorb.
INBOUND_DEDUPE_METADATA_KEYS = ("event_id", "message_id", "msg_id")
# Providers that persist connection.workspace_id = chat_id (telegram / feishu /
# wechat upsert_connection). Unbound inbound has no connection, so msg.workspace_id
# is unset; chat_id is still the tenant scope and is safe for the dedupe key.
# Slack is intentionally excluded: its channel ids are not globally unique.
CHAT_SCOPED_WORKSPACE_CHANNELS = frozenset({"telegram", "feishu", "wechat"})
CHANNEL_CAPABILITIES = {
"dingtalk": {"supports_streaming": False},
@@ -1170,7 +1175,15 @@ class ChannelManager:
# Fail closed: without a workspace/team/guild identifier we cannot tell two
# workspaces apart (e.g. Slack channel ids are not globally unique), so
# skip dedupe rather than risk collapsing distinct workspaces' messages.
workspace_id = msg.workspace_id or metadata.get("workspace_id") or metadata.get("team_id") or metadata.get("guild_id") or metadata.get("aibotid")
# Both fallbacks are appended last and gated on every earlier source being
# absent, so they can only turn "no key" into a key — never change one.
# A conversation_id not reused across a provider's own redelivery would
# degrade to today's no-dedupe behaviour, never collapse two conversations
# (chat_id and message_id stay in the tuple). conversation_id covers
# DingTalk (group + P2P); chat-scoped providers fall back to chat_id.
workspace_id = msg.workspace_id or metadata.get("workspace_id") or metadata.get("team_id") or metadata.get("guild_id") or metadata.get("aibotid") or metadata.get("conversation_id")
if not workspace_id and msg.channel_name in CHAT_SCOPED_WORKSPACE_CHANNELS:
workspace_id = msg.chat_id or None
if not workspace_id:
return None
return (msg.channel_name, str(workspace_id), msg.chat_id, message_id)
@@ -1632,6 +1645,10 @@ class ChannelManager:
except Exception as exc:
if _is_thread_busy_error(exc):
logger.warning("[Manager] thread busy (concurrent run rejected): thread_id=%s", thread_id)
# Swallowed like the generic handler would not be: release the
# key so the provider's redelivery can retry once the thread
# frees, instead of being dropped for the dedupe TTL.
self._release_inbound_dedupe_key(msg)
await self._send_error(msg, THREAD_BUSY_MESSAGE)
return
raise
@@ -1647,6 +1664,9 @@ class ChannelManager:
except Exception as exc:
if _is_thread_busy_error(exc):
logger.warning("[Manager] thread busy (concurrent run rejected): thread_id=%s", thread_id)
# Same reason as the fire-and-forget branch above: this error is
# handled here rather than re-raised, so release explicitly.
self._release_inbound_dedupe_key(msg)
await self._send_error(msg, THREAD_BUSY_MESSAGE)
return
else:
@@ -1818,6 +1838,12 @@ class ChannelManager:
metadata=_response_metadata(msg.metadata, pending_clarification=pending_clarification),
)
)
if stream_error is not None:
# This path swallows its own errors, so _handle_message's generic
# handler never runs and never releases the key. Release only
# after publishing the final outbound so a provider redelivery
# cannot overtake this attempt's terminal reply.
self._release_inbound_dedupe_key(msg)
# -- command handling --------------------------------------------------
backend/tests/test_channels.py
+355
View File
@@ -629,6 +629,20 @@ def _make_stream_part(event: str, data):
return SimpleNamespace(event=event, data=data)
def _ok_stream_events():
"""Minimal successful streaming run: one text chunk plus a final values frame."""
return [
_make_stream_part(
"messages-tuple",
[{"id": "ai-1", "content": "Hello", "type": "AIMessageChunk"}, {"langgraph_node": "agent"}],
),
_make_stream_part(
"values",
{"messages": [{"type": "human", "content": "hi"}, {"type": "ai", "content": "Hello"}], "artifacts": []},
),
]
def _make_async_iterator(items):
async def iterator():
for item in items:
@@ -1058,6 +1072,347 @@ class TestChannelManager:
)
assert ChannelManager._inbound_dedupe_key(without_workspace) is None
def test_inbound_dedupe_key_uses_chat_id_for_chat_scoped_providers_when_unbound(self):
"""Unbound telegram/feishu/wechat must still form a dedupe key via chat_id.
Those adapters persist connection.workspace_id = chat_id, but
attach_connection_identity only sets msg.workspace_id when a connection
exists. Provider redeliveries on unbound (or not-yet-bound) chats would
otherwise skip the entire inbound dedupe path and run the agent N times.
"""
from app.channels.manager import ChannelManager
for channel, chat_id, message_id in (
("telegram", "12345", "42"),
("feishu", "oc_abc", "om_1"),
("wechat", "wx_user_1", "m1"),
):
unbound = InboundMessage(
channel_name=channel,
chat_id=chat_id,
user_id="u1",
text="hi",
metadata={"message_id": message_id},
)
assert ChannelManager._inbound_dedupe_key(unbound) == (channel, chat_id, chat_id, message_id)
# Bound shape (workspace already on the message) must keep the same key
# so bound and unbound redeliveries of the same chat share the cache.
bound = InboundMessage(
channel_name=channel,
chat_id=chat_id,
user_id="u1",
text="hi",
workspace_id=chat_id,
metadata={"message_id": message_id},
)
assert ChannelManager._inbound_dedupe_key(bound) == (channel, chat_id, chat_id, message_id)
def test_inbound_dedupe_key_uses_dingtalk_conversation_id_when_unbound(self):
"""DingTalk stamps conversation_id on every inbound; use it when unbound.
Group connections store workspace_id=conversation_id; P2P stores None.
Without a metadata fallback, unbound groups and all P2P traffic skipped
dedupe entirely (including bound P2P, whose connection.workspace_id is
None). conversation_id is already on the message and is the natural
tenant scope — same role as Slack team_id / Discord guild_id.
"""
from app.channels.manager import ChannelManager
group_unbound = InboundMessage(
channel_name="dingtalk",
chat_id="cid123",
user_id="staff1",
text="hi",
metadata={
"conversation_type": "2",
"conversation_id": "cid123",
"message_id": "mid1",
},
)
assert ChannelManager._inbound_dedupe_key(group_unbound) == ("dingtalk", "cid123", "cid123", "mid1")
p2p = InboundMessage(
channel_name="dingtalk",
chat_id="staff1",
user_id="staff1",
text="hi",
# Bound P2P still has workspace_id=None on the connection record.
connection_id="conn1",
owner_user_id="owner1",
workspace_id=None,
metadata={
"conversation_type": "1",
"conversation_id": "cid_p2p",
"message_id": "mid1",
},
)
assert ChannelManager._inbound_dedupe_key(p2p) == ("dingtalk", "cid_p2p", "staff1", "mid1")
def test_inbound_dedupe_chat_scoped_fallback_does_not_collapse_distinct_chats(self):
"""newly_missed guard: chat_id fallback must not cross-dedupe two chats.
Same stable message_id string in two different chats is legitimate and
must produce distinct keys (message_ids are only unique per chat on
Telegram/Feishu/WeChat).
"""
from app.channels.manager import ChannelManager
a = InboundMessage(
channel_name="telegram",
chat_id="111",
user_id="u1",
text="hi",
metadata={"message_id": "42"},
)
b = InboundMessage(
channel_name="telegram",
chat_id="222",
user_id="u2",
text="hi",
metadata={"message_id": "42"},
)
assert ChannelManager._inbound_dedupe_key(a) == ("telegram", "111", "111", "42")
assert ChannelManager._inbound_dedupe_key(b) == ("telegram", "222", "222", "42")
assert ChannelManager._inbound_dedupe_key(a) != ChannelManager._inbound_dedupe_key(b)
@pytest.mark.parametrize(
("channel", "chat_id"),
(
("wechat", "wx_user_1"),
("telegram", "12345"),
("feishu", "oc_abc"),
),
)
def test_dispatch_loop_dedupes_unbound_chat_scoped_redelivery(self, tmp_path, monkeypatch, channel, chat_id):
"""Provider redelivery of an unbound chat-scoped message runs the agent once.
Shaped like wechat.py / telegram.py inbound metadata (message_id only, no
workspace_id / team_id) before attach_connection_identity finds a binding.
Parametrized across all three CHAT_SCOPED_WORKSPACE_CHANNELS so the
streaming dispatch path (telegram/feishu) is covered end-to-end too, not
only WeChat's runs.wait path.
"""
monkeypatch.setattr("app.channels.manager.STREAM_UPDATE_MIN_INTERVAL_SECONDS", 0.0)
from app.channels.manager import ChannelManager
streaming = ChannelManager._channel_supports_streaming(channel)
async def go():
bus = MessageBus()
store = ChannelStore(path=tmp_path / "store.json")
manager = ChannelManager(bus=bus, store=store)
manager._client = _make_mock_langgraph_client()
manager._client.runs.stream = MagicMock(side_effect=lambda *a, **kw: _make_async_iterator(_ok_stream_events()))
outbound_received: list[OutboundMessage] = []
async def capture_outbound(msg: OutboundMessage) -> None:
outbound_received.append(msg)
bus.subscribe_outbound(capture_outbound)
await manager.start()
# The mock the channel's dispatch path actually drives.
run_call = manager._client.runs.stream if streaming else manager._client.runs.wait
def _inbound(message_id: str) -> InboundMessage:
return InboundMessage(
channel_name=channel,
chat_id=chat_id,
user_id="u1",
text=f"hello from {channel}",
metadata={"message_id": message_id},
)
await bus.publish_inbound(_inbound("m-1"))
await bus.publish_inbound(_inbound("m-1"))
await _wait_for(lambda: run_call.call_count == 1 and any(m.is_final for m in outbound_received))
await asyncio.sleep(0.05)
assert run_call.call_count == 1
# Distinct message_id still processes (negative control / newly_missed).
await bus.publish_inbound(_inbound("m-2"))
await _wait_for(lambda: run_call.call_count == 2)
await asyncio.sleep(0.05)
await manager.stop()
assert run_call.call_count == 2
_run(go())
def test_streaming_transient_failure_releases_dedupe_key(self, tmp_path, monkeypatch):
"""Release a swallowed streaming error only after its final outbound.
_release_inbound_dedupe_key lives in _handle_message's `except Exception`
handler, but _handle_streaming_chat handles its own errors and never
re-raises — so without an explicit release the key recorded on receipt
survives the full dedupe TTL and the provider's redelivery (the retry
that would recover the failure) is silently dropped. Releasing before
the final outbound would let that retry overtake the terminal reply.
"""
monkeypatch.setattr("app.channels.manager.STREAM_UPDATE_MIN_INTERVAL_SECONDS", 0.0)
from app.channels.manager import ChannelManager
async def go():
bus = MessageBus()
store = ChannelStore(path=tmp_path / "store.json")
manager = ChannelManager(bus=bus, store=store)
outbound_received: list[OutboundMessage] = []
key_present_during_final_publish: list[bool] = []
async def capture_outbound(msg: OutboundMessage) -> None:
outbound_received.append(msg)
if msg.is_final:
key = manager._inbound_dedupe_key(_inbound())
key_present_during_final_publish.append(key in manager._recent_inbound_events)
bus.subscribe_outbound(capture_outbound)
def _failing_stream(*args, **kwargs):
async def gen():
yield _make_stream_part(
"messages-tuple",
[{"id": "ai-1", "content": "Partial", "type": "AIMessageChunk"}, {"langgraph_node": "agent"}],
)
raise ConnectionError("stream broken")
return gen()
manager._client = _make_mock_langgraph_client()
manager._client.runs.stream = MagicMock(side_effect=_failing_stream)
await manager.start()
def _inbound() -> InboundMessage:
return InboundMessage(
channel_name="feishu",
chat_id="chat1",
user_id="u1",
text="hi",
metadata={"message_id": "m-1"},
)
await bus.publish_inbound(_inbound())
await _wait_for(lambda: any(m.is_final for m in outbound_received))
await asyncio.sleep(0.05)
assert manager._client.runs.stream.call_count == 1
assert key_present_during_final_publish == [True]
# The provider redelivers the same message after the failure.
await bus.publish_inbound(_inbound())
await _wait_for(lambda: manager._client.runs.stream.call_count == 2)
await manager.stop()
assert manager._client.runs.stream.call_count == 2
_run(go())
def test_thread_busy_releases_dedupe_key(self, tmp_path):
"""A busy thread is transient, so its redelivery must stay reprocessable.
runs.wait's ConflictError is handled in place (busy message, no re-raise),
so it bypasses _handle_message's release just like the streaming path.
"""
import httpx
from langgraph_sdk.errors import ConflictError
from app.channels.manager import THREAD_BUSY_MESSAGE, ChannelManager
async def go():
bus = MessageBus()
store = ChannelStore(path=tmp_path / "store.json")
manager = ChannelManager(bus=bus, store=store)
outbound_received: list[OutboundMessage] = []
async def capture_outbound(msg: OutboundMessage) -> None:
outbound_received.append(msg)
bus.subscribe_outbound(capture_outbound)
request = httpx.Request("POST", "http://127.0.0.1:2024/threads/t/runs")
conflict = ConflictError(
"Thread is already running a task.",
response=httpx.Response(409, request=request),
body={"message": "Thread is already running a task."},
)
manager._client = _make_mock_langgraph_client()
manager._client.runs.wait = AsyncMock(side_effect=conflict)
await manager.start()
def _inbound() -> InboundMessage:
return InboundMessage(
channel_name="wechat",
chat_id="wx_user_1",
user_id="wx_user_1",
text="hi",
metadata={"message_id": "m-1"},
)
await bus.publish_inbound(_inbound())
await _wait_for(lambda: any(m.text == THREAD_BUSY_MESSAGE for m in outbound_received))
await asyncio.sleep(0.05)
assert manager._client.runs.wait.call_count == 1
await bus.publish_inbound(_inbound())
await _wait_for(lambda: manager._client.runs.wait.call_count == 2)
await manager.stop()
assert manager._client.runs.wait.call_count == 2
_run(go())
def test_fire_and_forget_thread_busy_releases_dedupe_key(self, tmp_path):
"""Same invariant on the third swallow site: runs.create's busy branch."""
import httpx
from langgraph_sdk.errors import ConflictError
import app.gateway.github.run_policy # noqa: F401 — register policy
from app.channels.manager import THREAD_BUSY_MESSAGE, ChannelManager
async def go():
bus = MessageBus()
store = ChannelStore(path=tmp_path / "store.json")
manager = ChannelManager(bus=bus, store=store)
outbound_received: list[OutboundMessage] = []
async def capture_outbound(msg: OutboundMessage) -> None:
outbound_received.append(msg)
bus.subscribe_outbound(capture_outbound)
request = httpx.Request("POST", "http://127.0.0.1:2024/threads/t/runs")
conflict = ConflictError(
"Thread is already running a task.",
response=httpx.Response(409, request=request),
body={"message": "Thread is already running a task."},
)
manager._client = _make_mock_langgraph_client()
manager._client.runs.create = AsyncMock(side_effect=conflict)
await manager.start()
def _inbound() -> InboundMessage:
return InboundMessage(
channel_name="github",
chat_id="owner/repo",
user_id="dev",
owner_user_id="agent-owner-1",
workspace_id="owner/repo",
text="hi",
metadata={"message_id": "delivery-1:dev:agent"},
)
await bus.publish_inbound(_inbound())
await _wait_for(lambda: any(m.text == THREAD_BUSY_MESSAGE for m in outbound_received))
await asyncio.sleep(0.05)
assert manager._client.runs.create.call_count == 1
await bus.publish_inbound(_inbound())
await _wait_for(lambda: manager._client.runs.create.call_count == 2)
await manager.stop()
assert manager._client.runs.create.call_count == 2
_run(go())
def test_github_redelivery_is_deduped_like_other_channels(self, tmp_path):
"""A redelivered GitHub webhook must dispatch the agent only once.