diff --git a/CHANGELOG.md b/CHANGELOG.md index ec9c8608fc..8505a8f218 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,13 @@ This section accumulates work toward the **2.3.0** milestone ### Fixed +- **channels:** A DingTalk reply in AI card mode (`card_template_id`) is no + longer lost when a card request fails, for example without card permission + or during a DingTalk outage. dingtalk-stream logs a failed card request and + returns as if it succeeded, so neither the text running reply nor the + `sampleMarkdown` fallback for the final answer ever ran. The channel now + sends card requests itself and falls back when one fails. ([#7781]) + - **harness:** Retrying a failed item no longer reopens a cancelled durable subagent batch. `retry_item` kept only a paused batch as it was, so a failed item of a cancelled batch put the batch back to `queued`, the worker claimed @@ -15989,4 +15996,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7732]: https://github.com/bytedance/deer-flow/pull/7732 [#7762]: https://github.com/bytedance/deer-flow/pull/7762 [#7772]: https://github.com/bytedance/deer-flow/pull/7772 +[#7781]: https://github.com/bytedance/deer-flow/pull/7781 [#7792]: https://github.com/bytedance/deer-flow/pull/7792 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index bf8ef3c752..d1dd152155 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -20,6 +20,10 @@ ### 修复 +- **渠道：** 钉钉 AI 卡片模式（`card_template_id`）下，卡片请求失败（例如缺少卡片权限或钉钉服务异常）时 + 不再丢失回复。dingtalk-stream 会记录失败的卡片请求后按成功返回，因此文本形式的运行中提示和最终答案的 + `sampleMarkdown` 兜底都不会执行。现在渠道自行发送卡片请求，并在失败时回退。([#7781]) + - **harness：** 重试失败条目不再重新打开已取消的持久子智能体批次。`retry_item` 此前只保留已暂停批次的状态，因此重试已取消批次中的失败条目会把批次改回 `queued`，worker 随即认领并执行该条目，批次最终变为 `completed`。现在对已取消 @@ -12739,4 +12743,5 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#7732]: https://github.com/bytedance/deer-flow/pull/7732 [#7762]: https://github.com/bytedance/deer-flow/pull/7762 [#7772]: https://github.com/bytedance/deer-flow/pull/7772 +[#7781]: https://github.com/bytedance/deer-flow/pull/7781 [#7792]: https://github.com/bytedance/deer-flow/pull/7792 diff --git a/backend/app/channels/dingtalk.py b/backend/app/channels/dingtalk.py index 17c44e4840..c837d888e5 100644 --- a/backend/app/channels/dingtalk.py +++ b/backend/app/channels/dingtalk.py @@ -10,6 +10,7 @@ import re import stat import threading import time +import uuid from pathlib import Path from typing import Any from urllib.parse import quote_plus @@ -202,7 +203,6 @@ class DingTalkChannel(Channel): self._stream_lock = threading.Lock() self._incoming_messages: dict[str, Any] = {} self._incoming_messages_lock = threading.Lock() - self._card_repliers: dict[str, Any] = {} # Serialize inbound-file writes into the uploads directory to avoid # racing writers clobbering one another (mirrors FeishuChannel). self._file_write_lock = threading.Lock() @@ -263,7 +263,6 @@ class DingTalkChannel(Channel): self._dingtalk_client = None with self._incoming_messages_lock: self._incoming_messages.clear() - self._card_repliers.clear() self._card_track_ids.clear() if self._thread: # The thread exits once its loop has stopped; keep the join off the @@ -351,12 +350,10 @@ class DingTalkChannel(Channel): logger.warning("[DingTalk] card stream failed, falling back to sampleMarkdown") if msg.is_final: self._card_track_ids.pop(source_key, None) - self._card_repliers.pop(out_track_id, None) await self._send_markdown_fallback(robot_code, conversation_type, sender_staff_id, conversation_id, msg.text) return if msg.is_final: self._card_track_ids.pop(source_key, None) - self._card_repliers.pop(out_track_id, None) return async def send_markdown() -> None: @@ -1244,30 +1241,42 @@ class DingTalkChannel(Channel): logger.warning("[DingTalk] SDK client or chatbot_message unavailable, skipping AI card") return None + out_track_id = uuid.uuid4().hex try: - from dingtalk_stream.card_replier import AICardReplier - except ImportError: - logger.warning("[DingTalk] dingtalk-stream card_replier not available") - return None - - try: - 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}, - ) - if not card_instance_id: - return None - - self._card_repliers[card_instance_id] = replier - logger.info("[DingTalk] AI card created: outTrackId=%s", card_instance_id) - return card_instance_id + await self._card_request("POST", "/v1.0/card/instances/createAndDeliver", self._card_create_body(out_track_id, chatbot_message, initial_text)) except Exception: logger.exception("[DingTalk] failed to create AI card") return None + logger.info("[DingTalk] AI card created: outTrackId=%s", out_track_id) + return out_track_id + + def _card_create_body(self, out_track_id: str, chatbot_message: Any, initial_text: str) -> dict[str, Any]: + """The ``createAndDeliver`` body dingtalk-stream's ``AICardReplier`` sends.""" + body: dict[str, Any] = { + "cardTemplateId": self._card_template_id, + "outTrackId": out_track_id, + "cardData": {"cardParamMap": {"content": initial_text}}, + "callbackType": "STREAM", + "imGroupOpenSpaceModel": {"supportForward": True}, + "imRobotOpenSpaceModel": {"supportForward": True}, + } + hosting_context = getattr(chatbot_message, "hosting_context", None) + conversation_type = str(chatbot_message.conversation_type) + if conversation_type == _CONVERSATION_TYPE_GROUP: + body["openSpaceId"] = f"dtv1.card//IM_GROUP.{chatbot_message.conversation_id}" + deliver_model: dict[str, Any] = {"robotCode": self._client_id} + body["imGroupOpenDeliverModel"] = deliver_model + elif conversation_type == _CONVERSATION_TYPE_P2P: + body["openSpaceId"] = f"dtv1.card//IM_ROBOT.{chatbot_message.sender_staff_id}" + deliver_model = {"spaceType": "IM_ROBOT"} + body["imRobotOpenDeliverModel"] = deliver_model + else: + return body + if hosting_context is not None: + deliver_model["extension"] = {"hostingRepliedContext": json.dumps({"userId": hosting_context.user_id})} + return body + async def _stream_update_card( self, out_track_id: str, @@ -1276,20 +1285,35 @@ class DingTalkChannel(Channel): is_finalize: bool = False, is_error: bool = False, ) -> None: - replier = self._card_repliers.get(out_track_id) - 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", - content_value=content, - append=False, - finished=is_finalize, - failed=is_error, + await self._card_request( + "PUT", + "/v1.0/card/streaming", + { + "outTrackId": out_track_id, + "guid": str(uuid.uuid4()), + "key": "content", + "content": content, + "isFull": True, + "isFinalize": is_finalize, + "isError": is_error, + }, ) + async def _card_request(self, method: str, path: str, body: dict[str, Any]) -> None: + """Send an AI card request with the channel token; raise if it fails. + + dingtalk-stream's ``AICardReplier`` logs a failed card request and + returns as if it succeeded; its create call even returns the card ID. + Neither the running reply's text fallback nor the sampleMarkdown + fallback in ``send`` could then run, so the reply was lost. + """ + token = await self._get_access_token() + async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client: + response = await client.request(method, f"{DINGTALK_API_BASE}{path}", headers=self._api_headers(token), json=body) + if response.is_error: + logger.warning("[DingTalk] card request %s %s returned %s: %s", method, path, response.status_code, response.text[:500]) + response.raise_for_status() + # -- media upload -------------------------------------------------------- async def _upload_media(self, file_path: str | Path, media_type: str) -> str | None: @@ -1318,26 +1342,6 @@ 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 index 18ee568ea6..a4eeaf1cca 100644 --- a/backend/tests/blocking_io/test_dingtalk_card_token.py +++ b/backend/tests/blocking_io/test_dingtalk_card_token.py @@ -1,26 +1,23 @@ """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. +dingtalk-stream's ``get_access_token()`` refreshes through ``requests.post`` +with no timeout. Card creation (``_send_running_reply``) and card streaming +(``send``) both run on the Gateway loop, so a card call through the SDK's +``AICardReplier`` blocked the loop for the whole refresh, about every two +hours. The channel now sends card requests through its own httpx client with +its own token. -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. +This anchor drives the real channel methods with a real +``DingTalkStreamClient`` whose token cache is empty, under the strict gate, and +asserts the card requests reached the local OpenAPI peer with the channel's +token. The SDK's endpoints point away from the real API (its token endpoint at +the discard port), so a call that slipped back to the SDK fails fast instead of +leaving the machine or 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 @@ -67,36 +64,3 @@ async def test_card_create_and_stream_never_fetch_the_sdk_token_on_the_loop(monk ("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/test_dingtalk_channel.py b/backend/tests/test_dingtalk_channel.py index 93ba100c65..a48f149799 100644 --- a/backend/tests/test_dingtalk_channel.py +++ b/backend/tests/test_dingtalk_channel.py @@ -1427,7 +1427,6 @@ class TestCardMode: channel = DingTalkChannel(bus, config={}) assert channel._card_template_id == "" assert channel._card_track_ids == {} - assert channel._card_repliers == {} assert channel._incoming_messages == {} assert channel._dingtalk_client is None @@ -1767,7 +1766,7 @@ class TestCardMode: _run(go()) - def test_card_replier_cleanup_on_final(self): + def test_card_track_cleanup_on_final(self): async def go(): bus = MessageBus() channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"}) @@ -1777,7 +1776,6 @@ class TestCardMode: source_key = f"{_CONVERSATION_TYPE_P2P}:user_001::msg_001" channel._card_track_ids[source_key] = "track_001" - channel._card_repliers["track_001"] = MagicMock() msg = OutboundMessage( channel_name="dingtalk", @@ -1796,7 +1794,6 @@ class TestCardMode: await channel.send(msg) assert source_key not in channel._card_track_ids - assert "track_001" not in channel._card_repliers _run(go()) @@ -1828,21 +1825,11 @@ class TestCardMode: _run(go()) - def test_stream_update_card_raises_without_replier(self): - async def go(): - bus = MessageBus() - channel = DingTalkChannel(bus, config={"card_template_id": "tpl_123"}) - - with pytest.raises(RuntimeError, match="No AICardReplier found"): - await channel._stream_update_card("nonexistent_track", "content") - - _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. + """Card requests send 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. + timeout, which on the Gateway loop would block every channel. """ pytest.importorskip("dingtalk_stream") import dingtalk_stream @@ -1885,21 +1872,148 @@ class TestCardMode: _run(go()) - def test_stream_update_card_raises_when_the_channel_token_fails(self): + @staticmethod + def _card_channel(monkeypatch, base: str) -> DingTalkChannel: + """A card-mode channel holding a valid token, so the API sees only card calls.""" + monkeypatch.setattr("app.channels.dingtalk.DINGTALK_API_BASE", base) + # The SDK too, so card calls can never reach the real DingTalk API. + monkeypatch.setattr("dingtalk_stream.card_replier.DINGTALK_OPENAPI_ENDPOINT", base) + channel = DingTalkChannel(MessageBus(), config={"card_template_id": "tpl_123"}) + channel._client_id = "robot_001" + channel._cached_token = "tok" + channel._token_expires_at = time.monotonic() + 3600 + channel._dingtalk_client = MagicMock() + return channel + + @staticmethod + def _incoming(conversation_type: str = _CONVERSATION_TYPE_P2P): + dingtalk_stream = pytest.importorskip("dingtalk_stream") + incoming = dingtalk_stream.ChatbotMessage() + incoming.conversation_type = conversation_type + incoming.sender_staff_id = "user_001" + incoming.conversation_id = "cid_001" + incoming.message_id = "msg_001" + return incoming + + @staticmethod + async def _reply_turn(channel: DingTalkChannel, incoming) -> None: + """Running reply, one streamed chunk and the final answer, as the manager sends them.""" + metadata = {"conversation_type": _CONVERSATION_TYPE_P2P, "sender_staff_id": "user_001", "conversation_id": "", "message_id": "msg_001"} + inbound = channel._make_inbound(chat_id="user_001", user_id="user_001", text="hello", metadata=metadata) + channel._incoming_messages[channel._make_card_source_key(inbound)] = incoming + await channel._send_running_reply("user_001", inbound) + for text, is_final in [("Partial", False), ("Final answer.", True)]: + await channel.send(OutboundMessage(channel_name="dingtalk", chat_id="user_001", thread_id="t1", text=text, is_final=is_final, thread_ts="msg_001", metadata=metadata)) + + @pytest.mark.parametrize("conversation_type", [_CONVERSATION_TYPE_P2P, _CONVERSATION_TYPE_GROUP], ids=["p2p", "group"]) + @pytest.mark.parametrize("hosted", [False, True], ids=["direct", "hosted"]) + def test_card_requests_match_the_sdk_request_bodies(self, monkeypatch, conversation_type, hosted): + """The channel sends the bodies dingtalk-stream's ``AICardReplier`` sends.""" + from dingtalk_stream import card_replier + from dingtalk_stream.chatbot import HostingContext + from support.local_sockets import LocalHttpServer + + incoming = self._incoming(conversation_type) + if hosted: + incoming.hosting_context = HostingContext() + incoming.hosting_context.user_id = "host_001" + + async def go(): + async with LocalHttpServer() as api: + base = api.url.removesuffix("/open") + monkeypatch.setattr(card_replier, "DINGTALK_OPENAPI_ENDPOINT", base) + sdk_client = MagicMock() + sdk_client.credential.client_id = "robot_001" + sdk_client.get_access_token.return_value = "tok" + sdk = card_replier.AICardReplier(sdk_client, incoming) + sdk_track = await sdk.async_create_and_deliver_card(card_template_id="tpl_123", card_data={"content": "Working on it..."}) + await sdk.async_streaming(card_instance_id=sdk_track, content_key="content", content_value="done", append=False, finished=True, failed=False) + + channel = self._card_channel(monkeypatch, base) + track = await channel._create_and_deliver_card("Working on it...", chatbot_message=incoming) + await channel._stream_update_card(track, "done", is_finalize=True) + + assert [(method, path) for method, path, _ in api.calls] == [("POST", "/v1.0/card/instances/createAndDeliver"), ("PUT", "/v1.0/card/streaming")] * 2 + sdk_create, sdk_stream, create, stream = api.requests + assert create.pop("outTrackId") == track + assert stream.pop("outTrackId") == track + sdk_create.pop("outTrackId") + sdk_stream.pop("outTrackId") + assert stream.pop("guid") != sdk_stream.pop("guid") + assert create == sdk_create + assert stream == sdk_stream + + _run(go()) + + @pytest.mark.parametrize("failure", [403, 500, "unreachable"]) + def test_failed_card_creation_still_delivers_the_reply(self, monkeypatch, failure): + """dingtalk-stream returns a card ID even when the API refused the card.""" + from support.local_sockets import LocalHttpServer + + async def deliver(base: str) -> DingTalkChannel: + channel = self._card_channel(monkeypatch, base) + channel._send_text_message_to_user = AsyncMock() + channel._send_p2p_message = AsyncMock() + await self._reply_turn(channel, self._incoming()) + return channel + + async def go(): + if failure == "unreachable": + async with LocalHttpServer() as closed: + base = closed.url.removesuffix("/open") + channel = await deliver(base) + else: + async with LocalHttpServer(status=failure, body={"code": "Forbidden.AccessDenied"}) as api: + channel = await deliver(api.url.removesuffix("/open")) + assert [(method, path) for method, path, _ in api.calls] == [("POST", "/v1.0/card/instances/createAndDeliver")] + + assert channel._card_track_ids == {} + channel._send_text_message_to_user.assert_awaited_once_with("robot_001", "user_001", "\u23f3 Working on it...") + # Without a card the streamed chunk is skipped and the final answer is sent once. + channel._send_p2p_message.assert_awaited_once_with("robot_001", "user_001", "Final answer.") + + _run(go()) + + def test_failed_card_stream_still_delivers_the_final_reply(self, monkeypatch): + """The SDK's streaming call also logs a failure and returns.""" + from support.local_sockets import LocalHttpServer + + async def go(): + async with LocalHttpServer() as api: + channel = self._card_channel(monkeypatch, api.url.removesuffix("/open")) + channel._send_text_message_to_user = AsyncMock() + channel._send_p2p_message = AsyncMock() + original_create = channel._create_and_deliver_card + + async def create_then_refuse(*args, **kwargs): + track = await original_create(*args, **kwargs) + api.status = 403 + return track + + channel._create_and_deliver_card = create_then_refuse + await self._reply_turn(channel, self._incoming()) + + assert [(method, path) for method, path, _ in api.calls] == [("POST", "/v1.0/card/instances/createAndDeliver")] + [("PUT", "/v1.0/card/streaming")] * 2 + assert channel._card_track_ids == {} + channel._send_text_message_to_user.assert_not_awaited() + channel._send_p2p_message.assert_awaited_once_with("robot_001", "user_001", "Final answer.") + + _run(go()) + + def test_stream_update_card_raises_when_the_channel_token_fails(self, monkeypatch): """A token failure surfaces, so ``send`` falls back instead of dropping the update.""" + http_client = MagicMock(side_effect=AssertionError("no card request without a token")) + monkeypatch.setattr("app.channels.dingtalk.httpx.AsyncClient", http_client) 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()) + http_client.assert_not_called() def test_stop_clears_card_state(self): async def go(): @@ -1908,14 +2022,12 @@ class TestCardMode: channel._running = True channel._dingtalk_client = MagicMock() channel._incoming_messages["key"] = MagicMock() - channel._card_repliers["track"] = MagicMock() channel._card_track_ids["source"] = "track" await channel.stop() assert channel._dingtalk_client is None assert channel._incoming_messages == {} - assert channel._card_repliers == {} assert channel._card_track_ids == {} _run(go())