diff --git a/README.md b/README.md index 596b9a263e..414d5c6acd 100644 --- a/README.md +++ b/README.md @@ -1346,6 +1346,8 @@ Scheduled-task result and lifecycle notices can also be pushed to a bound Feishu DingTalk's Markdown fallback displays tables as labeled blockquotes, preserving escaped pipes (`\|`) within cells and empty leading or trailing cells so values stay under the correct column labels. +DingTalk token caching uses a two-hour fallback when the provider's token lifetime is invalid or no longer than the five-minute refresh margin, avoiding a refresh on every request. + When DeerFlow runs in Docker Compose, IM channels execute inside the `gateway` container. In that case, do not point `channels.langgraph_url` or `channels.gateway_url` at `localhost`; use container service names such as `http://gateway:8001/api` and `http://gateway:8001`, or set `DEER_FLOW_CHANNELS_LANGGRAPH_URL` and `DEER_FLOW_CHANNELS_GATEWAY_URL`. **Commands** diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 99a7b275ae..bf8205c9c4 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -22,7 +22,7 @@ Finalize asyncgens; keep primary errors; propagate lone Bot errors. - `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`: the shared Gateway loop 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**: WeCom, Feishu, WeChat and QQ C2C share the outbox; `capabilities.py` gates enqueue/UI. Unsupported rows fail terminally. Await ACK; `ChannelUnavailable` parks, business rejection counts. WeCom parks socket `RuntimeError` except `Reply ack error`. Feishu/QQ/WeCom group hooks use owner connection IDs; Feishu p2p resolves owner. `group_name`: OAuth state only, no suffix/raw-ID labels. WeCom legacy-group fallback excludes private bindings. QQ: owner/current-App C2C only, never member OpenIDs/passive context. WeChat: encrypted owner/bot/private context across restart; ignore public tokens. `store_connection_context` locks/checks the active identity before merging encrypted extra; browser responses strip legacy tokens. Preferences gate enqueue/send/retry, cancel pending/sending and fence late writes; reenable never replays. In-flight sends may finish. Binding lookup errors park; revoked targets fail before liveness. `_on_finalization` commits one notice with the outcome (stop > pause > finish > run); recovery dedupes, idle uses `idle:`, trials/interrupts stay silent. Web/background events default off: confirm durable terminal state; recheck Human Input before send/retry. Text: localized, no IDs/links. Delivery starts after channels/stops first; failure detaches enqueue. [IM guide](../../docs/IM_CHANNEL_CONNECTIONS.md#proactive-notification-settings). - `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. Feishu/DingTalk `stop()` stop the SDK thread's private loop via `_request_*_shutdown` (Feishu first disarms lark's `_auto_reconnect`; DingTalk runs its own SDK connect-loop copy), 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. Leader lease (`leadership.py`): only the `channel_leases` holder starts a non-GitHub transport (`_start_channel` checks `holds()`; callers acquire first); `_supervise` renews at a fixed rate, runs starts/stops per provider, stops a transport before its lease lapses, re-applies runtime config whose `version` changed; reloads read a disconnect as its tombstone. -- `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()`; 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`) +- `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()`; 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`: `card_template_id` enables in-place AI Card streams; `receive_file` downloads `picture`/`richText` images and `file` documents via `downloadCode` to thread uploads, like Feishu. Invalid `expireIn` or <=300s uses 7200s before caching) - `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. **Backlog paging:** freeze the read floor and track `_backlog_until`; stop/warn on stalled timestamps, never skip a tied second. Removal/live close clears paging and closes backlog; ignore stale frames. Reset cursors per socket. Bounds: [subscription model](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-subscription-model). **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/dingtalk.py b/backend/app/channels/dingtalk.py index 4acea4c793..9c5a7bdda2 100644 --- a/backend/app/channels/dingtalk.py +++ b/backend/app/channels/dingtalk.py @@ -1101,14 +1101,20 @@ class DingTalkChannel(Channel): raise ValueError("DingTalk access token response did not contain a usable accessToken") raw_expires_in = data.get("expireIn", 7200) + now = time.monotonic() try: expires_in = int(raw_expires_in) - except (TypeError, ValueError): + # A lifetime within the refresh margin is already stale. + if expires_in <= _TOKEN_REFRESH_MARGIN_SECONDS: + raise ValueError + expires_at = now + expires_in - _TOKEN_REFRESH_MARGIN_SECONDS + except (TypeError, ValueError, OverflowError): logger.warning("[DingTalk] invalid expireIn value %r, using default 7200s", raw_expires_in) expires_in = 7200 + expires_at = now + expires_in - _TOKEN_REFRESH_MARGIN_SECONDS self._cached_token = access_token.strip() - self._token_expires_at = time.monotonic() + expires_in - _TOKEN_REFRESH_MARGIN_SECONDS + self._token_expires_at = expires_at return self._cached_token @staticmethod diff --git a/backend/tests/test_dingtalk_channel.py b/backend/tests/test_dingtalk_channel.py index 75b0b6e1f5..8d58d9d5dd 100644 --- a/backend/tests/test_dingtalk_channel.py +++ b/backend/tests/test_dingtalk_channel.py @@ -1023,22 +1023,23 @@ class TestAccessTokenValidation: _run(go()) - def test_invalid_expire_in_uses_default(self): + @pytest.mark.parametrize("expire_in", ["invalid", float("inf"), 10**400, 0, -1, 1, 100, 299, 300, "300"]) + def test_invalid_expire_in_uses_default(self, expire_in): async def go(): - import time from unittest.mock import patch bus = MessageBus() channel = DingTalkChannel(bus, config={}) channel._client_id = "k" channel._client_secret = "s" + call_count = 0 class FakeResponse: def raise_for_status(self): pass def json(self): - return {"accessToken": "tok_ok", "expireIn": "invalid"} + return {"accessToken": "tok_ok", "expireIn": expire_in} class FakeClient: async def __aenter__(self): @@ -1048,20 +1049,29 @@ class TestAccessTokenValidation: pass async def post(self, url, **kwargs): + nonlocal call_count + call_count += 1 return FakeResponse() - before = time.monotonic() - with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()): + with ( + patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()), + patch("app.channels.dingtalk.time.monotonic", return_value=1000.0) as clock, + ): token = await channel._get_access_token() + clock.return_value = 1005.0 + cached_token = await channel._get_access_token() assert token == "tok_ok" - assert channel._token_expires_at > before + assert cached_token == "tok_ok" + assert channel._token_expires_at == 7900.0 + assert call_count == 1 _run(go()) class TestTokenCaching: - def test_token_is_cached_across_calls(self): + @pytest.mark.parametrize(("expire_in", "expected_expiry"), [(301, 1001.0), (3600, 4300.0), (7200, 7900.0)]) + def test_token_is_cached_across_calls(self, expire_in, expected_expiry): async def go(): from unittest.mock import patch @@ -1077,7 +1087,7 @@ class TestTokenCaching: pass def json(self): - return {"accessToken": "tok_abc", "expireIn": 7200} + return {"accessToken": "tok_abc", "expireIn": expire_in} class FakeClient: async def __aenter__(self): @@ -1091,13 +1101,22 @@ class TestTokenCaching: call_count += 1 return FakeResponse() - with patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()): + with ( + patch("app.channels.dingtalk.httpx.AsyncClient", return_value=FakeClient()), + patch("app.channels.dingtalk.time.monotonic", return_value=1000.0) as clock, + ): t1 = await channel._get_access_token() + clock.return_value = 1000.5 t2 = await channel._get_access_token() + assert channel._token_expires_at == expected_expiry + assert call_count == 1 + clock.return_value = expected_expiry + t3 = await channel._get_access_token() assert t1 == "tok_abc" assert t2 == "tok_abc" - assert call_count == 1 + assert t3 == "tok_abc" + assert call_count == 2 _run(go())