diff --git a/CHANGELOG.md b/CHANGELOG.md index 13bb77f71c..add7931a14 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1402,6 +1402,15 @@ As of 2026-10-21, the milestone includes **829 merged pull requests**. ### Fixed +- **channels:** DingTalk AI cards no longer fetch an access token on the + Gateway event loop. Card creation and card streaming went through + dingtalk-stream's synchronous `get_access_token()`, which posts with + `requests` and no timeout, so the first card and every SDK token refresh + (about every two hours) stalled every channel, request and stream for the + whole request, indefinitely if DingTalk stalled. Card calls now use the + channel's own bounded token, and a token failure while streaming falls back + to a Markdown reply instead of dropping the update. ([#7597]) + - **web fetch:** Cache unavailable Readability.js probes and stop retrying the Node extractor after a process-launch failure. Installations without a working Node/npm toolchain now use the existing link-preserving Python @@ -13220,4 +13229,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7591]: https://github.com/bytedance/deer-flow/pull/7591 [#7593]: https://github.com/bytedance/deer-flow/pull/7593 [#7594]: https://github.com/bytedance/deer-flow/pull/7594 +[#7597]: https://github.com/bytedance/deer-flow/pull/7597 [#7606]: https://github.com/bytedance/deer-flow/pull/7606 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 30f7f5afd4..200daaa7ee 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -1116,6 +1116,12 @@ ### 修复 +- **渠道：** 钉钉 AI 卡片不再在 Gateway 事件循环上获取 access token。此前创建卡片和流式更新卡片 + 都会调用 dingtalk-stream 的同步 `get_access_token()`，它用 `requests` 发请求且没有超时，因此 + 首张卡片以及之后每次 SDK token 刷新（约每两小时一次）都会在整个请求期间阻塞所有渠道、请求和 + 流；钉钉无响应时会一直阻塞。现在卡片调用使用渠道自身带超时的 token，流式更新时 token 获取失败 + 会回退为 Markdown 回复，而不是丢弃这次更新。([#7597]) + - **网页抓取：** 缓存 Readability.js 不可用的探测结果，并在 Node 抽取器首次出现 进程启动失败后停止重试。没有可用 Node/npm 工具链的安装环境会在后续抓取中直接使用 现有且保留链接的 Python 回退，不再重复执行失败的子进程探测；单篇文档导致的非零退出 @@ -10542,4 +10548,5 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#7591]: https://github.com/bytedance/deer-flow/pull/7591 [#7593]: https://github.com/bytedance/deer-flow/pull/7593 [#7594]: https://github.com/bytedance/deer-flow/pull/7594 +[#7597]: https://github.com/bytedance/deer-flow/pull/7597 [#7606]: https://github.com/bytedance/deer-flow/pull/7606 diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index b15c773019..e96ccb64aa 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -48,7 +48,7 @@ Shared attachment feedback uses public `attachment_delivery_text`, preserves rou 6b. GitHub chat (`ChannelRunPolicy.fire_and_forget=True`): `runs.create()` returns once the run is `pending`; the manager does not wait for the final state and does not publish an outbound; the agent posts its own reply via sandbox `gh`. `ConflictError` on a busy thread still trips the standard `THREAD_BUSY_MESSAGE` path (log-only on GitHub); with `buffer_followups_on_busy=True` (GitHub's default) the triggering message is captured into a per-thread buffer so a concurrent comment is not silently dropped (see "Follow-up buffering while busy"). 7. Feishu patches one card per source (`config.update_multi=true`), with topic previews and queued → running → final states. Opt-in cards show context/question; bounded tokens preserve request ID/source. Originator/option checks consume once; consumption fails closed across workers (`feishu_card_interaction.py`). Card replies keep bound aliases in fields/metadata; ordinary replies preserve `/new`. Accepted intake retires only that card's buttons/text hint before terminal PATCH; rejected intake keeps text fallback. Non-final chunks store each thread alias once per reply; the final rewrites all. 8. Telegram edits the "Working on it..." stream target: 1s private/3s group throttle (20 msg/min), 4096-char truncation, rate-limited updates dropped; final edits split overflow. Text/preview/photo/document reply sends use `allow_sending_without_reply` only for `chat_id > 0`: missing private targets degrade atomically; group/forum routing and transport retries stay unchanged. -9. DingTalk AI Cards (`card_template_id`): `runs.stream()` → create card with initial text → `PUT /v1.0/card/streaming` → finalize (`is_final=True`). Creation/streaming failure uses `sampleMarkdown`; tables preserve escaped pipes and empty edge cells. +9. DingTalk AI Cards (`card_template_id`): `runs.stream()` → create card with initial text → `PUT /v1.0/card/streaming` → finalize (`is_final=True`). Creation/streaming failure uses `sampleMarkdown`; tables preserve escaped pipes and empty edge cells. Card calls get the channel token, not the SDK's. 10. For commands (`/new`, `/status`, `/models`, `/model`, `/memory`, `/goal`, `/agent`, `/help`): handle locally or query Gateway API. `/agent list` reads the effective owner's Custom Agents; `/agent use ` validates in the same owner bucket, creates a new thread, and persists `channel_agent_name` in its metadata. A Custom Agent writes canonical `metadata.agent_name` so thread-search routes Web continuation to `/workspace/agents//chats/`; `lead_agent` omits it and stays on the ordinary chat route. `/model ` persists `channel_model_name` in the *current* thread's metadata (in place, no fresh thread); `/model` reports the effective model, `/model default` clears it (stale pins self-clear). The `channel_agent_name` cache reloads after restart before a resumed turn routes. An explicit selection also normalizes `agent_name` across the top-level run context and both RunnableConfig carriers (`context`, `configurable`) — cleared for `lead_agent` — before Gateway's `setdefault` merge, so stale channel defaults cannot silently win. diff --git a/backend/app/channels/dingtalk.py b/backend/app/channels/dingtalk.py index 2bd2286503..17c44e4840 100644 --- a/backend/app/channels/dingtalk.py +++ b/backend/app/channels/dingtalk.py @@ -1251,7 +1251,9 @@ class DingTalkChannel(Channel): return None try: - replier = AICardReplier(self._dingtalk_client, chatbot_message) + token_client = _CardTokenClient(self._dingtalk_client.credential) + token_client.access_token = await self._get_access_token() + replier = AICardReplier(token_client, chatbot_message) card_instance_id = await replier.async_create_and_deliver_card( card_template_id=self._card_template_id, card_data={"content": initial_text}, @@ -1278,6 +1280,7 @@ class DingTalkChannel(Channel): if not replier: raise RuntimeError(f"No AICardReplier found for track ID {out_track_id}") + replier.dingtalk_client.access_token = await self._get_access_token() await replier.async_streaming( card_instance_id=out_track_id, content_key="content", @@ -1315,6 +1318,26 @@ class DingTalkChannel(Channel): return None +class _CardTokenClient: + """The client ``AICardReplier`` sees: it serves the channel's access token. + + The replier's async methods call ``dingtalk_client.get_access_token()`` + inline, and dingtalk-stream's version refreshes through ``requests.post`` + with no timeout, so on the Gateway loop every refresh (about every two + hours) blocked the loop for the whole request, indefinitely if DingTalk + stalled. The channel sets ``access_token`` from its own bounded + ``_get_access_token()`` before every replier call, so the replier never + fetches a token itself. + """ + + def __init__(self, credential: Any) -> None: + self.credential = credential + self.access_token = "" + + def get_access_token(self) -> str: + return self.access_token + + class _DingTalkMessageHandler: """Callback handler registered with dingtalk-stream.""" diff --git a/backend/tests/blocking_io/test_dingtalk_card_token.py b/backend/tests/blocking_io/test_dingtalk_card_token.py new file mode 100644 index 0000000000..18ee568ea6 --- /dev/null +++ b/backend/tests/blocking_io/test_dingtalk_card_token.py @@ -0,0 +1,102 @@ +"""Regression anchor: DingTalk AI card calls must not fetch a token on the loop. + +``AICardReplier``'s async methods call ``dingtalk_client.get_access_token()`` +inline, and dingtalk-stream's version refreshes through ``requests.post`` with +no timeout. Card creation (``_send_running_reply``) and card streaming +(``send``) both run on the Gateway loop, so every SDK token refresh, about +every two hours, blocked the loop for the whole request. The channel now hands +the replier its own httpx token instead. + +These anchors drive the real channel methods and the real ``AICardReplier`` +with a real ``DingTalkStreamClient`` whose token cache is empty, under the +strict gate. The SDK swallows every error from its token fetch (a gate +``BlockingError`` included) and then skips the request, so each anchor asserts +the card requests reached the local OpenAPI peer with the channel's token. The +SDK's own token endpoint points at the discard port, so a fetch that slipped +past the gate fails fast instead of hanging the loop. +""" + +from __future__ import annotations + +import pytest +import requests +from blockbuster import BlockingError +from dingtalk_stream import card_replier, stream +from dingtalk_stream.chatbot import ChatbotMessage +from dingtalk_stream.credential import Credential +from dingtalk_stream.stream import DingTalkStreamClient +from support.local_sockets import LocalHttpServer + +from app.channels import dingtalk as dingtalk_module +from app.channels.dingtalk import _CONVERSATION_TYPE_P2P, DingTalkChannel +from app.channels.message_bus import MessageBus + +pytestmark = pytest.mark.asyncio + +_UNREACHABLE_SDK_ENDPOINT = "http://127.0.0.1:9" + + +def _channel() -> DingTalkChannel: + channel = DingTalkChannel(MessageBus(), config={"card_template_id": "tpl_123"}) + channel._client_id, channel._client_secret = "id", "secret" + channel._dingtalk_client = DingTalkStreamClient(Credential("id", "secret")) + return channel + + +def _incoming() -> ChatbotMessage: + incoming = ChatbotMessage() + incoming.conversation_type = _CONVERSATION_TYPE_P2P + incoming.sender_staff_id = "user_001" + return incoming + + +async def test_card_create_and_stream_never_fetch_the_sdk_token_on_the_loop(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(stream, "DINGTALK_OPENAPI_ENDPOINT", _UNREACHABLE_SDK_ENDPOINT) + async with LocalHttpServer(body={"accessToken": "tok", "expireIn": 7200}) as api: + base = api.url.removesuffix("/open") + monkeypatch.setattr(dingtalk_module, "DINGTALK_API_BASE", base) + monkeypatch.setattr(card_replier, "DINGTALK_OPENAPI_ENDPOINT", base) + channel = _channel() + + track = await channel._create_and_deliver_card("Working on it...", chatbot_message=_incoming()) + assert track + await channel._stream_update_card(track, "done", is_finalize=True) + + assert [(method, path, headers.get("x-acs-dingtalk-access-token")) for method, path, headers in api.calls] == [ + ("POST", "/v1.0/oauth2/accessToken", None), + ("POST", "/v1.0/card/instances/createAndDeliver", "tok"), + ("PUT", "/v1.0/card/streaming", "tok"), + ] + + +async def test_blocking_probe_sdk_token_fetch_actually_trips_the_gate(monkeypatch: pytest.MonkeyPatch) -> None: + """Meta-check: a replier on the SDK client (``main``'s wiring) trips the gate. + + The SDK catches the ``BlockingError`` itself, so the probe records what its + ``requests.post`` raised. It goes through ``_stream_update_card`` because + the gate only flags IO whose stack passes through ``app.*``. + """ + monkeypatch.setattr(stream, "DINGTALK_OPENAPI_ENDPOINT", _UNREACHABLE_SDK_ENDPOINT) + raised: list[BaseException] = [] + post = requests.post + + def recording_post(*args, **kwargs): + try: + return post(*args, **kwargs) + except BaseException as exc: + raised.append(exc) + raise + + monkeypatch.setattr(requests, "post", recording_post) + channel = _channel() + channel._get_access_token = _static_token + channel._card_repliers["track"] = card_replier.AICardReplier(channel._dingtalk_client, _incoming()) + + await channel._stream_update_card("track", "done") + + assert len(raised) == 1 + assert isinstance(raised[0], BlockingError) + + +async def _static_token() -> str: + return "tok" diff --git a/backend/tests/support/local_sockets.py b/backend/tests/support/local_sockets.py index e1606dd8b6..59ad646def 100644 --- a/backend/tests/support/local_sockets.py +++ b/backend/tests/support/local_sockets.py @@ -8,7 +8,8 @@ Both run on the calling test's event loop and only ever bind ``127.0.0.1:0``. 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. + body is recorded so tests can assert on what the SDK sent, and ``calls`` + keeps each request's method, path and headers (names lowercased). """ from __future__ import annotations @@ -96,6 +97,7 @@ class LocalHttpServer: self.body = body if body is not None else {} self.stall = stall self.requests: list[dict[str, Any]] = [] + self.calls: list[tuple[str, str, dict[str, str]]] = [] self.accepted = asyncio.Event() self.closed = asyncio.Event() self.port = 0 @@ -123,6 +125,10 @@ class LocalHttpServer: 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"" + request_line, *header_lines = head.decode("latin-1").split("\r\n") + method, path, _ = request_line.split(" ", 2) + headers = {name.strip().lower(): value.strip() for name, sep, value in (line.partition(":") for line in header_lines) if sep} + self.calls.append((method, path, headers)) try: self.requests.append(json.loads(raw) if raw else {}) except ValueError: diff --git a/backend/tests/test_dingtalk_channel.py b/backend/tests/test_dingtalk_channel.py index 8682c191f8..93ba100c65 100644 --- a/backend/tests/test_dingtalk_channel.py +++ b/backend/tests/test_dingtalk_channel.py @@ -1838,6 +1838,69 @@ class TestCardMode: _run(go()) + def test_card_calls_use_the_channel_token_not_the_sdk_fetch(self, monkeypatch): + """The AI card replier sends the channel's httpx token; the SDK never fetches one. + + dingtalk-stream's ``get_access_token()`` posts with ``requests`` and no + timeout, inline in the replier's async methods on the Gateway loop. + """ + pytest.importorskip("dingtalk_stream") + import dingtalk_stream + from dingtalk_stream import card_replier + from support.local_sockets import LocalHttpServer + + async def go(): + async with LocalHttpServer(body={"accessToken": "tok-1", "expireIn": 7200}) as api: + base = api.url.removesuffix("/open") + monkeypatch.setattr("app.channels.dingtalk.DINGTALK_API_BASE", base) + monkeypatch.setattr(card_replier, "DINGTALK_OPENAPI_ENDPOINT", base) + channel = DingTalkChannel(MessageBus(), config={"card_template_id": "tpl_123"}) + channel._client_id, channel._client_secret = "id", "secret" + sdk_client = MagicMock() + sdk_client.credential.client_id = "id" + sdk_client.get_access_token.side_effect = AssertionError("the SDK must not fetch a token") + channel._dingtalk_client = sdk_client + incoming = dingtalk_stream.ChatbotMessage() + incoming.conversation_type = _CONVERSATION_TYPE_P2P + incoming.sender_staff_id = "user_001" + + track = await channel._create_and_deliver_card("Working on it...", chatbot_message=incoming) + assert track + + # An expired channel token is refreshed before the streamed update. + channel._token_expires_at = 0.0 + api.body = {"accessToken": "tok-2", "expireIn": 7200} + await channel._stream_update_card(track, "done", is_finalize=True) + + sdk_client.get_access_token.assert_not_called() + assert [(method, path, headers.get("x-acs-dingtalk-access-token")) for method, path, headers in api.calls] == [ + ("POST", "/v1.0/oauth2/accessToken", None), + ("POST", "/v1.0/card/instances/createAndDeliver", "tok-1"), + ("POST", "/v1.0/oauth2/accessToken", None), + ("PUT", "/v1.0/card/streaming", "tok-2"), + ] + assert api.requests[0] == {"appKey": "id", "appSecret": "secret"} + assert api.requests[3]["outTrackId"] == track + assert api.requests[3]["isFinalize"] is True + + _run(go()) + + def test_stream_update_card_raises_when_the_channel_token_fails(self): + """A token failure surfaces, so ``send`` falls back instead of dropping the update.""" + + async def go(): + channel = DingTalkChannel(MessageBus(), config={"card_template_id": "tpl_123"}) + replier = MagicMock() + replier.async_streaming = AsyncMock() + channel._card_repliers["track_001"] = replier + channel._get_access_token = AsyncMock(side_effect=ValueError("bad token response")) + + with pytest.raises(ValueError, match="bad token response"): + await channel._stream_update_card("track_001", "content", is_finalize=True) + replier.async_streaming.assert_not_awaited() + + _run(go()) + def test_stop_clears_card_state(self): async def go(): bus = MessageBus()