fix(channels): give DingTalk AI cards the channel token, not the SDK fetch (#7597)

* fix(channels): give DingTalk AI cards the channel token, not the SDK fetch

AICardReplier's async card calls read dingtalk_client.get_access_token()
inline, and dingtalk-stream refreshes it with requests.post and no
timeout. Card creation and card streaming both run on the Gateway loop,
so the first card and every SDK refresh after it (about every two hours)
blocked the loop for the whole token request, without bound if DingTalk
stalled.

Hand the replier a _CardTokenClient that serves the channel's own
httpx token (10s timeout), set before card creation and before every
streamed update. A token failure while streaming now raises into send's
existing sampleMarkdown fallback instead of being skipped silently.

Pin it with a strict blocking-I/O anchor over a real DingTalkStreamClient
and AICardReplier, a gate meta-check, and unit tests for token rotation
and failure. LocalHttpServer also records each request's method, path
and headers.

* docs(changelog): reference #7597 for the DingTalk AI card token fetch

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

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

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

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 <name>` 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/<name>/chats/<thread_id>`; `lead_agent` omits it and stays on the ordinary chat route. 
 `/model <name>` 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.

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."""
 

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"

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:

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()