# -*- coding: utf-8 -*-
# pylint: disable=too-many-statements,too-many-branches
# pylint: disable=too-many-return-statements,too-many-instance-attributes
"""WeChat (iLink Bot) Channel.

Uses the official WeChat iLink Bot HTTP API to receive and send messages.
Incoming messages are fetched via long-polling (getupdates); replies are sent
via sendmessage. Supports text, image, voice (ASR text), and file messages.

Authentication:
  - If bot_token is configured, it is used directly.
  - If bot_token is absent, a QR code login is triggered on start(); the
    resulting token is persisted to bot_token_file for future runs.
"""

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import os
import sys
import threading
import time
from collections import OrderedDict
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple

import base64 as _b64

from qwenpaw.schemas import (
    AgentRequest,
    FileContent,
    ImageContent,
    TextContent,
    VideoContent,
)

from ....exceptions import ChannelError
from ....constant import DEFAULT_MEDIA_DIR, WORKING_DIR
from ..renderer import ChannelDisplayConfig
from ..base import (
    BaseChannel,
    ContentType,
    OnReplySent,
    OutgoingContentPart,
    ProcessHandler,
)
from ..utils import file_url_to_local_path, split_text
from .client import ILinkClient, _DEFAULT_BASE_URL

logger = logging.getLogger(__name__)

# Max dedup set size
_WECHAT_PROCESSED_IDS_MAX = 2000

# Time window (seconds) for content-based dedup (same user + same text)
_TEXT_DEDUP_TTL = 30.0

# Default token file path
_DEFAULT_TOKEN_FILE = WORKING_DIR / "wechat_bot_token"


