From 5841c8e9b783d5c06468efe78528dd3113afeeef Mon Sep 17 00:00:00 2001 From: Nan Gao Date: Tue, 13 Oct 2026 22:24:46 +0800 Subject: [PATCH] fix(channels): make Feishu and DingTalk transports stoppable in-process (#6798) * fix(channels): make Feishu and DingTalk transports stoppable in-process FeishuChannel.stop() and DingTalkChannel.stop() could not end their SDK threads. Both SDKs run on a thread-private event loop that only returned on a fatal error: lark-oapi's ws.Client.start() parks forever in run_until_complete(_select()) and has no public stop, and the channel never kept the client handle; dingtalk-stream's start_forever() is a while-True loop that also catches CancelledError, and the disconnect() the channel called does not exist on the SDK. So every stop joined for 5s, raised ChannelStopTimeout, and ChannelService retained the instance while the platform connection stayed alive on the old process. That burns the full budget on shutdown and on POST /api/channels/{name}/restart, and it blocks the per-provider leader lease: a demoted leader could not release the transport before a peer took over. Both channels now publish the SDK thread's loop and client under a lock; stop() schedules a shutdown on that loop and then joins. Feishu disarms lark's _auto_reconnect before closing the socket and stopping the loop, because start() otherwise swallows the interrupted run_until_complete in its reconnect branch and dials again. DingTalk drives its own copy of the SDK connect loop that re-checks the running flag before every dial, runs the blocking ticket request off the loop, cancels keepalive with the socket, and logs the endpoint rather than the ticket-bearing payload. A shared shutdown_private_loop cancels the SDK's leftover tasks with a bounded drain and closes the loop; a stop requested before the client exists never dials. ChannelStopTimeout now only fires for a thread that genuinely wedges inside an SDK call. Telegram already stopped this way. Tests: a fake lark ws.Client with the SDK's start() shape (including the reconnect branch that swallows a bare loop stop), a real local websockets server that observes the DingTalk close, the stop-before-client race for both, and a blocking-io anchor that stop() ends a live private loop without blocking the Gateway loop. All eight were red on main. * docs(changelog): link the stoppable-transports entry to #6798 * fix(channels): release the socket on a stalled close and bound the DingTalk ticket request Review follow-ups on #6798 (Totoro-qaq): Feishu: the bounded _disconnect() wait cancelled websockets' close before its own close-timeout abort ran, and lark clears _conn in its finally, so a peer that never answers the close frame left the socket to garbage collection. Worse, a stop landing while _connect() still held lark's _lock snapshotted _conn as None. _shutdown_ws now settles the in-flight dial by taking that lock under one shared close deadline, snapshots the connection, runs _disconnect() under the same deadline, and aborts the transport itself (abort_connection_transport in base.py) when the handshake did not finish, yielding once so connection_lost runs before loop.stop(). DingTalk's websocket.close() path gets the same abort. DingTalk: the SDK's open_connection() calls requests.post with no timeout, and running it in an executor let a stalled ticket endpoint block interpreter exit (executor workers are joined at shutdown). The channel now sends the same request through httpx on the stream loop under a 30s deadline, so stop() cancels it and no worker thread lingers. Tests drive the real SDK clients against loopback peers: a WebSocket peer that completes the handshake and never ACKs the close (both channels assert the connection's socket reports fileno() == -1 right after stop(); red with the abort disabled), a real websockets server for the clean path, and a loopback ticket server (ok / 500 / never responds) with a no-lingering-non-daemon-thread check and a request-shape check. --- CHANGELOG.md | 26 ++ backend/app/channels/AGENTS.md | 12 +- backend/app/channels/base.py | 44 ++++ backend/app/channels/dingtalk.py | 183 ++++++++++++-- backend/app/channels/feishu.py | 112 ++++++++- .../blocking_io/test_channel_stop_offloop.py | 36 ++- backend/tests/support/local_sockets.py | 142 +++++++++++ backend/tests/test_dingtalk_channel.py | 178 +++++++++++++ backend/tests/test_feishu_parser.py | 237 ++++++++++++++++++ 9 files changed, 941 insertions(+), 29 deletions(-) create mode 100644 backend/tests/support/local_sockets.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a9e6459269..6c7187487c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1226,6 +1226,31 @@ This release closes that milestone with **621 merged pull requests**. ### Fixed +- **channels:** `FeishuChannel.stop()` and `DingTalkChannel.stop()` now end + their SDK threads instead of timing out. Both SDKs run on a thread-private + event loop that only returned on a fatal error, so every stop joined for + 5s, raised `ChannelStopTimeout`, and left the platform connection alive on + the old process — a restart, a redeploy, and the upcoming per-provider + leader hand-off could not release the connection. Each channel now + publishes that loop under a lock; `stop()` schedules a shutdown on it, then + joins. Feishu disarms lark-oapi's `_auto_reconnect` before closing the + socket and stopping the loop, because the SDK's `start()` otherwise swallows + the interrupted loop in its reconnect branch and dials again; DingTalk + drives its own copy of dingtalk-stream's connect loop (the SDK's is + `while True` and catches `CancelledError`), re-checks the running flag + before every dial, sends the ticket request itself through `httpx` with a + deadline (the SDK's `requests.post` has none, so a stalled endpoint could + pin the thread or process exit), and logs only the endpoint rather than + the ticket-bearing connection payload. When a peer never answers the close + frame, both channels abort the transport themselves instead of leaving the + socket to garbage collection, because the bounded close wait cancels + websockets' own close-timeout abort. A shared `shutdown_private_loop` + cancels the SDK's leftover tasks with a bounded drain and closes the loop, + and a stop requested before the client exists never dials. + `ChannelStopTimeout` is now reserved for a thread that genuinely wedges + inside an SDK call. Telegram already stopped its loop this way and is + unchanged. ([#6798]) + - **frontend:** The user-management settings page no longer renders a doomed query on the static-website demo, and a non-admin `?settings=users` deep link no longer opens the admin-only panel. The page gated only on @@ -11608,3 +11633,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6702]: https://github.com/bytedance/deer-flow/pull/6702 [#6706]: https://github.com/bytedance/deer-flow/pull/6706 [#6746]: https://github.com/bytedance/deer-flow/pull/6746 +[#6798]: https://github.com/bytedance/deer-flow/pull/6798 diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 529c96af69..ff15bce9e2 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -5,9 +5,7 @@ IM channels reach DeerFlow through `langgraph-sdk` HTTP; Gateway owns thread lif WeChat `_read_outbound_bytes` reads cap+1 off-loop; overflow returns `None` before encryption/upload; caps <=0 are unlimited. Cursor writes drain before cancellation; token expiry drains cursor/auth clearing under `_auth_lock`. -Telegram stop retries share live cleanup. If its loop dies, warn once and -drop only the matching future under `_shutdown_resources_lock`; client closure -remains unconfirmed. +Telegram stop retries share live cleanup. If its loop dies, warn once and drop only the matching future under `_shutdown_resources_lock`; client closure stays unconfirmed. **Auth**: SDK and raw-httpx `/goal` calls send process-local auth plus `_csrf_headers()`; internal auth requires CSRF. @@ -16,11 +14,11 @@ remains unconfirmed. - `store.py` - `ChannelStore` protocol (async `get_thread_id`/`set_thread_id`/`remove`/`list_entries`) for `channel_name:chat_id[:topic_id]` → `thread_id`. `SqlChannelStore` (`channel_thread_bindings`, harness `persistence/channel_thread_bindings/`; upsert keeps `created_at`, topic-less `remove` deletes by key prefix) for sqlite/postgres, so replicas share bindings; `JsonChannelStore` (`{base_dir}/channels/store.json`, loaded once per process, file I/O via `to_thread` under `_lock`) only for `database.backend: memory`. `resolve_channel_store` picks; `ChannelService.start()` runs `SqlChannelStore.import_legacy_json` once (file + empty table → `INSERT … ON CONFLICT DO NOTHING`, rename to `store.json.migrated`, a peer's rename tolerated; populated table → file kept). Feishu's sync lark callback (on the SDK's own loop) bridges `_resolve_routing_bounded` to the Gateway loop via `_run_store_call` (tracked future; refuses only a wait from the Gateway loop itself). Store calls run under `FEISHU_STORE_LOOKUP_TIMEOUT_SECONDS` (3s; the SDK loop also services ws keepalives); timeout or store error = *unresolved* (`StoreLookupUnresolved`), never a miss: the message joins a per-chat FIFO drained on the Gateway loop with backoff up to `FEISHU_STORE_RESOLVE_TOTAL_BUDGET_SECONDS` (30s), then is dropped with a warning, never routed to a new conversation; each deferred message holds a bus intake reservation (the `inbound_queue_maxsize` bound) and its 30s budget runs from arrival. `discord.py` reads `channels_dir`. Tests: `test_channel_thread_binding_store.py` (also under `blocking_io/`). - `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 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. + **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 (a `` fact block; a verbatim echo of the user's message as the 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`: every run and channel 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 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`) +- `service.py` - Drain manager work before transports; retain timed-out instances for retry. Feishu/DingTalk `stop()` stop the SDK thread's private loop via `_request_*_shutdown` (Feishu disarms lark's `_auto_reconnect` first, else its reconnect branch eats the stop; DingTalk runs its own copy of the SDK connect loop), then join; a live join or Slack cleanup raises `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 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()` (first failed tick logs 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: 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. - `github.py` - Webhook-driven GitHub channel. Inbound messages come from `POST /api/webhooks/github`; outbound is log-only because GitHub agents post explicitly with `gh` from their sandbox when they choose to comment or create a PR diff --git a/backend/app/channels/base.py b/backend/app/channels/base.py index 6e4482bb2d..2bd56e26ea 100644 --- a/backend/app/channels/base.py +++ b/backend/app/channels/base.py @@ -43,6 +43,50 @@ class ChannelStopTimeout(RuntimeError): """A channel still owns a live provider worker after bounded teardown.""" +PRIVATE_LOOP_DRAIN_TIMEOUT_SECONDS = 1.0 + + +def shutdown_private_loop(loop: asyncio.AbstractEventLoop, *, drain_timeout: float = PRIVATE_LOOP_DRAIN_TIMEOUT_SECONDS) -> None: + """Cancel what is still scheduled on a thread-private SDK loop, then close it. + + Runs on the thread that owns ``loop`` once its ``run_until_complete`` or + ``run_forever`` has returned. Provider SDKs leave their own tasks behind + (lark-oapi's ping and receive loops, dingtalk-stream's keepalive); cancelling + them with a bounded drain keeps the thread exit inside the channel's stop + budget, and a task that refuses to finish is abandoned with the loop rather + than joined. + """ + try: + pending = [task for task in asyncio.all_tasks(loop) if not task.done()] + for task in pending: + task.cancel() + if pending: + loop.run_until_complete(asyncio.wait(pending, timeout=drain_timeout)) + loop.run_until_complete(loop.shutdown_asyncgens()) + except Exception: + logger.debug("Error draining a private channel loop before close", exc_info=True) + finally: + asyncio.set_event_loop(None) + loop.close() + + +def abort_connection_transport(connection: Any) -> bool: + """Tear down a ``websockets`` connection's transport if its close handshake did not finish. + + ``connection.close()`` waits for the peer's close frame and only aborts the + transport after its own ``close_timeout``; a caller that bounds that wait more + tightly (channel stop budgets do) cancels before the abort, and the socket + then lives until garbage collection. Returns True when a transport was + aborted; the caller should yield to its loop once so ``connection_lost`` + runs before the loop is stopped. + """ + transport = getattr(connection, "transport", None) + if transport is None or transport.is_closing(): + return False + transport.abort() + return True + + @dataclass(eq=False, slots=True) class _ThreadsafeSubmission: coroutine: Coroutine[Any, Any, Any] diff --git a/backend/app/channels/dingtalk.py b/backend/app/channels/dingtalk.py index 0814035e1c..ec33ab1ffa 100644 --- a/backend/app/channels/dingtalk.py +++ b/backend/app/channels/dingtalk.py @@ -5,16 +5,18 @@ from __future__ import annotations import asyncio import json import logging +import platform import re import stat import threading import time from pathlib import Path from typing import Any +from urllib.parse import quote_plus import httpx -from app.channels.base import Channel, ChannelStopTimeout +from app.channels.base import Channel, ChannelStopTimeout, abort_connection_transport, shutdown_private_loop from app.channels.commands import is_known_channel_command, strip_leading_mentions from app.channels.connection_identity import attach_connection_identity from app.channels.message_bus import InboundMessage, InboundMessageType, InboundReservation, MessageBus, OutboundMessage, ResolvedAttachment @@ -36,6 +38,16 @@ DINGTALK_API_BASE = "https://api.dingtalk.com" _TOKEN_REFRESH_MARGIN_SECONDS = 300 +# stop() budget for the dingtalk-stream thread: ask its private loop to close +# the socket (bounded by the close timeout) and stop, then join the thread. +DINGTALK_STOP_TIMEOUT_SECONDS = 5 +DINGTALK_STREAM_CLOSE_TIMEOUT_SECONDS = 2 +# Deadline for the Stream ticket request (the SDK's own has none). +DINGTALK_STREAM_TICKET_TIMEOUT_SECONDS = 30 +# Retry pacing copied from DingTalkStreamClient.start() (dingtalk-stream 0.24). +_DINGTALK_STREAM_RETRY_SECONDS = 10 +_DINGTALK_STREAM_ERROR_RETRY_SECONDS = 3 + _CONVERSATION_TYPE_P2P = "1" _CONVERSATION_TYPE_GROUP = "2" @@ -182,7 +194,11 @@ class DingTalkChannel(Channel): self._card_template_id: str = config.get("card_template_id", "") self._card_track_ids: dict[str, str] = {} self._dingtalk_client: Any = None + # The SDK thread's private loop and client, published under _stream_lock + # once constructed so stop() can shut them down from the Gateway loop. self._stream_client: Any = None + self._stream_loop: asyncio.AbstractEventLoop | None = None + self._stream_lock = threading.Lock() self._incoming_messages: dict[str, Any] = {} self._incoming_messages_lock = threading.Lock() self._card_repliers: dict[str, Any] = {} @@ -233,33 +249,64 @@ class DingTalkChannel(Channel): async def stop(self) -> None: self._running = False self.bus.unsubscribe_outbound(self._on_outbound) + # Signal the SDK thread first so its loop winds down while the + # Gateway-side drain below runs. + self._request_stream_shutdown() await self._close_and_drain_threadsafe_futures() - stream_client = self._stream_client - if stream_client is not None: - try: - if hasattr(stream_client, "disconnect"): - stream_client.disconnect() - except Exception: - logger.debug("[DingTalk] error disconnecting stream client", exc_info=True) - self._dingtalk_client = None - self._stream_client = None with self._incoming_messages_lock: self._incoming_messages.clear() self._card_repliers.clear() self._card_track_ids.clear() if self._thread: - # The SDK thread only returns on a fatal error, so this join - # normally waits out its full timeout; keep it off the event loop. + # The thread exits once its loop has stopped; keep the join off the + # event loop because a wedged SDK call can still use up the budget. thread = self._thread - await asyncio.to_thread(thread.join, timeout=5) + await asyncio.to_thread(thread.join, timeout=DINGTALK_STOP_TIMEOUT_SECONDS) if thread.is_alive(): raise ChannelStopTimeout("DingTalk SDK thread is still running after stop timeout") if self._thread is thread: self._thread = None + with self._stream_lock: + self._stream_loop = None + self._stream_client = None logger.info("DingTalk channel stopped") + def _request_stream_shutdown(self) -> None: + """Ask the SDK thread to close its socket and stop its loop (non-blocking).""" + with self._stream_lock: + loop = self._stream_loop + client = self._stream_client + if loop is None: + return + try: + loop.call_soon_threadsafe(self._begin_stream_shutdown, loop, client) + except RuntimeError: + # The loop closed between the snapshot and the call: the thread is + # already on its way out and the join below observes that. + logger.debug("[DingTalk] stream loop already closed before shutdown could be scheduled") + + def _begin_stream_shutdown(self, loop: asyncio.AbstractEventLoop, client: Any) -> None: + loop.create_task(self._shutdown_stream(loop, client)) + + async def _shutdown_stream(self, loop: asyncio.AbstractEventLoop, client: Any) -> None: + """Runs on the SDK loop: close the live websocket, then stop the loop.""" + try: + websocket = getattr(client, "websocket", None) if client is not None else None + if websocket is not None: + try: + await asyncio.wait_for(websocket.close(), timeout=DINGTALK_STREAM_CLOSE_TIMEOUT_SECONDS) + except Exception: + logger.debug("[DingTalk] stream close did not complete cleanly", exc_info=True) + if abort_connection_transport(websocket): + logger.warning("[DingTalk] stream close handshake did not finish within %ss; transport aborted", DINGTALK_STREAM_CLOSE_TIMEOUT_SECONDS) + await asyncio.sleep(0) # let connection_lost run before the loop stops + except asyncio.CancelledError: + # The thread is already draining this loop; leave it to that. + raise + loop.stop() + def _resolve_routing(self, msg: OutboundMessage) -> tuple[str, str, str]: """Return (conversation_type, sender_staff_id, conversation_id). @@ -398,22 +445,126 @@ class DingTalkChannel(Channel): # -- stream client (runs in dedicated thread) -------------------------- def _run_stream(self, client_id: str, client_secret: str) -> None: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) try: import dingtalk_stream credential = dingtalk_stream.Credential(client_id, client_secret) client = dingtalk_stream.DingTalkStreamClient(credential) - self._stream_client = client client.register_callback_handler( dingtalk_stream.chatbot.ChatbotMessage.TOPIC, _DingTalkMessageHandler(self), ) - client.start_forever() + with self._stream_lock: + if not self._running: + # stop() ran before the client existed: never dial. + return + self._stream_loop = loop + self._stream_client = client + loop.run_until_complete(self._serve_stream(client)) except Exception: if self._running: logger.exception("DingTalk Stream Push error") + else: + logger.debug("[DingTalk] stream loop exited after stop", exc_info=True) + self._running = False finally: - self._stream_client = None + with self._stream_lock: + if self._stream_loop is loop: + self._stream_loop = None + self._stream_client = None + shutdown_private_loop(loop) + + async def _serve_stream(self, client: Any) -> None: + """Run dingtalk-stream's connection loop so ``stop()`` can end it. + + Mirrors ``DingTalkStreamClient.start()`` (dingtalk-stream 0.24): open a + connection ticket, hold the websocket, hand each frame to the SDK's + ``background_task``, and retry after 10s on a lost connection or 3s on + anything else. The SDK's own loop is ``while True`` and also catches + ``CancelledError``, so it can only be left by killing the thread; this + copy re-checks ``_running`` before every dial and fetches the ticket + asynchronously with a deadline so ``stop()`` can cancel it. + """ + import websockets + + client.pre_start() + while self._running: + try: + connection = await self._open_stream_connection(client) + if not self._running: + break + if not connection: + logger.error("[DingTalk] open connection failed") + await asyncio.sleep(_DINGTALK_STREAM_RETRY_SECONDS) + continue + # The ticket in ``connection`` is a credential; log the endpoint only. + logger.info("[DingTalk] stream endpoint is %s", connection.get("endpoint")) + uri = f"{connection['endpoint']}?ticket={quote_plus(connection['ticket'])}" + async with websockets.connect(uri) as websocket: + client.websocket = websocket + keepalive = asyncio.create_task(client.keepalive(websocket)) + try: + async for raw_message in websocket: + asyncio.create_task(client.background_task(json.loads(raw_message))) + finally: + keepalive.cancel() + except websockets.exceptions.ConnectionClosedError as exc: + if not self._running: + break + logger.error("[DingTalk] stream connection lost: %s", exc) + await asyncio.sleep(_DINGTALK_STREAM_RETRY_SECONDS) + except Exception: + if not self._running: + break + logger.exception("[DingTalk] stream error") + await asyncio.sleep(_DINGTALK_STREAM_ERROR_RETRY_SECONDS) + + async def _open_stream_connection(self, client: Any) -> dict[str, Any] | None: + """Fetch a Stream ticket: ``DingTalkStreamClient.open_connection()`` with a deadline. + + The SDK's version calls ``requests.post`` with no timeout, so a stalled + ticket endpoint pins whichever thread runs it (and, from an executor, + Gateway process exit, which joins executor workers). This sends the same + request through ``httpx`` on the stream loop under + ``DINGTALK_STREAM_TICKET_TIMEOUT_SECONDS``, so ``stop()`` cancels it like + any other task. Returns ``None`` on failure like the SDK does. + """ + from dingtalk_stream.version import VERSION_STRING + + topics: list[dict[str, str]] = [] + if getattr(client, "_is_event_required", False): + topics.append({"type": "EVENT", "topic": "*"}) + for topic in client.callback_handler_map: + topics.append({"type": "CALLBACK", "topic": topic}) + headers = { + "Content-Type": "application/json", + "Accept": "application/json", + "User-Agent": f"DingTalkStream/1.0 SDK/{VERSION_STRING} Python/{platform.python_version()} (+https://github.com/open-dingtalk/dingtalk-stream-sdk-python)", + } + body = { + "clientId": client.credential.client_id, + "clientSecret": client.credential.client_secret, + "subscriptions": topics, + "ua": f"dingtalk-sdk-python/v{VERSION_STRING}-union", + "localIp": client.get_host_ip(), + } + try: + async with httpx.AsyncClient(timeout=DINGTALK_STREAM_TICKET_TIMEOUT_SECONDS) as http: + response = await http.post(client.OPEN_CONNECTION_API, headers=headers, json=body) + response.raise_for_status() + payload = response.json() + except httpx.HTTPStatusError as exc: + logger.error("[DingTalk] open connection failed: HTTP %s", exc.response.status_code) + return None + except (httpx.HTTPError, ValueError) as exc: + logger.error("[DingTalk] open connection failed: %s", type(exc).__name__) + return None + if not isinstance(payload, dict) or not payload.get("endpoint") or not payload.get("ticket"): + logger.error("[DingTalk] open connection returned no endpoint/ticket") + return None + return payload def _on_chatbot_message(self, message: Any) -> None: if not self._running: diff --git a/backend/app/channels/feishu.py b/backend/app/channels/feishu.py index 0b1e578e1f..22d6f19c9d 100644 --- a/backend/app/channels/feishu.py +++ b/backend/app/channels/feishu.py @@ -16,7 +16,7 @@ from concurrent.futures import Future from dataclasses import dataclass, field from typing import Any, Literal, TypeVar -from app.channels.base import Channel, ChannelStopTimeout, ChannelUnavailable +from app.channels.base import Channel, ChannelStopTimeout, ChannelUnavailable, abort_connection_transport, shutdown_private_loop from app.channels.commands import is_known_channel_command, strip_leading_mentions from app.channels.connection_identity import attach_connection_identity, lookup_thread_id from app.channels.feishu_card_interaction import ( @@ -50,6 +50,10 @@ from deerflow.uploads.manager import ( logger = logging.getLogger(__name__) PENDING_CLARIFICATION_TTL_SECONDS = 30 * 60 +# stop() budget for the lark-oapi thread: ask its private loop to close the +# socket (bounded by the close timeout) and stop, then join the thread. +FEISHU_STOP_TIMEOUT_SECONDS = 5 +FEISHU_WS_CLOSE_TIMEOUT_SECONDS = 2 _PENDING_CARD_METADATA_KEY = "_feishu_pending_card_id" # Channel-store deadlines. ``_on_message`` runs on lark-oapi's private event loop, # which also carries the SDK's websocket reads, pongs and its 120s ping loop: @@ -168,6 +172,11 @@ class FeishuChannel(Channel): super().__init__(name="feishu", bus=bus, config=config) self._thread: threading.Thread | None = None self._main_loop: asyncio.AbstractEventLoop | None = None + # The SDK thread's private loop and client, published under _ws_lock once + # constructed so stop() can shut them down from the Gateway loop. + self._ws_loop: asyncio.AbstractEventLoop | None = None + self._ws_client: Any = None + self._ws_lock = threading.Lock() self._api_client = None self._CreateMessageReactionRequest = None self._CreateMessageReactionRequestBody = None @@ -363,6 +372,13 @@ class FeishuChannel(Channel): We work around this by creating a plain asyncio event loop for this thread and patching the SDK's module-level reference before calling ``start()``. + + ``stop()`` ends this thread through ``_request_ws_shutdown``: lark-oapi + 1.7 has no public stop, so ``start()`` only returns once this loop is + stopped from a callback scheduled on it — and only after that callback + disarmed ``_auto_reconnect``, because otherwise ``start()`` swallows the + interrupted ``run_until_complete`` in its reconnect branch and dials + again. """ loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) @@ -383,11 +399,93 @@ class FeishuChannel(Channel): log_level=lark.LogLevel.INFO, domain=domain, ) + with self._ws_lock: + if not self._running: + # stop() ran before the client existed: never dial. + return + self._ws_loop = loop + self._ws_client = ws_client ws_client.start() except Exception: if self._running: logger.exception("Feishu WebSocket error") + else: + logger.debug("[Feishu] SDK loop exited after stop", exc_info=True) self._running = False + finally: + with self._ws_lock: + if self._ws_loop is loop: + self._ws_loop = None + self._ws_client = None + shutdown_private_loop(loop) + + def _request_ws_shutdown(self) -> None: + """Ask the SDK thread to close its socket and stop its loop (non-blocking).""" + with self._ws_lock: + loop = self._ws_loop + ws_client = self._ws_client + if loop is None: + return + try: + loop.call_soon_threadsafe(self._begin_ws_shutdown, loop, ws_client) + except RuntimeError: + # The loop closed between the snapshot and the call: the thread is + # already on its way out and the join below observes that. + logger.debug("[Feishu] SDK loop already closed before shutdown could be scheduled") + + def _begin_ws_shutdown(self, loop: asyncio.AbstractEventLoop, ws_client: Any) -> None: + loop.create_task(self._shutdown_ws(loop, ws_client)) + + async def _shutdown_ws(self, loop: asyncio.AbstractEventLoop, ws_client: Any) -> None: + """Runs on the SDK loop: disarm reconnects, close the socket, stop the loop.""" + try: + if ws_client is not None: + # Must precede any await: once the loop stops, start() lands in + # its ``except Exception`` branch and consults this flag. + ws_client._auto_reconnect = False + deadline = loop.time() + FEISHU_WS_CLOSE_TIMEOUT_SECONDS + # _disconnect() clears _conn in its finally (also when cancelled + # mid-acquire), so take the handle first — once any in-flight + # dial has settled — and release the socket here if the close + # handshake does not finish, rather than whenever the + # connection object happens to be garbage-collected. + connection = await self._settled_ws_connection(ws_client, deadline) + disconnect = getattr(ws_client, "_disconnect", None) + if disconnect is not None: + try: + async with asyncio.timeout_at(deadline): + await disconnect() + except Exception: + logger.debug("[Feishu] WebSocket close did not complete cleanly", exc_info=True) + if abort_connection_transport(connection): + logger.warning("[Feishu] WebSocket close handshake did not finish within %ss; transport aborted", FEISHU_WS_CLOSE_TIMEOUT_SECONDS) + await asyncio.sleep(0) # let connection_lost run before the loop stops + except asyncio.CancelledError: + # The thread is already draining this loop; leave it to that. + raise + loop.stop() + + @staticmethod + async def _settled_ws_connection(ws_client: Any, deadline: float) -> Any: + """Return the SDK's live connection once any in-flight ``_connect()`` has settled. + + lark holds ``_lock`` across the whole dial and assigns ``_conn`` under + it, so a stop that lands mid-handshake would otherwise snapshot + ``None`` and leave the socket the dial produces to the SDK's cancelled + close. Waiting for the lock is bounded by the close deadline. + """ + lock = getattr(ws_client, "_lock", None) + if lock is None: + return getattr(ws_client, "_conn", None) + try: + async with asyncio.timeout_at(deadline): + await lock.acquire() + except TimeoutError: + return getattr(ws_client, "_conn", None) + try: + return getattr(ws_client, "_conn", None) + finally: + lock.release() def _on_ignored_message_event(self, event) -> None: logger.debug("[Feishu] ignoring non-content message event: %s", type(event).__name__) @@ -395,6 +493,9 @@ class FeishuChannel(Channel): async def stop(self) -> None: self._running = False self.bus.unsubscribe_outbound(self._on_outbound) + # Signal the SDK thread first so its loop winds down while the + # Gateway-side drain below runs. + self._request_ws_shutdown() await self._close_and_drain_threadsafe_futures() background_tasks = tuple(self._background_tasks) @@ -408,14 +509,17 @@ class FeishuChannel(Channel): self._background_tasks.clear() self._running_card_tasks.clear() if self._thread: - # The SDK thread only returns on a fatal error, so this join - # normally waits out its full timeout; keep it off the event loop. + # The thread exits once its loop has stopped; keep the join off the + # event loop because a wedged SDK call can still use up the budget. thread = self._thread - await asyncio.to_thread(thread.join, timeout=5) + await asyncio.to_thread(thread.join, timeout=FEISHU_STOP_TIMEOUT_SECONDS) if thread.is_alive(): raise ChannelStopTimeout("Feishu SDK thread is still running after stop timeout") if self._thread is thread: self._thread = None + with self._ws_lock: + self._ws_loop = None + self._ws_client = None logger.info("Feishu channel stopped") async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None: diff --git a/backend/tests/blocking_io/test_channel_stop_offloop.py b/backend/tests/blocking_io/test_channel_stop_offloop.py index 3f0c3ee684..df42b33faf 100644 --- a/backend/tests/blocking_io/test_channel_stop_offloop.py +++ b/backend/tests/blocking_io/test_channel_stop_offloop.py @@ -6,8 +6,8 @@ every run, stream and channel the Gateway serves: - Slack: ``SocketModeClient.close()`` joins the SDK's message-processor thread and waits for in-flight listeners. -- Feishu / DingTalk: ``stop()`` joins the SDK thread with a 5s timeout, and - that thread only returns on a fatal error. +- Feishu / DingTalk: ``stop()`` asks the SDK thread's private event loop to + shut down, then joins that thread with a 5s timeout. - Discord: ``stop()`` joins the client thread with a 10s timeout, which a timed-out client close or a slow ``_run_client()`` drain can use up. @@ -102,3 +102,35 @@ async def test_stop_joins_sdk_thread_off_loop(channel_cls) -> None: assert not sdk_thread.is_alive() assert channel._thread is None + + +@pytest.mark.parametrize( + ("channel_cls", "loop_attr"), + [(FeishuChannel, "_ws_loop"), (DingTalkChannel, "_stream_loop")], +) +async def test_stop_ends_live_sdk_loop_off_loop(channel_cls, loop_attr) -> None: + sdk_loop = asyncio.new_event_loop() + ready = threading.Event() + + def run_sdk_loop() -> None: + asyncio.set_event_loop(sdk_loop) + sdk_loop.call_soon(ready.set) + try: + sdk_loop.run_forever() + finally: + sdk_loop.close() + + sdk_thread = threading.Thread(target=run_sdk_loop, daemon=True) + sdk_thread.start() + assert await asyncio.to_thread(ready.wait, 5) + channel = channel_cls(MessageBus(), {}) + channel._thread = sdk_thread + channel._running = True + setattr(channel, loop_attr, sdk_loop) + + await channel.stop() + + assert not sdk_thread.is_alive() + assert sdk_loop.is_closed() + assert channel._thread is None + assert getattr(channel, loop_attr) is None diff --git a/backend/tests/support/local_sockets.py b/backend/tests/support/local_sockets.py new file mode 100644 index 0000000000..e1606dd8b6 --- /dev/null +++ b/backend/tests/support/local_sockets.py @@ -0,0 +1,142 @@ +"""Loopback peers for exercising SDK transports at the socket boundary. + +Both run on the calling test's event loop and only ever bind ``127.0.0.1:0``. + +- ``StalledWebSocketServer`` completes the WebSocket opening handshake, then + reads and discards every frame without answering, so a client's close + handshake never gets its ACK. ``closed`` is set once the peer observes EOF, + i.e. once the client actually released the TCP connection. +- ``LocalHttpServer`` answers every request with a fixed status and JSON body, + or (``stall=True``) accepts the connection and never responds. Each request + body is recorded so tests can assert on what the SDK sent. +""" + +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import json +from collections.abc import Callable +from typing import Any + +_WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + + +async def wait_until(predicate: Callable[[], Any], *, timeout: float = 5.0, interval: float = 0.01) -> Any: + """Poll ``predicate`` (which may read another thread's state) until it is truthy.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while True: + value = predicate() + if value: + return value + if loop.time() >= deadline: + raise TimeoutError("condition not met in time") + await asyncio.sleep(interval) + + +def transport_socket(connection: Any) -> Any: + """The socket object under a ``websockets`` connection, for release checks after stop.""" + return connection.transport.get_extra_info("socket") + + +def _header(request: bytes, name: str) -> str: + prefix = name.lower() + ":" + for line in request.decode("latin-1").split("\r\n"): + if line.lower().startswith(prefix): + return line.split(":", 1)[1].strip() + return "" + + +class StalledWebSocketServer: + def __init__(self) -> None: + self.accepted = asyncio.Event() + self.closed = asyncio.Event() + self.port = 0 + self._server: asyncio.base_events.Server | None = None + + async def __aenter__(self) -> StalledWebSocketServer: + self._server = await asyncio.start_server(self._handle, "127.0.0.1", 0) + self.port = self._server.sockets[0].getsockname()[1] + return self + + async def __aexit__(self, *exc_info: object) -> None: + assert self._server is not None + self._server.close() + try: + await asyncio.wait_for(self._server.wait_closed(), timeout=5) + except TimeoutError: + pass + + @property + def url(self) -> str: + return f"ws://127.0.0.1:{self.port}/" + + async def _handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + try: + request = await reader.readuntil(b"\r\n\r\n") + key = _header(request, "Sec-WebSocket-Key") + accept = base64.b64encode(hashlib.sha1((key + _WS_GUID).encode("ascii")).digest()).decode("ascii") + writer.write(("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n" + f"Sec-WebSocket-Accept: {accept}\r\n\r\n").encode("ascii")) + await writer.drain() + self.accepted.set() + while await reader.read(65536): + pass # swallow pings, data and the close frame; never answer + except (asyncio.IncompleteReadError, ConnectionError): + pass + finally: + self.closed.set() + writer.close() + + +class LocalHttpServer: + def __init__(self, *, status: int = 200, body: Any = None, stall: bool = False) -> None: + self.status = status + self.body = body if body is not None else {} + self.stall = stall + self.requests: list[dict[str, Any]] = [] + self.accepted = asyncio.Event() + self.closed = asyncio.Event() + self.port = 0 + self._server: asyncio.base_events.Server | None = None + + async def __aenter__(self) -> LocalHttpServer: + self._server = await asyncio.start_server(self._handle, "127.0.0.1", 0) + self.port = self._server.sockets[0].getsockname()[1] + return self + + async def __aexit__(self, *exc_info: object) -> None: + assert self._server is not None + self._server.close() + try: + await asyncio.wait_for(self._server.wait_closed(), timeout=5) + except TimeoutError: + pass + + @property + def url(self) -> str: + return f"http://127.0.0.1:{self.port}/open" + + async def _handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + try: + head = await reader.readuntil(b"\r\n\r\n") + length = int(_header(head, "Content-Length") or 0) + raw = await reader.readexactly(length) if length else b"" + try: + self.requests.append(json.loads(raw) if raw else {}) + except ValueError: + self.requests.append({"raw": raw.decode("latin-1")}) + self.accepted.set() + if self.stall: + while await reader.read(65536): + pass # hold the request open until the client gives up + return + payload = json.dumps(self.body).encode("utf-8") + writer.write((f"HTTP/1.1 {self.status} X\r\nContent-Type: application/json\r\nContent-Length: {len(payload)}\r\nConnection: close\r\n\r\n").encode("ascii") + payload) + await writer.drain() + except (asyncio.IncompleteReadError, ConnectionError): + pass + finally: + self.closed.set() + writer.close() diff --git a/backend/tests/test_dingtalk_channel.py b/backend/tests/test_dingtalk_channel.py index c9fdeddcfc..7ce5640dcf 100644 --- a/backend/tests/test_dingtalk_channel.py +++ b/backend/tests/test_dingtalk_channel.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio import logging import threading +import time from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock @@ -1808,6 +1809,183 @@ class TestStop: _run(go()) + @staticmethod + def _channel() -> DingTalkChannel: + return DingTalkChannel(MessageBus(), config={"client_id": "id", "client_secret": "secret"}) + + def test_stop_closes_live_stream_connection_and_exits_thread(self, monkeypatch): + pytest.importorskip("dingtalk_stream") + import dingtalk_stream + from support.local_sockets import LocalHttpServer + from websockets.asyncio.server import serve + + async def go(): + connected = asyncio.Event() + closed = asyncio.Event() + + async def handler(websocket): + connected.set() + try: + async for _ in websocket: + pass + finally: + closed.set() + + async with serve(handler, "127.0.0.1", 0) as server: + port = server.sockets[0].getsockname()[1] + async with LocalHttpServer(body={"endpoint": f"ws://127.0.0.1:{port}", "ticket": "ticket"}) as tickets: + monkeypatch.setattr(dingtalk_stream.DingTalkStreamClient, "OPEN_CONNECTION_API", tickets.url) + channel = self._channel() + await channel.start() + thread = channel._thread + assert thread is not None and thread.is_alive() + await asyncio.wait_for(connected.wait(), 5) + + started = time.monotonic() + await channel.stop() + elapsed = time.monotonic() - started + + assert elapsed < 5, f"stop() took {elapsed:.2f}s" + await asyncio.wait_for(closed.wait(), 5) + assert not thread.is_alive() + assert channel._thread is None + assert channel._stream_client is None + assert channel.is_running is False + + # The ticket request mirrors DingTalkStreamClient.open_connection(). + request = tickets.requests[0] + assert request["clientId"] == "id" + assert request["clientSecret"] == "secret" + assert {"type": "CALLBACK", "topic": dingtalk_stream.chatbot.ChatbotMessage.TOPIC} in request["subscriptions"] + assert request["ua"].startswith("dingtalk-sdk-python/v") + assert "localIp" in request + + _run(go()) + + def test_stop_aborts_transport_when_peer_never_acks_close(self, monkeypatch, caplog): + pytest.importorskip("dingtalk_stream") + import dingtalk_stream + from support.local_sockets import LocalHttpServer, StalledWebSocketServer, transport_socket, wait_until + + from app.channels import dingtalk as dingtalk_module + + caplog.set_level(logging.WARNING, logger="app.channels.dingtalk") + + async def go(): + async with StalledWebSocketServer() as peer, LocalHttpServer(body={"endpoint": peer.url, "ticket": "ticket"}) as tickets: + monkeypatch.setattr(dingtalk_stream.DingTalkStreamClient, "OPEN_CONNECTION_API", tickets.url) + monkeypatch.setattr(dingtalk_module, "DINGTALK_STREAM_CLOSE_TIMEOUT_SECONDS", 0.5) + channel = self._channel() + await channel.start() + thread = channel._thread + await asyncio.wait_for(peer.accepted.wait(), 5) + websocket = await wait_until(lambda: channel._stream_client is not None and channel._stream_client.websocket) + sock = transport_socket(websocket) + assert sock.fileno() != -1 + + started = time.monotonic() + await channel.stop() + elapsed = time.monotonic() - started + + assert elapsed < 5, f"stop() took {elapsed:.2f}s" + assert not thread.is_alive() + assert channel._thread is None + # The socket is released by stop() itself, not by a later GC pass. + assert sock.fileno() == -1 + assert "transport aborted" in caplog.text + await asyncio.wait_for(peer.closed.wait(), 5) + + _run(go()) + + def test_stop_interrupts_open_connection_retry_sleep(self, monkeypatch): + pytest.importorskip("dingtalk_stream") + import dingtalk_stream + from support.local_sockets import LocalHttpServer + + async def go(): + async with LocalHttpServer(status=500, body={"error": "nope"}) as tickets: + monkeypatch.setattr(dingtalk_stream.DingTalkStreamClient, "OPEN_CONNECTION_API", tickets.url) + channel = self._channel() + await channel.start() + thread = channel._thread + await asyncio.wait_for(tickets.accepted.wait(), 5) + + started = time.monotonic() + await channel.stop() + elapsed = time.monotonic() - started + + assert elapsed < 5, f"stop() took {elapsed:.2f}s" + assert not thread.is_alive() + assert channel._thread is None + + _run(go()) + + def test_stop_cancels_stalled_ticket_request_without_lingering_threads(self, monkeypatch): + pytest.importorskip("dingtalk_stream") + import dingtalk_stream + from support.local_sockets import LocalHttpServer + + def forbidden_open_connection(self): + raise AssertionError("the SDK's timeout-less open_connection() must not run") + + async def go(): + async with LocalHttpServer(stall=True) as tickets: + monkeypatch.setattr(dingtalk_stream.DingTalkStreamClient, "OPEN_CONNECTION_API", tickets.url) + monkeypatch.setattr(dingtalk_stream.DingTalkStreamClient, "open_connection", forbidden_open_connection) + await asyncio.to_thread(lambda: None) # warm the test loop's own executor before the baseline + baseline = {thread.ident for thread in threading.enumerate()} + channel = self._channel() + await channel.start() + thread = channel._thread + await asyncio.wait_for(tickets.accepted.wait(), 5) + + started = time.monotonic() + await channel.stop() + elapsed = time.monotonic() - started + + assert elapsed < 5, f"stop() took {elapsed:.2f}s" + assert not thread.is_alive() + # The cancelled request released its connection ... + await asyncio.wait_for(tickets.closed.wait(), 5) + # ... and nothing non-daemon is left to block interpreter exit. + lingering = [t.name for t in threading.enumerate() if t.ident not in baseline and t.is_alive() and not t.daemon] + assert lingering == [] + + _run(go()) + + def test_stop_before_stream_client_exists_never_connects(self, monkeypatch): + pytest.importorskip("dingtalk_stream") + import dingtalk_stream + from support.local_sockets import LocalHttpServer + + gate = threading.Event() + + class _GatedStreamClient(dingtalk_stream.DingTalkStreamClient): + def __init__(self, *args, **kwargs): + gate.wait(timeout=5) + super().__init__(*args, **kwargs) + + monkeypatch.setattr(dingtalk_stream, "DingTalkStreamClient", _GatedStreamClient) + + async def go(): + async with LocalHttpServer(body={"endpoint": "ws://127.0.0.1:1", "ticket": "ticket"}) as tickets: + monkeypatch.setattr(dingtalk_stream.DingTalkStreamClient, "OPEN_CONNECTION_API", tickets.url) + channel = self._channel() + await channel.start() + thread = channel._thread + + stop_task = asyncio.create_task(channel.stop()) + await asyncio.sleep(0.05) + assert not stop_task.done() + gate.set() + await stop_task + + assert not thread.is_alive() + assert channel._thread is None + assert tickets.requests == [] + + _run(go()) + def test_stop_joins_stream_thread_off_the_event_loop(self): async def go(): channel = DingTalkChannel(MessageBus(), config={}) diff --git a/backend/tests/test_feishu_parser.py b/backend/tests/test_feishu_parser.py index 888237eaad..333c5cb35e 100644 --- a/backend/tests/test_feishu_parser.py +++ b/backend/tests/test_feishu_parser.py @@ -1,5 +1,6 @@ import asyncio import json +import logging import tempfile import threading import time @@ -209,6 +210,242 @@ def test_feishu_stop_retains_live_ws_thread_after_join_timeout(): asyncio.run(go()) +class _FakeLarkWsClient: + """Stand-in for ``lark_oapi.ws.Client`` (1.7.x) with the same ``start()`` shape. + + The SDK drives the module-level ``lark_oapi.ws.client.loop`` itself: it runs + ``_connect()`` to completion, falls into a reconnect branch that swallows + *any* exception while ``_auto_reconnect`` is set, then parks forever in + ``_select()``. A stop that only calls ``loop.stop()`` is therefore absorbed + by that reconnect branch; the channel has to disarm it first. + """ + + def __init__(self, *, connect_forever: bool = False, construct_gate: threading.Event | None = None, **_kwargs) -> None: + self._auto_reconnect = True + self.connect_forever = connect_forever + self.loop: asyncio.AbstractEventLoop | None = None + self.start_calls = 0 + self.disconnect_calls = 0 + self.reconnect_calls = 0 + self.connecting = threading.Event() + self.selected = threading.Event() + if construct_gate is not None: + construct_gate.wait(timeout=5) + + async def _connect(self) -> None: + self.connecting.set() + if self.connect_forever: + await asyncio.Event().wait() + + async def _disconnect(self) -> None: + self.disconnect_calls += 1 + + async def _reconnect(self) -> None: + self.reconnect_calls += 1 + await asyncio.Event().wait() + + async def _select(self) -> None: + self.selected.set() + while True: + await asyncio.sleep(3600) + + def start(self) -> None: + import lark_oapi.ws.client as ws_client_mod + + loop = ws_client_mod.loop + self.loop = loop + self.start_calls += 1 + try: + loop.run_until_complete(self._connect()) + except Exception: + loop.run_until_complete(self._disconnect()) + if self._auto_reconnect: + loop.run_until_complete(self._reconnect()) + else: + raise + loop.run_until_complete(self._select()) + + +def _feishu_channel_with_fake_ws(monkeypatch, **client_kwargs): + import lark_oapi + + created: list[_FakeLarkWsClient] = [] + constructed = threading.Event() + + def factory(**_kwargs): + client = _FakeLarkWsClient(**client_kwargs) + created.append(client) + constructed.set() + return client + + monkeypatch.setattr(lark_oapi.ws, "Client", factory) + channel = FeishuChannel(MessageBus(), {"app_id": "test", "app_secret": "test"}) + return channel, created, constructed + + +def test_feishu_stop_disconnects_sdk_and_exits_its_thread(monkeypatch): + async def go(): + channel, created, constructed = _feishu_channel_with_fake_ws(monkeypatch) + await channel.start() + thread = channel._thread + assert thread is not None and thread.is_alive() + assert await asyncio.to_thread(constructed.wait, 5) + client = created[0] + assert await asyncio.to_thread(client.selected.wait, 5) + assert channel.is_running is True + + started = time.monotonic() + await channel.stop() + elapsed = time.monotonic() - started + + assert elapsed < 5, f"stop() took {elapsed:.2f}s" + assert not thread.is_alive() + assert channel._thread is None + assert channel.is_running is False + assert client.start_calls == 1 + assert client._auto_reconnect is False + assert client.disconnect_calls >= 1 + assert client.reconnect_calls == 0 + assert client.loop is not None and client.loop.is_closed() + + asyncio.run(go()) + + +def test_feishu_stop_interrupts_sdk_connect_without_reconnecting(monkeypatch): + async def go(): + channel, created, constructed = _feishu_channel_with_fake_ws(monkeypatch, connect_forever=True) + await channel.start() + thread = channel._thread + assert await asyncio.to_thread(constructed.wait, 5) + client = created[0] + assert await asyncio.to_thread(client.connecting.wait, 5) + + started = time.monotonic() + await channel.stop() + elapsed = time.monotonic() - started + + assert elapsed < 5, f"stop() took {elapsed:.2f}s" + assert not thread.is_alive() + assert channel._thread is None + assert client._auto_reconnect is False + assert client.reconnect_calls == 0 + assert client.loop is not None and client.loop.is_closed() + + asyncio.run(go()) + + +def test_feishu_stop_before_sdk_client_exists_never_starts_it(monkeypatch): + async def go(): + gate = threading.Event() + channel, created, constructed = _feishu_channel_with_fake_ws(monkeypatch, construct_gate=gate) + await channel.start() + thread = channel._thread + + stop_task = asyncio.create_task(channel.stop()) + await asyncio.sleep(0.05) + assert not stop_task.done() + gate.set() + await stop_task + + assert await asyncio.to_thread(constructed.wait, 5) + assert created[0].start_calls == 0 + assert not thread.is_alive() + assert channel._thread is None + + asyncio.run(go()) + + +def _point_lark_at(monkeypatch, ws_url: str) -> None: + """Skip lark's endpoint-discovery POST and dial ``ws_url`` with the real client.""" + import lark_oapi.ws.client as ws_client_mod + + monkeypatch.setattr(ws_client_mod.Client, "_get_conn_url", lambda self: f"{ws_url}?device_id=d1&service_id=1") + + +def test_feishu_stop_closes_real_sdk_connection(monkeypatch): + from websockets.asyncio.server import serve + + async def go(): + connected = asyncio.Event() + closed = asyncio.Event() + + async def handler(websocket): + connected.set() + try: + async for _ in websocket: + pass + finally: + closed.set() + + async with serve(handler, "127.0.0.1", 0) as server: + port = server.sockets[0].getsockname()[1] + _point_lark_at(monkeypatch, f"ws://127.0.0.1:{port}/") + channel = FeishuChannel(MessageBus(), {"app_id": "test", "app_secret": "test"}) + await channel.start() + thread = channel._thread + await asyncio.wait_for(connected.wait(), 5) + + started = time.monotonic() + await channel.stop() + elapsed = time.monotonic() - started + + assert elapsed < 5, f"stop() took {elapsed:.2f}s" + await asyncio.wait_for(closed.wait(), 5) + assert not thread.is_alive() + assert channel._thread is None + + asyncio.run(go()) + + +def test_feishu_stop_aborts_transport_when_peer_never_acks_close(monkeypatch, caplog): + """Real lark-oapi client against a peer that completes the handshake but never answers the close frame. + + The bounded ``_disconnect()`` wait is cancelled before websockets' own + ``close_timeout`` abort runs, so the channel must release the socket itself; + ``peer.closed`` only fires once the TCP connection is actually gone. + """ + from support.local_sockets import StalledWebSocketServer, transport_socket, wait_until + + caplog.set_level(logging.WARNING, logger="app.channels.feishu") + + async def go(): + async with StalledWebSocketServer() as peer: + _point_lark_at(monkeypatch, peer.url) + monkeypatch.setattr(feishu_module, "FEISHU_WS_CLOSE_TIMEOUT_SECONDS", 0.5) + channel = FeishuChannel(MessageBus(), {"app_id": "test", "app_secret": "test"}) + await channel.start() + thread = channel._thread + await asyncio.wait_for(peer.accepted.wait(), 5) + # Wait for lark to finish its dial (it assigns _conn under _lock). + connection = await wait_until(lambda: channel._ws_client is not None and not channel._ws_client._lock.locked() and channel._ws_client._conn) + sock = transport_socket(connection) + assert sock.fileno() != -1 + + started = time.monotonic() + await channel.stop() + elapsed = time.monotonic() - started + + assert elapsed < 5, f"stop() took {elapsed:.2f}s" + assert not thread.is_alive() + assert channel._thread is None + # The socket is released by stop() itself, not by a later GC pass. + assert sock.fileno() == -1 + assert "transport aborted" in caplog.text + await asyncio.wait_for(peer.closed.wait(), 5) + + asyncio.run(go()) + + +def test_feishu_stop_without_sdk_thread_is_a_no_op_for_the_loop_signal(): + async def go(): + channel = FeishuChannel(MessageBus(), {"app_id": "test", "app_secret": "test"}) + channel._running = True + await channel.stop() + assert channel._thread is None + + asyncio.run(go()) + + def test_feishu_event_handler_ignores_non_content_message_events(): import lark_oapi as lark