github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-10-26 13:16:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(channels): run Discord connection-repository work on the Gateway loop (#6214)

Browse Source 
* fix(channels): run Discord connection-repository work on the Gateway loop

discord.py delivers messages on a private loop in the client thread, and
the Discord adapter awaited the channel-connection repository there even
though its SQLAlchemy engine and pool belong to the Gateway loop. On
PostgreSQL the first Discord message after the Gateway used the pool
failed with "got Future attached to a different loop" and was dropped; on
SQLite a pool-exhausting burst failed with "Queue is bound to a different
event loop" and left the pool's wait queue bound to the Discord loop, so
the Gateway's own queries then failed the same way.

Resolve the identity together with the intake commit, and run /connect
binding, on the Gateway loop through _submit_threadsafe_coroutine, as
Telegram, Feishu and DingTalk already do. Bind replies return through
_run_on_discord_loop, and start()/stop() now open and drain the
cross-thread submissions.

* docs(changelog): reference #6214 for the Discord connection-loop fix

* fix(channels): correct the unscheduled Discord bind comment and pin it

When _submit_threadsafe_coroutine refuses the bind, base.py closes the
coroutine, so consume_oauth_state never runs and the one-time code stays
valid; only the Discord message is swallowed. Say that instead of
claiming the code is consumed, and pin both halves with a test: the
attempt is handled (never a chat turn) and the code still binds later.

* fix(channels): undo Discord typing when the Gateway hand-off fails

Since identity resolution moved to the Gateway loop, both inbound paths
started typing and the ack reaction as soon as the hand-off was
scheduled. A lookup that then raised (database outage, pool-checkout
timeout) dropped the message but left an endless typing loop and a
success reaction behind.

The hand-off now returns its completion future (new
Channel._submit_threadsafe_coroutine_future, which the bool helper wraps)
and the Discord side waits on it through asyncio.shield. Typing still
registers before the hand-off settles, so a fast reply's _stop_typing
cannot run first; the ack reaction waits for the commit. On failure, or
when stop() drains the hand-off, the indicator this message started is
stopped (never one another in-flight message owns) and no ack is added.

* fix(channels): keep shared Discord typing alive for dependent messages

Two messages to the same target share one typing task: the first starts
it, a later one reuses it. If the creator's hand-off failed after a
follower had committed, or while the follower was still pending, the
creator stopped the indicator the follower relied on, and nothing
restored it while the follower's request ran.

Count the hand-offs that depend on each typing task. A pending or
committed hand-off holds the task (a commit keeps its hold until a reply
stops the indicator), a failed one releases its hold, and the indicator
is stopped only when the failed message started it and no hold remains.
The count is dropped when the task ends or the channel cancels typing.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Hyeonsang Cho 2026-10-03 18:14:19 +09:00 committed by GitHub
parent c7e24e4d14
commit aeeee2994f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 665 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
CHANGELOG.md
View File

@ -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

14
CHANGELOG_zh.md
View File

@ -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

4
backend/app/channels/AGENTS.md
View File

File diff suppressed because one or more lines are too long

35
backend/app/channels/base.py
View File

@ -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."""

157
backend/app/channels/discord.py
View File

@ -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

10
backend/tests/test_discord_channel.py
View File

@ -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

469
backend/tests/test_discord_channel_connections.py
View File

@ -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()