deer-flow/backend/app/channels/dingtalk.py

"""DingTalk channel implementation."""

from __future__ import annotations

import asyncio
import json
import logging
import re
import threading
import time
from pathlib import Path
from typing import Any

import httpx

from app.channels.base import Channel
from app.channels.commands import is_known_channel_command
from app.channels.connection_identity import attach_connection_identity
from app.channels.message_bus import InboundMessage, InboundMessageType, MessageBus, OutboundMessage, ResolvedAttachment
from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.sandbox.sandbox_provider import get_sandbox_provider
from deerflow.uploads.manager import UnsafeUploadPathError, claim_unique_filename, normalize_filename, write_upload_file_no_symlink

logger = logging.getLogger(__name__)

DINGTALK_API_BASE = "https://api.dingtalk.com"

_TOKEN_REFRESH_MARGIN_SECONDS = 300

_CONVERSATION_TYPE_P2P = "1"
_CONVERSATION_TYPE_GROUP = "2"

_MAX_UPLOAD_SIZE_BYTES = 20 * 1024 * 1024

# Inbound attachments are buffered in memory before being persisted and synced
# into the sandbox, so bound them. DingTalk chats accept far larger files than
# an agent can usefully read; oversized ones surface as a failed-load marker.
_MAX_INBOUND_FILE_SIZE_BYTES = 50 * 1024 * 1024


def _normalize_conversation_type(raw: Any) -> str:
    """Normalize ``conversationType`` to ``"1"`` (P2P) or ``"2"`` (group).

    Stream payloads may send int or string values.
    """
    if raw is None:
        return _CONVERSATION_TYPE_P2P
    s = str(raw).strip()
    if s == _CONVERSATION_TYPE_GROUP:
        return _CONVERSATION_TYPE_GROUP
    return _CONVERSATION_TYPE_P2P


def _normalize_allowed_users(allowed_users: Any) -> set[str]:
    if allowed_users is None:
        return set()
    if isinstance(allowed_users, str):
        values = [allowed_users]
    elif isinstance(allowed_users, (list, tuple, set)):
        values = allowed_users
    else:
        logger.warning(
            "DingTalk allowed_users should be a list of user IDs; treating %s as one string value",
            type(allowed_users).__name__,
        )
        values = [allowed_users]
    return {str(uid) for uid in values if str(uid)}


def _is_dingtalk_command(text: str) -> bool:
    return is_known_channel_command(text)


def _display_filename(filename: str) -> str:
    """Collapse whitespace and cap length for embedding a filename in message text.

    ``fileName`` is webhook-supplied data: embedded raw in a failed-load marker,
    a newline could forge a standalone ``/mnt/user-data/uploads/...`` line and an
    over-long name would bloat the message text.
    """
    return re.sub(r"\s+", " ", filename).strip()[:80]


def _extract_text_from_rich_text(rich_text_list: list) -> str:
    parts: list[str] = []
    for item in rich_text_list:
        if isinstance(item, dict) and "text" in item:
            parts.append(item["text"])
    return " ".join(parts)


_FENCED_CODE_BLOCK_RE = re.compile(r"```(\w*)\n(.*?)```", re.DOTALL)
_INLINE_CODE_RE = re.compile(r"`([^`\n]+)`")
_HORIZONTAL_RULE_RE = re.compile(r"^-{3,}$", re.MULTILINE)
_TABLE_SEPARATOR_RE = re.compile(r"^\|[-:| ]+\|$", re.MULTILINE)


def _convert_markdown_table(text: str) -> str:
    # DingTalk sampleMarkdown does not render pipe-delimited tables.
    lines = text.split("\n")
    result: list[str] = []
    i = 0
    while i < len(lines):
        line = lines[i]
        # Detect table: header row followed by separator row
        if i + 1 < len(lines) and line.strip().startswith("|") and _TABLE_SEPARATOR_RE.match(lines[i + 1].strip()):
            headers = [h.strip() for h in line.strip().strip("|").split("|")]
            i += 2  # skip header + separator
            while i < len(lines) and lines[i].strip().startswith("|"):
                cells = [c.strip() for c in lines[i].strip().strip("|").split("|")]
                for h, c in zip(headers, cells):
                    result.append(f"> **{h}**: {c}")
                result.append("")
                i += 1
        else:
            result.append(line)
            i += 1
    return "\n".join(result)


def _adapt_markdown_for_dingtalk(text: str) -> str:
    """Adapt markdown for DingTalk's limited sampleMarkdown renderer."""

    def _code_block_to_quote(match: re.Match) -> str:
        lang = match.group(1)
        code = match.group(2).rstrip("\n")
        prefix = f"> **{lang}**\n" if lang else ""
        quoted_lines = "\n".join(f"> {line}" for line in code.split("\n"))
        return f"{prefix}{quoted_lines}\n"

    text = _FENCED_CODE_BLOCK_RE.sub(_code_block_to_quote, text)
    text = _INLINE_CODE_RE.sub(r"**\1**", text)
    text = _convert_markdown_table(text)
    text = _HORIZONTAL_RULE_RE.sub("───────────", text)
    return text


