diff --git a/CHANGELOG.md b/CHANGELOG.md index f42d2beec3..b3f7524bdc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -460,6 +460,24 @@ This release closes that milestone with **301 merged pull requests**. ### Fixed +- **channels:** Discord now runs its channel-connection database work on the + Gateway event loop. discord.py delivers messages on a private loop in the + client thread, and the Discord adapter awaited the connection repository there + even though its SQLAlchemy engine and pool belong to the Gateway loop. With + `channel_connections.enabled` on PostgreSQL, the first Discord message after + the Gateway had used the pool failed with `got Future … attached to a + different loop` and was dropped. With SQLite, a burst that exhausted the pool + failed with `Queue … is bound to a different event loop`, and the wait queue + stayed bound to the Discord loop, so the Gateway's own queries then failed the + same way. The identity lookup now runs together with the intake commit, and + `/connect` binding runs separately, both on the Gateway loop through + `_submit_threadsafe_coroutine` like Telegram, Feishu, and DingTalk. Bind + replies go back through the Discord loop, and `stop()` now drains that work + before tearing the client down. The typing indicator still registers before + the hand-off, and a failed lookup skips the ack reaction and stops the + indicator unless another message to the same target still relies on it, so a + dropped message never shows the bot as working. ([#6214]) + - **community:** The shared SSRF guard now refuses every non-global address, including the `100.64.0.0/10` shared address space that its flag checks let through. That range holds CGNAT and Tailscale hosts and Alibaba Cloud's @@ -7639,8 +7657,8 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6135]: https://github.com/bytedance/deer-flow/pull/6135 [#6138]: https://github.com/bytedance/deer-flow/pull/6138 [#6140]: https://github.com/bytedance/deer-flow/pull/6140 +[#6171]: https://github.com/bytedance/deer-flow/pull/6171 [#6201]: https://github.com/bytedance/deer-flow/pull/6201 [#6202]: https://github.com/bytedance/deer-flow/pull/6202 +[#6214]: https://github.com/bytedance/deer-flow/pull/6214 [#6212]: https://github.com/bytedance/deer-flow/pull/6212 - -[#6171]: https://github.com/bytedance/deer-flow/pull/6171 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 00ebf900b3..24becef623 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -419,6 +419,19 @@ ### 修复 +- **渠道：** Discord 的渠道连接数据库操作现在在 Gateway 事件循环上执行。 + discord.py 在客户端线程的私有事件循环上投递消息，而 Discord 适配器此前就在该 + 循环上 await 连接仓库，但仓库的 SQLAlchemy 引擎与连接池属于 Gateway 循环。在 + PostgreSQL 上开启 `channel_connections.enabled` 时，Gateway 使用过连接池之后的 + 第一条 Discord 消息会以 `got Future … attached to a different loop` 失败并被 + 丢弃；在 SQLite 上，耗尽连接池的突发流量会以 + `Queue … is bound to a different event loop` 失败，且等待队列会一直绑定在 + Discord 循环上，导致 Gateway 自身的查询随后也以同样方式失败。现在身份查询与入站 + 提交一起执行，`/connect` 绑定单独执行，二者都与 Telegram、飞书、钉钉一样经由 + `_submit_threadsafe_coroutine` 在 Gateway 循环上运行；绑定回复经 Discord 循环 + 发回，`stop()` 也会在关闭客户端前排空这些任务。输入中提示仍在交接前注册，查询失败时会跳过确认表情， + 并在同一目标没有其他消息依赖时停止该提示，被丢弃的消息不会让机器人显示为仍在处理。([#6214]) + - **社区工具：** 共享 SSRF 校验现在拒绝所有非全局地址，包括原先的标志位检查放行的 `100.64.0.0/10` 共享地址段。该地址段包含 CGNAT 与 Tailscale 主机以及阿里云 `100.100.100.200` 实例元数据端点，因此 `web_fetch`（crawl4ai、Browserless、 @@ -6387,3 +6400,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6201]: https://github.com/bytedance/deer-flow/pull/6201 [#6202]: https://github.com/bytedance/deer-flow/pull/6202 [#6212]: https://github.com/bytedance/deer-flow/pull/6212 +[#6214]: https://github.com/bytedance/deer-flow/pull/6214 diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 9d537129e4..a7a3a32ae5 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -10,10 +10,10 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk - `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) and `/agent` (`list` is owner-scoped; `use` creates a fresh thread and persists the Custom Agent selection under both the channel restart key and Web's canonical routing metadata so existing checkpoint lineage never changes agent across IM/Web), 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]`). -- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()`: it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. Blocking SDK teardown in `stop()` (a client `close()`, a thread `join()`) must run off the event loop through `asyncio.to_thread`, since every run and channel on the Gateway shares that loop and `stop()` also serves `POST /api/channels/{name}/restart`; Slack's `SocketModeClient.close()` is a shielded, tracked task so a cancelled stop leaves it running and a retried stop awaits the same close. Anchored by `tests/blocking_io/test_channel_stop_offloop.py`. **Proactive push** (`Channel.send_notification`): used by the scheduled-task notification outbox to push markdown to a bound external identity with no inbound frame. The base default raises `NotImplementedError` (after a `ChannelUnavailable` check when not running) so unsupported providers fail visibly in the outbox. Connectivity failures must raise `ChannelUnavailable` so the delivery worker parks without consuming retries; platform rejections of a well-formed send remain ordinary exceptions. **Only WeCom implements proactive push today** (`wecom.py` via `WSClient.send_message`); the SDK raises `RuntimeError` from that call for a dead socket and for a platform rejection alike, so only the `Reply ack error` prefix counts; other `RuntimeError`s, `ConnectionError`/`TimeoutError`/`OSError` and the `websockets` close exception park. Before sending, the worker re-checks the binding via `resolve_connections` (`ChannelConnectionRepository.list_connections`, owner-scoped, ahead of the channel-liveness check): a target no longer `connected` for the row's owner is finalized with `mark_failed(terminal=True)`; a lookup failure parks. Terminal outbox rows (`sent`/`failed`) ignore later status writes. Scheduler side (`app/scheduler/`): `handle_run_completion` enqueues only after `complete_run` recorded a success or failed outcome; manual triggers, interrupts and recovery-finalized occurrences stay silent. Gateway lifespan wires the enqueue side with the scheduler, then starts `NotificationDeliveryWorker` after the channel service (the scheduler starts first), detaching the enqueue side if delivery cannot start; the worker stops before the channel service. +- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()` (or `_submit_threadsafe_coroutine_future()` to await the outcome through `asyncio.shield`): it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. Blocking SDK teardown in `stop()` (a client `close()`, a thread `join()`) must run off the event loop through `asyncio.to_thread`, since every run and channel on the Gateway shares that loop and `stop()` also serves `POST /api/channels/{name}/restart`; Slack's `SocketModeClient.close()` is a shielded, tracked task so a cancelled stop leaves it running and a retried stop awaits the same close. Anchored by `tests/blocking_io/test_channel_stop_offloop.py`. **Proactive push** (`Channel.send_notification`): used by the scheduled-task notification outbox to push markdown to a bound external identity with no inbound frame. The base default raises `NotImplementedError` (after a `ChannelUnavailable` check when not running) so unsupported providers fail visibly in the outbox. Connectivity failures must raise `ChannelUnavailable` so the delivery worker parks without consuming retries; platform rejections of a well-formed send remain ordinary exceptions. **Only WeCom implements proactive push today** (`wecom.py` via `WSClient.send_message`); the SDK raises `RuntimeError` from that call for a dead socket and for a platform rejection alike, so only the `Reply ack error` prefix counts; other `RuntimeError`s, `ConnectionError`/`TimeoutError`/`OSError` and the `websockets` close exception park. Before sending, the worker re-checks the binding via `resolve_connections` (`ChannelConnectionRepository.list_connections`, owner-scoped, ahead of the channel-liveness check): a target no longer `connected` for the row's owner is finalized with `mark_failed(terminal=True)`; a lookup failure parks. Terminal outbox rows (`sent`/`failed`) ignore later status writes. Scheduler side (`app/scheduler/`): `handle_run_completion` enqueues only after `complete_run` recorded a success or failed outcome; manual triggers, interrupts and recovery-finalized occurrences stay silent. Gateway lifespan wires the enqueue side with the scheduler, then starts `NotificationDeliveryWorker` after the channel service (the scheduler starts first), detaching the enqueue side if delivery cannot start; the worker stops before the channel service. - `wecom.py` - WeCom WebSocket channel; implements `send_notification` for the outbox worker. Tracks `_ws_transport_up` across disconnect/reconnect so transport failures become `ChannelUnavailable` rather than burning the counted retry budget. - `service.py` - Manages lifecycle of all configured channels from `config.yaml`. Shutdown closes manager admission first and keeps transports alive until every manager worker/follow-up watcher has exited. A successful manager stop therefore owns no live handler; if the Gateway's outer timeout cancels shutdown, the service retains its channel objects and global singleton so unfinished resources are not detached and cleanup can be retried. -- `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` 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`) +- `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 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 diff --git a/backend/app/channels/base.py b/backend/app/channels/base.py index 3b6b6b6e51..1d470bf568 100644 --- a/backend/app/channels/base.py +++ b/backend/app/channels/base.py @@ -195,13 +195,42 @@ class Channel(ABC): reservation: InboundReservation | None = None, ) -> bool: """Submit provider-thread work while retaining its real asyncio Task.""" + return ( + self._submit_threadsafe_coroutine_future( + coroutine, + loop, + name=name, + msg_id=msg_id, + reservation=reservation, + ) + is not None + ) + + def _submit_threadsafe_coroutine_future( + self, + coroutine: Coroutine[Any, Any, T], + loop: asyncio.AbstractEventLoop | None, + *, + name: str, + msg_id: Any, + reservation: InboundReservation | None = None, + ) -> Future[T] | None: + """Like ``_submit_threadsafe_coroutine``, returning the completion future. + + ``None`` means the work was refused (and the reservation released). + The future settles on every path: the task's result or exception, or + cancellation when ``_close_and_drain_threadsafe_futures`` stops the + work before or while it runs. Await it from another loop through + ``asyncio.shield(asyncio.wrap_future(...))`` so cancelling the waiter + never cancels the shared future the finalizer still has to settle. + """ with self._threadsafe_submissions_lock: if not self._threadsafe_submission_intake_open or loop is None or not loop.is_running(): coroutine.close() if reservation is not None: reservation.release() - return False + return None submission = _ThreadsafeSubmission( coroutine=coroutine, @@ -219,8 +248,8 @@ class Channel(ABC): coroutine.close() if reservation is not None: reservation.release() - return False - return True + return None + return submission.completion def _start_threadsafe_submission(self, submission: _ThreadsafeSubmission) -> None: """Create the owned Task on its event loop or finish a pre-start cancel.""" diff --git a/backend/app/channels/discord.py b/backend/app/channels/discord.py index 8499a937cb..4c23f5b01b 100644 --- a/backend/app/channels/discord.py +++ b/backend/app/channels/discord.py @@ -7,6 +7,7 @@ import io import json import logging import threading +from concurrent.futures import Future from pathlib import Path from typing import Any @@ -85,6 +86,10 @@ class DiscordChannel(Channel): # Typing indicator management self._typing_tasks: dict[str, asyncio.Task] = {} + # Messages whose hand-off is pending or committed, per typing task. A + # failed hand-off may stop the indicator only when no other message + # still relies on it (see ``_acknowledge_after_handoff``). + self._typing_dependents: dict[asyncio.Task, int] = {} # Strong references for this channel's in-flight ack-reaction tasks. # The event loop keeps only weak references to scheduled tasks, so a @@ -127,6 +132,7 @@ class DiscordChannel(Channel): self._client = client self._discord_module = discord self._main_loop = asyncio.get_event_loop() + self._open_threadsafe_future_intake() @client.event async def on_message(message) -> None: @@ -208,6 +214,7 @@ class DiscordChannel(Channel): async def stop(self) -> None: self._running = False self.bus.unsubscribe_outbound(self._on_outbound) + await self._close_and_drain_threadsafe_futures() # Best-effort durability: flush in-memory thread mappings so the most # recent channel->thread mapping survives a hard shutdown (process @@ -358,13 +365,17 @@ class DiscordChannel(Channel): logger.exception("[Discord] failed to upload file: %s", attachment.filename) return False - async def _start_typing(self, channel, chat_id: str, thread_ts: str | None = None) -> None: - """Starts a loop to send periodic typing indicators.""" + async def _start_typing(self, channel, chat_id: str, thread_ts: str | None = None) -> asyncio.Task[None] | None: + """Starts a loop to send periodic typing indicators. + + Returns the loop's task when this call started it, ``None`` when the + channel is stopping or the target already shows typing. + """ if not self._running: - return + return None target_id = thread_ts or chat_id if target_id in self._typing_tasks: - return # Already typing for this target + return None # Already typing for this target async def _typing_loop(): # The loop's first failure logs at WARNING, so an indicator that @@ -386,6 +397,8 @@ class DiscordChannel(Channel): task = asyncio.create_task(_typing_loop()) self._typing_tasks[target_id] = task + task.add_done_callback(lambda done: self._typing_dependents.pop(done, None)) + return task async def _cancel_typing_tasks(self) -> None: """Cancel and await every typing task on their owning event loop.""" @@ -397,6 +410,7 @@ class DiscordChannel(Channel): if typing_tasks: await asyncio.gather(*(task for _, task in typing_tasks), return_exceptions=True) self._typing_tasks.clear() + self._typing_dependents.clear() def _discard_typing_tasks(self) -> None: """Forget stale typing tasks after their owning event loop has stopped.""" @@ -407,6 +421,7 @@ class DiscordChannel(Channel): target_id, ) self._typing_tasks.clear() + self._typing_dependents.clear() async def _cancel_ack_reaction_tasks(self) -> None: """Cancel in-flight ack reactions so stop() does not strand them. @@ -585,14 +600,11 @@ class DiscordChannel(Channel): }, ) inbound.topic_id = thread_id - inbound = await self._attach_connection_identity(inbound, guild_id=str(guild.id) if guild else None) - if not self._publish_reserved(inbound, reservation): + handoff = self._publish_reserved(inbound, reservation, guild_id=str(guild.id) if guild else None) + if handoff is None: return reservation_transferred = True - # Start typing indicator in the thread - if typing_target: - await self._start_typing(typing_target, chat_id, thread_id) - self._schedule_ack_reaction(message) + await self._acknowledge_after_handoff(message, handoff, typing_target=typing_target, chat_id=chat_id, thread_id=thread_id) return # Thread not tracked (orphaned) — create new thread and handle below @@ -694,31 +706,98 @@ class DiscordChannel(Channel): }, ) inbound.topic_id = thread_id - inbound = await self._attach_connection_identity(inbound, guild_id=str(guild.id) if guild else None) - if not self._publish_reserved(inbound, reservation): + handoff = self._publish_reserved(inbound, reservation, guild_id=str(guild.id) if guild else None) + if handoff is None: return reservation_transferred = True - - # Start typing/reaction only after bounded admission succeeds. - if typing_target: - await self._start_typing(typing_target, chat_id, thread_id) - self._schedule_ack_reaction(message) + await self._acknowledge_after_handoff(message, handoff, typing_target=typing_target, chat_id=chat_id, thread_id=thread_id) finally: if not reservation_transferred: reservation.release() - def _publish_reserved(self, inbound: InboundMessage, reservation: InboundReservation) -> bool: - """Transfer an already-reserved message to the Gateway loop.""" - if self._main_loop and self._main_loop.is_running(): - try: - # This is called from discord.py's private event-loop thread. - self._main_loop.call_soon_threadsafe(self._commit_reserved_inbound, reservation, inbound) - return True - except RuntimeError: - logger.info("[Discord] main loop stopped before reserved inbound could be scheduled") - reservation.release() - return False + def _publish_reserved(self, inbound: InboundMessage, reservation: InboundReservation, *, guild_id: str | None) -> Future[bool] | None: + """Transfer an already-reserved message to the Gateway loop. + + Called from discord.py's private event-loop thread. The connection + identity is resolved on the Gateway loop as well, because the + repository's SQLAlchemy engine and pool belong to that loop: awaiting + them from the Discord loop fails under asyncpg ("attached to a + different loop") and binds the pool's wait queue to the wrong loop + under aiosqlite, breaking the Gateway's own queries. + + Returns the hand-off's completion future (``True`` once committed), or + ``None`` when the Gateway loop refused it. + """ + handoff = self._submit_threadsafe_coroutine_future( + self._commit_reserved_inbound_with_identity(inbound, reservation, guild_id=guild_id), + self._main_loop, + name="commit_reserved_inbound", + msg_id=inbound.metadata.get("message_id"), + reservation=reservation, + ) + if handoff is None: + logger.info("[Discord] main loop stopped before reserved inbound could be scheduled") + return handoff + + async def _acknowledge_after_handoff( + self, + message, + handoff: Future[bool], + *, + typing_target, + chat_id: str, + thread_id: str | None, + ) -> None: + """Show progress for a handed-off message, undoing it if the hand-off fails. + + Typing starts before the hand-off resolves, as it did before the + identity lookup moved to the Gateway loop: ``_start_typing`` registers + without yielding, so a fast reply's ``_stop_typing`` cannot run first + and leave an indicator behind. The ack reaction waits for the commit. + If identity resolution or the commit fails, or ``stop()`` drains the + hand-off, the message was dropped and gets no ack. Another message to + the same target may have reused the indicator meanwhile, pending or + already committed, so a failed hand-off stops it only when this message + started it and no other message still depends on it. + """ + target_id = thread_id or chat_id + started_task = await self._start_typing(typing_target, chat_id, thread_id) if typing_target else None + typing_task = self._typing_tasks.get(target_id) if typing_target else None + if typing_task is not None: + self._typing_dependents[typing_task] = self._typing_dependents.get(typing_task, 0) + 1 + try: + committed = await asyncio.shield(asyncio.wrap_future(handoff)) + except asyncio.CancelledError: + if not handoff.cancelled(): + raise + committed = False # drained by stop() + except Exception: + committed = False # the submission finalizer already logged it + if committed: + # Keep this message's hold until a reply stops the indicator. + self._schedule_ack_reaction(message) + return + if typing_task is None: + return + remaining = self._typing_dependents.get(typing_task, 1) - 1 + if remaining > 0: + self._typing_dependents[typing_task] = remaining + return + self._typing_dependents.pop(typing_task, None) + if typing_task is started_task and self._typing_tasks.get(target_id) is typing_task: + await self._stop_typing(chat_id, thread_id) + + async def _commit_reserved_inbound_with_identity( + self, + inbound: InboundMessage, + reservation: InboundReservation, + *, + guild_id: str | None, + ) -> bool: + """Attach connection identity and commit the reservation on the Gateway loop.""" + inbound = await self._attach_connection_identity(inbound, guild_id=guild_id) + return self._commit_reserved_inbound(reservation, inbound) async def _attach_connection_identity(self, inbound: InboundMessage, guild_id: str | None = None) -> InboundMessage: return await attach_connection_identity( @@ -730,12 +809,28 @@ class DiscordChannel(Channel): ) async def _bind_connection_from_connect_code(self, message, code: str) -> bool: + """Dispatch the bind flow to the Gateway loop, where the repository's engine lives.""" if self._connection_repo is None or not code: return False + scheduled = self._submit_threadsafe_coroutine( + self._bind_connection_from_connect_code_on_main(message, code), + self._main_loop, + name="bind_connection", + msg_id=getattr(message, "id", None), + ) + if not scheduled: + logger.warning("[Discord] main loop not running, cannot bind channel connection") + # Handled either way so a bind attempt never reaches the agent as a chat + # turn; an unscheduled bind leaves the code unconsumed, so the user can + # retry it after the Gateway restarts. + return True + + async def _bind_connection_from_connect_code_on_main(self, message, code: str) -> bool: + """Run the bind flow on the Gateway loop; replies go back through the Discord loop.""" state = await self._connection_repo.consume_oauth_state(provider="discord", state=code) if state is None: - await self._send_connection_reply(message, "Discord connection code is invalid or expired.") + await self._run_on_discord_loop(self._send_connection_reply(message, "Discord connection code is invalid or expired.")) return True guild = getattr(message, "guild", None) @@ -743,7 +838,7 @@ class DiscordChannel(Channel): author = getattr(message, "author", None) user_id = str(getattr(author, "id", "") or "") if not user_id: - await self._send_connection_reply(message, "Discord connection could not be completed from this message.") + await self._run_on_discord_loop(self._send_connection_reply(message, "Discord connection could not be completed from this message.")) return True guild_id = str(getattr(guild, "id", "") or "") or None @@ -760,7 +855,7 @@ class DiscordChannel(Channel): }, status="connected", ) - await self._send_connection_reply(message, "Discord connected to DeerFlow.") + await self._run_on_discord_loop(self._send_connection_reply(message, "Discord connected to DeerFlow.")) return True @staticmethod diff --git a/backend/tests/test_discord_channel.py b/backend/tests/test_discord_channel.py index 9dd39e77cf..3a6869ef74 100644 --- a/backend/tests/test_discord_channel.py +++ b/backend/tests/test_discord_channel.py @@ -281,9 +281,8 @@ async def test_discord_bot_mention_slash_skill_routes_as_chat() -> None: channel._add_reaction = noop await channel._on_message(_make_discord_message("<@999> /data-analysis analyze uploads/foo.csv")) - await asyncio.sleep(0) - - inbound = bus.get_inbound_nowait() + # The commit runs as a task on the Gateway loop (identity lookup included). + inbound = await asyncio.wait_for(bus.get_inbound(), timeout=5) bus.inbound_task_done() assert inbound.text == "/data-analysis analyze uploads/foo.csv" assert inbound.msg_type == InboundMessageType.CHAT @@ -306,9 +305,8 @@ async def test_discord_bot_mention_known_command_routes_as_command() -> None: channel._add_reaction = noop await channel._on_message(_make_discord_message("<@999> /help")) - await asyncio.sleep(0) - - inbound = bus.get_inbound_nowait() + # The commit runs as a task on the Gateway loop (identity lookup included). + inbound = await asyncio.wait_for(bus.get_inbound(), timeout=5) bus.inbound_task_done() assert inbound.text == "/help" assert inbound.msg_type == InboundMessageType.COMMAND diff --git a/backend/tests/test_discord_channel_connections.py b/backend/tests/test_discord_channel_connections.py index 7dc7a7ce13..53e58a2934 100644 --- a/backend/tests/test_discord_channel_connections.py +++ b/backend/tests/test_discord_channel_connections.py @@ -2,8 +2,12 @@ from __future__ import annotations +import asyncio +import sys +import threading from datetime import UTC, datetime, timedelta -from unittest.mock import AsyncMock, MagicMock +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -74,7 +78,12 @@ async def test_discord_connect_command_binds_gateway_identity(repo): message.channel.id = 456 message.channel.send = AsyncMock() - handled = await channel._bind_connection_from_connect_code(message, state) + async def _passthrough(coro): + return await coro + + channel._run_on_discord_loop = AsyncMock(side_effect=_passthrough) + + handled = await channel._bind_connection_from_connect_code_on_main(message, state) connections = await repo.list_connections("deerflow-user-1") assert handled is True @@ -86,3 +95,459 @@ async def test_discord_connect_command_binds_gateway_identity(repo): assert connections[0]["workspace_name"] == "Deer Guild" assert connections[0]["metadata"]["channel_id"] == "456" message.channel.send.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# Connection repository calls stay on the Gateway loop +# --------------------------------------------------------------------------- +# +# discord.py runs ``_on_message`` on a private loop in the client thread, while +# the repository's SQLAlchemy engine and pool belong to the Gateway loop. +# Awaiting the repository from the Discord loop fails with asyncpg ("attached to +# a different loop") and binds the pool's wait queue to the wrong loop under +# aiosqlite contention, so every repository call must run on ``_main_loop``. + + +class _LoopRecordingRepo: + """Delegate to the real repository, recording the loop of every call.""" + + def __init__(self, repo) -> None: + self._repo = repo + self.calls: list[tuple[str, asyncio.AbstractEventLoop]] = [] + + def __getattr__(self, name: str): + method = getattr(self._repo, name) + + async def _recorded(*args, **kwargs): + self.calls.append((name, asyncio.get_running_loop())) + return await method(*args, **kwargs) + + return _recorded + + +def _start_discord_loop() -> tuple[asyncio.AbstractEventLoop, threading.Thread]: + """A real background loop standing in for discord.py's client loop.""" + loop = asyncio.new_event_loop() + ready = threading.Event() + + def _runner() -> None: + loop.call_soon(ready.set) + loop.run_forever() + + thread = threading.Thread(target=_runner, daemon=True) + thread.start() + ready.wait() + return loop, thread + + +def _stop_discord_loop(loop: asyncio.AbstractEventLoop, thread: threading.Thread) -> None: + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=5) + loop.close() + + +def _discord_channel_on_loops(bus: MessageBus, repo, discord_loop: asyncio.AbstractEventLoop) -> DiscordChannel: + channel = DiscordChannel(bus=bus, config={"bot_token": "token", "connection_repo": repo}) + channel._running = True + channel._client = SimpleNamespace(user=SimpleNamespace(id=999, mention="<@999>")) + channel._discord_module = SimpleNamespace(Thread=type("FakeThread", (), {})) + channel._main_loop = asyncio.get_running_loop() + channel._discord_loop = discord_loop + + async def noop(*_args, **_kwargs): + return None + + channel._start_typing = noop + channel._add_reaction = noop + return channel + + +def _discord_message(text: str, *, send=None): + return SimpleNamespace( + id=111, + content=text, + author=SimpleNamespace(id=987, bot=False, display_name="Alice", name="alice"), + guild=SimpleNamespace(id=123, name="Deer Guild"), + channel=SimpleNamespace(id=456, send=send or AsyncMock()), + add_reaction=lambda _emoji: None, + ) + + +async def _on_discord_loop(coro, loop: asyncio.AbstractEventLoop): + """Run *coro* on the Discord loop, as discord.py dispatches ``on_message``.""" + return await asyncio.wait_for(asyncio.wrap_future(asyncio.run_coroutine_threadsafe(coro, loop)), timeout=5) + + +@pytest.mark.anyio +async def test_discord_message_resolves_identity_on_gateway_loop(repo): + connection = await repo.upsert_connection( + owner_user_id="alice", + provider="discord", + external_account_id="987", + external_account_name="Alice", + status="connected", + ) + recording = _LoopRecordingRepo(repo) + bus = MessageBus() + discord_loop, thread = _start_discord_loop() + try: + channel = _discord_channel_on_loops(bus, recording, discord_loop) + await _on_discord_loop(channel._on_message(_discord_message("hello")), discord_loop) + inbound = await asyncio.wait_for(bus.get_inbound(), timeout=5) + bus.inbound_task_done() + finally: + _stop_discord_loop(discord_loop, thread) + + assert inbound.connection_id == connection["id"] + assert inbound.owner_user_id == "alice" + assert recording.calls, "identity lookup never reached the repository" + assert all(loop is channel._main_loop for _name, loop in recording.calls) + + +@pytest.mark.anyio +async def test_discord_connect_code_binds_on_gateway_loop_and_replies_on_discord_loop(repo): + state = "discord-bind-code" + await repo.create_oauth_state( + owner_user_id="deerflow-user-1", + provider="discord", + state=state, + expires_at=datetime.now(UTC) + timedelta(minutes=5), + ) + recording = _LoopRecordingRepo(repo) + reply_loops: list[asyncio.AbstractEventLoop] = [] + replied = asyncio.Event() + main_loop = asyncio.get_running_loop() + + async def _send(text: str) -> None: + reply_loops.append(asyncio.get_running_loop()) + main_loop.call_soon_threadsafe(replied.set) + + bus = MessageBus() + discord_loop, thread = _start_discord_loop() + try: + channel = _discord_channel_on_loops(bus, recording, discord_loop) + await _on_discord_loop(channel._on_message(_discord_message(f"/connect {state}", send=_send)), discord_loop) + await asyncio.wait_for(replied.wait(), timeout=5) + finally: + _stop_discord_loop(discord_loop, thread) + + connections = await repo.list_connections("deerflow-user-1") + assert [c["external_account_id"] for c in connections] == ["987"] + assert [name for name, _loop in recording.calls] == ["consume_oauth_state", "upsert_connection"] + assert all(loop is main_loop for _name, loop in recording.calls) + assert reply_loops == [discord_loop] + # A consumed bind code is a control-plane command, never an agent turn. + assert bus.inbound_queue.qsize() == 0 + + +@pytest.mark.anyio +async def test_discord_stop_cancels_pending_identity_lookup_and_releases_intake(repo): + lookup_started = asyncio.Event() + + class _BlockingRepo: + async def find_connection_by_external_identity(self, **_kwargs): + lookup_started.set() + await asyncio.Event().wait() + + bus = MessageBus(inbound_queue_maxsize=1) + discord_loop, thread = _start_discord_loop() + try: + channel = _discord_channel_on_loops(bus, _BlockingRepo(), discord_loop) + # _on_message waits on the hand-off, so it stays pending with the lookup. + on_message = asyncio.run_coroutine_threadsafe(channel._on_message(_discord_message("hello")), discord_loop) + await asyncio.wait_for(lookup_started.wait(), timeout=5) + + await channel._close_and_drain_threadsafe_futures() + # The drain settles the hand-off, so the Discord-side handler finishes too. + await asyncio.wait_for(asyncio.wrap_future(on_message), timeout=5) + finally: + _stop_discord_loop(discord_loop, thread) + + # The cancelled lookup must hand the intake slot back to the shared bus. + await bus.publish_inbound(InboundMessage(channel_name="slack", chat_id="C1", user_id="U1", text="capacity was released")) + assert bus.inbound_queue.qsize() == 1 + + +@pytest.mark.anyio +async def test_discord_unscheduled_bind_is_handled_but_leaves_code_unconsumed(repo): + state = "discord-bind-code" + await repo.create_oauth_state( + owner_user_id="deerflow-user-1", + provider="discord", + state=state, + expires_at=datetime.now(UTC) + timedelta(minutes=5), + ) + bus = MessageBus() + channel = DiscordChannel(bus=bus, config={"bot_token": "token", "connection_repo": repo}) + channel._main_loop = None # Gateway loop already gone + + handled = await channel._bind_connection_from_connect_code(_discord_message(f"/connect {state}"), state) + + # The message is swallowed rather than published as a chat turn, but the + # one-time code was never consumed, so a retry after restart still binds. + assert handled is True + assert bus.inbound_queue.qsize() == 0 + assert await repo.consume_oauth_state(provider="discord", state=state) is not None + + +class _FailingIdentityRepo: + async def find_connection_by_external_identity(self, **_kwargs): + raise RuntimeError("database unavailable") + + +def _channel_with_real_typing(bus: MessageBus, repo, discord_loop: asyncio.AbstractEventLoop, reactions: list[int]) -> DiscordChannel: + channel = _discord_channel_on_loops(bus, repo, discord_loop) + del channel._start_typing # exercise the real typing loop + + async def _record_reaction(message) -> None: + reactions.append(message.id) + + channel._add_reaction = _record_reaction + return channel + + +def _typing_message(text: str = "hello", *, author_id: int = 987): + message = _discord_message(text) + message.author.id = author_id + + async def _typing() -> None: + return None + + message.channel.typing = _typing + return message + + +@pytest.mark.anyio +async def test_discord_failed_identity_lookup_stops_typing_and_skips_ack(): + # A raising lookup drops the message, so nothing may keep reporting that + # the bot is working on it: no typing loop left behind, no ack reaction. + reactions: list[int] = [] + bus = MessageBus(inbound_queue_maxsize=1) + discord_loop, thread = _start_discord_loop() + try: + channel = _channel_with_real_typing(bus, _FailingIdentityRepo(), discord_loop, reactions) + await _on_discord_loop(channel._on_message(_typing_message()), discord_loop) + await _on_discord_loop(asyncio.sleep(0), discord_loop) + typing_targets = list(channel._typing_tasks) + finally: + _stop_discord_loop(discord_loop, thread) + + assert typing_targets == [] + assert reactions == [] + assert bus.inbound_queue.qsize() == 0 + await bus.publish_inbound(InboundMessage(channel_name="slack", chat_id="C1", user_id="U1", text="capacity was released")) + assert bus.inbound_queue.qsize() == 1 + + +@pytest.mark.anyio +async def test_discord_failed_identity_lookup_keeps_another_messages_typing(): + reactions: list[int] = [] + bus = MessageBus() + discord_loop, thread = _start_discord_loop() + try: + channel = _channel_with_real_typing(bus, _FailingIdentityRepo(), discord_loop, reactions) + + async def _register_in_flight_typing() -> asyncio.Task: + # An earlier message to the same target already shows typing. + await channel._start_typing(_typing_message().channel, "456") + return channel._typing_tasks["456"] + + in_flight = await _on_discord_loop(_register_in_flight_typing(), discord_loop) + await _on_discord_loop(channel._on_message(_typing_message()), discord_loop) + still_registered = channel._typing_tasks.get("456") is in_flight + await _on_discord_loop(channel._cancel_typing_tasks(), discord_loop) + finally: + _stop_discord_loop(discord_loop, thread) + + assert still_registered + + +@pytest.mark.anyio +@pytest.mark.parametrize("follower_settles_first", [True, False], ids=["follower-committed", "follower-pending"]) +async def test_discord_failed_typing_creator_keeps_typing_for_a_follower(follower_settles_first: bool): + # A starts typing and waits on its lookup; B, to the same target, reuses + # that indicator. When A's lookup then fails, B (committed or still + # pending) still depends on the indicator, so A must not stop it. + gates = {"1": asyncio.Event(), "2": asyncio.Event()} # creator, follower + lookups_started: set[str] = set() + + class _PerAuthorRepo: + async def find_connection_by_external_identity(self, *, external_account_id: str, **_kwargs): + lookups_started.add(external_account_id) + await gates[external_account_id].wait() + if external_account_id == "1": + raise RuntimeError("database unavailable") + return None + + async def _wait_for_lookup(author: str) -> None: + while author not in lookups_started: + await asyncio.sleep(0.01) + + reactions: list[int] = [] + bus = MessageBus() + discord_loop, thread = _start_discord_loop() + try: + channel = _channel_with_real_typing(bus, _PerAuthorRepo(), discord_loop, reactions) + creator = asyncio.run_coroutine_threadsafe(channel._on_message(_typing_message(author_id=1)), discord_loop) + await asyncio.wait_for(_wait_for_lookup("1"), timeout=5) + follower = asyncio.run_coroutine_threadsafe(channel._on_message(_typing_message(author_id=2)), discord_loop) + await asyncio.wait_for(_wait_for_lookup("2"), timeout=5) + + settle_order = [(gates["2"], follower), (gates["1"], creator)] if follower_settles_first else [(gates["1"], creator), (gates["2"], follower)] + for gate, handler in settle_order: + gate.set() + await asyncio.wait_for(asyncio.wrap_future(handler), timeout=5) + typing_targets = list(channel._typing_tasks) + await _on_discord_loop(channel._cancel_typing_tasks(), discord_loop) + inbound = await asyncio.wait_for(bus.get_inbound(), timeout=5) + bus.inbound_task_done() + finally: + _stop_discord_loop(discord_loop, thread) + + assert typing_targets == ["456"] + assert inbound.text == "hello" + assert reactions == [111] + + +@pytest.mark.anyio +async def test_discord_committed_message_keeps_typing_and_acks(repo): + await repo.upsert_connection(owner_user_id="alice", provider="discord", external_account_id="987", status="connected") + reactions: list[int] = [] + bus = MessageBus() + discord_loop, thread = _start_discord_loop() + try: + channel = _channel_with_real_typing(bus, repo, discord_loop, reactions) + await _on_discord_loop(channel._on_message(_typing_message()), discord_loop) + await _on_discord_loop(asyncio.sleep(0), discord_loop) + typing_targets = list(channel._typing_tasks) + held = sum(channel._typing_dependents.values()) + + async def _reply_stops_typing() -> None: + await channel._stop_typing("456") + await asyncio.sleep(0) # let the cancelled loop finish + + await _on_discord_loop(_reply_stops_typing(), discord_loop) + dependents_after_reply = dict(channel._typing_dependents) + inbound = await asyncio.wait_for(bus.get_inbound(), timeout=5) + bus.inbound_task_done() + finally: + _stop_discord_loop(discord_loop, thread) + + assert inbound.owner_user_id == "alice" + assert typing_targets == ["456"] + assert held == 1 # the committed message holds the indicator + assert dependents_after_reply == {} # released once a reply stops typing + assert reactions == [111] + + +@pytest.mark.anyio +async def test_discord_reply_stopping_typing_during_handoff_leaves_no_indicator(): + # A reply can stop typing while the hand-off is still settling. Typing is + # registered before the hand-off, so that stop wins; starting it after the + # hand-off resolved would leave an indicator nothing ever stops. + reactions: list[int] = [] + bus = MessageBus() + discord_loop, thread = _start_discord_loop() + try: + channel = None + + class _ReplyDuringLookupRepo: + async def find_connection_by_external_identity(self, **_kwargs): + asyncio.run_coroutine_threadsafe(channel._stop_typing("456"), discord_loop) + return None + + channel = _channel_with_real_typing(bus, _ReplyDuringLookupRepo(), discord_loop, reactions) + await _on_discord_loop(channel._on_message(_typing_message()), discord_loop) + await _on_discord_loop(asyncio.sleep(0), discord_loop) + typing_targets = list(channel._typing_tasks) + await _on_discord_loop(channel._cancel_typing_tasks(), discord_loop) + await asyncio.wait_for(bus.get_inbound(), timeout=5) + bus.inbound_task_done() + finally: + _stop_discord_loop(discord_loop, thread) + + assert typing_targets == [] + assert reactions == [111] + + +@pytest.mark.anyio +async def test_discord_cancelled_handler_leaves_the_shared_handoff_to_settle(): + # discord.py may cancel a handler mid-wait. That must not cancel the shared + # completion future, or the submission finalizer fails to settle it and the + # already-running commit is reported as an error on the Gateway loop. + main_loop = asyncio.get_running_loop() + lookup_started = asyncio.Event() + release = asyncio.Event() + loop_errors: list[dict] = [] + + class _GatedRepo: + async def find_connection_by_external_identity(self, **_kwargs): + lookup_started.set() + await release.wait() + return None + + previous_handler = main_loop.get_exception_handler() + main_loop.set_exception_handler(lambda _loop, context: loop_errors.append(context)) + bus = MessageBus() + discord_loop, thread = _start_discord_loop() + try: + channel = _discord_channel_on_loops(bus, _GatedRepo(), discord_loop) + on_message = asyncio.run_coroutine_threadsafe(channel._on_message(_discord_message("hello")), discord_loop) + await asyncio.wait_for(lookup_started.wait(), timeout=5) + on_message.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(asyncio.wrap_future(on_message), timeout=5) + + release.set() + inbound = await asyncio.wait_for(bus.get_inbound(), timeout=5) + bus.inbound_task_done() + await asyncio.sleep(0) + finally: + main_loop.set_exception_handler(previous_handler) + _stop_discord_loop(discord_loop, thread) + + assert inbound.text == "hello" + assert loop_errors == [] + + +def _fake_discord_module() -> SimpleNamespace: + class _Client: + def __init__(self, **_kwargs) -> None: + self.user = None + + def event(self, handler): + return handler + + return SimpleNamespace( + Intents=SimpleNamespace(default=lambda: SimpleNamespace()), + AllowedMentions=SimpleNamespace(none=lambda: None), + Client=_Client, + ) + + +@pytest.mark.anyio +async def test_discord_stop_closes_and_restart_reopens_threadsafe_submission_intake(): + channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + channel._running = True + channel._discord_loop = None + loop = asyncio.get_running_loop() + + async def _probe() -> None: + return None + + await channel.stop() + # stop() drains and closes intake, so a late SDK callback cannot start work. + assert channel._submit_threadsafe_coroutine(_probe(), loop, name="probe", msg_id=None) is False + + with ( + patch.dict(sys.modules, {"discord": _fake_discord_module()}), + patch.object(DiscordChannel, "_run_client") as run_client, + patch.object(DiscordChannel, "_load_active_threads"), + ): + await channel.start() + await asyncio.to_thread(channel._thread.join, 5) + run_client.assert_called_once() + + # A restarted channel must accept work again, or every message is dropped. + assert channel._submit_threadsafe_coroutine(_probe(), loop, name="probe", msg_id=None) is True + await channel._close_and_drain_threadsafe_futures()