diff --git a/README.md b/README.md index e9cbb37f1f..1b98e436fe 100644 --- a/README.md +++ b/README.md @@ -1050,6 +1050,7 @@ Notes: - IM channel workers call Gateway's LangGraph-compatible API internally and automatically attach process-local internal auth plus the CSRF cookie/header pair required for thread and run creation. - Inbound work is bounded to `inbound_queue_maxsize` pending messages plus `max_concurrency` active workers. When capacity is exhausted, socket/polling providers drop new messages before sending DeerFlow's working acknowledgment and emit a rate-limited warning. Buzz leaves its replay cursor unchanged and reconnects for relay replay; GitHub webhooks return `503`, marking the delivery failed for manual/API redelivery. Shutdown closes admission immediately, keeps channel transports available while accepted messages drain for up to `shutdown_grace_period_seconds`, then cancels and awaits active handlers before closing provider resources; the Gateway's outer timeout can cancel an incomplete shutdown without detaching those resources. - Telegram shutdown retains ongoing cleanup across a timeout or cancellation so a retry can finish it. If the Telegram loop has already stopped, shutdown warns that download clients may remain open and releases the abandoned cleanup state; later stops do not repeat that warning. +- Slack stop waits up to five seconds for an in-flight Socket Mode connection and its cleanup. If cleanup is still pending, the previous instance stays tracked and a restart can be retried after it finishes. SDK cleanup failures report their original error and remain retryable. - Feishu/Lark now queues rapid follow-up messages per mapped DeerFlow `thread_id` instead of immediately surfacing the generic busy reply, and topic replies keep a per-message card with a compact source-message preview across queued/running/final patches. - Streaming IM channels treat backend `error` events like transport failures and follow the same reply and retry path. The channel logs the error type and message for diagnosis. diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 7c0764066d..529c96af69 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -17,9 +17,9 @@ remains unconfirmed. - `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn) and `/agent` (`list` is owner-scoped; `use` creates a fresh thread and persists the Custom Agent selection under both the channel restart key and Web's canonical routing metadata so existing checkpoint lineage never changes agent across IM/Web), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout` Streaming `error` frames use the same failure path as transport exceptions. Publish the final outbound before releasing the inbound dedupe key, so provider redelivery can retry without overtaking the terminal reply. Log the frame's error name and message at WARNING with the thread ID; keep the synthetic error generic so payload details do not change reply selection. **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()` (or `_submit_threadsafe_coroutine_future()` to await the outcome through `asyncio.shield`): it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. Blocking SDK teardown in `stop()` (a client `close()`, a thread `join()`) must run off the event loop through `asyncio.to_thread`, since every run and channel on the Gateway shares that loop and `stop()` also serves `POST /api/channels/{name}/restart`; Slack's `SocketModeClient.close()` is a shielded, tracked task so a cancelled stop leaves it running and a retried stop awaits the same close. Anchored by `tests/blocking_io/test_channel_stop_offloop.py`. **Proactive push** (`Channel.send_notification`): the scheduled-task outbox pushes markdown to a bound identity with no inbound frame. The base default raises `NotImplementedError` (after a `ChannelUnavailable` check when not running). `capabilities.py` gates enqueue/UI; unsupported legacy rows fail terminally (`test_channel_capabilities.py`). Await ACK; `ChannelUnavailable` parks transport failures, rejections count. Group targets use owner-bound connection IDs and a separate group hook; see [group delivery](../../docs/IM_CHANNEL_CONNECTIONS.md#feishu-group-notifications). **WeCom/Feishu push**; Feishu uses owner-scoped p2p routing. WeCom uses `WSClient.send_message`; its SDK raises `RuntimeError` for a dead socket and a rejection alike, so only the `Reply ack error` prefix counts; other `RuntimeError`s, `ConnectionError`/`TimeoutError`/`OSError` and the `websockets` close exception park. Before sending, the worker re-checks the binding (owner-scoped `resolve_connections`, before liveness): a gone target is `mark_failed(terminal=True)`; a lookup failure parks. Terminal rows ignore later status writes. Scheduler side: `_on_finalization` enqueues each notice in the outcome's transaction, so recovery notifies once (unique key; idle ends use `idle:`); one message per occurrence (stop > pause > finish > run outcome); plain trials and interrupts stay silent; text (`scheduler/notification_text.py`) is localized, without IDs or links. Lifespan starts `NotificationDeliveryWorker` after the channel service (detaching enqueue if it cannot) and stops it first. +- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()` (or `_submit_threadsafe_coroutine_future()` to await the outcome through `asyncio.shield`): it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. Blocking SDK teardown in `stop()` (a client `close()`, a thread `join()`) must run off the event loop through `asyncio.to_thread`, since every run and channel on the Gateway shares that loop and `stop()` also serves `POST /api/channels/{name}/restart`; Slack cleanup awaits connect then closes off-loop; a 5s deadline/cancellation retains it for retry. SDK close errors propagate. Anchored by `tests/blocking_io/test_channel_stop_offloop.py`. **Proactive push** (`Channel.send_notification`): the scheduled-task outbox pushes markdown to a bound identity with no inbound frame. The base default raises `NotImplementedError` (after a `ChannelUnavailable` check when not running). `capabilities.py` gates enqueue/UI; unsupported legacy rows fail terminally (`test_channel_capabilities.py`). Await ACK; `ChannelUnavailable` parks transport failures, rejections count. Group targets use owner-bound connection IDs and a separate group hook; see [group delivery](../../docs/IM_CHANNEL_CONNECTIONS.md#feishu-group-notifications). **WeCom/Feishu push**; Feishu uses owner-scoped p2p routing. WeCom uses `WSClient.send_message`; its SDK raises `RuntimeError` for a dead socket and a rejection alike, so only the `Reply ack error` prefix counts; other `RuntimeError`s, `ConnectionError`/`TimeoutError`/`OSError` and the `websockets` close exception park. Before sending, the worker re-checks the binding (owner-scoped `resolve_connections`, before liveness): a gone target is `mark_failed(terminal=True)`; a lookup failure parks. Terminal rows ignore later status writes. Scheduler side: `_on_finalization` enqueues each notice in the outcome's transaction, so recovery notifies once (unique key; idle ends use `idle:`); one message per occurrence (stop > pause > finish > run outcome); plain trials and interrupts stay silent; text (`scheduler/notification_text.py`) is localized, without IDs or links. Lifespan starts `NotificationDeliveryWorker` after the channel service (detaching enqueue if it cannot) and stops it first. - `wecom.py` - WeCom WebSocket channel with proactive `send_notification`; `_ws_transport_up` maps disconnects to `ChannelUnavailable` (no retry used). `capabilities.py` - SDK-free `CHANNEL_CAPABILITIES`. -- `service.py` - Drain manager work before transports; retain timed-out instances for retry. Live Feishu/DingTalk/Telegram joins raise `ChannelStopTimeout`: retain, defer replacement/restart, log WARNING without traceback; wedged instances report not running. Reload returns snapshots; restart owns cached fallback/publication under the channel lock. `_stopping` fences starts/publication (including after pre-start I/O) until shutdown succeeds. Lock manager startup through stale cleanup; capture the shutdown generation before locking to fence queued starts. `stop()` stays unlocked to fence in-flight startup. +- `service.py` - Drain manager work before transports; retain timed-out instances for retry. Live Feishu/DingTalk/Telegram joins and Slack cleanup raise `ChannelStopTimeout`: retain, defer replacement/restart, log WARNING without traceback; wedged instances report not running. Reload returns snapshots; restart owns cached fallback/publication under the channel lock. `_stopping` fences starts/publication (including after pre-start I/O) until shutdown succeeds. Lock manager startup through stale cleanup; capture the shutdown generation before locking to fence queued starts. `stop()` stays unlocked to fence in-flight startup. - `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, edits the "Working on it..." stream target in place via `editMessageText`, and can optionally send final Markdown replies as Rich Messages through `channels.telegram.rich_messages` — the Rich path only fires when `rich_messages` is on **and** the text contains a rich construct (fenced code, emphasis, a table, a task list, `
`, block math, or a link), so structured command/error replies (none of these) stay plain text and their newlines and `` tokens survive; `discord.py` runs `_on_message` on discord.py's private loop, so connection-repository work (identity lookup with the reservation commit, `/connect` binding) goes to the Gateway loop through `_submit_threadsafe_coroutine` — the SQLAlchemy engine is bound there — and bind replies return through `_run_on_discord_loop`; typing registers before that hand-off settles, and a failed hand-off skips the ack and stops the indicator it started unless another message still relies on it; `discord.py` registers typing loops before inbound handling yields and `_start_typing()` refuses work once `_running` is false; each tick sends the indicator with discord.py 2.x's `await channel.typing()` (`trigger_typing()` was removed in 2.0, and each typing loop logs its first failed tick at WARNING, later ones at DEBUG); because `stop()` runs on the main loop while typing tasks belong to `_discord_loop`, normal cross-thread cancellation, awaiting, and map cleanup are scheduled there with a bounded wait, while `_run_client()` drains tasks in its `finally` block before an exception/disconnect can make that loop unusable; an already-stopped foreign loop must never have its tasks awaited from the main loop; to that end every outbound cross-loop call (`send` / `send_file` / `_get_channel_or_thread`) goes through `_run_on_discord_loop`, which bounds the await (`DISCORD_OUTBOUND_TIMEOUT_SECONDS` 30s; `DISCORD_UPLOAD_TIMEOUT_SECONDS` 120s for file uploads, whose unbounded-size payload needs room for a slow uplink plus 429 retry-after) and fails fast with a `RuntimeError` when the client loop is missing or not running — a dead client becomes a logged send failure instead of a permanently hung `ChannelManager` worker — and `is_running` reports client-thread aliveness (like `feishu.py`) so `ensure_channel_ready` can restart the channel after `_run_client()` exits on a fatal error; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured, and overrides `receive_file` to download inbound images (`picture`/`richText`) and documents (`file`) by `downloadCode` into the thread uploads bucket, mirroring `feishu.py`) - `buzz.py` - Buzz (Nostr relay) implementation: one NIP-42-authenticated websocket, pubkey-allowlist + mention/DM/thread-follow gating (thread-follow reads via the manager's `connection_identity.lookup_thread_id`, so bound authors' repo-only threads count), streaming replies via in-place kind-40003 edits; requires the `buzz` dependency extra. Its durable seen-event replay guard uses async `aseen()` / `arecord()` / `aflush()` boundaries: initial JSON reads and coalesced atomic writes run off the Gateway event loop, and `stop()` awaits the final flush before returning. **Subscription model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-subscription-model)): the relay fans kind-9 chat events out **only** to channel-scoped subscriptions, proved on a live relay: `REQ {"kinds":[9]}` is accepted and answered with `EOSE` but never receives an event, `REQ {"kinds":[9],"#h":[uuid]}` works, and a multi-value `#h` matches nothing, so it is strictly one REQ per channel. 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** — 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 (a one-off hiccup is the common case); later ones back off 1s then 2s, awaited inline in the read loop, not in a background task that could outlive its 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 a false operator-facing warning that discovery/membership tracking was DOWN while discovery then completed normally. 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), `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 so one can be replaced or closed without disturbing the others, 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:** 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: one shared cursor is the newest event seen in *any* channel, so a busy channel would drag it past a quiet channel's unread messages, skipping them on the next reconnect. 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. diff --git a/backend/app/channels/slack.py b/backend/app/channels/slack.py index 0cf6e1cac4..51a78c1db8 100644 --- a/backend/app/channels/slack.py +++ b/backend/app/channels/slack.py @@ -10,12 +10,13 @@ from typing import Any from markdown_to_mrkdwn import SlackMarkdownConverter -from app.channels.base import Channel +from app.channels.base import Channel, ChannelStopTimeout from app.channels.commands import is_known_channel_command from app.channels.connection_identity import attach_connection_identity from app.channels.message_bus import InboundMessageType, InboundReservation, MessageBus, OutboundMessage, ResolvedAttachment logger = logging.getLogger(__name__) +SLACK_STOP_TIMEOUT_SECONDS = 5.0 _slack_md_converter = SlackMarkdownConverter() @@ -103,6 +104,7 @@ class SlackChannel(Channel): def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None: super().__init__(name="slack", bus=bus, config=config) self._socket_client = None + self._socket_connect_future: asyncio.Future[None] | None = None self._socket_close_task: asyncio.Task[None] | None = None self._web_client = None self._loop: asyncio.AbstractEventLoop | None = None @@ -115,6 +117,10 @@ class SlackChannel(Channel): async def start(self) -> None: if self._running: return + if self._socket_client is not None or self._socket_close_task is not None: + # A timed-out or cancelled stop still owns its SDK work. Reuse its + # bounded wait rather than replacing a client that can still dial. + await self.stop() try: from slack_sdk import WebClient @@ -152,12 +158,11 @@ class SlackChannel(Channel): self._running = True self.bus.subscribe_outbound(self._on_outbound) - # Start socket mode in background thread. Go through - # ``_connect_socket_mode`` so a connect failure gets logged: nothing - # awaits the future ``run_in_executor`` returns, so an exception stored - # on it would never surface. + # Start socket mode in a background thread. + # ``_connect_socket_mode`` logs startup failures. Retain the future so + # stop() can wait for an in-flight dial before closing its socket. socket_client = self._socket_client - asyncio.get_event_loop().run_in_executor(None, self._connect_socket_mode, socket_client) + self._socket_connect_future = asyncio.get_event_loop().run_in_executor(None, self._connect_socket_mode, socket_client) logger.info("Slack channel started") def _connect_socket_mode(self, socket_client: Any) -> None: @@ -166,7 +171,7 @@ class SlackChannel(Channel): Mirrors the Telegram and Discord channels, whose background threads log their errors instead of letting them vanish with the thread. """ - if socket_client is not self._socket_client: + if not self._running or socket_client is not self._socket_client: return try: socket_client.connect() @@ -177,24 +182,35 @@ class SlackChannel(Channel): self._running = False self.bus.unsubscribe_outbound(self._on_outbound) await self._close_and_drain_threadsafe_futures() - # ``SocketModeClient.close()`` joins the SDK's message-processor thread - # (up to ~1s) and waits for in-flight listeners, whose blocking Web API - # calls run up to the WebClient timeout, so it runs off the event loop. - # The client is detached first so a still-queued connect skips it. The - # close task is shielded and tracked: a cancelled stop() leaves it - # running, and a retried stop() awaits it instead of closing twice. - socket_client, self._socket_client = self._socket_client, None - if socket_client is not None: - self._socket_close_task = asyncio.create_task(asyncio.to_thread(socket_client.close)) + # The SDK can publish a session after close() if connect() is already + # waiting for its URL. One owned task drains that connect and then + # closes off-loop. A timeout or cancellation keeps it available for + # retry; cancelling an executor future would not stop its thread. + if self._socket_close_task is None and self._socket_client is not None: + self._socket_close_task = asyncio.create_task(self._close_socket_client(self._socket_client, self._socket_connect_future)) close_task = self._socket_close_task if close_task is not None: try: - await asyncio.shield(close_task) - finally: + await asyncio.wait_for(asyncio.shield(close_task), timeout=SLACK_STOP_TIMEOUT_SECONDS) + except TimeoutError as exc: if close_task.done(): + close_task.result() + else: + raise ChannelStopTimeout("Slack Socket Mode cleanup is still running") from exc + finally: + if close_task.done() and self._socket_close_task is close_task: self._socket_close_task = None logger.info("Slack channel stopped") + async def _close_socket_client(self, socket_client: Any, connect_future: asyncio.Future[None] | None) -> None: + if connect_future is not None: + await asyncio.shield(connect_future) + await asyncio.to_thread(socket_client.close) + if self._socket_client is socket_client: + self._socket_client = None + if self._socket_connect_future is connect_future: + self._socket_connect_future = None + async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None: web_client = await self._get_web_client_for_message(msg) if not web_client: diff --git a/backend/tests/blocking_io/test_channel_stop_offloop.py b/backend/tests/blocking_io/test_channel_stop_offloop.py index b4baffbd30..3f0c3ee684 100644 --- a/backend/tests/blocking_io/test_channel_stop_offloop.py +++ b/backend/tests/blocking_io/test_channel_stop_offloop.py @@ -18,10 +18,12 @@ gate raises ``BlockingError`` from the thread-join lock wait. from __future__ import annotations +import asyncio import threading import pytest +from app.channels.base import ChannelStopTimeout from app.channels.dingtalk import DingTalkChannel from app.channels.discord import DiscordChannel from app.channels.feishu import FeishuChannel @@ -47,6 +49,45 @@ async def test_slack_stop_closes_real_socket_mode_client_off_loop() -> None: assert channel._socket_client is None +async def test_slack_stop_keeps_connect_cleanup_off_loop(monkeypatch) -> None: + started, release = threading.Event(), threading.Event() + loop_thread = threading.get_ident() + close_threads = [] + + class SocketClient: + def connect(self): + started.set() + release.wait(5) + + def close(self): + close_threads.append(threading.get_ident()) + + monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 0.02) + channel = SlackChannel(MessageBus(), {}) + client = SocketClient() + channel._socket_client = client + channel._running = True + future = asyncio.get_running_loop().run_in_executor(None, channel._connect_socket_mode, client) + channel._socket_connect_future = future + try: + assert await asyncio.to_thread(started.wait, 2) + with pytest.raises(ChannelStopTimeout): + await channel.stop() + assert not future.cancelled() + assert close_threads == [] + release.set() + await asyncio.wait_for(asyncio.shield(channel._socket_close_task), 2) + await channel.stop() + assert len(close_threads) == 1 + assert close_threads[0] != loop_thread + assert channel._socket_client is None + finally: + release.set() + await asyncio.wait_for(asyncio.shield(future), 5) + monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 3) + await channel.stop() + + @pytest.mark.parametrize("channel_cls", [FeishuChannel, DingTalkChannel, DiscordChannel]) async def test_stop_joins_sdk_thread_off_loop(channel_cls) -> None: release = threading.Event() diff --git a/backend/tests/test_slack_channel_connections.py b/backend/tests/test_slack_channel_connections.py index cec537d22f..3b908fac84 100644 --- a/backend/tests/test_slack_channel_connections.py +++ b/backend/tests/test_slack_channel_connections.py @@ -179,8 +179,8 @@ def test_slack_socket_mode_connect_failure_is_logged(monkeypatch, caplog): }, ) - # start() drops the future for the executor-backed connect; capture it - # here so the test can wait for the connect attempt to finish. + # Capture the executor-backed connect so the test can wait for the + # failing attempt before checking its logged error. loop = asyncio.get_running_loop() real_run_in_executor = loop.run_in_executor connect_futures = [] @@ -252,25 +252,31 @@ def test_slack_socket_mode_queued_connect_does_not_target_replacement_client(mon # same executor path and must still run. if func != channel._connect_socket_mode: return real_run_in_executor(executor, func, *args) - queued_callbacks.append((func, args)) - return loop.create_future() + future = loop.create_future() + queued_callbacks.append((func, args, future)) + return future monkeypatch.setattr(loop, "run_in_executor", queue_run_in_executor) await channel.start() - await channel.stop() + stopping = asyncio.create_task(channel.stop()) + await asyncio.sleep(0) + stale_func, stale_args, stale_future = queued_callbacks[0] + stale_func(*stale_args) + stale_future.set_result(None) + await stopping await channel.start() assert len(clients) == 2 assert len(queued_callbacks) == 2 - stale_func, stale_args = queued_callbacks[0] stale_func(*stale_args) assert clients[0].connect_calls == 0 assert clients[1].connect_calls == 0 - current_func, current_args = queued_callbacks[1] + current_func, current_args, current_future = queued_callbacks[1] current_func(*current_args) + current_future.set_result(None) assert clients[0].connect_calls == 0 assert clients[1].connect_calls == 1 diff --git a/backend/tests/test_slack_channel_lifecycle.py b/backend/tests/test_slack_channel_lifecycle.py new file mode 100644 index 0000000000..d200589764 --- /dev/null +++ b/backend/tests/test_slack_channel_lifecycle.py @@ -0,0 +1,243 @@ +"""Slack stop retains executor-backed connects until their sockets are closed.""" + +from __future__ import annotations + +import asyncio +import contextlib +import socket +import threading +from contextlib import asynccontextmanager +from unittest.mock import MagicMock + +import pytest + +from app.channels.base import ChannelStopTimeout +from app.channels.message_bus import MessageBus +from app.channels.service import ChannelService +from app.channels.slack import SlackChannel + +pytestmark = pytest.mark.asyncio + + +def _capture_connect(monkeypatch, channel): + loop = asyncio.get_running_loop() + run_in_executor = loop.run_in_executor + futures = [] + + def capture(executor, func, *args): + future = run_in_executor(executor, func, *args) + if func == channel._connect_socket_mode: + futures.append(future) + return future + + monkeypatch.setattr(loop, "run_in_executor", capture) + return futures + + +@asynccontextmanager +async def _blocked_connect(monkeypatch): + import slack_sdk.socket_mode + + started, release = threading.Event(), threading.Event() + clients = [] + + class SocketClient: + def __init__(self, **kwargs): + self.socket_mode_request_listeners = [] + self.connect_finished = False + self.connected = False + self.close_calls = 0 + clients.append(self) + + def connect(self): + started.set() + if not release.wait(5): + raise TimeoutError("test did not release connection") + self.connected = True + self.connect_finished = True + + def close(self): + self.close_calls += 1 + self.connected = False + + monkeypatch.setattr(slack_sdk.socket_mode, "SocketModeClient", SocketClient) + channel = SlackChannel(MessageBus(), {"bot_token": "test", "app_token": "test", "bot_user_id": "UBOT", "web_client_factory": lambda **kwargs: object()}) + futures = _capture_connect(monkeypatch, channel) + try: + await channel.start() + assert await asyncio.to_thread(started.wait, 2) + yield channel, clients, futures[0], release + finally: + release.set() + for future in futures: + await asyncio.wait_for(asyncio.shield(future), 5) + monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 3, raising=False) + await asyncio.wait_for(channel.stop(), 4) + + +async def test_stop_preserves_sdk_close_timeout_and_retries(): + error = TimeoutError("SDK close failed") + socket_client = MagicMock() + socket_client.close.side_effect = [error, None] + channel = SlackChannel(MessageBus(), {}) + channel._socket_client = socket_client + + stopping = asyncio.create_task(channel.stop()) + await asyncio.sleep(0) + cleanup = channel._socket_close_task + assert cleanup is not None + with pytest.raises(TimeoutError, match="SDK close failed") as caught: + await stopping + # Python may copy TimeoutError across the executor boundary; stop must + # preserve the exception actually produced by its owned cleanup task. + assert caught.value is cleanup.exception() + assert type(caught.value) is TimeoutError + assert caught.value.__cause__ is not caught.value + assert channel._socket_client is socket_client + assert channel._socket_close_task is None + assert socket_client.close.call_count == 1 + + await channel.stop() + assert socket_client.close.call_count == 2 + assert channel._socket_client is None + assert channel._socket_close_task is None + + +async def test_stop_timeout_retains_connect_and_same_cleanup_for_retry(monkeypatch): + monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 0.02, raising=False) + async with _blocked_connect(monkeypatch) as (channel, clients, connect, release): + with pytest.raises(ChannelStopTimeout, match="cleanup is still running"): + await channel.stop() + cleanup = channel._socket_close_task + assert cleanup is not None and not cleanup.done() + assert channel._socket_client is clients[0] + assert channel._socket_connect_future is connect + assert not connect.done() + assert clients[0].close_calls == 0 + + with pytest.raises(ChannelStopTimeout): + await channel.start() + assert channel._socket_close_task is cleanup + assert len(clients) == 1 + assert not connect.cancelled() + + release.set() + await asyncio.wait_for(asyncio.shield(cleanup), 2) + await channel.stop() + assert clients[0].connect_finished + assert clients[0].close_calls == 1 + assert not clients[0].connected + assert channel._socket_client is None + assert channel._socket_connect_future is None + assert channel._socket_close_task is None + + +async def test_cancelling_stop_does_not_cancel_connect_or_duplicate_cleanup(monkeypatch): + async with _blocked_connect(monkeypatch) as (channel, clients, connect, release): + stopping = asyncio.create_task(channel.stop()) + await asyncio.sleep(0) + stopping.cancel() + with pytest.raises(asyncio.CancelledError): + await stopping + cleanup = channel._socket_close_task + assert cleanup is not None and not cleanup.done() + assert channel._socket_client is clients[0] + assert not connect.cancelled() + assert clients[0].close_calls == 0 + + release.set() + await channel.stop() + assert clients[0].close_calls == 1 + assert channel._socket_close_task is None + + +async def test_service_retains_slack_while_connect_is_still_owned(monkeypatch): + monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 0.02, raising=False) + async with _blocked_connect(monkeypatch) as (channel, clients, connect, release): + service = ChannelService({"slack": {"enabled": True}}, store=MagicMock()) + service._running = True + service._channels["slack"] = channel + assert await service.restart_channel("slack", reload_config=False) is False + assert service._channels["slack"] is channel + assert channel._socket_client is clients[0] + assert not connect.cancelled() + assert len(clients) == 1 + + release.set() + await asyncio.wait_for(asyncio.shield(channel._socket_close_task), 2) + assert await service.remove_channel("slack") is True + assert "slack" not in service._channels + assert clients[0].close_calls == 1 + + +async def test_real_sdk_late_connection_is_closed_before_stop_completes(monkeypatch): + from websockets.asyncio.server import serve + + for key in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy"): + monkeypatch.delenv(key, raising=False) + monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 0.02, raising=False) + discovery_started, release_discovery = threading.Event(), threading.Event() + accepted, peer_closed = asyncio.Event(), asyncio.Event() + peers = [] + + async def handler(peer): + peers.append(peer) + accepted.set() + await peer.wait_closed() + peer_closed.set() + + async with serve(handler, "127.0.0.1", 0, ping_interval=None, close_timeout=0.2) as server: + port = server.sockets[0].getsockname()[1] + + class WebClient: + ssl = None + + def __init__(self, **kwargs): + pass + + def apps_connections_open(self, **kwargs): + discovery_started.set() + if not release_discovery.wait(5): + raise TimeoutError("test did not release local URL discovery") + return {"url": f"ws://127.0.0.1:{port}/socket-mode"} + + channel = SlackChannel(MessageBus(), {"bot_token": "test", "app_token": "test", "bot_user_id": "UBOT", "web_client_factory": WebClient}) + futures = _capture_connect(monkeypatch, channel) + sdk = None + try: + await channel.start() + sdk = channel._socket_client + assert await asyncio.to_thread(discovery_started.wait, 2) + with pytest.raises(ChannelStopTimeout): + await channel.stop() + assert channel._socket_client is sdk + assert not futures[0].cancelled() + release_discovery.set() + monkeypatch.setattr("app.channels.slack.SLACK_STOP_TIMEOUT_SECONDS", 5) + await asyncio.wait_for(channel.stop(), 6) + await asyncio.wait_for(accepted.wait(), 2) + await asyncio.wait_for(peer_closed.wait(), 2) + assert not sdk.is_connected() + assert sdk.closed is True + assert sdk.auto_reconnect_enabled is False + assert channel._socket_client is None + assert channel._socket_connect_future is None + finally: + release_discovery.set() + for future in futures: + await asyncio.wait_for(asyncio.shield(future), 5) + if sdk is not None: + # slack-sdk 3.41.0 leaves its receiver interval runner alive + # after close; explicitly reap the test-owned runner as well. + sdk.auto_reconnect_enabled = False + sdk.default_auto_reconnect_enabled = False + if sdk.current_session is not None: + sdk.current_session_state.terminated = True + sock = sdk.current_session.sock + if sock is not None: + with contextlib.suppress(OSError): + sock.shutdown(socket.SHUT_RDWR) + await asyncio.wait_for(asyncio.to_thread(sdk.close), 5) + await asyncio.wait_for(asyncio.to_thread(sdk.current_session_runner.shutdown), 3) + for peer in peers: + peer.transport.abort()