DingTalk adapter compatibility fixes and reconnect issues

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #11471 fix(dingtalk): support dingtalk-stream 0.24+ SDK (async process, CallbackMessage, oapi webhooks, TextContent)

  merged to main

  files: gateway/platforms/dingtalk.py, scripts/release.py, tests/gateway/test_dingtalk.py

  Show the diff

  • gateway/platforms/dingtalk.py

    @@ -54,7 +54,7 @@
     MAX_MESSAGE_LENGTH = 20000
     RECONNECT_BACKOFF = [2, 5, 10, 30, 60]
     _SESSION_WEBHOOKS_MAX = 500
    -_DINGTALK_WEBHOOK_RE = re.compile(r'^https://api\.dingtalk\.com/')
    +_DINGTALK_WEBHOOK_RE = re.compile(r'^https://(?:api|oapi)\.dingtalk\.com/')
     
     
     def check_dingtalk_requirements() -> bool:
    @@ -128,12 +128,12 @@ async def connect(self) -> bool:
                 return False
     
         async def _run_stream(self) -> None:
    -        """Run the blocking stream client with auto-reconnection."""
    +        """Run the stream client with auto-reconnection."""
             backoff_idx = 0
             while self._running:
                 try:
                     logger.debug("[%s] Starting stream client...", self.name)
    -                await asyncio.to_thread(self._stream_client.start)
    +                await self._stream_client.start()
                 except asyncio.CancelledError:
                     return
                 except Exception as e:
    @@ -238,18 +238,35 @@ async def _on_message(self, message: "ChatbotMessage") -> None:
     
         @staticmethod
         def _extract_text(message: "ChatbotMessage") -> str:
    -        """Extract plain text from a DingTalk chatbot message."""
    -        text = getattr(message, "text", None) or ""
    -        if isinstance(text, dict):
    -            content = text.get("content", "").strip()
    -        else:
    -            content = str(text).strip()
    -
    -        # Fall back to rich text if present
    +        """Extract plain text from a DingTalk chatbot message.
    +
    +        Handles both legacy and current dingtalk-stream SDK payload shapes:
    +          * legacy: ``message.text`` was a dict ``{"content": "..."}``
    +          * >= 0.20: ``message.text`` is a ``TextContent`` dataclass whose
    +            ``__str__`` returns ``"TextContent(content=...)"`` — never fall
    +            back to ``str(text)`` without extracting ``.content`` first.
    +          * rich text moved from ``message.rich_text`` (list) to
    +            ``message.rich_text_content.rich_text_list`` (list of dicts).
    +        """
    +        text = getattr(message, "text", None)
    +        content = ""
    +        if text is not None:
    +            if isinstance(text, dict):
    +                content = (text.get("content") or "").strip()
    +            elif hasattr(text, "content"):
    +                content = str(text.content or "").strip()
    +            else:
    +                content = str(text).strip()
    +
             if not content:
    -            rich_text = getattr(message, "rich_text", None)
    -            if rich_text and isinstance(rich_text, list):
    -                parts = [item["text"] for item in rich_text
    +            rich_list = None
    +            rtc = getattr(message, "rich_text_content", None)
    +            if rtc is not None and hasattr(rtc, "rich_text_list"):
    +                rich_list = rtc.rich_text_list
    +            if rich_list is None:
    +                rich_list = getattr(message, "rich_text", None)
    +            if rich_list and isinstance(rich_list, list):
    +                parts = [item["text"] for item in rich_list
                              if isinstance(item, dict) and item.get("text")]
                     content = " ".join(parts).strip()
             return content
    @@ -314,19 +331,16 @@ def __init__(self, adapter: DingTalkAdapter, loop: asyncio.AbstractEventLoop):
             self._adapter = adapter
             self._loop = loop
     
    -    def process(self, message: "ChatbotMessage"):
    -        """Called by dingtalk-stream in its thread when a message arrives.
    +    async def process(self, callback_message):
    +        """Called by dingtalk-stream when a message arrives.
     
    -        Schedules the async handler on the main event loop.
    +        dingtalk-stream >= 0.24 passes a CallbackMessage whose `.data` contains
    +        the chatbot payload. Convert it to ChatbotMessage and await the adapter
    +        handler directly on the main event loop.
             """
    -        loop = self._loop
    -        if loop is None or loop.is_closed():
    -            logger.error("[DingTalk] Event loop unavailable, cannot dispatch message")
    -            return dingtalk_stream.AckMessage.STATUS_OK, "OK"
    -
    -        future = asyncio.run_coroutine_threadsafe(self._adapter._on_message(message), loop)
             try:
    -            future.result(timeout=60)
    +            chatbot_msg = ChatbotMessage.from_dict(callback_message.data)
    +            await self._adapter._on_message(chatbot_msg)
             except Exception:
                 logger.exception("[DingTalk] Error processing incoming message")
     

  • scripts/release.py

    @@ -82,6 +82,7 @@
         "[redacted:email]": "aydnOktay",
         "[redacted:email]": "Farukest",
         "[redacted:email]": "rovle",
    +    "[redacted:email]": "kevinskysunny",
         "[redacted:email]": "teyrebaz33",
         "[redacted:email]": "Ruzzgar",
         "[redacted:email]": "alireza78a",

  • tests/gateway/test_dingtalk.py

    @@ -273,3 +273,133 @@ class TestPlatformEnum:
     
         def test_dingtalk_in_platform_enum(self):
             assert Platform.DINGTALK.value == "dingtalk"
    +
    +
    +# ---------------------------------------------------------------------------
    +# SDK compatibility regression tests (dingtalk-stream >= 0.20 / 0.24)
    +# ---------------------------------------------------------------------------
    +
    +
    +class TestWebhookDomainAllowlist:
    +    """Guard the webhook origin allowlist against regression.
    +
    +    The SDK started returning reply webhooks on ``oapi.dingtalk.com`` in
    +    addition to ``api.dingtalk.com``. Both must be accepted, and hostile
    +    lookalikes must still be rejected (SSRF defence-in-depth).
    +    """
    +
    +    def test_api_domain_accepted(self):
    +        from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE
    +        assert _DINGTALK_WEBHOOK_RE.match(
    +            "https://api.dingtalk.com/robot/send?access_token=x"
    +        )
    +
    +    def test_oapi_domain_accepted(self):
    +        from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE
    +        assert _DINGTALK_WEBHOOK_RE.match(
    +            "https://oapi.dingtalk.com/robot/send?access_token=x"
    +        )
    +
    +    def test_http_rejected(self):
    +        from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE
    +        assert not _DINGTALK_WEBHOOK_RE.match("http://api.dingtalk.com/robot/send")
    +
    +    def test_suffix_attack_rejected(self):
    +        from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE
    +        assert not _DINGTALK_WEBHOOK_RE.match(
    +            "https://api.dingtalk.com.evil.example/"
    +        )
    +
    +    def test_unsanctioned_subdomain_rejected(self):
    +        from gateway.platforms.dingtalk import _DINGTALK_WEBHOOK_RE
    +        # Only api.* and oapi.* are allowed — e.g. eapi.dingtalk.com must not slip through
    +        assert not _DINGTALK_WEBHOOK_RE.match("https://eapi.dingtalk.com/robot/send")
    +
    +
    +class TestHandlerProcessIsAsync:
    +    """dingtalk-stream >= 0.20 requires ``process`` to be a coroutine."""
    +
    +    def test_process_is_coroutine_function(self):
    +        from gateway.platforms.dingtalk import _IncomingHandler
    +        assert asyncio.iscoroutinefunction(_IncomingHandler.process)
    +
    +
    +class TestExtractText:
    +    """_extract_text must handle both legacy and current SDK payload shapes.
    +
    +    Before SDK 0.20 ``message.text`` was a ``dict`` with a ``content`` key.
    +    From 0.20 onward it is a ``TextContent`` dataclass whose ``__str__``
    +    returns ``"TextContent(content=...)"`` — falling back to ``str(text)``
    +    leaks that repr into the agent's input.
    +    """
    +
    +    def test_text_as_dict_legacy(self):
    +        from gateway.platforms.dingtalk import DingTalkAdapter
    +        msg = MagicMock()
    +        msg.text = {"content": "hello world"}
    +        msg.rich_text_content = None
    +        msg.rich_text = None
    +        assert DingTalkAdapter._extract_text(msg) == "hello world"
    +
    +    def test_text_as_textcontent_object(self):
    +        """SDK >= 0.20 shape: object with ``.content`` attribute."""
    +        from gateway.platforms.dingtalk import DingTalkAdapter
    +
    +        class FakeTextContent:
    +            content = "hello from new sdk"
    +
    +            def __str__(self):  # mimic real SDK repr
    +                return f"TextContent(content={self.content})"
    +
    +        msg = MagicMock()
    +        msg.text = FakeTextContent()
    +        msg.rich_text_content = None
    +        msg.rich_text = None
    +        result = DingTalkAdapter._extract_text(msg)
    +        assert result == "hello from new sdk"
    +        assert "TextContent(" not in result
    +
    +    def test_text_content_attr_with_empty_string(self):
    +        from gateway.platforms.dingtalk import DingTalkAdapter
    +
    +        class FakeTextContent:
    +            content = ""
    +
    +        msg = MagicMock()
    +        msg.text = FakeTextContent()
    +        msg.rich_text_content = None
    +        msg.rich_text = None
    +        assert DingTalkAdapter._extract_text(msg) == ""
    +
    +    def test_rich_text_content_new_shape(self):
    +        """SDK >= 0.20 exposes rich text as ``message.rich_text_content.rich_text_list``."""
    +        from gateway.platforms.dingtalk import DingTalkAdapter
    +
    +        class FakeRichText:
    +            rich_text_list = [{"text": "hello "}, {"text": "world"}]
    +
    +        msg = MagicMock()
    +        msg.text = None
    +        msg.rich_text_content = FakeRichText()
    +        msg.rich_text = None
    +        result = DingTalkAdapter._extract_text(msg)
    +        assert "hello" in result and "world" in result
    +
    +    def test_rich_text_legacy_shape(self):
    +        """Legacy ``message.rich_text`` list remains supported."""
    +        from gateway.platforms.dingtalk import DingTalkAdapter
    +        msg = MagicMock()
    +        msg.text = None
    +        msg.rich_text_content = None
    +        msg.rich_text = [{"text": "legacy "}, {"text": "rich"}]
    +        result = DingTalkAdapter._extract_text(msg)
    +        assert "legacy" in result and "rich" in result
    +
    +    def test_empty_message(self):
    +        from gateway.platforms.dingtalk import DingTalkAdapter
    +        msg = MagicMock()
    +        msg.text = None
    +        msg.rich_text_content = None
    +        msg.rich_text = None
    +        assert DingTalkAdapter._extract_text(msg) == ""
    +

• #121768 fix(dingtalk): stop the stream reconnect log storm and fail closed on a websockets/SDK mismatch

  merged to main

  files: plugins/platforms/dingtalk/adapter.py, tests/gateway/test_dingtalk_reconnect_circuit_breaker.py