Summary

The DingTalk platform adapter fails to process any incoming messages in stream mode because it reads business fields (msgId, senderId, sessionWebhook, text, etc.) from the wrong layer of the SDK's message object. This PR fixes the data extraction to match the actual dingtalk-stream SDK wire format.

Root Cause

The dingtalk-stream Python SDK delivers incoming messages as CallbackMessage objects. The actual business payload (sender info, message text, conversation context, session webhook) is nested inside message.data (a dict/JSON string), not exposed as top-level attributes.

The existing code used getattr(message, "sender_id", ""), getattr(message, "text", ""), etc. — all of which silently fell through to empty defaults. The result:

Additionally, DingTalkStreamClient.start() is a native async coroutine in the current SDK, but the code wrapped it in asyncio.to_thread() as if it were blocking — this caused event loop conflicts and the _CallbackHandler.process() had to bridge between threads unnecessarily.

Fix

Data extraction — Read all business fields from message.data dict instead of top-level attributes:

# Before (broken)
msg_id = getattr(message, "message_id", None)
text = getattr(message, "text", None)
session_webhook = getattr(message, "session_webhook", None)

# After (correct)
data = getattr(message, "data", None) or {}
msg_id = data.get("msgId")
text = data.get("text", "")
session_webhook = data.get("sessionWebhook", "")

Async handling — Removed thread-bridging (asyncio.to_thread, run_coroutine_threadsafe) since start() is already async:

# Before
await asyncio.to_thread(self._stream_client.start)
# _CallbackHandler.process() used run_coroutine_threadsafe

# After
await self._stream_client.start()
# _CallbackHandler.process() is async, awaits directly

Diagnostic logging — Added session_webhook logging to aid future debugging of reply routing issues.

Changed Methods

Commits

Test Plan

• Verified CallbackMessage.data dict contains all expected fields (msgId, senderId, senderNick, text, sessionWebhook, conversationId, conversationType, conversationTitle, createAt, senderStaffId, richText)
• Tested 1:1 DM → bot receives text, extracts sender info, replies successfully
• Tested group @mention → bot processes mention, replies via session webhook
• Verified rich text fallback (richText field) still works
• Verified deduplication by msgId still functions correctly
• Confirmed _stream_client.start() runs as native coroutine without event loop conflicts
• Checked diagnostic logging output shows correct session_webhook and chat_id values

Risk Assessment

Medium risk — changes the core message processing path for the DingTalk adapter. However:

• The fix aligns the code with the actual SDK wire format (the current code is completely non-functional in stream mode)
• All field names match the DingTalk Robot callback documentation
• The async refactoring removes a layer of complexity (thread bridging) rather than adding it
• Backward compatibility: data extraction falls back to {} if the attribute is missing, and getattr fallbacks remain for message_id

Recommendation: This PR should be merged together with the webhook domain fix PR (which adds oapi.dingtalk.com to the allowed webhook regex), as both are required for end-to-end DingTalk stream functionality.