class WeChatChannel(BaseChannel):
    """WeChat iLink Bot channel: long-poll receive, HTTP send.

    Session IDs:
        - Private chat:  wechat:<from_user_id>
        - Group chat:    wechat:group:<group_id>
    """

    channel = "wechat"

    def __init__(
        self,
        process: ProcessHandler,
        enabled: bool,
        bot_token: str = "",
        bot_token_file: str = "",
        base_url: str = "",
        bot_prefix: str = "",
        media_dir: str = "",
        workspace_dir: Path | None = None,
        on_reply_sent: OnReplySent = None,
        display_config: ChannelDisplayConfig | None = None,
        no_text_debounce: bool = True,
        dm_policy: str = "open",
        group_policy: str = "open",
        allow_from: Optional[List[str]] = None,
        deny_message: str = "",
        message_merge_enabled: bool = False,
        message_merge_delay_ms: int = 0,
        access_control_dm: bool = False,
        access_control_group: bool = False,
    ):
        super().__init__(
            process,
            on_reply_sent=on_reply_sent,
            display_config=display_config,
            no_text_debounce=no_text_debounce,
            dm_policy=dm_policy,
            group_policy=group_policy,
            allow_from=allow_from,
            deny_message=deny_message,
            access_control_dm=access_control_dm,
            access_control_group=access_control_group,
        )
        self.enabled = enabled
        self.bot_token = bot_token
        self.bot_prefix = bot_prefix
        self._base_url = base_url or _DEFAULT_BASE_URL
        self._bot_token_file = (
            Path(bot_token_file).expanduser()
            if bot_token_file
            else _DEFAULT_TOKEN_FILE
        )
        self._context_tokens_file = (
            self._bot_token_file.parent / "wechat_context_tokens.json"
        )
        self._workspace_dir = (
            Path(workspace_dir).expanduser() if workspace_dir else None
        )
        # Use workspace-specific media dir if workspace_dir is provided
        if not media_dir and self._workspace_dir:
            self._media_dir = self._workspace_dir / "media"
        elif media_dir:
            self._media_dir = Path(media_dir).expanduser()
        else:
            self._media_dir = DEFAULT_MEDIA_DIR

        # Message merge settings (mitigates 10-msg context_token limit)
        self._message_merge_enabled = message_merge_enabled
        self._message_merge_delay_ms = max(message_merge_delay_ms, 0)
        # Merge buffer (WeChat iLink is single-chat only, one buffer suffices)
        self._merge_buffer: List[OutgoingContentPart] = []
        self._merge_meta: Optional[Dict[str, Any]] = None
        self._merge_timer: Optional[asyncio.TimerHandle] = None
        self._merge_to_handle: str = ""

        self._client: Optional[ILinkClient] = None
        self._loop: Optional[asyncio.AbstractEventLoop] = None
        self._poll_loop: Optional[asyncio.AbstractEventLoop] = None
        self._poll_task: Optional[asyncio.Task] = None
        self._poll_thread: Optional[threading.Thread] = None
        self._stop_event = threading.Event()
        self._loop_accepting = threading.Event()  # cleared on stop

        # Cursor for long-polling (get_updates_buf)
        self._cursor: str = ""

        # Message dedup (context_token or derived id)
        self._processed_ids: OrderedDict[str, None] = OrderedDict()
        self._processed_ids_lock = threading.Lock()

        # Content-based dedup: {user_id:content_hash -> timestamp}
        self._text_dedup: OrderedDict[str, float] = OrderedDict()

        # Cache last context_token per user for proactive sends
        self._user_context_tokens: Dict[str, str] = {}

        # Cache typing tickets per user (24h TTL)
        self._typing_tickets: Dict[
            str,
            Tuple[str, float],
        ] = {}  # user_id -> (ticket, expiry_time)
        self._typing_lock = threading.Lock()
        # Store stop functions for active typing indicators
        self._typing_stop_funcs: Dict[
            str,
            Callable[[], None],
        ] = {}  # user_id -> stop function
        self._typing_stop_lock = threading.Lock()

    # ------------------------------------------------------------------
    # Thread-safe helpers
    # ------------------------------------------------------------------

    def _dispatch_to_main_loop(
        self,
        coro: Any,
        *,
        description: str = "",
    ) -> bool:
        """Safely dispatch a coroutine to the main event loop from poll thread.

        Returns True if successfully dispatched, False if loop is unavailable.
        Prevents RuntimeError when the main loop has already stopped.
        If dispatch fails, the coroutine is closed to suppress
        'coroutine was never awaited' warnings.
        """
        if not self._loop_accepting.is_set():
            logger.debug(
                "wechat: skipping dispatch (loop not accepting): %s",
                description,
            )
            coro.close()
            return False
        loop = self._loop
        if loop is None or loop.is_closed():
            coro.close()
            return False
        try:
            asyncio.run_coroutine_threadsafe(coro, loop)
            return True
        except RuntimeError:
            logger.debug(
                "wechat: dispatch failed (loop stopped): %s",
                description,
            )
            coro.close()
            return False

    # ------------------------------------------------------------------
    # Factory methods
    # ------------------------------------------------------------------

    @classmethod
    def from_env(
        cls,
        process: ProcessHandler,
        on_reply_sent: OnReplySent = None,
    ) -> "WeChatChannel":
        allow_from_env = os.getenv("WECHAT_ALLOW_FROM", "")
        allow_from = (
            [s.strip() for s in allow_from_env.split(",") if s.strip()]
            if allow_from_env
            else []
        )
        return cls(
            process=process,
            enabled=os.getenv("WECHAT_CHANNEL_ENABLED", "0") == "1",
            bot_token=os.getenv("WECHAT_BOT_TOKEN", ""),
            bot_token_file=os.getenv("WECHAT_BOT_TOKEN_FILE", ""),
            base_url=os.getenv("WECHAT_BASE_URL", ""),
            bot_prefix=os.getenv("WECHAT_BOT_PREFIX", ""),
            media_dir=os.getenv("WECHAT_MEDIA_DIR", ""),
            on_reply_sent=on_reply_sent,
            dm_policy=os.getenv("WECHAT_DM_POLICY", "open"),
            group_policy=os.getenv("WECHAT_GROUP_POLICY", "open"),
            allow_from=allow_from,
            deny_message=os.getenv("WECHAT_DENY_MESSAGE", ""),
        )

    @classmethod
    def from_config(
        cls,
        process: ProcessHandler,
        config: Any,
        on_reply_sent: OnReplySent = None,
        display_config: ChannelDisplayConfig | None = None,
        no_text_debounce: bool = True,
        workspace_dir: Path | None = None,
    ) -> "WeChatChannel":
        return cls(
            process=process,
            enabled=getattr(config, "enabled", False),
            bot_token=getattr(config, "bot_token", "") or "",
            bot_token_file=getattr(config, "bot_token_file", "") or "",
            base_url=getattr(config, "base_url", "") or "",
            bot_prefix=getattr(config, "bot_prefix", "") or "",
            media_dir=getattr(config, "media_dir", None) or "",
            workspace_dir=workspace_dir,
            on_reply_sent=on_reply_sent,
            display_config=display_config
            or ChannelDisplayConfig.from_config(config),
            no_text_debounce=no_text_debounce,
            dm_policy=getattr(config, "dm_policy", "open") or "open",
            group_policy=getattr(config, "group_policy", "open") or "open",
            allow_from=getattr(config, "allow_from", []) or [],
            deny_message=getattr(config, "deny_message", "") or "",
            message_merge_enabled=getattr(
                config,
                "message_merge_enabled",
                False,
            ),
            message_merge_delay_ms=getattr(
                config,
                "message_merge_delay_ms",
                0,
            )
            or 0,
            access_control_dm=bool(
                getattr(config, "access_control_dm", False),
            ),
            access_control_group=bool(
                getattr(config, "access_control_group", False),
            ),
        )

    # ------------------------------------------------------------------
    # Session helpers
    # ------------------------------------------------------------------

    def resolve_session_id(
        self,
        sender_id: str,
        channel_meta: Optional[Dict[str, Any]] = None,
    ) -> str:
        meta = channel_meta or {}
        group_id = (meta.get("wechat_group_id") or "").strip()
        if group_id:
            return f"wechat:group:{group_id}"
        return f"wechat:{sender_id}" if sender_id else "wechat:unknown"

    @staticmethod
    def _parse_user_id_from_handle(to_handle: str) -> str:
        h = (to_handle or "").strip()
        if h.startswith("wechat:group:"):
            return h[len("wechat:group:") :]
        if h.startswith("wechat:"):
            return h[len("wechat:") :]
        return h

    def to_handle_from_target(self, *, user_id: str, session_id: str) -> str:
        return session_id or f"wechat:{user_id}"

    def get_to_handle_from_request(self, request: Any) -> str:
        session_id = getattr(request, "session_id", "") or ""
        user_id = getattr(request, "user_id", "") or ""
        return session_id or f"wechat:{user_id}"

    def get_on_reply_sent_args(self, request: Any, to_handle: str) -> tuple:
        return (
            getattr(request, "user_id", "") or "",
            getattr(request, "session_id", "") or "",
        )

    def build_agent_request_from_native(
        self,
        native_payload: Any,
    ) -> "AgentRequest":
        payload = native_payload if isinstance(native_payload, dict) else {}
        channel_id = payload.get("channel_id") or self.channel
        sender_id = payload.get("sender_id") or ""
        content_parts = payload.get("content_parts") or []
        meta = payload.get("meta") or {}
        session_id = payload.get("session_id") or self.resolve_session_id(
            sender_id,
            meta,
        )
        user_id = payload.get("user_id", sender_id)
        request = self.build_agent_request_from_user_content(
            channel_id=channel_id,
            sender_id=user_id,
            session_id=session_id,
            content_parts=content_parts,
            channel_meta=meta,
        )
        setattr(request, "channel_meta", meta)
        return request

    def merge_native_items(self, items: List[Any]) -> Any:
        if not items:
            return None
        first = items[0] if isinstance(items[0], dict) else {}
        merged_parts: List[Any] = []
        for it in items:
            p = it if isinstance(it, dict) else {}
            merged_parts.extend(p.get("content_parts") or [])
        last = items[-1] if isinstance(items[-1], dict) else {}
        return {
            "channel_id": first.get("channel_id") or self.channel,
            "sender_id": last.get("sender_id", first.get("sender_id", "")),
            "acl_sender_id": first.get("acl_sender_id") or "",
            "user_id": last.get("user_id", first.get("user_id", "")),
            "session_id": last.get("session_id", first.get("session_id", "")),
            "content_parts": merged_parts,
            "meta": dict(last.get("meta") or {}),
        }

    # ------------------------------------------------------------------
    # Token persistence
    # ------------------------------------------------------------------

    def _load_token_from_file(self) -> str:
        """Try to load persisted bot_token from token file."""
        try:
            if self._bot_token_file.exists():
                token = self._bot_token_file.read_text(
                    encoding="utf-8",
                ).strip()
                if token:
                    logger.info(
                        "wechat: loaded bot_token from %s",
                        self._bot_token_file,
                    )
                    return token
        except Exception:
            logger.debug("wechat: failed to read token file", exc_info=True)
        return ""

    def _save_token_to_file(self, token: str) -> None:
        """Persist bot_token to token file."""
        try:
            self._bot_token_file.parent.mkdir(parents=True, exist_ok=True)
            self._bot_token_file.write_text(token, encoding="utf-8")
            logger.info("wechat: bot_token saved to %s", self._bot_token_file)
        except Exception:
            logger.warning("wechat: failed to save token file", exc_info=True)

    def _load_context_tokens(self) -> None:
        """Load persisted context_tokens from file into memory."""
        try:
            if self._context_tokens_file.exists():
                data = json.loads(
                    self._context_tokens_file.read_text(encoding="utf-8"),
                )
                if isinstance(data, dict):
                    self._user_context_tokens = {
                        k: v
                        for k, v in data.items()
                        if isinstance(k, str) and isinstance(v, str)
                    }
                    logger.info(
                        "wechat: loaded %d context_tokens from %s",
                        len(self._user_context_tokens),
                        self._context_tokens_file,
                    )
        except Exception:
            logger.debug(
                "wechat: failed to load context_tokens file",
                exc_info=True,
            )

    def _save_context_tokens(self) -> None:
        """Persist current context_tokens dict to file."""
        try:
            self._context_tokens_file.parent.mkdir(
                parents=True,
                exist_ok=True,
            )
            self._context_tokens_file.write_text(
                json.dumps(self._user_context_tokens, ensure_ascii=False),
                encoding="utf-8",
            )
        except Exception:
            logger.debug(
                "wechat: failed to save context_tokens file",
                exc_info=True,
            )

    # ------------------------------------------------------------------
    # Message dedup
    # ------------------------------------------------------------------

    def _is_duplicate(self, msg_id: str) -> bool:
        with self._processed_ids_lock:
            if msg_id in self._processed_ids:
                return True
            self._processed_ids[msg_id] = None
            while len(self._processed_ids) > _WECHAT_PROCESSED_IDS_MAX:
                self._processed_ids.popitem(last=False)
        return False

    def _is_text_duplicate(self, from_user_id: str, text: str) -> bool:
        """Content-based dedup within a short time window.

        Catches duplicates that slip past ``_is_duplicate`` when the iLink
        API delivers the same message across two polls with different
        ``context_token`` / ``msg_id`` values.
        """
        content_hash = hashlib.md5(text.encode()).hexdigest()[:16]
        key = f"{from_user_id}:{content_hash}"
        now = time.time()
        with self._processed_ids_lock:
            prev_time = self._text_dedup.get(key)
            if prev_time is not None and now - prev_time < _TEXT_DEDUP_TTL:
                return True
            self._text_dedup[key] = now
            # Evict old entries to bound memory
            while len(self._text_dedup) > _WECHAT_PROCESSED_IDS_MAX:
                self._text_dedup.popitem(last=False)
        return False

    # ------------------------------------------------------------------
    # QR code login
    # ------------------------------------------------------------------

    async def _do_qrcode_login(self) -> bool:
        """Perform QR code login and update self.bot_token.

        Prints QR code URL to logger (INFO) for the user to scan.
        Returns True if login succeeded.
        """
        if not self._client:
            return False
        try:
            qr_data = await self._client.get_bot_qrcode()
            qrcode = qr_data.get("qrcode", "")
            qrcode_url = qr_data.get("url") or qr_data.get(
                "qrcode_img_content",
                "",
            )
            logger.info(
                "wechat: Please scan the QR code to log in.\n  QR URL: %s",
                qrcode_url or "(see qrcode_img_content in debug log)",
            )
            if logger.isEnabledFor(logging.DEBUG):
                img_b64 = qr_data.get("qrcode_img_content", "")
                if img_b64:
                    logger.debug(
                        "wechat: QR code base64 PNG: %s",
                        img_b64[:80],
                    )

            logger.info("wechat: waiting for QR code scan (up to 300s)…")
            token, base_url = await self._client.wait_for_login(qrcode)
            self.bot_token = token
            self._client.bot_token = token
            if base_url and base_url != self._client.base_url:
                self._client.base_url = base_url.rstrip("/")
                self._base_url = base_url.rstrip("/")
            self._save_token_to_file(token)
            logger.info("wechat: QR code login succeeded")
            return True
        except Exception:
            logger.exception("wechat: QR code login failed")
            return False

    # ------------------------------------------------------------------
    # Long-poll loop (runs in background thread)
    # ------------------------------------------------------------------

    def _run_poll_forever(self) -> None:
        """Background thread: run long-poll loop in a dedicated event loop."""
        if sys.platform == "darwin":
            poll_loop = asyncio.SelectorEventLoop()
        else:
            poll_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(poll_loop)
        self._poll_loop = poll_loop
        try:
            # Wrap in a task so stop() can cancel it gracefully
            self._poll_task = poll_loop.create_task(self._poll_loop_async())
            poll_loop.run_until_complete(self._poll_task)
        except asyncio.CancelledError:
            logger.info("wechat: poll task cancelled (graceful stop)")
        except Exception:
            logger.exception("wechat: poll thread failed")
        finally:
            self._poll_task = None
            try:
                pending = asyncio.all_tasks(poll_loop)
                for task in pending:
                    task.cancel()
                if pending:
                    poll_loop.run_until_complete(
                        asyncio.gather(*pending, return_exceptions=True),
                    )
                poll_loop.run_until_complete(poll_loop.shutdown_asyncgens())
                poll_loop.close()
            except Exception:
                pass
            self._poll_loop = None

    async def _poll_loop_async(self) -> None:
        """Async long-poll loop: continuously call getupdates."""
        # Create a per-thread HTTP client
        client = ILinkClient(
            bot_token=self.bot_token,
            base_url=self._base_url,
        )
        await client.start()
        cursor = self._cursor

        # Circuit breaker: exponential backoff on consecutive failures
        consecutive_failures = 0
        max_backoff_seconds = 120  # cap at 2 minutes

        try:
            while not self._stop_event.is_set():
                try:
                    data = await client.getupdates(cursor)
                    ret = data.get("ret", -1)
                    new_cursor = data.get("get_updates_buf")
                    if new_cursor is not None:
                        cursor = new_cursor
                        self._cursor = cursor
                    msgs: List[Dict[str, Any]] = data.get("msgs") or []
                    for msg in msgs:
                        await self._on_message(msg, client)

                    # Reset circuit breaker on any successful poll
                    consecutive_failures = 0

                    # ret=-1 is normal long-poll timeout (no new messages)
                    if ret != 0 and not msgs:
                        if ret == -1:
                            logger.debug(
                                "wechat getupdates timeout (ret=-1)"
                                ", continue polling",
                            )
                        else:
                            logger.warning(
                                "wechat getupdates non-zero ret=%s"
                                " (no msgs), retry in 3s",
                                ret,
                            )
                            await asyncio.sleep(3)
                except asyncio.CancelledError:
                    break
                except Exception:
                    consecutive_failures += 1
                    backoff = min(
                        5 * (2 ** (consecutive_failures - 1)),
                        max_backoff_seconds,
                    )
                    logger.exception(
                        "wechat poll error (%d consecutive), retry in %ds",
                        consecutive_failures,
                        backoff,
                    )
                    if not self._stop_event.is_set():
                        await asyncio.sleep(backoff)
        finally:
            await client.stop()

    # ------------------------------------------------------------------
    # Inbound message handler
    # ------------------------------------------------------------------

    async def _on_message(
        self,
        msg: Dict[str, Any],
        client: ILinkClient,
    ) -> None:
        """Parse one inbound WeChatMessage and enqueue for processing."""
        try:
            from_user_id = msg.get("from_user_id", "")
            to_user_id = msg.get("to_user_id", "")
            context_token = msg.get("context_token", "")
            group_id = msg.get("group_id", "")
            msg_type = msg.get("message_type", 0)

            # Only process user→bot messages (message_type == 1)
            if msg_type != 1:
                return

            # Dedup: use context_token as unique id
            dedup_key = (
                context_token or f"{from_user_id}_{msg.get('msg_id', '')}"
            )
            if dedup_key and self._is_duplicate(dedup_key):
                logger.debug(
                    "wechat: duplicate message skipped: %s",
                    dedup_key[:40],
                )
                return

            # Content-based dedup: catch duplicates that arrive with
            # different context_token / msg_id across separate polls.
            raw_text = "".join(
                (item.get("text_item") or {}).get("text", "")
                for item in (msg.get("item_list") or [])
                if item.get("type", 0) == 1
            ).strip()
            if raw_text and self._is_text_duplicate(from_user_id, raw_text):
                logger.debug(
                    "wechat: content-duplicate message skipped: "
                    "user=%s text_len=%d",
                    from_user_id[:12],
                    len(raw_text),
                )
                return

            content_parts: List[Any] = []
            text_parts: List[str] = []

            item_list: List[Dict[str, Any]] = msg.get("item_list") or []
            for item in item_list:
                item_type = item.get("type", 0)

                if item_type == 1:
                    # Text
                    text = (
                        (item.get("text_item") or {}).get("text", "").strip()
                    )
                    # Filter out empty text or text that looks like a filename
                    # (e.g., "document.pdf", "image.jpg") to avoid triggering
                    # immediate agent replies for file-only messages.
                    # This allows BaseChannel._apply_no_text_debounce to work
                    # correctly for media-only messages.
                    if text:
                        # Check if text looks like a filename (has extension)
                        # Common file extensions to filter out
                        filename_extensions = (
                            ".txt",
                            ".doc",
                            ".docx",
                            ".pdf",
                            ".jpg",
                            ".jpeg",
                            ".png",
                            ".gif",
                            ".mp4",
                            ".avi",
                            ".mov",
                            ".mp3",
                            ".wav",
                            ".zip",
                            ".rar",
                            ".xlsx",
                            ".xls",
                            ".ppt",
                            ".pptx",
                        )
                        is_filename = any(
                            text.lower().endswith(ext)
                            for ext in filename_extensions
                        )
                        # Only add text if it's not just a filename
                        if not is_filename:
                            text_parts.append(text)

                elif item_type == 2:
                    # Image (AES-128-ECB encrypted on CDN)
                    img_item = item.get("image_item") or {}
                    media = img_item.get("media") or {}
                    encrypt_query_param = media.get("encrypt_query_param", "")
                    # Key priority: image_item.aeskey (hex) > media.aes_key
                    # Per official SDK: hex aeskey → base64 for decryption
                    aeskey_hex = img_item.get("aeskey", "")
                    if aeskey_hex:
                        aes_key = _b64.b64encode(
                            bytes.fromhex(aeskey_hex),
                        ).decode()
                    else:
                        aes_key = media.get("aes_key", "")
                    if encrypt_query_param:
                        path = await self._download_media(
                            client,
                            "",
                            aes_key,
                            "image.jpg",
                            encrypt_query_param=encrypt_query_param,
                        )
                        if path:
                            content_parts.append(
                                ImageContent(
                                    type=ContentType.IMAGE,
                                    image_url=path,
                                ),
                            )
                        else:
                            text_parts.append("[image: download failed]")
                    else:
                        text_parts.append("[image: no url]")

                elif item_type == 3:
                    # Voice — use ASR transcription text
                    voice_item = item.get("voice_item") or {}
                    asr_text = (
                        voice_item.get("text_item", {}).get("text", "").strip()
                        if isinstance(voice_item.get("text_item"), dict)
                        else voice_item.get("text", "").strip()
                    )
                    if asr_text:
                        text_parts.append(asr_text)
                    else:
                        text_parts.append("[voice: no transcription]")

                elif item_type == 4:
                    # File attachment
                    file_item = item.get("file_item") or {}
                    filename = (
                        file_item.get("file_name", "file.bin") or "file.bin"
                    )
                    media = file_item.get("media") or {}
                    encrypt_query_param = media.get("encrypt_query_param", "")
                    aes_key = media.get(
                        "aes_key",
                        "",
                    )  # base64(Format A or B), handled by aes_ecb_decrypt
                    if encrypt_query_param:
                        path = await self._download_media(
                            client,
                            "",
                            aes_key,
                            filename,
                            encrypt_query_param=encrypt_query_param,
                        )
                        if path:
                            content_parts.append(
                                FileContent(
                                    type=ContentType.FILE,
                                    file_url=path,
                                ),
                            )
                        else:
                            text_parts.append("[file: download failed]")
                    else:
                        text_parts.append("[file: no url]")

                elif item_type == 5:
                    # Video
                    video_item = item.get("video_item") or {}
                    media = video_item.get("media") or {}
                    encrypt_query_param = media.get("encrypt_query_param", "")
                    aes_key = media.get("aes_key", "")
                    if encrypt_query_param:
                        path = await self._download_media(
                            client,
                            "",
                            aes_key,
                            "video.mp4",
                            encrypt_query_param=encrypt_query_param,
                        )
                        if path:
                            content_parts.append(
                                VideoContent(
                                    type=ContentType.VIDEO,
                                    video_url=path,
                                ),
                            )
                        else:
                            text_parts.append("[video: download failed]")
                    else:
                        text_parts.append("[video: no url]")
                else:
                    text_parts.append(f"[unsupported type: {item_type}]")

                # Handle quoted (replied-to) message if present.
                # When a user replies to a message in WeChat, the item
                # contains a ``ref_msg`` field whose ``message_item``
                # mirrors the structure of a normal item (type 1-5).
                ref_msg = item.get("ref_msg")
                if ref_msg:
                    await self._process_quoted_ref_msg(
                        ref_msg,
                        text_parts,
                        content_parts,
                        client,
                    )

            text = "\n".join(text_parts).strip()
            if text:
                content_parts.insert(
                    0,
                    TextContent(type=ContentType.TEXT, text=text),
                )
            if not content_parts:
                return

            is_group = bool(group_id)
            meta: Dict[str, Any] = {
                "wechat_from_user_id": from_user_id,
                "wechat_to_user_id": to_user_id,
                "wechat_context_token": context_token,
                "wechat_group_id": group_id,
                "is_group": is_group,
            }

            # Save latest context_token for proactive sends (heartbeat/cron)
            if from_user_id and context_token:
                self._user_context_tokens[from_user_id] = context_token
                self._save_context_tokens()

            # Start "typing..." indicator immediately and keep refreshing
            # until send_content_parts finishes the reply.
            if from_user_id and context_token:

                async def _start_typing_on_receive():
                    # Stop any existing typing indicator first
                    with self._typing_stop_lock:
                        old_stop = self._typing_stop_funcs.pop(
                            from_user_id,
                            None,
                        )
                    if old_stop:
                        old_stop()

                    stop_func = await self.start_typing(
                        from_user_id,
                        context_token,
                    )
                    with self._typing_stop_lock:
                        self._typing_stop_funcs[from_user_id] = stop_func

                self._dispatch_to_main_loop(
                    _start_typing_on_receive(),
                    description="start typing indicator",
                )

            session_id = self.resolve_session_id(from_user_id, meta)
            native = {
                "channel_id": self.channel,
                "sender_id": from_user_id,
                "user_id": "" if is_group else from_user_id,
                "session_id": session_id,
                "content_parts": content_parts,
                "meta": meta,
            }
            logger.info(
                "wechat recv: from=%s group=%s text_len=%s",
                (from_user_id or "")[:20],
                (group_id or "")[:20],
                len(text),
            )
            if self._enqueue is not None:
                self._enqueue(native)

        except Exception:
            logger.exception("wechat _on_message failed")

    async def _process_quoted_ref_msg(
        self,
        ref_msg: Dict[str, Any],
        text_parts: List[str],
        content_parts: List[Any],
        client: ILinkClient,
    ) -> None:
        """Process a quoted (replied-to) message from ``ref_msg``.

        Extracts text and media from the referenced message and prepends
        quoted text to *text_parts* / appends media to *content_parts*,
        following the same pattern used by the WeCom and Feishu channels.

        Args:
            ref_msg: The ``ref_msg`` dict from an inbound item.
            text_parts: Mutable list of text strings to prepend quoted text.
            content_parts: Mutable list of content parts to append media.
            client: The ILinkClient used for downloading media.
        """
        quoted_item = ref_msg.get("message_item") or {}
        quoted_type = quoted_item.get("type", 0)

        if quoted_type == 1:
            # Quoted text
            quoted_text = (
                (quoted_item.get("text_item") or {}).get("text", "").strip()
            )
            if quoted_text:
                text_parts.insert(0, f"[quoted message: {quoted_text}]")

        elif quoted_type == 2:
            # Quoted image
            img_item = quoted_item.get("image_item") or {}
            media = img_item.get("media") or {}
            encrypt_query_param = media.get("encrypt_query_param", "")
            aeskey_hex = img_item.get("aeskey", "")
            if aeskey_hex:
                aes_key = _b64.b64encode(
                    bytes.fromhex(aeskey_hex),
                ).decode()
            else:
                aes_key = media.get("aes_key", "")
            if encrypt_query_param:
                path = await self._download_media(
                    client,
                    "",
                    aes_key,
                    "image.jpg",
                    encrypt_query_param=encrypt_query_param,
                )
                if path:
                    content_parts.append(
                        ImageContent(
                            type=ContentType.IMAGE,
                            image_url=path,
                        ),
                    )
                else:
                    text_parts.insert(0, "[quoted image: download failed]")
            else:
                text_parts.insert(0, "[quoted image: no url]")

        elif quoted_type == 3:
            # Quoted voice — use ASR transcription
            voice_item = quoted_item.get("voice_item") or {}
            asr_text = (
                voice_item.get("text_item", {}).get("text", "").strip()
                if isinstance(voice_item.get("text_item"), dict)
                else voice_item.get("text", "").strip()
            )
            if asr_text:
                text_parts.insert(0, f"[quoted voice: {asr_text}]")
            else:
                text_parts.insert(0, "[quoted voice: no transcription]")

        elif quoted_type == 4:
            # Quoted file
            file_item = quoted_item.get("file_item") or {}
            filename = file_item.get("file_name", "file.bin") or "file.bin"
            media = file_item.get("media") or {}
            encrypt_query_param = media.get("encrypt_query_param", "")
            aes_key = media.get("aes_key", "")
            if encrypt_query_param:
                path = await self._download_media(
                    client,
                    "",
                    aes_key,
                    filename,
                    encrypt_query_param=encrypt_query_param,
                )
                if path:
                    content_parts.append(
                        FileContent(
                            type=ContentType.FILE,
                            file_url=path,
                        ),
                    )
                else:
                    text_parts.insert(0, "[quoted file: download failed]")
            else:
                text_parts.insert(0, "[quoted file: no url]")

        elif quoted_type == 5:
            # Quoted video
            video_item = quoted_item.get("video_item") or {}
            media = video_item.get("media") or {}
            encrypt_query_param = media.get("encrypt_query_param", "")
            aes_key = media.get("aes_key", "")
            if encrypt_query_param:
                path = await self._download_media(
                    client,
                    "",
                    aes_key,
                    "video.mp4",
                    encrypt_query_param=encrypt_query_param,
                )
                if path:
                    content_parts.append(
                        VideoContent(
                            type=ContentType.VIDEO,
                            video_url=path,
                        ),
                    )
                else:
                    text_parts.insert(0, "[quoted video: download failed]")
            else:
                text_parts.insert(0, "[quoted video: no url]")

        else:
            if quoted_type:
                text_parts.insert(
                    0,
                    f"[quoted message: unsupported type {quoted_type}]",
                )

    # ------------------------------------------------------------------
    # Media download helper
    # ------------------------------------------------------------------

    async def _download_media(
        self,
        client: ILinkClient,
        url: str,
        aes_key: str = "",
        filename_hint: str = "file.bin",
        encrypt_query_param: str = "",
    ) -> Optional[str]:
        """Download and optionally decrypt a CDN media file.

        Returns local file path, or None on failure.
        """
        try:
            data = await client.download_media(
                url,
                aes_key,
                encrypt_query_param,
            )
            self._media_dir.mkdir(parents=True, exist_ok=True)
            safe_name = (
                "".join(c for c in filename_hint if c.isalnum() or c in "-_.")
                or "media"
            )
            url_hash = hashlib.md5(
                (encrypt_query_param or url).encode(),
            ).hexdigest()[:8]
            path = self._media_dir / f"wechat_{url_hash}_{safe_name}"
            path.write_bytes(data)
            return str(path)
        except Exception:
            logger.exception("wechat _download_media failed url=%s", url[:60])
            return None

    # ------------------------------------------------------------------
    # Send helpers
    # ------------------------------------------------------------------

    async def _send_text_direct(
        self,
        to_user_id: str,
        text: str,
        context_token: str,
        client: Optional[ILinkClient] = None,
        api_initiated: bool = False,
        send_meta: Optional[Dict[str, Any]] = None,
    ) -> None:
        """Send text using the shared ILinkClient (or create a temp one).

        Args:
            api_initiated: If True, raise ChannelError on send failure.
                Used by /api/messages/send to provide accurate error feedback.
            send_meta: If provided, used to track context_token invalidation.
                When ret=-2, sets send_meta["_wechat_token_invalid"] = True
                so subsequent sends in the same request are skipped.
        """
        _client = client or self._client
        if not _client or not to_user_id or not text:
            return
        try:
            resp = await _client.send_text(to_user_id, text, context_token)
        except Exception:
            logger.exception("wechat _send_text_direct failed")
            if api_initiated:
                raise
            return
        if isinstance(resp, dict):
            ret = resp.get("ret", 0)
            errcode = resp.get("errcode", 0)
            if ret != 0 or errcode != 0:
                logger.warning(
                    "wechat send_text rejected: "
                    "ret=%s errcode=%s to_user_id=%s",
                    ret,
                    errcode,
                    to_user_id,
                )
                if api_initiated:
                    raise ChannelError(
                        channel_name="wechat",
                        message=(
                            f"iLink API rejected: ret={ret} "
                            f"errcode={errcode} response={resp}"
                        ),
                    )
                # ret=-2 means context_token is invalid/consumed;
                # mark meta so subsequent sends in this request are skipped.
                if ret == -2 and send_meta is not None:
                    send_meta["_wechat_token_invalid"] = True

    async def _send_media_file(
        self,
        to_user_id: str,
        context_token: str,
        file_path: str,
        content_type: ContentType,
        send_meta: Optional[Dict[str, Any]] = None,
    ) -> None:
        """Send a media file (image/file/video) to WeChat.

        Args:
            to_user_id: Recipient user ID.
            context_token: Context token from inbound message.
            file_path: Local path to the media file.
            content_type: Type of media (IMAGE/FILE/VIDEO).
            send_meta: If provided, used to track context_token invalidation.
        """
        if not self._client or not to_user_id or not context_token:
            logger.warning(
                "wechat _send_media_file: missing required parameters",
            )
            return

        try:
            # Convert URL to local path if it's a file:// URL
            file_path = file_url_to_local_path(file_path) or file_path

            # Check if file exists
            path_obj = Path(file_path)
            if not path_obj.exists():
                logger.warning(
                    "wechat _send_media_file: file not found: %s",
                    file_path,
                )
                return

            # Send based on content type
            resp: Optional[Dict[str, Any]] = None
            if content_type == ContentType.IMAGE:
                resp = await self._client.send_image(
                    to_user_id,
                    str(path_obj),
                    context_token,
                )
            elif content_type == ContentType.FILE:
                filename = path_obj.name
                resp = await self._client.send_file(
                    to_user_id,
                    str(path_obj),
                    filename,
                    context_token,
                )
            elif content_type == ContentType.VIDEO:
                resp = await self._client.send_video(
                    to_user_id,
                    str(path_obj),
                    context_token,
                )
            else:
                logger.warning(
                    "wechat _send_media_file: unsupported content type: %s",
                    content_type,
                )
                return

            # Check response for errors (same logic as _send_text_direct)
            if isinstance(resp, dict):
                ret = resp.get("ret", 0)
                errcode = resp.get("errcode", 0)
                if ret != 0 or errcode != 0:
                    logger.warning(
                        "wechat send_media rejected: "
                        "ret=%s errcode=%s to_user_id=%s type=%s",
                        ret,
                        errcode,
                        to_user_id,
                        content_type,
                    )
                    if ret == -2 and send_meta is not None:
                        send_meta["_wechat_token_invalid"] = True
        except Exception:
            logger.exception(
                "wechat _send_media_file failed type=%s path=%s",
                content_type,
                file_path[:60],
            )

    def _stop_typing_for_user(self, user_id: str) -> None:
        """Stop typing indicator and remove from tracking dict."""
        with self._typing_stop_lock:
            stop_func = self._typing_stop_funcs.pop(user_id, None)
        if stop_func:
            stop_func()

    # ------------------------------------------------------------------
    # Message merge helpers (mitigates 10-msg context_token limit)
    #
    # WeChat iLink Bot is single-chat only (no group chat), so a single
    # instance-level buffer is sufficient — no per-session keying needed.
    # ------------------------------------------------------------------

    async def _buffer_parts_for_merge(
        self,
        to_handle: str,
        parts: List[OutgoingContentPart],
        meta: Optional[Dict[str, Any]] = None,
    ) -> None:
        """Append parts to the merge buffer.

        In full-merge mode (delay_ms == 0) nothing is flushed here; the
        caller is responsible for flushing at the end of the request.

        In delay-merge mode (delay_ms > 0) a timer is (re)started; when
        it fires the buffer is flushed automatically.
        """
        # Insert a newline separator between existing buffer and new parts
        # so that merged messages are visually separated in the final output.
        if self._merge_buffer and parts:
            self._merge_buffer.append(
                TextContent(type=ContentType.TEXT, text="\n"),
            )
        self._merge_buffer.extend(parts)
        self._merge_meta = dict(meta or {})
        self._merge_to_handle = to_handle

        if self._message_merge_delay_ms > 0:
            if self._merge_timer is not None:
                self._merge_timer.cancel()

            loop = asyncio.get_running_loop()
            delay_sec = self._message_merge_delay_ms / 1000.0
            self._merge_timer = loop.call_later(
                delay_sec,
                lambda: asyncio.ensure_future(
                    self._flush_merge_buffer(to_handle),
                ),
            )
        elif not self._merge_meta.get("wechat_context_token"):
            # Full-merge mode: cron/proactive calls won't trigger
            # _on_process_completed, so flush immediately.
            await self._flush_merge_buffer(to_handle)

    async def _flush_merge_buffer(
        self,
        to_handle: str,
    ) -> None:
        """Flush the merge buffer: send all accumulated parts at once."""
        if self._merge_timer is not None:
            self._merge_timer.cancel()
            self._merge_timer = None

        buffered_parts = self._merge_buffer
        buffered_meta = self._merge_meta
        self._merge_buffer = []
        self._merge_meta = None

        if not buffered_parts:
            return

        await self._send_content_parts_immediate(
            to_handle,
            buffered_parts,
            buffered_meta,
        )

    async def send_content_parts(
        self,
        to_handle: str,
        parts: List[OutgoingContentPart],
        meta: Optional[Dict[str, Any]] = None,
    ) -> None:
        """Send agent response content back to the WeChat user.

        When message merging is enabled, text parts are buffered and
        sent together (either at the end of the request or after a
        configurable delay window).  Media parts are always sent
        immediately since they cannot be merged into text.
        """
        if not self.enabled:
            return

        if self._message_merge_enabled:
            # Separate text/refusal parts (mergeable) from media (immediate)
            text_parts: List[OutgoingContentPart] = []
            media_parts: List[OutgoingContentPart] = []
            for part in parts:
                part_type = getattr(part, "type", None) or (
                    part.get("type") if isinstance(part, dict) else None
                )
                if part_type in (ContentType.TEXT, ContentType.REFUSAL):
                    text_parts.append(part)
                else:
                    media_parts.append(part)

            # Buffer text parts for later merge
            if text_parts:
                await self._buffer_parts_for_merge(
                    to_handle,
                    text_parts,
                    meta,
                )

            # Media parts are sent immediately (cannot be merged)
            if media_parts:
                await self._send_content_parts_immediate(
                    to_handle,
                    media_parts,
                    meta,
                )

            return

        # Merging disabled: send everything immediately
        await self._send_content_parts_immediate(to_handle, parts, meta)

    async def _send_content_parts_immediate(
        self,
        to_handle: str,
        parts: List[OutgoingContentPart],
        meta: Optional[Dict[str, Any]] = None,
    ) -> None:
        """Actually send content parts to WeChat (no merge buffering)."""
        m = meta or {}
        to_user_id = (
            m.get("wechat_from_user_id")
            or self._parse_user_id_from_handle(to_handle)
            or ""
        )
        context_token = m.get("wechat_context_token", "") or (
            self._user_context_tokens.get(to_user_id, "")
        )

        if not to_user_id:
            logger.warning("wechat send_content_parts: no to_user_id")
            return

        prefix = m.get("bot_prefix", "") or self.bot_prefix or ""
        text_parts: List[str] = []

        for p in parts:
            # Skip all sends once context_token is marked invalid
            if m.get("_wechat_token_invalid"):
                break

            t = getattr(p, "type", None) or (
                p.get("type") if isinstance(p, dict) else None
            )
            text_val = getattr(p, "text", None) or (
                p.get("text") if isinstance(p, dict) else None
            )
            refusal_val = getattr(p, "refusal", None) or (
                p.get("refusal") if isinstance(p, dict) else None
            )
            if t == ContentType.TEXT and text_val:
                text_parts.append(text_val)
            elif t == ContentType.REFUSAL and refusal_val:
                text_parts.append(refusal_val)
            elif t == ContentType.IMAGE:
                # Send image
                image_url = getattr(p, "image_url", None) or (
                    p.get("image_url") if isinstance(p, dict) else None
                )
                if image_url:
                    await self._send_media_file(
                        to_user_id,
                        context_token,
                        image_url,
                        ContentType.IMAGE,
                        send_meta=m,
                    )
            elif t == ContentType.FILE:
                # Send file
                file_url = getattr(p, "file_url", None) or (
                    p.get("file_url") if isinstance(p, dict) else None
                )
                if file_url:
                    await self._send_media_file(
                        to_user_id,
                        context_token,
                        file_url,
                        ContentType.FILE,
                        send_meta=m,
                    )
            elif t == ContentType.VIDEO:
                # Send video
                video_url = getattr(p, "video_url", None) or (
                    p.get("video_url") if isinstance(p, dict) else None
                )
                if video_url:
                    await self._send_media_file(
                        to_user_id,
                        context_token,
                        video_url,
                        ContentType.VIDEO,
                        send_meta=m,
                    )
            elif t == ContentType.AUDIO:
                # Send audio as file (WeChat has no dedicated audio send)
                audio_url = getattr(p, "data", None) or (
                    p.get("data") if isinstance(p, dict) else None
                )
                if audio_url and not audio_url.startswith("data:"):
                    await self._send_media_file(
                        to_user_id,
                        context_token,
                        audio_url,
                        ContentType.FILE,
                        send_meta=m,
                    )

        body = "\n".join(text_parts).strip()
        if prefix and body:
            body = prefix + "  " + body

        if not body or m.get("_wechat_token_invalid"):
            return

        api_send = bool(m.get("_api_send"))
        for chunk in split_text(body):
            if m.get("_wechat_token_invalid"):
                return
            await self._send_text_direct(
                to_user_id,
                chunk,
                context_token,
                api_initiated=api_send,
                send_meta=m,
            )

    async def _on_process_completed(
        self,
        request: Any,
        to_handle: str,
        send_meta: Dict[str, Any],
    ) -> None:
        """Flush merge buffer (if any) and stop typing indicator."""
        # Flush any remaining merged messages before finishing
        if self._message_merge_enabled:
            await self._flush_merge_buffer(to_handle)

        user_id = (
            (send_meta or {}).get("wechat_from_user_id")
            or self._parse_user_id_from_handle(to_handle)
            or ""
        )
        if user_id:
            self._stop_typing_for_user(user_id)

    async def _on_consume_error(
        self,
        request: Any,
        to_handle: str,
        err_text: str,
    ) -> None:
        """Flush merge buffer, stop typing, and send error message."""
        # Flush any buffered messages before sending the error
        if self._message_merge_enabled:
            await self._flush_merge_buffer(to_handle)

        user_id = self._parse_user_id_from_handle(to_handle) or ""
        if user_id:
            self._stop_typing_for_user(user_id)
        await super()._on_consume_error(request, to_handle, err_text)

    async def send(
        self,
        to_handle: str,
        text: str,
        meta: Optional[Dict[str, Any]] = None,
    ) -> None:
        """Proactive send (e.g. from scheduled jobs)."""
        if not self.enabled:
            return
        m = meta or {}
        to_user_id = (
            m.get("wechat_from_user_id")
            or self._parse_user_id_from_handle(to_handle)
            or ""
        )
        context_token = m.get("wechat_context_token", "") or (
            self._user_context_tokens.get(to_user_id, "")
        )
        prefix = m.get("bot_prefix", "") or self.bot_prefix or ""
        body = (prefix + "  " + text) if prefix and text else text
        if not body or not to_user_id:
            return
        send_state: Dict[str, Any] = {}
        for chunk in split_text(body):
            if send_state.get("_wechat_token_invalid"):
                return
            await self._send_text_direct(
                to_user_id,
                chunk,
                context_token,
                send_meta=send_state,
            )

    # ------------------------------------------------------------------
    # Typing Indicator
    # ------------------------------------------------------------------

    async def _get_typing_ticket(
        self,
        user_id: str,
        context_token: str,
    ) -> str:
        """Get or fetch typing ticket for a user.

        Args:
            user_id: User ID
            context_token: Context token for the user

        Returns:
            Typing ticket string (empty if failed)
        """
        now = time.time()
        cache_ttl = 24 * 3600  # 24 hours

        logger.debug(
            "wechat _get_typing_ticket called for user_id="
            f"{user_id}, context_token="
            f"{context_token[:20] if context_token else 'NONE'}...",
        )

        with self._typing_lock:
            # Check cache
            if user_id in self._typing_tickets:
                ticket, expiry = self._typing_tickets[user_id]
                if now < expiry:
                    logger.debug(
                        f"wechat using cached typing_ticket for {user_id}",
                    )
                    return ticket
                # Expired, remove from cache
                del self._typing_tickets[user_id]

        # Fetch new ticket from API
        try:
            logger.info(f"wechat calling getconfig API for {user_id}")
            resp = await self._client.getconfig(
                ilink_user_id=user_id,
                context_token=context_token,
            )
            ret = resp.get("ret", 1)
            errcode = resp.get("errcode") or 0  # Treat None as 0
            logger.info(
                f"wechat getconfig response: ret={ret}, "
                f"errcode={resp.get('errcode')}, "
                f"ticket={'FOUND' if resp.get('typing_ticket') else 'EMPTY'}",
            )
            if ret == 0 and errcode == 0:
                ticket = resp.get("typing_ticket", "").strip()
                if ticket:
                    with self._typing_lock:
                        self._typing_tickets[user_id] = (
                            ticket,
                            now + cache_ttl,
                        )
                    logger.info(
                        f"wechat got typing_ticket for {user_id}: "
                        f"{ticket[:20]}... (length={len(ticket)})",
                    )
                    return ticket
                else:
                    logger.warning(
                        "wechat getconfig returned no typing_ticket",
                    )
            else:
                logger.warning(
                    f"wechat getconfig failed: ret={ret}, "
                    f"errcode={resp.get('errcode')}",
                )
        except Exception as e:
            logger.warning(f"wechat getconfig failed: {e}")

        return ""

    async def start_typing(
        self,
        user_id: str,
        context_token: str,
    ) -> Callable[[], None]:
        """Start typing indicator for a user.

        Args:
            user_id: User ID
            context_token: Context token for the user

        Returns:
            A stop function that cancels the typing indicator
        """
        logger.info(f"wechat start_typing called for user_id={user_id}")
        ticket = await self._get_typing_ticket(user_id, context_token)
        if not ticket:
            logger.debug("wechat start_typing: no ticket for %s", user_id)
            return lambda: None

        stop_event = asyncio.Event()
        stop_called = False

        async def refresh_typing():
            """Refresh typing indicator every 5 seconds."""
            logger.debug("wechat refresh_typing started for %s", user_id)
            while not stop_event.is_set():
                client = self._client
                if client is None:
                    logger.debug(
                        "wechat refresh_typing: client gone, exiting "
                        "for %s",
                        user_id,
                    )
                    break
                try:
                    await client.sendtyping(user_id, ticket, status=1)
                except Exception as exc:
                    logger.debug(
                        "wechat sendtyping refresh failed: %s",
                        exc,
                    )
                try:
                    await asyncio.wait_for(
                        stop_event.wait(),
                        timeout=5.0,
                    )
                except asyncio.TimeoutError:
                    pass
            logger.debug("wechat refresh_typing stopped for %s", user_id)

        # Create the background refresh task.
        # The reference is held by the stop() closure via stop_event;
        # we do not need to store it separately.
        asyncio.create_task(refresh_typing())

        # Send initial typing status
        client = self._client
        if client:
            try:
                await client.sendtyping(user_id, ticket, status=1)
            except Exception as exc:
                logger.debug("wechat sendtyping initial failed: %s", exc)

        def stop(send_cancel: bool = True):
            """Stop the typing indicator and cancel the refresh task."""
            nonlocal stop_called
            if stop_called:
                return
            stop_called = True
            stop_event.set()

            if send_cancel:
                client = self._client
                if client:

                    async def _cancel():
                        try:
                            await client.sendtyping(
                                user_id,
                                ticket,
                                status=2,
                            )
                        except Exception:
                            pass

                    try:
                        asyncio.ensure_future(_cancel())
                    except RuntimeError:
                        pass

        return stop

    # ------------------------------------------------------------------
    # Lifecycle
    # ------------------------------------------------------------------

    async def health_check(self) -> Dict[str, Any]:
        """Check WeChat long-poll client status."""
        if not self.enabled:
            return {
                "channel": self.channel,
                "status": "disabled",
                "detail": "WeChat channel is disabled.",
            }
        issues = []
        if self._client is None:
            issues.append("WeChat client not initialized")
        if not self.bot_token:
            issues.append("Bot token not available (not logged in)")
        poll_thread_alive = (
            self._poll_thread is not None and self._poll_thread.is_alive()
        )
        if not poll_thread_alive:
            issues.append("Long-poll thread is not running")
        if issues:
            return {
                "channel": self.channel,
                "status": "unhealthy",
                "detail": "; ".join(issues),
            }
        return {
            "channel": self.channel,
            "status": "healthy",
            "detail": "WeChat client is connected and polling.",
        }

    async def start(self) -> None:
        if not self.enabled:
            logger.debug("wechat channel disabled")
            return

        # Resolve token: config > token file
        if not self.bot_token:
            self.bot_token = self._load_token_from_file()

        # Load persisted context_tokens for proactive sends
        self._load_context_tokens()

        # If still no token, do QR code login with a temporary client
        if not self.bot_token:
            login_client = ILinkClient(base_url=self._base_url)
            await login_client.start()
            try:
                self._client = login_client
                ok = await self._do_qrcode_login()
                if not ok:
                    raise ChannelError(
                        channel_name="wechat",
                        message=(
                            "WeChat QR code login failed. "
                            "Please provide a valid bot_token in config"
                        ),
                    )
                # Login succeeded; login_client becomes the long-lived client
            except Exception:
                await login_client.stop()
                self._client = None
                raise
        else:
            # Token already known — create the long-lived client now
            self._client = ILinkClient(
                bot_token=self.bot_token,
                base_url=self._base_url,
            )
            await self._client.start()

        self._loop = asyncio.get_running_loop()
        self._stop_event.clear()
        self._loop_accepting.set()  # main loop is ready to accept tasks

        # Launch background long-poll thread
        self._poll_thread = threading.Thread(
            target=self._run_poll_forever,
            daemon=True,
            name="wechat-poll",
        )
        self._poll_thread.start()
        logger.info(
            "wechat channel started (token=%s…)",
            (self.bot_token or "")[:12],
        )

    async def stop(self) -> None:
        if not self.enabled:
            return
        # Signal poll thread to stop accepting new work BEFORE stopping loop
        self._loop_accepting.clear()
        self._stop_event.set()
        # Cancel the poll task gracefully instead of brute-force stopping loop
        if self._poll_loop is not None and self._poll_task is not None:
            try:
                self._poll_loop.call_soon_threadsafe(self._poll_task.cancel)
            except Exception:
                pass
        if self._poll_thread:
            self._poll_thread.join(timeout=10)
        self._poll_thread = None

        # Stop all active typing indicators before closing the client
        with self._typing_stop_lock:
            stop_funcs = list(self._typing_stop_funcs.values())
            self._typing_stop_funcs.clear()
        for func in stop_funcs:
            try:
                func()
            except Exception:
                pass

        if self._client:
            await self._client.stop()
        self._client = None
        logger.info("wechat channel stopped")
