fix(channels): follow Buzz threads mapped in the connection repository (#6232)

* fix(channels): follow Buzz threads mapped in the connection repository

With channel_connections enabled, ChannelManager stores a bound author's
thread mapping only in the connection repository, but Buzz's thread-follow
gate read only the JSON ChannelStore. engaged_thread was therefore always
False for authors bound with /connect, and every unmentioned reply in a
thread the bot was already answering was silently dropped.

Buzz now resolves the connection identity before the mention gate and looks
the thread up through lookup_thread_id, a helper extracted from
ChannelManager._lookup_thread_id that both sides now share, so the gate and
the manager can no longer disagree about which store owns a mapping.

* fix(channels): drop the manager's own JSON-store read in the skill whitelist

_resolve_available_skill_names fell back to reading the JSON ChannelStore
when the caller's lookup_thread_id found nothing. For a bound message that
lookup is repo-only, so a legacy unbound mapping for the same chat decided
which custom agent's skill whitelist applied. The parameter is now required
and the caller passes "" when nothing is mapped.

Also pin the Buzz thread-follow fix against the real SQL connection
repository (bound through a real /connect) instead of a fake.

* docs(changelog): reference #6232 for the Buzz thread-follow fix

* fix(channels): log Buzz drops of unmentioned replies with no engaged thread

The thread-follow gate was the one drop in _handle_chat_event without a
log line. Log it at DEBUG with the thread root and whether the author was
bound, matching the allowlist drop, so a report that the bot stopped
following replies can be told apart from a lost bind or missing mapping.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -466,6 +466,18 @@ This release closes that milestone with **301 merged pull requests**.
 
 ### Fixed
 
+- **channels:** Buzz now follows a thread without a fresh mention for authors
+  bound with `/connect`. With `channel_connections.enabled`, the manager maps a
+  bound author's threads only in the connection repository, but Buzz's
+  thread-follow gate read only the JSON channel store, so every unmentioned
+  reply in a thread the bot was already answering was silently dropped. Buzz
+  now resolves the connection before the mention gate and looks the thread up
+  through the same helper as the manager (`lookup_thread_id`), so a bound
+  author's engaged thread is the one the manager will reuse, and a legacy JSON
+  mapping no longer counts for that author. The manager's slash-skill whitelist
+  check had its own copy of the same read and is fixed with it: a bound user
+  with no thread yet was checked against the agent of a legacy JSON thread for
+  the same chat and could be told an enabled skill was not available. ([#6232])
 - **memory:** Reading DeerMem agent memory no longer fails while another write
   deletes a fact. `load()`, `reload()`, and the full `rebuild_index()` scan list
   the fact files without the storage locks, so a delete committed between the
@@ -7696,5 +7708,6 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6214]: https://github.com/bytedance/deer-flow/pull/6214
 [#6229]: https://github.com/bytedance/deer-flow/pull/6229
 [#6230]: https://github.com/bytedance/deer-flow/pull/6230
+[#6232]: https://github.com/bytedance/deer-flow/pull/6232
 [#6238]: https://github.com/bytedance/deer-flow/pull/6238
 [#6255]: https://github.com/bytedance/deer-flow/pull/6255

CHANGELOG_zh.md

@@ -423,6 +423,14 @@
 
 ### 修复
 
+- **渠道：** 通过 `/connect` 绑定的 Buzz 作者在已参与的话题中回复时，无需再次提及
+  机器人。开启 `channel_connections.enabled` 后，管理器只在连接仓库中记录已绑定作者
+  的话题映射，而 Buzz 的话题跟随判断只读取 JSON 渠道存储，导致机器人正在回复的话题
+  中所有未提及的回复都被静默丢弃。现在 Buzz 会在提及判断之前解析连接，并通过与管理器
+  相同的辅助函数（`lookup_thread_id`）查找话题，因此已绑定作者的已参与话题就是管理器
+  将复用的话题，旧的 JSON 映射也不再对该作者生效。管理器的斜杠技能白名单检查中有同一读取逻辑
+  的副本，也一并修复：尚无话题的已绑定用户此前会按同一会话旧 JSON 话题的智能体进行检查，
+  可能被告知已启用的技能不可用。([#6232])
 - **记忆：** 读取 DeerMem 智能体记忆时，不再因另一写入同时删除事实而失败。
   `load()`、`reload()` 与全量 `rebuild_index()` 扫描在不持有存储锁的情况下列
   出事实文件，若删除恰好在列出之后、打开文件之前提交，就会对完好的数据抛出
@@ -6426,5 +6434,6 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6214]: https://github.com/bytedance/deer-flow/pull/6214
 [#6229]: https://github.com/bytedance/deer-flow/pull/6229
 [#6230]: https://github.com/bytedance/deer-flow/pull/6230
+[#6232]: https://github.com/bytedance/deer-flow/pull/6232
 [#6238]: https://github.com/bytedance/deer-flow/pull/6238
 [#6255]: https://github.com/bytedance/deer-flow/pull/6255

backend/app/channels/buzz.py

@@ -52,7 +52,7 @@ from app.channels import buzz_nostr
 from app.channels.base import Channel
 from app.channels.buzz_seen_events import BuzzSeenEventStore
 from app.channels.commands import is_known_channel_command
-from app.channels.connection_identity import attach_connection_identity
+from app.channels.connection_identity import attach_connection_identity, lookup_thread_id
 from app.channels.message_bus import InboundMessage, InboundMessageType, InboundQueueFullError, MessageBus, OutboundMessage
 
 logger = logging.getLogger(__name__)
@@ -1316,12 +1316,6 @@ class BuzzChannel(Channel):
         # and returned above, so this gate only ever sees ordinary chat text.
         thread_root = self._thread_root(ev)
         mentioned = self._keys.pubkey_hex in buzz_nostr.tag_values(ev, "p")
-        store = self.config.get("channel_store")
-        engaged_thread = bool(thread_root and store is not None and store.get_thread_id(self.name, channel_id, topic_id=thread_root))
-        allowed_without_mention = (not self._require_mention) or channel_id in self._mention_free or self._is_dm(channel_id) or engaged_thread
-        if not mentioned and not allowed_without_mention:
-            return
-
         if mentioned:
             text = self._strip_own_mention(text)
 
@@ -1329,7 +1323,22 @@ class BuzzChannel(Channel):
         inbound: InboundMessage = self._make_inbound(chat_id=channel_id, user_id=author, text=text, msg_type=msg_type, thread_ts=thread_root, metadata={"event_id": str(ev.get("id", ""))})
         inbound.topic_id = thread_root
         inbound.workspace_id = self._workspace_id
+        # Resolved before the mention gate, not after: the manager maps a bound
+        # author's threads in the connection repository only, so the thread-follow
+        # check below can find them only through the resolved connection_id.
         inbound = await self._attach_connection_identity(inbound)
+
+        allowed_without_mention = (not self._require_mention) or channel_id in self._mention_free or self._is_dm(channel_id)
+        if not mentioned and not allowed_without_mention:
+            engaged_thread = bool(thread_root and await lookup_thread_id(inbound, repo=self._connection_repo, store=self.config.get("channel_store")))
+            if not engaged_thread:
+                # Logged like the allowlist drop above: "the bot stopped following my
+                # replies" must be triageable as require_mention working vs a lost
+                # bind or a missing thread mapping.
+                channel_name = self._channel_meta.get(channel_id, {}).get("name") or "<unnamed>"
+                logger.debug("[buzz] dropped unmentioned chat event in channel %s (%s): thread=%s not engaged (bound=%s)", channel_name, channel_id, thread_root, bool(inbound.connection_id))
+                return
+
         self._last_requester[(channel_id, thread_root)] = author
         await self._publish(inbound)
         # Only a fully accepted-and-published event advances the cursor, and only

backend/app/channels/connection_identity.py

@@ -42,3 +42,19 @@ async def attach_connection_identity(
         return inbound
 
     return inbound
+
+
+async def lookup_thread_id(msg: InboundMessage, *, repo: Any, store: Any) -> str | None:
+    """Return the DeerFlow thread mapped to *msg*'s conversation/topic, if any.
+
+    A message carrying a resolved connection is mapped in the connection repository
+    and nowhere else (``ChannelManager._store_thread_id`` writes only there); every
+    other message is mapped in the JSON ``ChannelStore``. Adapters that gate on "is
+    this thread already ours?" must use this rather than reading the store directly,
+    or bound users' threads look unengaged.
+    """
+    if msg.connection_id and repo is not None:
+        return await repo.get_thread_id(msg.connection_id, msg.chat_id, msg.topic_id)
+    if store is None:
+        return None
+    return store.get_thread_id(msg.channel_name, msg.chat_id, topic_id=msg.topic_id)

backend/app/channels/manager.py

@@ -24,6 +24,7 @@ from langgraph_sdk.errors import ConflictError
 from app.channels import buzz_run_policy as _buzz_run_policy  # noqa: F401
 from app.channels import feishu_run_policy as _feishu_run_policy  # noqa: F401
 from app.channels.commands import KNOWN_CHANNEL_COMMANDS
+from app.channels.connection_identity import lookup_thread_id
 from app.channels.dedupe_store import InboundDedupeStore, MemoryInboundDedupeStore
 from app.channels.message_bus import (
     INBOUND_FILE_CONTENT_KEY,
@@ -1781,13 +1782,9 @@ class ChannelManager:
                 )
         return policy
 
-    def _resolve_available_skill_names(
-        self,
-        msg: InboundMessage,
-        thread_id: str | None = None,
-    ) -> set[str] | None:
-        if thread_id is None:
-            thread_id = self.store.get_thread_id(msg.channel_name, msg.chat_id, topic_id=msg.topic_id) or ""
+    def _resolve_available_skill_names(self, msg: InboundMessage, thread_id: str) -> set[str] | None:
+        """*thread_id* comes from ``_lookup_thread_id`` (``""`` when unmapped); never
+        re-read the JSON store here, which holds no mapping for bound messages."""
         _, _, run_context = self._resolve_run_params(msg, thread_id)
         if run_context.get("is_bootstrap"):
             return {"bootstrap"}
@@ -2199,13 +2196,7 @@ class ChannelManager:
         await self.bus.publish_outbound(outbound)
 
     async def _lookup_thread_id(self, msg: InboundMessage) -> str | None:
-        if msg.connection_id and self._connection_repo is not None:
-            return await self._connection_repo.get_thread_id(
-                msg.connection_id,
-                msg.chat_id,
-                msg.topic_id,
-            )
-        return self.store.get_thread_id(msg.channel_name, msg.chat_id, topic_id=msg.topic_id)
+        return await lookup_thread_id(msg, repo=self._connection_repo, store=self.store)
 
     async def _store_thread_id(self, msg: InboundMessage, thread_id: str) -> None:
         if msg.connection_id and msg.owner_user_id and self._connection_repo is not None:
@@ -2807,7 +2798,7 @@ class ChannelManager:
             slash_resolution = await asyncio.to_thread(
                 lambda: _resolve_slash_skill_command(
                     raw_text,
-                    self._resolve_available_skill_names(msg, thread_id),
+                    self._resolve_available_skill_names(msg, thread_id or ""),
                     self._get_skill_storage,
                 )
             )

backend/tests/test_buzz_channel.py

@@ -325,6 +325,9 @@ class FakeConnectionRepo:
         self.lookups.append({"provider": provider, "external_account_id": external_account_id, "workspace_id": workspace_id})
         return self._connections.get((provider, external_account_id, workspace_id))
 
+    async def get_thread_id(self, connection_id, external_conversation_id, external_topic_id=None):
+        return None  # no thread mapped yet; the real-repository round trip is pinned separately
+
 
 def test_connect_code_binds_and_never_publishes_even_for_unauthorized_author():
     """FINDING 1 (Critical): a valid /connect code from a non-allowlisted pubkey
@@ -930,6 +933,82 @@ def test_unbound_pubkey_inbound_message_carries_no_connection_identity():
     assert [lookup["workspace_id"] for lookup in repo.lookups] == ["buzz.example.com"]
 
 
+# -- Thread-follow gate must read the mapping from where the manager wrote it ------
+
+
+@pytest.fixture
+async def sql_connection_repo(tmp_path):
+    from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelCredentialCipher
+    from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
+
+    await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'channels.db'}", sqlite_dir=str(tmp_path))
+    try:
+        yield ChannelConnectionRepository(get_session_factory(), cipher=ChannelCredentialCipher.from_key("test-encryption-key"))
+    finally:
+        await close_engine()
+
+
+@pytest.mark.anyio
+async def test_bound_pubkey_unmentioned_reply_follows_a_thread_the_manager_mapped_in_the_repo(sql_connection_repo, tmp_path):
+    """With channel connections enabled, ``ChannelManager._store_thread_id`` writes a
+    bound author's thread mapping to the connection repository *only*. The gate used
+    to read just the JSON ``ChannelStore``, so ``engaged_thread`` was always False for
+    bound authors and their unmentioned follow-ups were silently dropped. Driven
+    through the real manager, JSON store, and SQL repository (bound via a real
+    ``/connect``) so the write and read sides cannot drift apart."""
+    from datetime import UTC, datetime, timedelta
+
+    from app.channels.manager import ChannelManager
+    from app.channels.store import ChannelStore
+
+    repo = sql_connection_repo
+    await repo.create_oauth_state(owner_user_id="owner-bound", provider="buzz", state="tok-bind", expires_at=datetime.now(UTC) + timedelta(minutes=10))
+    store = ChannelStore(tmp_path / "store.json")
+    ch, captured = _started(connection_repo=repo, channel_store=store)
+    ch._transport = FakeTransport()
+    manager = ChannelManager(bus=ch.bus, store=store, connection_repo=repo)
+    root = "aa" * 32
+
+    async def dispatch(ev):
+        await ch.handle_relay_frame(json.dumps(["EVENT", "sub1", ev]))
+
+    await dispatch(_event(sk=SK_OWNER, content="/connect tok-bind", mentions=()))
+    await dispatch(_event(sk=SK_OWNER, reply_to=root, created_at=1700000200))
+    assert len(captured) == 1 and captured[0].owner_user_id == "owner-bound"
+    connection_id = captured[0].connection_id
+    assert connection_id
+    await manager._store_thread_id(captured[0], "thread-1")
+    assert store.get_thread_id("buzz", CHANNEL, topic_id=root) is None  # the manager wrote the repo only
+
+    await dispatch(_event(sk=SK_OWNER, content="follow-up", mentions=(), reply_to=root, created_at=1700000300))
+    assert len(captured) == 2
+    assert captured[1].text == "follow-up" and captured[1].topic_id == root
+    assert captured[1].connection_id == connection_id
+    assert await manager._lookup_thread_id(captured[1]) == "thread-1"
+
+
+def test_bound_pubkey_thread_follow_ignores_a_json_mapping_the_manager_would_not_read(caplog):
+    """The gate resolves the mapping exactly as ``ChannelManager._lookup_thread_id``
+    does: for a bound author that is the repository alone. A legacy JSON-store row for
+    the same topic (e.g. written before the pubkey was bound) is not this connection's
+    thread, so it must not relax ``require_mention``. The drop is logged at DEBUG with
+    the bind state, so "the bot stopped following my replies" stays triageable."""
+
+    class JsonStoreWithMapping:
+        def get_thread_id(self, channel_name, chat_id, topic_id=None):
+            return "legacy-thread"
+
+    repo = FakeConnectionRepo(states={"tok-bind": "owner-bound"})
+    ch, captured = _started(connection_repo=repo, channel_store=JsonStoreWithMapping())
+    ch._transport = FakeTransport()
+
+    _dispatch(ch, _event(sk=SK_OWNER, content="/connect tok-bind", mentions=()))
+    with caplog.at_level(logging.DEBUG, logger="app.channels.buzz"):
+        _dispatch(ch, _event(sk=SK_OWNER, content="follow-up", mentions=(), reply_to="aa" * 32))
+    assert captured == []
+    assert f"dropped unmentioned chat event in channel <unnamed> ({CHANNEL}): thread={'aa' * 32} not engaged (bound=True)" in caplog.text
+
+
 # -- FINAL REVIEW FINDING 3 (Important): oversize streaming re-posted the tail ------
 
 

backend/tests/test_channels.py

@@ -3128,11 +3128,71 @@ class TestChannelManager:
 
         expected_owner = manager._resolve_run_params(msg, "")[2].get("user_id")
 
-        manager._resolve_available_skill_names(msg)
+        manager._resolve_available_skill_names(msg, "")
 
         assert expected_owner and expected_owner != "default"
         assert captured["user_id"] == expected_owner
 
+    def test_slash_skill_whitelist_for_a_bound_message_ignores_the_json_store_mapping(self, monkeypatch, tmp_path):
+        """A bound message's thread lives in the connection repository only
+        (``lookup_thread_id``). When the repository has no mapping yet, the whitelist
+        pre-check used to fall back to reading the JSON ``ChannelStore`` itself, so a
+        legacy unbound mapping for the same chat — here a thread pinned to a
+        frontend-only custom agent — decided which skills the bound user could run.
+        """
+        from app.channels.manager import ChannelManager
+
+        loaded_agents: list[str] = []
+
+        def spy_load_agent_config(name, *, user_id=None):
+            loaded_agents.append(name)
+            return SimpleNamespace(skills=["frontend-design"])
+
+        monkeypatch.setattr("app.channels.manager.load_agent_config", spy_load_agent_config)
+
+        class EmptyConnectionRepo:
+            async def get_thread_id(self, connection_id, external_conversation_id, external_topic_id=None):
+                return None
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=tmp_path / "store.json")
+            store.set_thread_id("test", "chat1", "legacy-thread", user_id="platform-user")
+            manager = ChannelManager(bus=bus, store=store, connection_repo=EmptyConnectionRepo())
+            manager._remember_thread_agent("legacy-thread", "frontend-only")
+            manager._skill_storage = _make_channel_skill_storage([_make_channel_skill(tmp_path, "data-analysis")])
+
+            routed_to_chat: list[InboundMessage] = []
+
+            async def record_chat(msg, **kwargs):
+                routed_to_chat.append(msg)
+
+            manager._handle_chat = record_chat
+            outbound_received: list[OutboundMessage] = []
+
+            async def capture_outbound(msg):
+                outbound_received.append(msg)
+
+            bus.subscribe_outbound(capture_outbound)
+
+            await manager._handle_command(
+                InboundMessage(
+                    channel_name="test",
+                    chat_id="chat1",
+                    user_id="platform-user",
+                    connection_id="conn-1",
+                    owner_user_id="owner-alice",
+                    text="/data-analysis go",
+                    msg_type=InboundMessageType.COMMAND,
+                )
+            )
+
+            assert outbound_received == []
+            assert len(routed_to_chat) == 1
+            assert loaded_agents == []
+
+        _run(go())
+
     def test_handle_command_slash_skill_reports_disabled_skill(self, tmp_path):
         from app.channels.manager import ChannelManager