From 37b03a3811398f08a7b0ce1a8659694360297c87 Mon Sep 17 00:00:00 2001 From: hataa <79907651+hata33@users.noreply.github.com> Date: Thu, 10 Sep 2026 14:36:55 +0800 Subject: [PATCH] fix(channels): bound Discord outbound cross-loop awaits and restart dead clients, fixes #5226 (#5227) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(channels): bound Discord outbound cross-loop awaits and restart dead clients, fixes #5226 * docs: reduce inherited agent guidance size * fix(channels): stop half-started channels before discarding them _start_channel now tears the instance down (stop + untrack) whenever start() raises or the channel never reaches is_running, so an outbound listener subscribed before the transport was confirmed cannot outlive its channel. Addresses the review on #5227. * fix(channels): retain half-started channels until failed-start cleanup completes Ownership in _stop_and_discard_channel now mirrors ChannelService.stop(): the instance is dropped only after its stop() completes. A cancellation arriving mid-cleanup (or a stop() that raises) leaves it tracked, so a retried readiness attempt stops it again before replacing it and service shutdown can still reach it — untracking first orphaned resources nobody could clean up. Addresses the round-3 review on #5227. * fix(channels): defer replacement when a retained channel fails to stop The pre-retry stop in ensure_channel_ready popped unconditionally, so a retained instance whose second stop() raised was untracked with its outbound listener still subscribed — the same orphan one hop later. restart_channel (del after failed stop) and remove_channel (pop before stop) had the same shape. All three now route through _stop_and_discard_channel and decline the operation for that round when the instance is retained, so _start_channel can never overwrite a still-listening channel. Addresses the review on #5227. * fix(channels): enforce the retention guarantee inside the readiness attempt loop A failed attempt whose cleanup retained the instance used to let the next attempt (attempts=2 is the production default) construct a fresh instance and overwrite the retained one via _start_channel's unconditional assignment — orphaning the first instance's subscribed listener one hop earlier than the cross-round guard covers. The guard now lives at the mechanism: _start_channel refuses to install while the name is still tracked, and ensure_channel_ready ends the loop on retention. The shared discard helper's log message is path-neutral. Addresses the review on #5227. * fix(channels): make the retained-instance guard message path-neutral The guard can fire for any still-tracked instance, not only failed cleanup, so the message must not assume the cause. --------- Co-authored-by: Willem Jiang --- backend/app/channels/AGENTS.md | 2 +- backend/app/channels/discord.py | 65 ++++- backend/app/channels/service.py | 105 ++++++-- backend/tests/test_channels.py | 361 ++++++++++++++++++++++++++ backend/tests/test_discord_channel.py | 135 ++++++++++ 5 files changed, 640 insertions(+), 28 deletions(-) diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 6115d5831..7650d4877 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -12,7 +12,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). 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. - `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`; `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`) +- `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; 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/discord.py b/backend/app/channels/discord.py index 4a7a52ad2..d3ae4651e 100644 --- a/backend/app/channels/discord.py +++ b/backend/app/channels/discord.py @@ -19,6 +19,19 @@ logger = logging.getLogger(__name__) _DISCORD_MAX_MESSAGE_LEN = 2000 +# Bound for outbound work scheduled onto the Discord client's event loop. +# Discord API calls normally return in well under a second; anything still +# pending after this is a dead or wedged client, not a slow response. +DISCORD_OUTBOUND_TIMEOUT_SECONDS = 30.0 + +# File uploads carry an unbounded-size payload with no per-channel size cap +# (unlike Feishu/Telegram's send_file limits), so they get their own, larger +# bound: a 50 MB artifact over a ~5 Mbps uplink takes ~80 s to push, and a +# large 429 retry-after inside discord.py can extend that further. Cancelling +# a healthy upload mid-transfer would report it as failed, so the bound here +# only exists to convert a wedged client into a logged failure. +DISCORD_UPLOAD_TIMEOUT_SECONDS = 120.0 + class DiscordChannel(Channel): """Discord bot channel. @@ -239,10 +252,46 @@ class DiscordChannel(Channel): self._discord_module = None logger.info("Discord channel stopped") + @property + def is_running(self) -> bool: + """Running means the client thread is still alive, not just started. + + ``_run_client`` exits when discord.py gives up for good (invalidated + token, unrecoverable close) while ``_running`` stays True, so the base + flag alone would keep reporting a healthy channel forever. Mirrors + ``FeishuChannel.is_running`` so ``ChannelService.ensure_channel_ready`` + can restart the channel after its client thread dies. + """ + if not self._running: + return False + return self._thread is not None and self._thread.is_alive() + + async def _run_on_discord_loop(self, coro, *, timeout: float = DISCORD_OUTBOUND_TIMEOUT_SECONDS): + """Schedule *coro* on the Discord loop and await it with a bound. + + The Discord client runs on a dedicated thread whose loop is stopped but + not closed when the client dies, so ``call_soon_threadsafe`` keeps + queueing callbacks that never run and an unbounded ``wrap_future`` + await would hang a ChannelManager worker forever. ``stop()`` already + bounds its identical cross-loop awaits with ``wait_for``; this extends + that pattern to the outbound path. Failing fast when the loop is + missing or not running turns a dead client into a logged send failure + instead of a wedged worker. + """ + loop = self._discord_loop + if loop is None or not loop.is_running(): + coro.close() + raise RuntimeError("Discord client event loop is not running") + future = asyncio.run_coroutine_threadsafe(coro, loop) + try: + return await asyncio.wait_for(asyncio.wrap_future(future), timeout=timeout) + except TimeoutError: + future.cancel() + raise + async def send(self, msg: OutboundMessage) -> None: # Stop typing indicator once we're sending the response - stop_future = asyncio.run_coroutine_threadsafe(self._stop_typing(msg.chat_id, msg.thread_ts), self._discord_loop) - await asyncio.wrap_future(stop_future) + await self._run_on_discord_loop(self._stop_typing(msg.chat_id, msg.thread_ts)) target = await self._resolve_target(msg) if target is None: @@ -251,12 +300,10 @@ class DiscordChannel(Channel): text = msg.text or "" for chunk in self._split_text(text): - send_future = asyncio.run_coroutine_threadsafe(target.send(chunk), self._discord_loop) - await asyncio.wrap_future(send_future) + await self._run_on_discord_loop(target.send(chunk)) async def send_file(self, msg: OutboundMessage, attachment: ResolvedAttachment) -> bool: - stop_future = asyncio.run_coroutine_threadsafe(self._stop_typing(msg.chat_id, msg.thread_ts), self._discord_loop) - await asyncio.wrap_future(stop_future) + await self._run_on_discord_loop(self._stop_typing(msg.chat_id, msg.thread_ts)) target = await self._resolve_target(msg) if target is None: @@ -274,8 +321,7 @@ class DiscordChannel(Channel): # success and failure paths. data = await asyncio.to_thread(self._read_attachment_bytes, str(attachment.actual_path)) file = self._discord_module.File(io.BytesIO(data), filename=attachment.filename) - send_future = asyncio.run_coroutine_threadsafe(target.send(file=file), self._discord_loop) - await asyncio.wrap_future(send_future) + await self._run_on_discord_loop(target.send(file=file), timeout=DISCORD_UPLOAD_TIMEOUT_SECONDS) logger.info("[Discord] file uploaded: %s", attachment.filename) return True except Exception: @@ -773,9 +819,8 @@ class DiscordChannel(Channel): except (TypeError, ValueError): return None - get_future = asyncio.run_coroutine_threadsafe(self._fetch_channel(target_id), self._discord_loop) try: - return await asyncio.wrap_future(get_future) + return await self._run_on_discord_loop(self._fetch_channel(target_id)) except Exception: logger.exception("[Discord] failed to resolve target id=%s", raw_id) return None diff --git a/backend/app/channels/service.py b/backend/app/channels/service.py index ea0a36f17..8cf430e81 100644 --- a/backend/app/channels/service.py +++ b/backend/app/channels/service.py @@ -255,11 +255,15 @@ class ChannelService: return True if channel is not None: - try: - await channel.stop() - except Exception: - logger.exception("Error stopping non-running channel before readiness retry") - self._channels.pop(name, None) + # Ownership-preserving cleanup: the instance is retained when + # its stop() fails or is cancelled, and this round must NOT + # start a replacement over it — _start_channel would overwrite + # the tracked entry and orphan the still-subscribed listener + # one hop later (the gap this closes from the review on 5227). + await self._stop_and_discard_channel(name, channel) + if self._channels.get(name) is channel: + logger.warning("Readiness retry deferred: previous %s channel failed to stop and remains tracked", name) + return False max_attempts = max(1, attempts) for attempt in range(max_attempts): @@ -267,6 +271,13 @@ class ChannelService: logger.info("Retrying channel startup after readiness check") if await self._start_channel(name, channel_config): return True + # A failed attempt whose cleanup retained the instance ends + # the loop for this round: the next attempt would be refused + # by _start_channel's retained-instance guard anyway, and the + # still-tracked channel must not be replaced one hop later. + if self._channels.get(name) is not None: + logger.warning("Readiness retries deferred: %s channel failed to clean up after a failed start and remains tracked", name) + return False return False async def stop(self) -> None: @@ -329,11 +340,14 @@ class ChannelService: async def restart_channel(self, name: str, *, reload_config: bool = True) -> bool: """Restart a specific channel. Returns True if successful.""" if name in self._channels: - try: - await self._channels[name].stop() - except Exception: - logger.exception("Error stopping channel for restart") - del self._channels[name] + channel = self._channels[name] + # Same ownership rule as readiness retries: retain an instance + # whose stop() fails, and decline the restart rather than + # overwriting a still-tracked (still-listening) channel. + await self._stop_and_discard_channel(name, channel) + if self._channels.get(name) is channel: + logger.warning("Restart deferred: %s channel failed to stop and remains tracked", name) + return False if reload_config: # Reading config.yaml and the runtime store is disk IO; keep it @@ -364,16 +378,58 @@ class ChannelService: async def remove_channel(self, name: str) -> bool: """Remove runtime config for a channel and stop it if currently running.""" self._config.pop(name, None) - channel = self._channels.pop(name, None) + channel = self._channels.get(name) if channel is None: return True + # Stop-then-drop with the shared ownership rule: a channel whose + # stop() fails stays tracked (and returns False) instead of being + # popped first and leaking its subscribed listener on failure. + await self._stop_and_discard_channel(name, channel) + if self._channels.get(name) is channel: + logger.warning("Removal incomplete: %s channel failed to stop and remains tracked", name) + return False + logger.info("Channel stopped and removed") + return True + + async def _stop_and_discard_channel(self, name: str, channel: Channel) -> None: + """Stop a channel and drop it only once its ``stop()`` has completed. + + This is the single ownership-preserving cleanup every discard path + routes through (failed startup, readiness retry, restart, removal). + ``start()`` subscribes the outbound listener before the transport is + up, so an instance that never reached ``is_running`` — or a running + one being torn down — must be ``stop()``-ed before it is discarded: + otherwise the bus keeps a strong reference to the dead listener and + every future outbound for this channel name fans out to it, while + repeated attempts accumulate more stale listeners the service can no + longer clean up (the instances are untracked by then). Discord's + fail-fast ``is_running`` makes this reachable for a client thread that + dies immediately (invalid token); the same hygiene applies to any + channel that subscribes before its transport is confirmed. + + Ownership mirrors ``ChannelService.stop()``: the instance is dropped + only after its ``stop()`` actually completes. A cancellation arriving + mid-cleanup (or a ``stop()`` that raises) leaves it tracked, so a + retried readiness attempt stops it again before replacing it and + service shutdown can still reach it — untracking first would orphan + resources nobody can clean up anymore. Callers check for retention + (``self._channels.get(name) is channel``) and defer starting or + removing a replacement for that round, so startup cannot silently + overwrite a still-listening retained instance; ``ensure_channel_ready`` + additionally serializes on the per-channel readiness lock. + """ try: await channel.stop() - logger.info("Channel stopped and removed") - return True + except asyncio.CancelledError: + # Keep this transport owned by the service: the Gateway deadline + # interrupted cleanup, so detaching it here would hide resources + # that may still be in use (mirrors ChannelService.stop()). + raise except Exception: - logger.exception("Error stopping channel for removal") - return False + logger.exception("Error stopping channel %s during discard", name) + return + if self._channels.get(name) is channel: + self._channels.pop(name, None) async def _start_channel(self, name: str, config: dict[str, Any]) -> bool: """Instantiate and start a single channel.""" @@ -382,6 +438,17 @@ class ChannelService: logger.warning("Unknown channel type") return False + # Never install a fresh instance over a retained one: a channel whose + # failed cleanup kept it tracked still holds a subscribed outbound + # listener, and overwriting the entry here is the one remaining way to + # orphan it (nothing would be able to stop it afterwards). Callers + # decline the operation when they see the name still tracked; this + # guard makes the invariant hold at the mechanism itself. + retained = self._channels.get(name) + if retained is not None: + logger.warning("Refusing to start %s: another channel instance is still tracked under this name (previous cleanup incomplete, or the instance is still running)", name) + return False + try: from deerflow.reflection import resolve_class @@ -390,6 +457,7 @@ class ChannelService: logger.exception("Failed to import channel class") return False + channel: Channel | None = None try: config = dict(config) config["channel_store"] = self.store @@ -407,14 +475,17 @@ class ChannelService: self._channels[name] = channel await channel.start() if not channel.is_running: - self._channels.pop(name, None) logger.error("Channel did not enter a running state after start()") + await self._stop_and_discard_channel(name, channel) return False logger.info("Channel started") return True except Exception: - self._channels.pop(name, None) logger.exception("Failed to start channel") + if channel is not None: + await self._stop_and_discard_channel(name, channel) + else: + self._channels.pop(name, None) return False def get_status(self) -> dict[str, Any]: diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index 52f3906e9..971089b90 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -7519,6 +7519,367 @@ class TestChannelService: _run(go()) + def test_readiness_retry_defers_when_old_instance_fails_to_stop(self): + """A retained channel whose stop() fails must not be replaced this round. + + The pre-retry cleanup retains the instance when stop() raises; the + readiness attempt then declines instead of letting _start_channel + overwrite the still-tracked, still-listening channel — the one-hop- + later orphan shape from the review. + """ + from app.channels.base import Channel + from app.channels.service import ChannelService + + class FailingStopChannel(Channel): + def __init__(self, bus, config): + super().__init__(name="telegram", bus=bus, config=config) + self.stop_calls = 0 + self.bus.subscribe_outbound(self._on_outbound) + + async def start(self): + self._running = True + + async def stop(self): + self.stop_calls += 1 + self._running = False + raise RuntimeError("stop boom") + + async def send(self, msg): + raise NotImplementedError + + async def _on_outbound(self, msg): + raise AssertionError("a listener slated for cleanup must never receive outbounds") + + async def go(): + service = ChannelService(channels_config={"telegram": {"enabled": True, "bot_token": "x"}}) + await service.manager.start() + service._running = True + + stale = FailingStopChannel(bus=service.bus, config={}) + service._channels["telegram"] = stale + + ready = await service.ensure_channel_ready("telegram", attempts=2) + + assert ready is False + assert service._channels.get("telegram") is stale # retained, not overwritten + assert stale.stop_calls == 1 # each retry stops the retained instance again + # The listener stays subscribed precisely because the instance is + # retained: only a completed stop may unsubscribe it. + assert any(getattr(listener, "__self__", None) is stale for listener in service.bus._outbound_listeners) + + # Shutdown reports the retained channel's failing stop (ExceptionGroup) + # instead of silently orphaning it — expected here by construction. + with pytest.raises(Exception): + await service.stop() + + _run(go()) + + def test_readiness_attempts_do_not_replace_retained_instance(self, monkeypatch): + """Within one ensure_channel_ready loop, a failed attempt whose cleanup + retains the instance must end the loop instead of being overwritten. + + The reviewer repro on #5227: with attempts=2 (the production default), + a channel whose start() never reaches is_running AND whose stop() + raises used to let attempt 2 construct a fresh instance and overwrite + the retained one — returning True while the first instance's outbound + listener stayed subscribed forever. + """ + import deerflow.reflection as reflection_module + from app.channels.base import Channel + from app.channels.service import ChannelService + + async def go(): + service = ChannelService(channels_config={"telegram": {"enabled": True, "bot_token": "x"}}) + await service.manager.start() + service._running = True + + created = [] + + class FailFastAndUncleanChannel(Channel): + def __init__(self, bus, config): + super().__init__(name="telegram", bus=bus, config=config) + self.stop_calls = 0 + created.append(self) + + async def start(self): + # Subscribe the listener, then report a client thread that + # died before start() returned (the Discord invalid-token + # shape). + self.bus.subscribe_outbound(self._on_outbound) + self._running = True + + @property + def is_running(self) -> bool: + return False + + async def stop(self): + self.stop_calls += 1 + self._running = False + raise RuntimeError("stop boom") + + async def send(self, msg): + raise NotImplementedError + + async def _on_outbound(self, msg): + raise AssertionError("a listener slated for cleanup must never receive outbounds") + + monkeypatch.setattr(reflection_module, "resolve_class", lambda path, base_class=None: FailFastAndUncleanChannel) + + ready = await service.ensure_channel_ready("telegram", attempts=2) + + assert ready is False + assert len(created) == 1 # attempt 2 never constructed a replacement + retained = created[0] + assert service._channels.get("telegram") is retained + assert retained.stop_calls == 1 + assert any(getattr(listener, "__self__", None) is retained for listener in service.bus._outbound_listeners) + + with pytest.raises(Exception): + await service.stop() + + _run(go()) + + def test_restart_and_remove_retain_channel_when_stop_fails(self): + """restart_channel and remove_channel defer instead of orphaning a failed stop.""" + from app.channels.base import Channel + from app.channels.service import ChannelService + + class FailingStopChannel(Channel): + def __init__(self, bus, config): + super().__init__(name="telegram", bus=bus, config=config) + self.stop_calls = 0 + self.bus.subscribe_outbound(self._on_outbound) + + async def start(self): + self._running = True + + async def stop(self): + self.stop_calls += 1 + self._running = False + raise RuntimeError("stop boom") + + async def send(self, msg): + raise NotImplementedError + + async def _on_outbound(self, msg): + raise AssertionError("a listener slated for cleanup must never receive outbounds") + + async def go(): + service = ChannelService(channels_config={"telegram": {"enabled": True, "bot_token": "x"}}) + await service.manager.start() + service._running = True + + original = FailingStopChannel(bus=service.bus, config={}) + service._channels["telegram"] = original + + assert await service.restart_channel("telegram") is False + assert service._channels.get("telegram") is original + assert original.stop_calls == 1 + assert any(getattr(listener, "__self__", None) is original for listener in service.bus._outbound_listeners) + + assert await service.remove_channel("telegram") is False + assert service._channels.get("telegram") is original + assert original.stop_calls == 2 + assert any(getattr(listener, "__self__", None) is original for listener in service.bus._outbound_listeners) + + with pytest.raises(Exception): + await service.stop() + + _run(go()) + + def test_failed_channel_startup_is_transactional(self, monkeypatch): + """A channel that never reaches is_running must be stopped before discard. + + start() subscribes the outbound listener before the transport is + confirmed up, so a client thread that dies immediately (the Discord + invalid-token shape) must not leave a stale listener behind on the + bus — repeated readiness attempts would otherwise accumulate dead + listeners the service can no longer clean up. + """ + import deerflow.reflection as reflection_module + from app.channels.base import Channel + from app.channels.service import ChannelService + + async def go(): + service = ChannelService(channels_config={"telegram": {"enabled": True, "bot_token": "x"}}) + await service.manager.start() + service._running = True + + created = [] + + class DeadOnArrivalChannel(Channel): + def __init__(self, bus, config): + super().__init__(name="telegram", bus=bus, config=config) + self.stop_calls = 0 + created.append(self) + + async def start(self): + # Every adapter subscribes outbound before its transport is up. + self.bus.subscribe_outbound(self._on_outbound) + self._running = True + + @property + def is_running(self) -> bool: + return False + + async def stop(self): + self.stop_calls += 1 + self._running = False + self.bus.unsubscribe_outbound(self._on_outbound) + + async def send(self, msg): + raise NotImplementedError + + async def _on_outbound(self, msg): + raise AssertionError("a dead listener must never receive outbounds") + + monkeypatch.setattr(reflection_module, "resolve_class", lambda path, base_class=None: DeadOnArrivalChannel) + + ready = await service.ensure_channel_ready("telegram", attempts=3) + + assert ready is False + assert len(created) == 3 + assert all(channel.stop_calls == 1 for channel in created) + assert service.bus._outbound_listeners == [] + assert "telegram" not in service._channels + + await service.stop() + + _run(go()) + + def test_cancelled_failed_start_cleanup_retains_channel_until_cleaned(self, monkeypatch): + """Cancellation during failed-start cleanup must not orphan the channel. + + ``_stop_and_discard_channel`` keeps the half-started instance tracked + until its ``stop()`` completes: cancelling the readiness request + mid-cleanup (review repro) leaves the instance reachable, so a later + readiness retry stops it again before replacing it and service + shutdown can still clean it up. Untracking first would leave the + subscribed outbound listener owned by nobody — stop count stuck at + one and the listener still registered after ``service.stop()``. + """ + import deerflow.reflection as reflection_module + from app.channels.base import Channel + from app.channels.service import ChannelService + + async def go(): + service = ChannelService(channels_config={"telegram": {"enabled": True, "bot_token": "x"}}) + await service.manager.start() + service._running = True + + release = asyncio.Event() + created = [] + + class SuspendableStopChannel(Channel): + def __init__(self, bus, config): + super().__init__(name="telegram", bus=bus, config=config) + self.stop_calls = 0 + self.stop_entered = asyncio.Event() + created.append(self) + + async def start(self): + # Every adapter subscribes outbound before its transport is up. + self.bus.subscribe_outbound(self._on_outbound) + self._running = True + + @property + def is_running(self) -> bool: + return False + + async def stop(self): + self.stop_calls += 1 + self.stop_entered.set() + if not release.is_set(): + await release.wait() + self._running = False + self.bus.unsubscribe_outbound(self._on_outbound) + + async def send(self, msg): + raise NotImplementedError + + monkeypatch.setattr(reflection_module, "resolve_class", lambda path, base_class=None: SuspendableStopChannel) + + # Phase 1: readiness cancelled while the failed-start cleanup is + # suspended inside stop(). + readiness = asyncio.ensure_future(service.ensure_channel_ready("telegram", attempts=1)) + while not created: + await asyncio.sleep(0.01) + await created[0].stop_entered.wait() + readiness.cancel() + try: + await readiness + except asyncio.CancelledError: + pass + + retained = service._channels.get("telegram") + assert retained is created[0] # retained for retry/shutdown, not orphaned + assert retained.stop_calls == 1 # first cleanup was interrupted + assert service.bus._outbound_listeners # listener still registered + + # Phase 2: a later readiness retry stops the retained instance + # before replacing it — never swaps an uncleaned channel out. + release.set() + ready = await service.ensure_channel_ready("telegram", attempts=1) + assert ready is False + assert len(created) == 2 + assert created[0].stop_calls == 2 # cleanup completed on retry + assert created[1].stop_calls == 1 # replacement got its own teardown + assert service.bus._outbound_listeners == [] + assert "telegram" not in service._channels + + await service.stop() + + _run(go()) + + def test_start_channel_exception_stops_and_discards(self, monkeypatch): + """A start() that raises mid-way must also stop the half-started channel.""" + import deerflow.reflection as reflection_module + from app.channels.base import Channel + from app.channels.service import ChannelService + + async def go(): + service = ChannelService(channels_config={"telegram": {"enabled": True, "bot_token": "x"}}) + await service.manager.start() + service._running = True + + created = [] + + class StartRaisesChannel(Channel): + def __init__(self, bus, config): + super().__init__(name="telegram", bus=bus, config=config) + self.stop_calls = 0 + created.append(self) + + async def start(self): + self.bus.subscribe_outbound(self._on_outbound) + self._running = True + raise RuntimeError("simulated invalid token") + + async def stop(self): + self.stop_calls += 1 + self._running = False + self.bus.unsubscribe_outbound(self._on_outbound) + + async def send(self, msg): + raise NotImplementedError + + async def _on_outbound(self, msg): + raise AssertionError("a discarded listener must never receive outbounds") + + monkeypatch.setattr(reflection_module, "resolve_class", lambda path, base_class=None: StartRaisesChannel) + + ready = await service.ensure_channel_ready("telegram", attempts=2) + + assert ready is False + assert len(created) == 2 + assert all(channel.stop_calls == 1 for channel in created) + assert service.bus._outbound_listeners == [] + assert "telegram" not in service._channels + + await service.stop() + + _run(go()) + def test_session_config_is_forwarded_to_manager(self): from app.channels.service import ChannelService diff --git a/backend/tests/test_discord_channel.py b/backend/tests/test_discord_channel.py index dbeb45f09..ec9208029 100644 --- a/backend/tests/test_discord_channel.py +++ b/backend/tests/test_discord_channel.py @@ -433,3 +433,138 @@ async def test_stop_wiring_drains_ack_tasks_across_loops() -> None: assert not channel._ack_reaction_tasks finally: _stop_bg_loop(bg_loop, bg_thread) + + +# --------------------------------------------------------------------------- +# Dead-client fail-fast and is_running thread-aliveness +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_send_fails_fast_when_discord_loop_is_not_running() -> None: + """A stopped (not closed) Discord loop must fail the send, not hang the worker. + + ``_run_client`` leaves the loop stopped-but-unclosed when the client dies, + which is exactly the state where ``call_soon_threadsafe`` queues callbacks + that never run — the permanent-hang case this guards against. + """ + channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + channel._discord_loop = asyncio.new_event_loop() # created, never run + channel._running = True + msg = OutboundMessage(channel_name="discord", chat_id="c1", thread_id="t1", text="hello") + + try: + with pytest.raises(RuntimeError, match="event loop is not running"): + await channel.send(msg) + finally: + channel._discord_loop.close() + + +@pytest.mark.asyncio +async def test_outbound_loop_call_times_out_when_never_completes() -> None: + """Even on a live loop, an outbound call that never resolves is bounded by the timeout.""" + bg_loop, bg_thread = _start_bg_loop() + try: + channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + channel._discord_loop = bg_loop + + async def _never_completes() -> None: + await asyncio.sleep(3600) + + with pytest.raises(TimeoutError): + await channel._run_on_discord_loop(_never_completes(), timeout=0.1) + + # The cancelled call leaves its task parked on the bg loop (concurrent + # cancellation cannot reach a running run_coroutine_threadsafe task); + # clean it up so stopping the loop has nothing pending. + async def _cancel_leftovers() -> None: + current = asyncio.current_task() + for task in asyncio.all_tasks(): + if task is not current: + task.cancel() + + cleanup = asyncio.run_coroutine_threadsafe(_cancel_leftovers(), bg_loop) + cleanup.result(timeout=5) + finally: + _stop_bg_loop(bg_loop, bg_thread) + + +def test_is_running_tracks_thread_aliveness() -> None: + """``is_running`` reflects the client thread, so readiness can restart a dead channel.""" + channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + assert channel.is_running is False # never started + + channel._running = True + assert channel.is_running is False # started but the thread object is gone + + finished = threading.Thread(target=lambda: None) + finished.start() + finished.join() + channel._thread = finished + assert channel.is_running is False # dead client thread (fatal exit) + + channel._thread = threading.current_thread() + assert channel.is_running is True # live thread -> healthy + + +def _close_unawaited_mock_coroutines(run_mock) -> None: + """Close the coroutines handed to an AsyncMock that never awaited them. + + ``_run_on_discord_loop`` receives already-created coroutine objects; an + AsyncMock stand-in records them without awaiting, so they must be closed + explicitly or GC warns about never-awaited coroutines. + """ + for call in run_mock.await_args_list: + call.args[0].close() + + +@pytest.mark.asyncio +async def test_send_file_upload_call_uses_the_dedicated_upload_timeout(tmp_path) -> None: + """Pin the upload call site to DISCORD_UPLOAD_TIMEOUT_SECONDS. + + Regressing ``send_file``'s upload call to the 30 s control-plane default + (the exact bug round 1 of this review caught) keeps every helper-level + test green; only the call site's ``timeout=`` kwarg can guard it. + """ + from app.channels.discord import DISCORD_UPLOAD_TIMEOUT_SECONDS + + channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + channel._discord_module = SimpleNamespace(File=lambda fp, filename=None: fp) + run_mock = AsyncMock(return_value=None) + channel._run_on_discord_loop = run_mock # type: ignore[method-assign] + channel._resolve_target = _resolve_to(SimpleNamespace(send=_noop_coro)) + + path = tmp_path / "upload.txt" + path.write_bytes(b"hello") + att = ResolvedAttachment("/mnt/user-data/outputs/upload.txt", path, "upload.txt", "text/plain", 5, False) + msg = OutboundMessage(channel_name="discord", chat_id="c1", thread_id="t1", text="t") + + try: + assert await channel.send_file(msg, att) is True + finally: + _close_unawaited_mock_coroutines(run_mock) + + assert len(run_mock.await_args_list) == 2 # stop_typing, then the upload + stop_call, upload_call = run_mock.await_args_list + assert "timeout" not in stop_call.kwargs # control-plane call keeps the 30 s default + assert upload_call.kwargs.get("timeout") == DISCORD_UPLOAD_TIMEOUT_SECONDS + + +@pytest.mark.asyncio +async def test_send_control_calls_keep_the_default_outbound_bound() -> None: + """``send``'s typing-stop and message sends rely on the 30 s default, not an override.""" + channel = DiscordChannel(bus=MessageBus(), config={"bot_token": "token"}) + run_mock = AsyncMock(return_value=None) + channel._run_on_discord_loop = run_mock # type: ignore[method-assign] + channel._resolve_target = _resolve_to(SimpleNamespace(send=_noop_coro)) + + msg = OutboundMessage(channel_name="discord", chat_id="c1", thread_id="t1", text="hello") + + try: + await channel.send(msg) + finally: + _close_unawaited_mock_coroutines(run_mock) + + assert len(run_mock.await_args_list) == 2 # stop_typing + one text chunk + for call in run_mock.await_args_list: + assert "timeout" not in call.kwargs