From 12fb53e5decd3733c8670e091cf34fda98054f76 Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Sun, 4 Oct 2026 21:38:43 +0900 Subject: [PATCH] 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 --- CHANGELOG.md | 13 ++++ CHANGELOG_zh.md | 9 +++ backend/app/channels/AGENTS.md | 2 +- backend/app/channels/buzz.py | 23 ++++-- backend/app/channels/connection_identity.py | 16 +++++ backend/app/channels/manager.py | 21 ++---- backend/tests/test_buzz_channel.py | 79 +++++++++++++++++++++ backend/tests/test_channels.py | 62 +++++++++++++++- 8 files changed, 201 insertions(+), 24 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8eaca55051..cc795935a7 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 8f438e6a7b..633a6f4350 100644 --- a/CHANGELOG_zh.md +++ b/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 diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index f4ed5c4e36..78855f53c7 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -14,7 +14,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk - `wecom.py` - WeCom WebSocket channel with proactive `send_notification`; `_ws_transport_up` maps disconnects to `ChannelUnavailable` without consuming retry budget. - `service.py` - Owns configured channel lifecycles. Shutdown drains manager work before transports and retains objects on timeout for retry. If a bounded Feishu/DingTalk join leaves the SDK thread alive, `ChannelStopTimeout` retains that instance and defers replacement/restart; log WARNING without traceback. The wedge still reports `is_running`/status false. - `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` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, edits the "Working on it..." stream target in place via `editMessageText`, and can optionally send final Markdown replies as Rich Messages through `channels.telegram.rich_messages` — the Rich path only fires when `rich_messages` is on **and** the text contains a rich construct (fenced code, emphasis, a table, a task list, `
`, block math, or a link), so structured command/error replies (none of these) stay plain text and their newlines and `` tokens survive; `discord.py` runs `_on_message` on discord.py's private loop, so connection-repository work (identity lookup with the reservation commit, `/connect` binding) goes to the Gateway loop through `_submit_threadsafe_coroutine` — the SQLAlchemy engine is bound there — and bind replies return through `_run_on_discord_loop`; typing registers before that hand-off settles, and a failed hand-off skips the ack and stops the indicator it started unless another message still relies on it; `discord.py` registers typing loops before inbound handling yields and `_start_typing()` refuses work once `_running` is false; each tick sends the indicator with discord.py 2.x's `await channel.typing()` (`trigger_typing()` was removed in 2.0, and each typing loop logs its first failed tick at WARNING, later ones at DEBUG); because `stop()` runs on the main loop while typing tasks belong to `_discord_loop`, normal cross-thread cancellation, awaiting, and map cleanup are scheduled there with a bounded wait, while `_run_client()` drains tasks in its `finally` block before an exception/disconnect can make that loop unusable; an already-stopped foreign loop must never have its tasks awaited from the main loop; to that end every outbound cross-loop call (`send` / `send_file` / `_get_channel_or_thread`) goes through `_run_on_discord_loop`, which bounds the await (`DISCORD_OUTBOUND_TIMEOUT_SECONDS` 30s; `DISCORD_UPLOAD_TIMEOUT_SECONDS` 120s for file uploads, whose unbounded-size payload needs room for a slow uplink plus 429 retry-after) and fails fast with a `RuntimeError` when the client loop is missing or not running — a dead client becomes a logged send failure instead of a permanently hung `ChannelManager` worker — and `is_running` reports client-thread aliveness (like `feishu.py`) so `ensure_channel_ready` can restart the channel after `_run_client()` exits on a fatal error; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured, and overrides `receive_file` to download inbound images (`picture`/`richText`) and documents (`file`) by `downloadCode` into the thread uploads bucket, mirroring `feishu.py`) -- `buzz.py` - Buzz (Nostr relay) implementation: one NIP-42-authenticated websocket, pubkey-allowlist + mention/DM/thread-follow gating, streaming replies via in-place kind-40003 edits; requires the `buzz` dependency extra. Its durable seen-event replay guard uses async `aseen()` / `arecord()` / `aflush()` boundaries: initial JSON reads and coalesced atomic writes run off the Gateway event loop, and `stop()` awaits the final flush before returning. **Subscription model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-subscription-model)): the relay fans kind-9 chat events out **only** to channel-scoped subscriptions, proved against a live relay — `REQ {"kinds":[9]}` is accepted and answered with `EOSE` but never receives an event (the connector authenticated and then sat silent forever), `REQ {"kinds":[9],"#h":[uuid]}` works, and a multi-value `#h` matches nothing, so it is strictly one REQ per channel (the same shape as Buzz's own `buzz-acp` harness). Every connection therefore rebuilds three kinds of subscription after NIP-42 auth: `buzz-discovery` (`{"kinds":[39000]}`, a historical query returning exactly the channels this identity belongs to, one stored event each then `EOSE` — adding `#p` returns zero, do not "narrow" it), `buzz-membership` (`{"kinds":[44100,44101],"#p":[us],"since":}`, the relay-signed member-added/member-removed notifications whose `p` tag names the affected member and `h` tag the channel), and one `buzz-chat-` per discovered channel. Chat subscriptions open as each kind-39000 arrives; the discovery `EOSE` is the completeness barrier that retries any that failed and warns when discovery found nothing. A kind-44100 for our pubkey subscribes to the new channel **live** (then re-issues discovery so its name/type reach the DM-detection cache, but only when that channel's metadata is actually missing — an unconditional refresh is how a burst of 44100s multiplied into one discovery pass each); a kind-44101 issues `buzz_nostr.close_frame` for exactly that channel's subscription and drops its metadata. **The membership subscription is scoped to LIVE events** and this is load-bearing: buzz-relay *stores* 44100/44101 and serves history newest-first (default limit 2000), so an unscoped filter replayed the whole membership history on every connect — every stored add read as live, re-running discovery once each (M+1 discovery passes × N stored kind-39000 events per connect, observed live as two `channel discovery complete` lines and one channel logged `` because the 44100 path subscribed before its metadata arrived), re-subscribing channels we have since been removed from, and letting a stored 44101 transiently unsubscribe a channel we are still in. `since` is anchored at the moment the socket opened (`_session_started_at`) minus `MEMBERSHIP_LOOKBACK_SECONDS` (60) of slack, which covers both relay clock skew and a membership change published *during* the connect/auth handshake; the slack can only cost an idempotent replay of the last minute. **A relay `CLOSED` frame is recovered, not merely forgotten** — every subscription on the socket fails silently when dropped, so `_handle_closed` re-issues it, bounded by `MAX_RESUBSCRIBE_ATTEMPTS` (3) per subscription id per connection *and per auth epoch* (`_resubscribe_attempts` is reset on session start, on re-auth, and by `stop()`, so pre-auth rejections — which the auth branch already recovers wholesale — never spend the authenticated session's budget). The first retry is immediate (the common case is a one-off hiccup); later ones back off 1s then 2s, awaited inline in the read loop rather than in a background task that could outlive its own socket. **`auth-required:` before this socket has completed its NIP-42 handshake is the expected bootstrap sequence, not a refusal**, and `_handle_closed` short-circuits it ahead of every recovery path: the connector opens its control REQs immediately in case the relay serves unauthenticated reads, a closed relay answers `auth-required:` plus an `AUTH` challenge, and the auth branch re-opens everything. That case is logged at DEBUG and consumes neither the permanent-refusal branch nor the retry budget (a chat subscription is still dropped from `_chat_subscriptions`, since it genuinely is not subscribed and discovery is what re-opens it). Treating it as permanent produced an operator-facing warning claiming discovery/membership tracking was DOWN in the same run where discovery then completed, every channel was subscribed, and a brand-new channel's kind-44100 was picked up one second later. The boundary is the per-socket `_auth_completed` flag, set once the signed AUTH event has been sent and cleared on session entry, session exit, and `stop()`; the same reason *after* that stays loud, and says the subscription is down until the relay's next AUTH challenge or the next reconnect rather than borrowing the non-auth wording. Then `_is_transient_close` decides whether to retry at all: NIP-01/NIP-42 `auth-required:`/`restricted:`/`blocked:`/`mute:`/`invalid:`/`pow:` prefixes and buzz-relay's own removal/revocation prose are permanent (do not fight the relay over a channel that is no longer ours; a post-auth `auth-required:` is recovered by the AUTH branch, not by re-issuing), `rate-limited:`/`error:`/no-reason-at-all and anything unrecognized are transient — the default resolves toward "keep listening" because going silently deaf is the failure this exists to remove, and the attempt budget bounds a wrong guess. A chat `CLOSED` is only ever recovered for a channel already in `_chat_subscriptions`: a `CLOSED` is relay-supplied, so acting on an unknown one would let a relay induce a subscription just by naming a channel. Every subscription that goes unlistened is logged at WARNING, never INFO. Subscription ids are deterministic per channel precisely so one can be replaced or closed without disturbing the others on the socket, and `_chat_subscriptions` is per-socket state cleared on session end, on re-auth (a pre-auth REQ may have been rejected), and by `stop()`. Three bounds on remote-fed state: `MAX_CACHED_CHANNELS` (512) caps the kind-39000 metadata cache, the watermark map, and the resubscribe-attempt map; `MAX_CHANNEL_SUBSCRIPTIONS` (256, well under buzz-relay's own 1024-per-connection ceiling) caps live chat subscriptions — at the cap new channels are refused and named in a warning rather than evicting a working subscription. **Known bound (documented, not fixed):** the relay caps historical delivery at 2000 events per subscription, newest-first, even with a `since`, so >2000 unread messages in a *single* channel across a disconnect loses the oldest — the relay never sends them and the watermark advances past them. That is the one remaining path that can skip; everything else fails toward replay. **Trust model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-trust-model)): every inbound `EVENT` is authenticated at the single `handle_relay_frame` choke point — the NIP-01 id is recomputed from the delivered payload and the BIP-340 Schnorr signature verified against the claimed `pubkey` (`buzz_nostr.verify_event`, pure and total: malformed input returns `False`, never raises) — so `ev["pubkey"]`, the authorization principal for both the allowlist and the `/connect` bind, cannot be forged by a relay the DeerFlow operator does not run. What remains trusted is the *authorship* of kind-39000 channel metadata: any member can sign one, and because per-channel subscriptions are now driven by discovery, a forged kind-39000 has two effects rather than one — it can mark a channel `type: "dm"` (relaxing `require_mention` for that channel) **and** it can induce a chat subscription for a channel of the forger's choosing, since the channels we listen to are exactly the channels we hold metadata for. Neither makes anything be *acted on*: `allowed_users` and per-event signature verification are independent gates, so an induced subscription only means the relay reads its own traffic back to a subscriber that drops it, bounded by `MAX_CHANNEL_SUBSCRIPTIONS` (which refuses rather than evicts, so it cannot displace a real channel). Same for a forged kind-44100, except its `p` tag is re-checked locally so it must at least name us. Closing this needs a configured trusted relay pubkey, which `relay_url` is not. `allowed_users` is deny-by-default (empty = nobody, unlike siblings' empty = everyone), so `start()` logs a WARNING when it is empty and each drop logs at DEBUG. The resubscribe cursor (`since`) is **per channel**, advances only for events that were actually processed, and never past `now + MAX_FUTURE_SKEW_SECONDS`, because it is peer-supplied (`created_at`) and a single future-dated event otherwise made the connector permanently deaf. Per channel rather than global is the safety-critical half: subscriptions are per channel, so one shared cursor is the newest event seen in *any* channel and a busy channel would drag it past a quiet channel's unread messages, skipping them on the next reconnect — measured on a live relay, three channels of one identity sat ~28h apart. Per-channel cursors can only ever cost duplicate delivery (absorbed by the manager's `event_id` dedupe), and an evicted cursor degrades to "no `since`", i.e. the relay's default backlog — both fail toward replay, never toward a miss. Streaming tracks every oversize chunk index (`_stream_targets` for chunk 0, `_stream_tails` for the rest), since the manager republishes cumulative text and reposting `chunks[1:]` per update flooded the channel; all of it is per-connection state cleared by `stop()`, and the remote-fed kind-39000 cache is capped at `MAX_CACHED_CHANNELS`. **`send()` refuses outright to publish text carrying a hidden model-context wrapper** (``, ``, `` — `_HIDDEN_CONTEXT_MARKERS`), logging at ERROR and clearing the stream bookkeeping on a blocked `is_final`. This is defense in depth behind the manager's allowlist, and it lives here rather than in a sibling connector because on Buzz a leak is permanent: every streaming update is an immutable public Nostr event, so a corrective edit only changes what clients render while the original leaked event stays on the relay. Matching is on the literal opening tag, so a reply that merely talks about memory is still published +- `buzz.py` - Buzz (Nostr relay) implementation: one NIP-42-authenticated websocket, pubkey-allowlist + mention/DM/thread-follow gating (thread-follow reads via the manager's `connection_identity.lookup_thread_id`, so bound authors' repo-only threads count), streaming replies via in-place kind-40003 edits; requires the `buzz` dependency extra. Its durable seen-event replay guard uses async `aseen()` / `arecord()` / `aflush()` boundaries: initial JSON reads and coalesced atomic writes run off the Gateway event loop, and `stop()` awaits the final flush before returning. **Subscription model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-subscription-model)): the relay fans kind-9 chat events out **only** to channel-scoped subscriptions, proved against a live relay — `REQ {"kinds":[9]}` is accepted and answered with `EOSE` but never receives an event (the connector authenticated and then sat silent forever), `REQ {"kinds":[9],"#h":[uuid]}` works, and a multi-value `#h` matches nothing, so it is strictly one REQ per channel (the same shape as Buzz's own `buzz-acp` harness). Every connection therefore rebuilds three kinds of subscription after NIP-42 auth: `buzz-discovery` (`{"kinds":[39000]}`, a historical query returning exactly the channels this identity belongs to, one stored event each then `EOSE` — adding `#p` returns zero, do not "narrow" it), `buzz-membership` (`{"kinds":[44100,44101],"#p":[us],"since":}`, the relay-signed member-added/member-removed notifications whose `p` tag names the affected member and `h` tag the channel), and one `buzz-chat-` per discovered channel. Chat subscriptions open as each kind-39000 arrives; the discovery `EOSE` is the completeness barrier that retries any that failed and warns when discovery found nothing. A kind-44100 for our pubkey subscribes to the new channel **live** (then re-issues discovery so its name/type reach the DM-detection cache, but only when that channel's metadata is actually missing — an unconditional refresh is how a burst of 44100s multiplied into one discovery pass each); a kind-44101 issues `buzz_nostr.close_frame` for exactly that channel's subscription and drops its metadata. **The membership subscription is scoped to LIVE events** and this is load-bearing: buzz-relay *stores* 44100/44101 and serves history newest-first (default limit 2000), so an unscoped filter replayed the whole membership history on every connect — every stored add read as live, re-running discovery once each (M+1 discovery passes × N stored kind-39000 events per connect, observed live as two `channel discovery complete` lines and one channel logged `` because the 44100 path subscribed before its metadata arrived), re-subscribing channels we have since been removed from, and letting a stored 44101 transiently unsubscribe a channel we are still in. `since` is anchored at the moment the socket opened (`_session_started_at`) minus `MEMBERSHIP_LOOKBACK_SECONDS` (60) of slack, which covers both relay clock skew and a membership change published *during* the connect/auth handshake; the slack can only cost an idempotent replay of the last minute. **A relay `CLOSED` frame is recovered, not merely forgotten** — every subscription on the socket fails silently when dropped, so `_handle_closed` re-issues it, bounded by `MAX_RESUBSCRIBE_ATTEMPTS` (3) per subscription id per connection *and per auth epoch* (`_resubscribe_attempts` is reset on session start, on re-auth, and by `stop()`, so pre-auth rejections — which the auth branch already recovers wholesale — never spend the authenticated session's budget). The first retry is immediate (the common case is a one-off hiccup); later ones back off 1s then 2s, awaited inline in the read loop rather than in a background task that could outlive its own socket. **`auth-required:` before this socket has completed its NIP-42 handshake is the expected bootstrap sequence, not a refusal**, and `_handle_closed` short-circuits it ahead of every recovery path: the connector opens its control REQs immediately in case the relay serves unauthenticated reads, a closed relay answers `auth-required:` plus an `AUTH` challenge, and the auth branch re-opens everything. That case is logged at DEBUG and consumes neither the permanent-refusal branch nor the retry budget (a chat subscription is still dropped from `_chat_subscriptions`, since it genuinely is not subscribed and discovery is what re-opens it). Treating it as permanent produced an operator-facing warning claiming discovery/membership tracking was DOWN in the same run where discovery then completed, every channel was subscribed, and a brand-new channel's kind-44100 was picked up one second later. The boundary is the per-socket `_auth_completed` flag, set once the signed AUTH event has been sent and cleared on session entry, session exit, and `stop()`; the same reason *after* that stays loud, and says the subscription is down until the relay's next AUTH challenge or the next reconnect rather than borrowing the non-auth wording. Then `_is_transient_close` decides whether to retry at all: NIP-01/NIP-42 `auth-required:`/`restricted:`/`blocked:`/`mute:`/`invalid:`/`pow:` prefixes and buzz-relay's own removal/revocation prose are permanent (do not fight the relay over a channel that is no longer ours; a post-auth `auth-required:` is recovered by the AUTH branch, not by re-issuing), `rate-limited:`/`error:`/no-reason-at-all and anything unrecognized are transient — the default resolves toward "keep listening" because going silently deaf is the failure this exists to remove, and the attempt budget bounds a wrong guess. A chat `CLOSED` is only ever recovered for a channel already in `_chat_subscriptions`: a `CLOSED` is relay-supplied, so acting on an unknown one would let a relay induce a subscription just by naming a channel. Every subscription that goes unlistened is logged at WARNING, never INFO. Subscription ids are deterministic per channel precisely so one can be replaced or closed without disturbing the others on the socket, and `_chat_subscriptions` is per-socket state cleared on session end, on re-auth (a pre-auth REQ may have been rejected), and by `stop()`. Three bounds on remote-fed state: `MAX_CACHED_CHANNELS` (512) caps the kind-39000 metadata cache, the watermark map, and the resubscribe-attempt map; `MAX_CHANNEL_SUBSCRIPTIONS` (256, well under buzz-relay's own 1024-per-connection ceiling) caps live chat subscriptions — at the cap new channels are refused and named in a warning rather than evicting a working subscription. **Known bound (documented, not fixed):** the relay caps historical delivery at 2000 events per subscription, newest-first, even with a `since`, so >2000 unread messages in a *single* channel across a disconnect loses the oldest — the relay never sends them and the watermark advances past them. That is the one remaining path that can skip; everything else fails toward replay. **Trust model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-trust-model)): every inbound `EVENT` is authenticated at the single `handle_relay_frame` choke point — the NIP-01 id is recomputed from the delivered payload and the BIP-340 Schnorr signature verified against the claimed `pubkey` (`buzz_nostr.verify_event`, pure and total: malformed input returns `False`, never raises) — so `ev["pubkey"]`, the authorization principal for both the allowlist and the `/connect` bind, cannot be forged by a relay the DeerFlow operator does not run. What remains trusted is the *authorship* of kind-39000 channel metadata: any member can sign one, and because per-channel subscriptions are now driven by discovery, a forged kind-39000 has two effects rather than one — it can mark a channel `type: "dm"` (relaxing `require_mention` for that channel) **and** it can induce a chat subscription for a channel of the forger's choosing, since the channels we listen to are exactly the channels we hold metadata for. Neither makes anything be *acted on*: `allowed_users` and per-event signature verification are independent gates, so an induced subscription only means the relay reads its own traffic back to a subscriber that drops it, bounded by `MAX_CHANNEL_SUBSCRIPTIONS` (which refuses rather than evicts, so it cannot displace a real channel). Same for a forged kind-44100, except its `p` tag is re-checked locally so it must at least name us. Closing this needs a configured trusted relay pubkey, which `relay_url` is not. `allowed_users` is deny-by-default (empty = nobody, unlike siblings' empty = everyone), so `start()` logs a WARNING when it is empty and each drop logs at DEBUG. The resubscribe cursor (`since`) is **per channel**, advances only for events that were actually processed, and never past `now + MAX_FUTURE_SKEW_SECONDS`, because it is peer-supplied (`created_at`) and a single future-dated event otherwise made the connector permanently deaf. Per channel rather than global is the safety-critical half: subscriptions are per channel, so one shared cursor is the newest event seen in *any* channel and a busy channel would drag it past a quiet channel's unread messages, skipping them on the next reconnect — measured on a live relay, three channels of one identity sat ~28h apart. Per-channel cursors can only ever cost duplicate delivery (absorbed by the manager's `event_id` dedupe), and an evicted cursor degrades to "no `since`", i.e. the relay's default backlog — both fail toward replay, never toward a miss. Streaming tracks every oversize chunk index (`_stream_targets` for chunk 0, `_stream_tails` for the rest), since the manager republishes cumulative text and reposting `chunks[1:]` per update flooded the channel; all of it is per-connection state cleared by `stop()`, and the remote-fed kind-39000 cache is capped at `MAX_CACHED_CHANNELS`. **`send()` refuses outright to publish text carrying a hidden model-context wrapper** (``, ``, `` — `_HIDDEN_CONTEXT_MARKERS`), logging at ERROR and clearing the stream bookkeeping on a blocked `is_final`. This is defense in depth behind the manager's allowlist, and it lives here rather than in a sibling connector because on Buzz a leak is permanent: every streaming update is an immutable public Nostr event, so a corrective edit only changes what clients render while the original leaked event stays on the relay. Matching is on the literal opening tag, so a reply that merely talks about memory is still published Buzz seen-event shutdown stays bounded and retryable: `aflush()` awaits any in-flight write, attempts at most one final snapshot, leaves a still-changing generation dirty for fail-open replay, and returns without a live persistence timer. A Gateway cancellation does not cancel the underlying worker-thread write; `BuzzChannel.stop()` tracks cleanup completion separately from transport admission so ChannelService can retry the retained channel without racing a second snapshot against that write. The store is quiesced before stop (including the already-stopped guard), so a timed-out relay task that records after stop only marks data dirty and cannot schedule detached file work; a repeated `stop()` still drains that dirty state, while `BuzzChannel.start()` explicitly resumes scheduling and flushes it automatically. - `github.py` - Webhook-driven GitHub channel. Inbound messages come from `POST /api/webhooks/github`; outbound is log-only because GitHub agents post explicitly with `gh` from their sandbox when they choose to comment or create a PR - `app/gateway/routers/channel_connections.py` - Browser-facing user connection and disconnect APIs diff --git a/backend/app/channels/buzz.py b/backend/app/channels/buzz.py index a768d7c616..35f056f9d8 100644 --- a/backend/app/channels/buzz.py +++ b/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 "" + 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 diff --git a/backend/app/channels/connection_identity.py b/backend/app/channels/connection_identity.py index 162498aff6..3acd291c72 100644 --- a/backend/app/channels/connection_identity.py +++ b/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) diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index b27f2d9c42..b52267c398 100644 --- a/backend/app/channels/manager.py +++ b/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, ) ) diff --git a/backend/tests/test_buzz_channel.py b/backend/tests/test_buzz_channel.py index cbb9f58258..dcd4f7585f 100644 --- a/backend/tests/test_buzz_channel.py +++ b/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 ({CHANNEL}): thread={'aa' * 32} not engaged (bound=True)" in caplog.text + + # -- FINAL REVIEW FINDING 3 (Important): oversize streaming re-posted the tail ------ diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index 3d07a3a500..f6748e1c09 100644 --- a/backend/tests/test_channels.py +++ b/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