From df01102dfc458559abb1acf29d9aee1fb6d9b30b Mon Sep 17 00:00:00 2001 From: AoHanBei Date: Tue, 11 Aug 2026 21:54:19 +0800 Subject: [PATCH] fix(discord): prevent typing tasks after stop (#4752) * fix(discord): prevent typing tasks after stop * fix(discord): serialize typing cleanup on event loop * fix(discord): harden typing cleanup on loop exit --- backend/AGENTS.md | 2 +- backend/README.md | 2 + backend/app/channels/discord.py | 89 ++++++++++--- backend/tests/test_channels.py | 213 ++++++++++++++++++++++++++++++++ 4 files changed, 289 insertions(+), 17 deletions(-) diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 68ed63091..ce3b28a84 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -982,7 +982,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk **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) - `service.py` - Manages lifecycle of all configured channels from `config.yaml` -- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` 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`; `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`; `discord.py` registers typing loops before inbound handling yields and `_start_typing()` refuses work once `_running` is false; 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; `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. **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 - `github.py` - Webhook-driven GitHub channel. Inbound messages come from `POST /api/webhooks/github`; outbound is log-only because GitHub agents post explicitly with `gh` from their sandbox when they choose to comment or create a PR - `app/gateway/routers/channel_connections.py` - Browser-facing user connection and disconnect APIs diff --git a/backend/README.md b/backend/README.md index 59a557776..2c4571f0e 100644 --- a/backend/README.md +++ b/backend/README.md @@ -136,6 +136,8 @@ FastAPI application providing REST endpoints for frontend integration: The IM bridge supports Feishu, Slack, and Telegram. Slack and Telegram still use the final `runs.wait()` response path, while Feishu now streams through `runs.stream(["messages-tuple", "values"])`, serializes rapid same-thread turns inside the channel manager, and updates a single in-thread card per source message in place. +Discord registers each typing-indicator loop before inbound message handling yields and refuses to start new typing work after the channel stops. Typing tasks are owned by the dedicated Discord event loop, so normal shutdown schedules bounded cancellation, awaiting, and map cleanup on that loop before closing the client. The Discord worker also drains the tasks in its `finally` block while its loop is still usable, covering disconnect and exception exits; if `stop()` encounters an already-stopped foreign loop, it never awaits those loop-bound tasks from the main loop. This serializes registration and cleanup across the main and Discord threads while preventing shutdown hangs and cross-loop `RuntimeError`s. + For Feishu card updates, DeerFlow stores the running card's `message_id` per inbound message and patches that same card until the run finishes, preserving the existing `OK` / `DONE` reaction flow. When a follow-up arrives inside an existing Feishu topic while another turn is still running, the later message now waits on the mapped DeerFlow `thread_id`, receives a queued/running card on that exact source message, and keeps a compact source-message blockquote in subsequent patches so rapid consecutive questions remain distinguishable. --- diff --git a/backend/app/channels/discord.py b/backend/app/channels/discord.py index a4a5a936f..e00e5d6ea 100644 --- a/backend/app/channels/discord.py +++ b/backend/app/channels/discord.py @@ -174,26 +174,56 @@ class DiscordChannel(Channel): self._running = False self.bus.unsubscribe_outbound(self._on_outbound) - # Cancel all active typing indicator tasks - for target_id, task in list(self._typing_tasks.items()): - if not task.done(): - task.cancel() - logger.debug("[Discord] cancelled typing task for target %s", target_id) - self._typing_tasks.clear() - - if self._client and self._discord_loop and self._discord_loop.is_running(): - close_future = asyncio.run_coroutine_threadsafe(self._client.close(), self._discord_loop) + discord_loop = self._discord_loop + current_loop = asyncio.get_running_loop() + if discord_loop is None or discord_loop is current_loop: + await self._cancel_typing_tasks() + elif discord_loop.is_running(): + # Serialize cleanup with _start_typing() on the owning loop. The + # loop may exit between is_running() and scheduling, so neither + # scheduling nor waiting is allowed to block the rest of stop(). + cleanup_coro = self._cancel_typing_tasks() try: - await asyncio.wait_for(asyncio.wrap_future(close_future), timeout=10) - except TimeoutError: - logger.warning("[Discord] client close timed out after 10s") - except Exception: - logger.exception("[Discord] error while closing client") + cleanup_future = asyncio.run_coroutine_threadsafe(cleanup_coro, discord_loop) + except RuntimeError: + cleanup_coro.close() + logger.warning("[Discord] event loop stopped before typing-task cleanup could be scheduled") + else: + try: + await asyncio.wait_for(asyncio.wrap_future(cleanup_future), timeout=10) + except TimeoutError: + cleanup_future.cancel() + logger.warning("[Discord] typing-task cleanup timed out after 10s") + except Exception: + logger.exception("[Discord] error while cleaning up typing tasks") + + if self._client and discord_loop and discord_loop.is_running(): + close_coro = self._client.close() + try: + close_future = asyncio.run_coroutine_threadsafe(close_coro, discord_loop) + except RuntimeError: + close_coro.close() + logger.warning("[Discord] event loop stopped before client close could be scheduled") + else: + try: + await asyncio.wait_for(asyncio.wrap_future(close_future), timeout=10) + except TimeoutError: + close_future.cancel() + logger.warning("[Discord] client close timed out after 10s") + except Exception: + logger.exception("[Discord] error while closing client") if self._thread: self._thread.join(timeout=10) self._thread = None + # _run_client() normally drains these tasks in its finally block. If + # the owning loop was stopped externally before that cleanup ran, only + # discard the stale references here; awaiting them from this loop would + # raise a cross-loop RuntimeError. + if discord_loop and discord_loop is not current_loop and not discord_loop.is_running(): + self._discard_typing_tasks() + self._client = None self._discord_loop = None self._discord_module = None @@ -244,6 +274,8 @@ class DiscordChannel(Channel): async def _start_typing(self, channel, chat_id: str, thread_ts: str | None = None) -> None: """Starts a loop to send periodic typing indicators.""" + if not self._running: + return target_id = thread_ts or chat_id if target_id in self._typing_tasks: return # Already typing for this target @@ -262,6 +294,27 @@ class DiscordChannel(Channel): task = asyncio.create_task(_typing_loop()) self._typing_tasks[target_id] = task + async def _cancel_typing_tasks(self) -> None: + """Cancel and await every typing task on their owning event loop.""" + typing_tasks = list(self._typing_tasks.items()) + for target_id, task in typing_tasks: + if not task.done(): + task.cancel() + logger.debug("[Discord] cancelled typing task for target %s", target_id) + if typing_tasks: + await asyncio.gather(*(task for _, task in typing_tasks), return_exceptions=True) + self._typing_tasks.clear() + + def _discard_typing_tasks(self) -> None: + """Forget stale typing tasks after their owning event loop has stopped.""" + for target_id, task in self._typing_tasks.items(): + if not task.done(): + logger.warning( + "[Discord] discarding pending typing task for stopped-loop target %s", + target_id, + ) + self._typing_tasks.clear() + async def _stop_typing(self, chat_id: str, thread_ts: str | None = None) -> None: """Stops the typing loop for a specific target.""" target_id = thread_ts or chat_id @@ -352,7 +405,7 @@ class DiscordChannel(Channel): self._publish(inbound) # Start typing indicator in the thread if typing_target: - asyncio.create_task(self._start_typing(typing_target, chat_id, thread_id)) + await self._start_typing(typing_target, chat_id, thread_id) asyncio.create_task(self._add_reaction(message)) return @@ -460,7 +513,7 @@ class DiscordChannel(Channel): # Start typing indicator in the correct target (thread or channel) if typing_target: - asyncio.create_task(self._start_typing(typing_target, chat_id, thread_id)) + await self._start_typing(typing_target, chat_id, thread_id) self._publish(inbound) asyncio.create_task(self._add_reaction(message)) @@ -534,6 +587,10 @@ class DiscordChannel(Channel): if self._running: logger.exception("Discord client error") finally: + try: + self._discord_loop.run_until_complete(self._cancel_typing_tasks()) + except Exception: + logger.exception("Error while cleaning up Discord typing tasks") try: if self._client and not self._client.is_closed(): self._discord_loop.run_until_complete(self._client.close()) diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index e5f54dd3e..745a7aadc 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -5692,6 +5692,219 @@ class TestHandleChatWithArtifacts: _run(go()) +class TestDiscordChannel: + def test_stop_prevents_queued_typing_starter_from_creating_task(self): + from app.channels.discord import DiscordChannel + + async def go(): + channel = DiscordChannel(MessageBus(), config={}) + channel._running = True + typing_target = SimpleNamespace(trigger_typing=AsyncMock()) + + # Queue the starter without yielding to it. stop() therefore runs + # first and must form a boundary that the delayed starter cannot + # cross by installing a fresh infinite typing loop afterwards. + starter = asyncio.create_task(channel._start_typing(typing_target, "chat-1")) + await channel.stop() + await starter + + try: + assert channel._typing_tasks == {} + finally: + leaked_tasks = list(channel._typing_tasks.values()) + for task in leaked_tasks: + task.cancel() + await asyncio.gather(*leaked_tasks, return_exceptions=True) + + _run(go()) + + def test_stop_serializes_typing_cleanup_with_discord_loop(self): + from app.channels.discord import DiscordChannel + + class BlockingTypingTasks(dict): + def __init__(self, lookup_started: threading.Event, release_lookup: threading.Event): + super().__init__() + self._lookup_started = lookup_started + self._release_lookup = release_lookup + self.registered_task = None + + def __contains__(self, key): + self._lookup_started.set() + if not self._release_lookup.wait(timeout=5): + raise TimeoutError("typing-task lookup was not released") + return super().__contains__(key) + + def __setitem__(self, key, value): + self.registered_task = value + super().__setitem__(key, value) + + async def go(): + channel = DiscordChannel(MessageBus(), config={}) + channel._running = True + typing_target = SimpleNamespace(trigger_typing=AsyncMock()) + + discord_loop = asyncio.new_event_loop() + loop_ready = threading.Event() + + def run_discord_loop(): + asyncio.set_event_loop(discord_loop) + loop_ready.set() + discord_loop.run_forever() + + discord_thread = threading.Thread(target=run_discord_loop, daemon=True) + discord_thread.start() + assert await asyncio.to_thread(loop_ready.wait, 5) + + lookup_started = threading.Event() + release_lookup = threading.Event() + stop_started = threading.Event() + channel._discord_loop = discord_loop + typing_tasks = BlockingTypingTasks(lookup_started, release_lookup) + channel._typing_tasks = typing_tasks + channel.bus.unsubscribe_outbound = MagicMock(side_effect=lambda _callback: stop_started.set()) + + starter = asyncio.run_coroutine_threadsafe(channel._start_typing(typing_target, "chat-1"), discord_loop) + stop_task = None + + try: + # Pause the Discord loop after _start_typing() has observed + # _running=True but before it can create and register the task. + assert await asyncio.to_thread(lookup_started.wait, 5) + + stop_task = asyncio.create_task(channel.stop()) + # unsubscribe_outbound() runs immediately after stop() flips + # _running to False. Once this fires, release the Discord + # loop so any queued cleanup can serialize after registration. + assert await asyncio.to_thread(stop_started.wait, 5) + release_lookup.set() + + await asyncio.wait_for(stop_task, timeout=5) + await asyncio.wait_for(asyncio.wrap_future(starter), timeout=5) + + assert typing_tasks == {} + assert typing_tasks.registered_task is not None + assert typing_tasks.registered_task.done() + finally: + release_lookup.set() + if stop_task is not None and not stop_task.done(): + stop_task.cancel() + await asyncio.gather(stop_task, return_exceptions=True) + + if not starter.done(): + starter.cancel() + await asyncio.gather(asyncio.wrap_future(starter), return_exceptions=True) + + async def cleanup_typing_tasks(): + leaked_tasks = list(channel._typing_tasks.values()) + for task in leaked_tasks: + task.cancel() + await asyncio.gather(*leaked_tasks, return_exceptions=True) + channel._typing_tasks.clear() + + cleanup = asyncio.run_coroutine_threadsafe(cleanup_typing_tasks(), discord_loop) + await asyncio.wait_for(asyncio.wrap_future(cleanup), timeout=5) + discord_loop.call_soon_threadsafe(discord_loop.stop) + await asyncio.to_thread(discord_thread.join, 5) + assert not discord_thread.is_alive() + discord_loop.close() + + _run(go()) + + def test_stop_does_not_await_typing_tasks_from_stopped_discord_loop(self): + from app.channels.discord import DiscordChannel + + async def go(): + channel = DiscordChannel(MessageBus(), config={}) + channel._running = True + typing_target = SimpleNamespace(trigger_typing=AsyncMock()) + + discord_loop = asyncio.new_event_loop() + loop_ready = threading.Event() + + def run_discord_loop(): + asyncio.set_event_loop(discord_loop) + loop_ready.set() + discord_loop.run_forever() + + discord_thread = threading.Thread(target=run_discord_loop, daemon=True) + discord_thread.start() + assert await asyncio.to_thread(loop_ready.wait, 5) + + channel._discord_loop = discord_loop + channel._thread = discord_thread + starter = asyncio.run_coroutine_threadsafe(channel._start_typing(typing_target, "chat-1"), discord_loop) + await asyncio.wait_for(asyncio.wrap_future(starter), timeout=5) + typing_task = channel._typing_tasks["chat-1"] + + discord_loop.call_soon_threadsafe(discord_loop.stop) + await asyncio.to_thread(discord_thread.join, 5) + assert not discord_thread.is_alive() + assert not discord_loop.is_running() + assert not typing_task.done() + + try: + # The task belongs to a loop that has already exited. stop() + # must not gather it from the main loop, and must still finish + # clearing all channel lifecycle state. + await channel.stop() + + assert channel._typing_tasks == {} + assert channel._thread is None + assert channel._discord_loop is None + finally: + + def drain_stopped_loop(): + asyncio.set_event_loop(discord_loop) + if not typing_task.done(): + typing_task.cancel() + discord_loop.run_until_complete(asyncio.gather(typing_task, return_exceptions=True)) + discord_loop.close() + + await asyncio.to_thread(drain_stopped_loop) + + _run(go()) + + def test_run_client_drains_typing_tasks_before_worker_loop_exits(self): + from app.channels.discord import DiscordChannel + + channel = DiscordChannel(MessageBus(), config={}) + channel._running = True + typing_target = SimpleNamespace(trigger_typing=AsyncMock()) + + class FailingClient: + def __init__(self): + self.closed = False + self.typing_task = None + + async def start(self, _token): + await channel._start_typing(typing_target, "chat-1") + self.typing_task = channel._typing_tasks["chat-1"] + raise RuntimeError("simulated disconnect") + + def is_closed(self): + return self.closed + + async def close(self): + self.closed = True + + client = FailingClient() + channel._client = client + channel._bot_token = "token" + + try: + with patch("app.channels.discord.logger.exception"): + channel._run_client() + + assert channel._typing_tasks == {} + assert client.typing_task is not None + assert client.typing_task.done() + assert client.closed + finally: + discord_loop = channel._discord_loop + if discord_loop is not None and not discord_loop.is_closed(): + discord_loop.close() + + class TestFeishuChannel: def test_prepare_inbound_publishes_without_waiting_for_running_card(self): from app.channels.feishu import FeishuChannel