class DingTalkChannel(Channel):
    """DingTalk IM channel using Stream Push (WebSocket, no public IP needed)."""

    def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None:
        super().__init__(name="dingtalk", bus=bus, config=config)
        self._thread: threading.Thread | None = None
        self._main_loop: asyncio.AbstractEventLoop | None = None
        self._client_id: str = ""
        self._client_secret: str = ""
        self._allowed_users: set[str] = _normalize_allowed_users(config.get("allowed_users"))
        self._cached_token: str = ""
        self._token_expires_at: float = 0.0
        self._token_lock = asyncio.Lock()
        self._card_template_id: str = config.get("card_template_id", "")
        self._card_track_ids: dict[str, str] = {}
        self._dingtalk_client: Any = None
        self._stream_client: Any = None
        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()

    @property
    def supports_streaming(self) -> bool:
        return bool(self._card_template_id)

    async def start(self) -> None:
        if self._running:
            return

        try:
            import dingtalk_stream  # noqa: F401
        except ImportError:
            logger.error("dingtalk-stream is not installed. Install it with: uv add dingtalk-stream")
            return

        client_id = self.config.get("client_id", "")
        client_secret = self.config.get("client_secret", "")

        if not client_id or not client_secret:
            logger.error("DingTalk channel requires client_id and client_secret")
            return

        self._client_id = client_id
        self._client_secret = client_secret
        self._main_loop = asyncio.get_running_loop()

        if self._card_template_id:
            logger.info("[DingTalk] AI Card mode enabled (template=%s)", self._card_template_id)

        self._running = True
        self.bus.subscribe_outbound(self._on_outbound)

        self._thread = threading.Thread(
            target=self._run_stream,
            args=(client_id, client_secret),
            daemon=True,
        )
        self._thread.start()
        logger.info("DingTalk channel started")

    async def stop(self) -> None:
        self._running = False
        self.bus.unsubscribe_outbound(self._on_outbound)

        stream_client = self._stream_client
        if stream_client is not None:
            try:
                if hasattr(stream_client, "disconnect"):
                    stream_client.disconnect()
            except Exception:
                logger.debug("[DingTalk] error disconnecting stream client", exc_info=True)

        self._dingtalk_client = None
        self._stream_client = None
        with self._incoming_messages_lock:
            self._incoming_messages.clear()
        self._card_repliers.clear()
        self._card_track_ids.clear()
        if self._thread:
            self._thread.join(timeout=5)
            self._thread = None
        logger.info("DingTalk channel stopped")

    def _resolve_routing(self, msg: OutboundMessage) -> tuple[str, str, str]:
        """Return (conversation_type, sender_staff_id, conversation_id).

        Uses msg.chat_id as the primary routing key; metadata as fallback.
        """
        conversation_type = _normalize_conversation_type(msg.metadata.get("conversation_type"))
        sender_staff_id = msg.metadata.get("sender_staff_id", "")
        conversation_id = msg.metadata.get("conversation_id", "")
        if conversation_type == _CONVERSATION_TYPE_GROUP:
            conversation_id = msg.chat_id or conversation_id
        else:
            sender_staff_id = msg.chat_id or sender_staff_id
        return conversation_type, sender_staff_id, conversation_id

    async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None:
        conversation_type, sender_staff_id, conversation_id = self._resolve_routing(msg)
        robot_code = self._client_id

        # Card mode: stream update to existing AI card
        source_key = self._make_card_source_key_from_outbound(msg)
        out_track_id = self._card_track_ids.get(source_key)

        # ``card_template_id`` enables ``runs.stream`` (non-final + final outbounds).
        # If card creation failed, skip non-final chunks to avoid duplicate messages.
        if self._card_template_id and not out_track_id and not msg.is_final:
            return

        if out_track_id:
            try:
                await self._stream_update_card(
                    out_track_id,
                    msg.text,
                    is_finalize=msg.is_final,
                )
            except Exception:
                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:
            if conversation_type == _CONVERSATION_TYPE_GROUP:
                await self._send_group_message(robot_code, conversation_id, msg.text, at_user_ids=[sender_staff_id] if sender_staff_id else None)
            else:
                await self._send_p2p_message(robot_code, sender_staff_id, msg.text)

        # Non-card mode: send sampleMarkdown with retry
        await self._send_with_retry(
            send_markdown,
            max_retries=_max_retries,
            log_prefix="[DingTalk]",
        )
        return

    async def _send_markdown_fallback(
        self,
        robot_code: str,
        conversation_type: str,
        sender_staff_id: str,
        conversation_id: str,
        text: str,
    ) -> None:
        try:
            if conversation_type == _CONVERSATION_TYPE_GROUP:
                await self._send_group_message(robot_code, conversation_id, text)
            else:
                await self._send_p2p_message(robot_code, sender_staff_id, text)
        except Exception:
            logger.exception("[DingTalk] markdown fallback also failed")
            raise

    async def send_file(self, msg: OutboundMessage, attachment: ResolvedAttachment) -> bool:
        if attachment.size > _MAX_UPLOAD_SIZE_BYTES:
            logger.warning("[DingTalk] file too large (%d bytes), skipping: %s", attachment.size, attachment.filename)
            return False

        conversation_type, sender_staff_id, conversation_id = self._resolve_routing(msg)
        robot_code = self._client_id

        try:
            media_id = await self._upload_media(attachment.actual_path, "image" if attachment.is_image else "file")
            if not media_id:
                return False

            if attachment.is_image:
                msg_key = "sampleImageMsg"
                msg_param = json.dumps({"photoURL": media_id})
            else:
                msg_key = "sampleFile"
                msg_param = json.dumps(
                    {
                        "fileUrl": media_id,
                        "fileName": attachment.filename,
                        "fileSize": str(attachment.size),
                    }
                )

            token = await self._get_access_token()
            async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
                if conversation_type == _CONVERSATION_TYPE_GROUP:
                    response = await client.post(
                        f"{DINGTALK_API_BASE}/v1.0/robot/groupMessages/send",
                        headers=self._api_headers(token),
                        json={
                            "msgKey": msg_key,
                            "msgParam": msg_param,
                            "robotCode": robot_code,
                            "openConversationId": conversation_id,
                        },
                    )
                else:
                    response = await client.post(
                        f"{DINGTALK_API_BASE}/v1.0/robot/oToMessages/batchSend",
                        headers=self._api_headers(token),
                        json={
                            "msgKey": msg_key,
                            "msgParam": msg_param,
                            "robotCode": robot_code,
                            "userIds": [sender_staff_id],
                        },
                    )
                response.raise_for_status()

            logger.info("[DingTalk] file sent: %s", attachment.filename)
            return True
        except (httpx.HTTPError, OSError, ValueError, TypeError, AttributeError):
            logger.exception("[DingTalk] failed to send file: %s", attachment.filename)
            return False

    # -- stream client (runs in dedicated thread) --------------------------

    def _run_stream(self, client_id: str, client_secret: str) -> None:
        try:
            import dingtalk_stream

            credential = dingtalk_stream.Credential(client_id, client_secret)
            client = dingtalk_stream.DingTalkStreamClient(credential)
            self._stream_client = client
            client.register_callback_handler(
                dingtalk_stream.chatbot.ChatbotMessage.TOPIC,
                _DingTalkMessageHandler(self),
            )
            client.start_forever()
        except Exception:
            if self._running:
                logger.exception("DingTalk Stream Push error")
        finally:
            self._stream_client = None

    def _on_chatbot_message(self, message: Any) -> None:
        if not self._running:
            return
        try:
            sender_staff_id = message.sender_staff_id or ""
            conversation_type = _normalize_conversation_type(message.conversation_type)
            conversation_id = message.conversation_id or ""
            msg_id = message.message_id or ""
            sender_nick = message.sender_nick or ""

            text = self._extract_text(message)
            files = self._extract_files(message)
            if not text and not files:
                logger.info("[DingTalk] empty message with no files, ignoring")
                return

            connect_code = self._pending_connect_code(text)
            if connect_code:
                if self._main_loop and self._main_loop.is_running():
                    fut = asyncio.run_coroutine_threadsafe(
                        self._bind_connection_from_connect_code(
                            conversation_type=conversation_type,
                            sender_staff_id=sender_staff_id,
                            sender_nick=sender_nick,
                            conversation_id=conversation_id,
                            code=connect_code,
                        ),
                        self._main_loop,
                    )
                    fut.add_done_callback(lambda f, mid=msg_id: self._log_future_error(f, "bind_connection", mid))
                else:
                    logger.warning("[DingTalk] main loop not running, cannot bind channel connection")
                return

            if self._allowed_users and sender_staff_id not in self._allowed_users:
                logger.debug("[DingTalk] ignoring message from non-allowed user: %s", sender_staff_id)
                return

            # Log only metadata (length, not content) so message text never reaches
            # INFO logs, and only after the allowed_users gate so blocked senders are
            # not logged at all.
            logger.info(
                "[DingTalk] parsed message: conv_type=%s, msg_id=%s, sender=%s(%s), text_len=%d, files=%d",
                conversation_type,
                msg_id,
                sender_staff_id,
                sender_nick,
                len(text or ""),
                len(files),
            )

            if _is_dingtalk_command(text):
                msg_type = InboundMessageType.COMMAND
            else:
                msg_type = InboundMessageType.CHAT

            # P2P: topic_id=None (single thread per user, like Telegram private chat)
            # Group: topic_id=msg_id (each new message starts a new topic, like Feishu)
            topic_id: str | None = msg_id if conversation_type == _CONVERSATION_TYPE_GROUP else None

            # chat_id uses conversation_id for groups, sender_staff_id for P2P
            chat_id = conversation_id if conversation_type == _CONVERSATION_TYPE_GROUP else sender_staff_id

            # An empty chat_id does not identify a conversation, and ChannelStore keys on it:
            # a P2P message with no sender keys to the literal "dingtalk:", so every such
            # message from every user shares one thread and one history. Mirrors the guard the
            # WeChat channel already applies (wechat.py, `if not chat_id: return`) and the one
            # this file's own bind path applies before persisting a connection.
            if not chat_id:
                logger.warning(
                    "[DingTalk] ignoring message with no conversation identity: conv_type=%s, msg_id=%s",
                    conversation_type,
                    msg_id,
                )
                return

            inbound = self._make_inbound(
                chat_id=chat_id,
                user_id=sender_staff_id,
                text=text,
                msg_type=msg_type,
                thread_ts=msg_id,
                files=files,
                metadata={
                    "conversation_type": conversation_type,
                    "conversation_id": conversation_id,
                    "sender_staff_id": sender_staff_id,
                    "sender_nick": sender_nick,
                    "message_id": msg_id,
                },
            )
            inbound.topic_id = topic_id

            if self._card_template_id:
                source_key = self._make_card_source_key(inbound)
                with self._incoming_messages_lock:
                    self._incoming_messages[source_key] = message

            if self._main_loop and self._main_loop.is_running():
                logger.info("[DingTalk] publishing inbound message to bus (type=%s, msg_id=%s)", msg_type.value, msg_id)
                fut = asyncio.run_coroutine_threadsafe(
                    self._prepare_inbound(chat_id, inbound),
                    self._main_loop,
                )
                fut.add_done_callback(lambda f, mid=msg_id: self._log_future_error(f, "prepare_inbound", mid))
            else:
                logger.warning("[DingTalk] main loop not running, cannot publish inbound message")
        except Exception:
            logger.exception("[DingTalk] error processing chatbot message")

    @staticmethod
    def _extract_text(message: Any) -> str:
        msg_type = message.message_type
        if msg_type == "text" and message.text:
            return message.text.content.strip()
        if msg_type == "richText" and message.rich_text_content:
            return _extract_text_from_rich_text(message.rich_text_content.rich_text_list).strip()
        return ""

    # -- inbound: file attachments -----------------------------------------

    @staticmethod
    def _extract_files(message: Any) -> list[dict[str, Any]]:
        """Extract inbound file/image descriptors from a DingTalk message.

        Returns a list of dicts shaped for ``InboundMessage.files``; each carries
        ``type`` (``"image"`` or ``"file"``), ``download_code``, and ``filename``.

        Images arrive as ``picture`` messages (a single ``downloadCode``) or as
        inline images inside a ``richText`` message. Documents arrive as ``file``
        messages, which ``dingtalk_stream.ChatbotMessage.from_dict`` does not
        parse — the descriptor (``downloadCode`` / ``fileName``) is read from the
        raw callback payload stashed on the message as ``_df_raw_data`` by
        ``_DingTalkMessageHandler.process``.
        """
        msg_type = getattr(message, "message_type", None)
        files: list[dict[str, Any]] = []

        if msg_type == "picture":
            image_content = getattr(message, "image_content", None)
            code = getattr(image_content, "download_code", None)
            if code:
                files.append({"type": "image", "download_code": code, "filename": "image.png"})
        elif msg_type == "richText":
            try:
                codes = message.get_image_list() or []
            except Exception:
                # Log rather than swallow: without this, a richText message whose
                # SDK image parse fails looks identical to one that simply has no
                # inline images, which is hard to diagnose if the SDK shape changes.
                logger.warning("[DingTalk] failed to read inline images from richText message", exc_info=True)
                codes = []
            for idx, code in enumerate(codes):
                if code:
                    files.append({"type": "image", "download_code": code, "filename": f"image_{idx}.png"})
        elif msg_type == "file":
            raw = getattr(message, "_df_raw_data", None)
            content = raw.get("content") if isinstance(raw, dict) else None
            if isinstance(content, dict):
                code = content.get("downloadCode")
                if code:
                    filename = content.get("fileName") or "file.bin"
                    files.append({"type": "file", "download_code": code, "filename": filename})

        return files

    async def receive_file(self, msg: InboundMessage, thread_id: str, *, user_id: str | None = None) -> InboundMessage:
        """Download inbound DingTalk files into the thread uploads directory.

        Mirrors :meth:`FeishuChannel.receive_file`: each descriptor in
        ``msg.files`` is downloaded by its ``downloadCode``, persisted under the
        thread's uploads bucket, synced into a non-local sandbox, and its sandbox
        virtual path is prepended to ``msg.text`` so downstream models can read
        the file by path. Descriptors are then cleared so the generic
        URL-based ``_ingest_inbound_files`` pass does not try to re-fetch them
        (DingTalk files are downloaded by code, not by URL).

        An attachment that fails to download contributes a short
        ``[failed to load ...]`` marker instead of a path, so the agent can tell
        the user something was sent but could not be read rather than silently
        ignoring it.
        """
        if not msg.files:
            return msg

        virtual_paths: list[str] = []
        failures: list[str] = []
        for f in msg.files:
            if not isinstance(f, dict):
                continue
            download_code = f.get("download_code")
            if not download_code:
                continue
            file_type = "image" if f.get("type") == "image" else "file"
            filename = f.get("filename") if isinstance(f.get("filename"), str) else ""
            # Per-attachment isolation: the manager awaits receive_file without a
            # try, so anything escaping here would kill the chat turn with no
            # reply — the silent-drop failure mode this feature exists to fix.
            try:
                virtual_path = await self._receive_single_file(download_code, file_type, filename, thread_id, user_id=user_id)
            except Exception:
                logger.exception("[DingTalk] unexpected error receiving inbound %s", file_type)
                virtual_path = ""
            if virtual_path:
                virtual_paths.append(virtual_path)
            else:
                display = _display_filename(filename)
                failures.append(f"[failed to load {file_type}: {display}]" if display else f"[failed to load {file_type}]")

        prefix_lines = virtual_paths + failures
        if prefix_lines:
            block = "\n".join(prefix_lines)
            msg.text = f"{block}\n\n{msg.text}".strip() if msg.text else block

        msg.files = []
        return msg

    async def _receive_single_file(
        self,
        download_code: str,
        file_type: str,
        filename: str,
        thread_id: str,
        *,
        user_id: str | None = None,
    ) -> str:
        """Download one file by ``downloadCode`` and persist it into uploads.

        Returns the sandbox virtual path on success, or ``""`` on failure.
        """
        content = await self._download_by_code(download_code)
        if not content:
            return ""

        paths = get_paths()
        effective_user_id = user_id or get_effective_user_id()

        default_ext = "png" if file_type == "image" else "bin"
        # ``download_code`` is attacker-controllable webhook data, so restrict it to
        # a safe character set before embedding it in the fallback name. This keeps
        # the fallback safe by construction instead of relying on a later write
        # failure to reject a name that escaped the uploads directory.
        code_token = re.sub(r"[^A-Za-z0-9_-]", "", download_code)[-12:] or "attachment"
        fallback_name = f"dingtalk_{code_token}.{default_ext}"
        # normalize_filename strips directory components and rejects traversal
        # patterns ("..", backslash paths, over-long names); fall back to the
        # sanitized generated name if the platform-supplied filename is rejected.
        try:
            safe_filename = normalize_filename(filename or fallback_name)
        except ValueError:
            safe_filename = fallback_name

        def _persist() -> Path:
            # Directory prep, the uniqueness claim, and the write are blocking
            # filesystem IO — the whole sequence stays off the event loop. The
            # claim and the write share one lock because generated names repeat
            # across messages ("image.png" for every picture message): without a
            # claim a later attachment silently overwrites an earlier one whose
            # path was already handed to the agent, and letting the claim and
            # write interleave would resolve two attachments to the same free name.
            paths.ensure_thread_dirs(thread_id, user_id=effective_user_id)
            uploads_dir = paths.sandbox_uploads_dir(thread_id, user_id=effective_user_id).resolve()
            with self._file_write_lock:
                seen = {entry.name for entry in uploads_dir.iterdir() if entry.is_file()}
                unique_name = claim_unique_filename(safe_filename, seen)
                # write_upload_file_no_symlink refuses a symlinked destination:
                # uploads dirs can be mounted into local sandboxes, so a sandbox
                # process could otherwise redirect this privileged write outside
                # the bucket.
                return write_upload_file_no_symlink(uploads_dir, unique_name, content)

        try:
            resolved_target = await asyncio.to_thread(_persist)
        except (OSError, UnsafeUploadPathError):
            logger.exception("[DingTalk] failed to persist downloaded file: %s", safe_filename)
            return ""

        virtual_path = f"{VIRTUAL_PATH_PREFIX}/uploads/{resolved_target.name}"

        try:
            sandbox_provider = get_sandbox_provider()
            # acquire_async keeps provider lifecycle work (Docker discovery,
            # readiness polls) off the event loop; update_file is blocking
            # transport IO on remote sandboxes, so it is offloaded too.
            sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id)
            if sandbox_id != "local":
                sandbox = sandbox_provider.get(sandbox_id)
                if sandbox is None:
                    # Mirror Feishu: the agent's non-local sandbox cannot see this
                    # file, so returning the virtual path would hand the model a
                    # path that reads as nothing — surface a failed-load marker.
                    logger.warning("[DingTalk] sandbox %s not found after acquire, dropping attachment: %s", sandbox_id, virtual_path)
                    return ""
                await asyncio.to_thread(sandbox.update_file, virtual_path, content)
        except Exception:
            # Same failure mode as the sandbox-is-None branch: the bytes never
            # reached the agent's sandbox, so the virtual path would read as
            # nothing. Mirror Feishu and surface a failed-load marker.
            logger.exception("[DingTalk] failed to sync downloaded file into non-local sandbox: %s", virtual_path)
            return ""

        return virtual_path

    async def _download_by_code(self, download_code: str) -> bytes | None:
        """Exchange a DingTalk ``downloadCode`` for the raw file bytes.

        Two steps per the DingTalk robot OpenAPI:
        1. ``POST /v1.0/robot/messageFiles/download`` -> ``{downloadUrl}``
        2. ``GET downloadUrl`` -> binary content
        """
        try:
            # Token acquisition stays inside the try: the manager awaits
            # receive_file without a try, so a token failure escaping here would
            # abort the whole chat turn instead of degrading to a failed-load marker.
            token = await self._get_access_token()
            async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
                response = await client.post(
                    f"{DINGTALK_API_BASE}/v1.0/robot/messageFiles/download",
                    headers={
                        "x-acs-dingtalk-access-token": token,
                        "Content-Type": "application/json",
                    },
                    json={"downloadCode": download_code, "robotCode": self._client_id},
                )
                if response.status_code != 200:
                    logger.warning("[DingTalk] messageFiles/download failed: status=%d, body=%s", response.status_code, response.text[:300])
                    return None
                download_url = response.json().get("downloadUrl")
                if not download_url:
                    logger.warning("[DingTalk] messageFiles/download returned no downloadUrl")
                    return None

                # Stream with a size cap: the bytes are buffered in memory before
                # being persisted, so an oversized attachment must be refused
                # before it is fully read, not after.
                chunks: list[bytes] = []
                total = 0
                async with client.stream("GET", download_url, follow_redirects=True) as file_response:
                    file_response.raise_for_status()
                    async for chunk in file_response.aiter_bytes():
                        total += len(chunk)
                        if total > _MAX_INBOUND_FILE_SIZE_BYTES:
                            logger.warning("[DingTalk] inbound file exceeds %d bytes, dropping", _MAX_INBOUND_FILE_SIZE_BYTES)
                            return None
                        chunks.append(chunk)
                return b"".join(chunks)
        except (httpx.HTTPError, ValueError):
            logger.exception("[DingTalk] failed to download file by code")
            return None

    async def _prepare_inbound(self, chat_id: str, inbound: InboundMessage) -> None:
        inbound = await self._attach_connection_identity(inbound)
        # Running reply must finish before publish_inbound so AI card tracks are
        # registered before the manager emits streaming outbounds.
        await self._send_running_reply(chat_id, inbound)
        await self.bus.publish_inbound(inbound)

    @staticmethod
    def _connection_workspace_id(conversation_type: str, conversation_id: str) -> str | None:
        if conversation_type == _CONVERSATION_TYPE_GROUP and conversation_id:
            return conversation_id
        return None

    async def _attach_connection_identity(self, inbound: InboundMessage) -> InboundMessage:
        conversation_type = str(inbound.metadata.get("conversation_type") or _CONVERSATION_TYPE_P2P)
        conversation_id = str(inbound.metadata.get("conversation_id") or "")
        return await attach_connection_identity(
            inbound,
            repo=self._connection_repo,
            provider="dingtalk",
            workspace_id=self._connection_workspace_id(conversation_type, conversation_id),
            fallback_without_workspace=True,
        )

    async def _bind_connection_from_connect_code(
        self,
        *,
        conversation_type: str,
        sender_staff_id: str,
        sender_nick: str,
        conversation_id: str,
        code: str,
    ) -> bool:
        if self._connection_repo is None or not code:
            return False

        state = await self._connection_repo.consume_oauth_state(provider="dingtalk", state=code)
        if state is None:
            await self._send_connection_reply(
                conversation_type,
                sender_staff_id,
                conversation_id,
                "DingTalk connection code is invalid or expired.",
            )
            return True

        if not sender_staff_id:
            await self._send_connection_reply(
                conversation_type,
                sender_staff_id,
                conversation_id,
                "DingTalk connection could not be completed from this message.",
            )
            return True

        await self._connection_repo.upsert_connection(
            owner_user_id=state["owner_user_id"],
            provider="dingtalk",
            external_account_id=sender_staff_id,
            external_account_name=sender_nick or None,
            workspace_id=self._connection_workspace_id(conversation_type, conversation_id),
            metadata={
                "conversation_type": conversation_type,
                "conversation_id": conversation_id,
            },
            status="connected",
        )
        await self._send_connection_reply(
            conversation_type,
            sender_staff_id,
            conversation_id,
            "DingTalk connected to DeerFlow.",
        )
        return True

    async def _send_connection_reply(
        self,
        conversation_type: str,
        sender_staff_id: str,
        conversation_id: str,
        text: str,
    ) -> None:
        robot_code = self._client_id
        if conversation_type == _CONVERSATION_TYPE_GROUP:
            if conversation_id:
                await self._send_text_message_to_group(robot_code, conversation_id, text)
            return
        if sender_staff_id:
            await self._send_text_message_to_user(robot_code, sender_staff_id, text)

    async def _send_running_reply(self, chat_id: str, inbound: InboundMessage) -> None:
        conversation_type = inbound.metadata.get("conversation_type", _CONVERSATION_TYPE_P2P)
        sender_staff_id = inbound.metadata.get("sender_staff_id", "")
        conversation_id = inbound.metadata.get("conversation_id", "")
        text = "\u23f3 Working on it..."

        try:
            if self._card_template_id:
                source_key = self._make_card_source_key(inbound)
                with self._incoming_messages_lock:
                    chatbot_message = self._incoming_messages.pop(source_key, None)
                out_track_id = await self._create_and_deliver_card(
                    text,
                    chatbot_message=chatbot_message,
                )
                if out_track_id:
                    self._card_track_ids[source_key] = out_track_id
                    logger.info("[DingTalk] AI card running reply sent for chat=%s", chat_id)
                    return

            robot_code = self._client_id
            if conversation_type == _CONVERSATION_TYPE_GROUP:
                await self._send_text_message_to_group(robot_code, conversation_id, text)
            else:
                await self._send_text_message_to_user(robot_code, sender_staff_id, text)
            logger.info("[DingTalk] 'Working on it...' reply sent for chat=%s", chat_id)
        except Exception:
            logger.exception("[DingTalk] failed to send running reply for chat=%s", chat_id)

    # -- DingTalk API helpers ----------------------------------------------

    async def _get_access_token(self) -> str:
        if self._cached_token and time.monotonic() < self._token_expires_at:
            return self._cached_token
        async with self._token_lock:
            if self._cached_token and time.monotonic() < self._token_expires_at:
                return self._cached_token
            async with httpx.AsyncClient(timeout=httpx.Timeout(10.0)) as client:
                response = await client.post(
                    f"{DINGTALK_API_BASE}/v1.0/oauth2/accessToken",
                    json={"appKey": self._client_id, "appSecret": self._client_secret},  # DingTalk API field names
                )
                response.raise_for_status()
                data = response.json()

                if not isinstance(data, dict):
                    raise ValueError(f"DingTalk access token response must be a JSON object, got {type(data).__name__}")

                access_token = data.get("accessToken")
                if not isinstance(access_token, str) or not access_token.strip():
                    raise ValueError("DingTalk access token response did not contain a usable accessToken")

                raw_expires_in = data.get("expireIn", 7200)
                try:
                    expires_in = int(raw_expires_in)
                except (TypeError, ValueError):
                    logger.warning("[DingTalk] invalid expireIn value %r, using default 7200s", raw_expires_in)
                    expires_in = 7200

                self._cached_token = access_token.strip()
                self._token_expires_at = time.monotonic() + expires_in - _TOKEN_REFRESH_MARGIN_SECONDS
                return self._cached_token

    @staticmethod
    def _api_headers(token: str) -> dict[str, str]:
        return {
            "x-acs-dingtalk-access-token": token,
            "Content-Type": "application/json",
        }

    async def _send_text_message_to_user(self, robot_code: str, user_id: str, text: str) -> None:
        token = await self._get_access_token()
        async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
            response = await client.post(
                f"{DINGTALK_API_BASE}/v1.0/robot/oToMessages/batchSend",
                headers=self._api_headers(token),
                json={
                    "msgKey": "sampleText",
                    "msgParam": json.dumps({"content": text}),
                    "robotCode": robot_code,
                    "userIds": [user_id],
                },
            )
            response.raise_for_status()

    async def _send_text_message_to_group(self, robot_code: str, conversation_id: str, text: str) -> None:
        token = await self._get_access_token()
        async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
            response = await client.post(
                f"{DINGTALK_API_BASE}/v1.0/robot/groupMessages/send",
                headers=self._api_headers(token),
                json={
                    "msgKey": "sampleText",
                    "msgParam": json.dumps({"content": text}),
                    "robotCode": robot_code,
                    "openConversationId": conversation_id,
                },
            )
            response.raise_for_status()

    async def _send_p2p_message(self, robot_code: str, user_id: str, text: str) -> None:
        text = _adapt_markdown_for_dingtalk(text)
        token = await self._get_access_token()
        async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
            response = await client.post(
                f"{DINGTALK_API_BASE}/v1.0/robot/oToMessages/batchSend",
                headers=self._api_headers(token),
                json={
                    "msgKey": "sampleMarkdown",
                    "msgParam": json.dumps({"title": "DeerFlow", "text": text}),
                    "robotCode": robot_code,
                    "userIds": [user_id],
                },
            )
            response.raise_for_status()
            data = response.json()
            if data.get("processQueryKey"):
                logger.info("[DingTalk] P2P message sent to user=%s", user_id)
            else:
                logger.warning("[DingTalk] P2P send response: %s", data)

    async def _send_group_message(
        self,
        robot_code: str,
        conversation_id: str,
        text: str,
        *,
        at_user_ids: list[str] | None = None,  # noqa: ARG002
    ) -> None:
        # at_user_ids accepted for call-site compatibility but not passed to the API
        # (sampleMarkdown does not support @mentions).
        text = _adapt_markdown_for_dingtalk(text)
        token = await self._get_access_token()

        async with httpx.AsyncClient(timeout=httpx.Timeout(30.0)) as client:
            response = await client.post(
                f"{DINGTALK_API_BASE}/v1.0/robot/groupMessages/send",
                headers=self._api_headers(token),
                json={
                    "msgKey": "sampleMarkdown",
                    "msgParam": json.dumps({"title": "DeerFlow", "text": text}),
                    "robotCode": robot_code,
                    "openConversationId": conversation_id,
                },
            )
            response.raise_for_status()
            data = response.json()
            if data.get("processQueryKey"):
                logger.info("[DingTalk] group message sent to conversation=%s", conversation_id)
            else:
                logger.warning("[DingTalk] group send response: %s", data)

    # -- AI Card streaming helpers -------------------------------------------

    def _make_card_source_key(self, inbound: InboundMessage) -> str:
        m = inbound.metadata
        return f"{m.get('conversation_type', '')}:{m.get('sender_staff_id', '')}:{m.get('conversation_id', '')}:{m.get('message_id', '')}"

    def _make_card_source_key_from_outbound(self, msg: OutboundMessage) -> str:
        m = msg.metadata
        correlation_id = m.get("message_id") or msg.thread_ts or ""
        return f"{m.get('conversation_type', '')}:{m.get('sender_staff_id', '')}:{m.get('conversation_id', '')}:{correlation_id}"

    async def _create_and_deliver_card(
        self,
        initial_text: str,
        *,
        chatbot_message: Any = None,
    ) -> str | None:
        if self._dingtalk_client is None or chatbot_message is None:
            logger.warning("[DingTalk] SDK client or chatbot_message unavailable, skipping AI card")
            return None

        try:
            from dingtalk_stream.card_replier import AICardReplier
        except ImportError:
            logger.warning("[DingTalk] dingtalk-stream card_replier not available")
            return None

        try:
            replier = AICardReplier(self._dingtalk_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
        except Exception:
            logger.exception("[DingTalk] failed to create AI card")
            return None

    async def _stream_update_card(
        self,
        out_track_id: str,
        content: str,
        *,
        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}")

        await replier.async_streaming(
            card_instance_id=out_track_id,
            content_key="content",
            content_value=content,
            append=False,
            finished=is_finalize,
            failed=is_error,
        )

    # -- media upload --------------------------------------------------------

    async def _upload_media(self, file_path: str | Path, media_type: str) -> str | None:
        try:
            file_bytes = await asyncio.to_thread(Path(file_path).read_bytes)
            token = await self._get_access_token()
            async with httpx.AsyncClient(timeout=httpx.Timeout(60.0)) as client:
                response = await client.post(
                    f"{DINGTALK_API_BASE}/v1.0/files/upload",
                    headers={"x-acs-dingtalk-access-token": token},
                    files={"file": ("upload", file_bytes)},
                    data={"type": media_type},
                )
                response.raise_for_status()
                try:
                    payload = response.json()
                except json.JSONDecodeError:
                    logger.exception("[DingTalk] failed to decode upload response JSON: %s", file_path)
                    return None
                if not isinstance(payload, dict):
                    logger.warning("[DingTalk] unexpected upload response type %s for %s", type(payload).__name__, file_path)
                    return None
                return payload.get("mediaId")
        except (httpx.HTTPError, OSError):
            logger.exception("[DingTalk] failed to upload media: %s", file_path)
            return None


class _DingTalkMessageHandler:
    """Callback handler registered with dingtalk-stream."""

    def __init__(self, channel: DingTalkChannel) -> None:
        self._channel = channel

    def pre_start(self) -> None:
        if hasattr(self, "dingtalk_client") and self.dingtalk_client is not None:
            self._channel._dingtalk_client = self.dingtalk_client

    async def raw_process(self, callback_message: Any) -> Any:
        import dingtalk_stream
        from dingtalk_stream.frames import Headers

        code, message = await self.process(callback_message)
        ack_message = dingtalk_stream.AckMessage()
        ack_message.code = code
        ack_message.headers.message_id = callback_message.headers.message_id
        ack_message.headers.content_type = Headers.CONTENT_TYPE_APPLICATION_JSON
        ack_message.data = {"response": message}
        return ack_message

    async def process(self, callback: Any) -> tuple[int, str]:
        import dingtalk_stream

        incoming_message = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
        # Stash the raw callback payload: dingtalk_stream does not parse ``file``
        # (document) messages, so DingTalkChannel._extract_files reads the file
        # descriptor (downloadCode / fileName) from it.
        incoming_message._df_raw_data = callback.data
        self._channel._on_chatbot_message(incoming_message)
        return dingtalk_stream.AckMessage.STATUS_OK, "OK"