EvoScientist-Multi/EvoScientist/channels/feishu/channel.py

"""Feishu (飞书/Lark) channel implementation.

Receives messages via HTTP event subscription webhook (aiohttp) or
WebSocket long connection (lark-oapi SDK), sends replies via Feishu
Open API REST endpoints.

Feishu Open API docs: https://open.feishu.cn/document

Authentication:
  - App ID + App Secret → tenant_access_token (2-hour TTL, auto-refreshed)

Event subscription (two modes):
  - **Webhook**: URL verification challenge on first request,
    ``im.message.receive_v1`` events via HTTP POST callback.
    Requires a publicly reachable URL.
  - **WebSocket**: outbound long connection via ``lark-oapi`` SDK.
    No public IP required.

Send API:
  - ``POST /open-apis/im/v1/messages?receive_id_type=chat_id``
"""

from __future__ import annotations

import asyncio
import base64
import hashlib
import json
import logging
import queue
import re
import threading
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, ClassVar

if TYPE_CHECKING:
    from aiohttp import web

from ..base import Channel, ChannelError, RawIncoming
from ..capabilities import FEISHU as FEISHU_CAPS
from ..config import BaseChannelConfig
from ..mixins import TokenMixin, WebhookMixin

logger = logging.getLogger(__name__)


# ── Markdown → Feishu Post conversion ────────────────────────────


def _parse_inline_text(text: str) -> list[dict]:
    """Parse inline Markdown elements into Feishu post tag dicts.

    Handles: `code`, **bold**, ~~strikethrough~~, [link](url), _italic_.
    """
    elements: list[dict] = []
    # Pattern order matters: code first (protect content), then bold, strikethrough, link, italic
    pattern = re.compile(
        r"`([^`]+)`"  # inline code
        r"|\*\*(.+?)\*\*"  # bold
        r"|~~(.+?)~~"  # strikethrough
        r"|\[([^\]]+)\]\(([^)]+)\)"  # link
        r"|_(.+?)_"  # italic
    )
    pos = 0
    for m in pattern.finditer(text):
        # Plain text before this match
        if m.start() > pos:
            elements.append({"tag": "text", "text": text[pos : m.start()]})

        if m.group(1) is not None:
            # inline code → code_block would be block-level; use text with style
            elements.append(
                {
                    "tag": "text",
                    "text": m.group(1),
                    "style": ["code_block"],
                }
            )
        elif m.group(2) is not None:
            elements.append(
                {
                    "tag": "text",
                    "text": m.group(2),
                    "style": ["bold"],
                }
            )
        elif m.group(3) is not None:
            elements.append(
                {
                    "tag": "text",
                    "text": m.group(3),
                    "style": ["strikethrough"],
                }
            )
        elif m.group(4) is not None:
            elements.append(
                {
                    "tag": "a",
                    "text": m.group(4),
                    "href": m.group(5),
                }
            )
        elif m.group(6) is not None:
            elements.append(
                {
                    "tag": "text",
                    "text": m.group(6),
                    "style": ["italic"],
                }
            )
        pos = m.end()

    # Remaining plain text
    if pos < len(text):
        elements.append({"tag": "text", "text": text[pos:]})
    return elements


def _parse_inline_elements(line: str) -> list[dict]:
    """Parse a single Markdown line into a list of Feishu post elements.

    Handles headings (→ bold), blockquotes (→ italic with prefix),
    list items (→ bullet prefix), and plain lines.
    """
    # Heading: # Title → bold text
    heading_match = re.match(r"^(#{1,6})\s+(.+)$", line)
    if heading_match:
        return [{"tag": "text", "text": heading_match.group(2), "style": ["bold"]}]

    # Blockquote: > text → italic with "▎" prefix
    quote_match = re.match(r"^>\s*(.*)$", line)
    if quote_match:
        inner = quote_match.group(1)
        elements = [{"tag": "text", "text": "▎", "style": ["italic"]}]
        elements.extend(_parse_inline_text(inner))
        return elements

    # Unordered list: - item or * item → "• " prefix
    list_match = re.match(r"^[\-\*]\s+(.+)$", line)
    if list_match:
        elements = [{"tag": "text", "text": "• "}]
        elements.extend(_parse_inline_text(list_match.group(1)))
        return elements

    # Ordered list: 1. item → keep number prefix
    ol_match = re.match(r"^(\d+)\.\s+(.+)$", line)
    if ol_match:
        elements = [{"tag": "text", "text": f"{ol_match.group(1)}. "}]
        elements.extend(_parse_inline_text(ol_match.group(2)))
        return elements

    # Plain line
    return _parse_inline_text(line)


def _markdown_to_feishu_post(text: str) -> dict | None:
    """Convert Markdown text to Feishu post (rich text) JSON structure.

    Returns a dict like {"zh_cn": {"content": [[...]]}} suitable for
    Feishu msg_type="post", or None if the text is empty.
    """
    if not text or not text.strip():
        return None

    paragraphs: list[list[dict]] = []
    current_paragraph: list[dict] = []
    in_code_block = False
    code_lines: list[str] = []
    code_lang = ""

    for line in text.split("\n"):
        # Code block fences
        if line.startswith("```"):
            if not in_code_block:
                # Flush any pending paragraph
                if current_paragraph:
                    paragraphs.append(current_paragraph)
                    current_paragraph = []
                in_code_block = True
                code_lang = line[3:].strip()
                code_lines = []
            else:
                # End of code block
                code_text = "\n".join(code_lines)
                paragraphs.append(
                    [
                        {
                            "tag": "code_block",
                            "language": code_lang or "plain",
                            "text": code_text,
                        }
                    ]
                )
                in_code_block = False
                code_lines = []
                code_lang = ""
            continue

        if in_code_block:
            code_lines.append(line)
            continue

        # Empty line → new paragraph
        if not line.strip():
            if current_paragraph:
                paragraphs.append(current_paragraph)
                current_paragraph = []
            continue

        # Non-empty line
        elements = _parse_inline_elements(line)
        if elements:
            # Each visual line becomes its own paragraph in Feishu post
            if current_paragraph:
                paragraphs.append(current_paragraph)
            current_paragraph = elements

    # Flush remaining
    if in_code_block and code_lines:
        code_text = "\n".join(code_lines)
        paragraphs.append(
            [
                {
                    "tag": "code_block",
                    "language": code_lang or "plain",
                    "text": code_text,
                }
            ]
        )
    elif current_paragraph:
        paragraphs.append(current_paragraph)

    if not paragraphs:
        return None

    return {"zh_cn": {"content": paragraphs}}


@dataclass
class FeishuConfig(BaseChannelConfig):
    app_id: str = ""
    app_secret: str = ""
    verification_token: str = ""
    encrypt_key: str = ""
    webhook_port: int = 9000
    text_chunk_limit: int = 4096
    feishu_domain: str = "https://open.feishu.cn"
    subscription_mode: str = "webhook"  # "webhook" | "websocket"


class FeishuChannel(Channel, WebhookMixin, TokenMixin):
    capabilities = FEISHU_CAPS
    """Feishu channel using Open API + event subscription webhook."""

    name = "feishu"
    _ready_attrs = ("_http_client", "_access_token")
    _non_retryable_patterns = (
        *Channel._non_retryable_patterns,
        "app_access_token is empty",  # invalid credentials
        "10003",  # invalid app_id
        "10014",  # invalid app_secret
        "99991401",  # permission denied
        "99991663",  # no permission
        "99991672",  # feature not enabled
    )
    _rate_limit_patterns = ("99991400", "rate limit", "频率限制")
    _rate_limit_delay = 2.0

    def __init__(self, config: FeishuConfig):
        super().__init__(config)
        self._mention_names: list[str] = []  # bot mention keys from events
        self._main_loop: asyncio.AbstractEventLoop | None = None
        self._lark_ws_thread: threading.Thread | None = None
        self._ws_event_queue: queue.Queue | None = None
        self._ws_consumer_task: asyncio.Task | None = None

    # ── WebhookMixin overrides ────────────────────────────────────

    def _get_webhook_port(self) -> int:
        return self.config.webhook_port

    def _webhook_routes(self) -> list[tuple[str, Any]]:
        return [("POST", "/webhook/event", self._handle_event)]

    # ── TokenMixin overrides ──────────────────────────────────────

    async def _fetch_token(self) -> tuple[str, int]:
        """Fetch Feishu tenant_access_token."""
        url = f"{self.config.feishu_domain}/open-apis/auth/v3/tenant_access_token/internal"
        body = {
            "app_id": self.config.app_id,
            "app_secret": self.config.app_secret,
        }
        try:
            resp = await self._http_client.post(url, json=body)
            data = resp.json()
        except Exception as e:
            raise ChannelError(f"Failed to get Feishu access token: {e}") from e

        if data.get("code") != 0:
            raise ChannelError(f"Feishu auth error: {data.get('msg', 'unknown')}")
        return data["tenant_access_token"], data.get("expire", 7200)

    # ── Lifecycle ─────────────────────────────────────────────────

    _VALID_SUBSCRIPTION_MODES = ("webhook", "websocket")

    async def start(self) -> None:
        if not self.config.app_id:
            raise ChannelError("Feishu app_id is required")
        if not self.config.app_secret:
            raise ChannelError("Feishu app_secret is required")
        if self.config.subscription_mode not in self._VALID_SUBSCRIPTION_MODES:
            raise ChannelError(
                f"Invalid feishu_subscription_mode: {self.config.subscription_mode!r}. "
                f"Must be one of {self._VALID_SUBSCRIPTION_MODES}"
            )

        if self.config.subscription_mode == "websocket":
            await self._start_websocket_mode()
        else:
            await self._start_webhook_mode()

    async def _start_webhook_mode(self) -> None:
        try:
            import httpx  # noqa: F401
            from aiohttp import web  # noqa: F401
        except ImportError:
            raise ChannelError(
                "aiohttp or httpx not installed. "
                "Install with: pip install aiohttp httpx"
            ) from None

        # Start webhook server (sets up self._http_client)
        await self._start_webhook_server()

        # Verify credentials by fetching initial token
        await self._refresh_token()

        self._running = True
        logger.info(
            f"Feishu channel started (webhook on port {self.config.webhook_port})"
        )

    async def _start_websocket_mode(self) -> None:
        try:
            import lark_oapi as lark
        except ImportError:
            raise ChannelError(
                "lark-oapi not installed. Install with: pip install 'lark-oapi>=1.4.0'"
            ) from None

        import httpx

        proxy = getattr(self.config, "proxy", None) or None
        self._http_client = httpx.AsyncClient(timeout=15, proxy=proxy)

        # Verify credentials by fetching initial token
        await self._refresh_token()

        self._main_loop = asyncio.get_running_loop()

        # Thread-safe queue: SDK thread puts events, main loop consumes
        self._ws_event_queue = queue.Queue()

        # Set _running BEFORE creating consumer task — the task checks
        # `while self._running` and would exit immediately otherwise.
        self._running = True
        self._ws_consumer_task = asyncio.create_task(self._consume_ws_events())

        # Build SDK event handler
        handler = (
            lark.EventDispatcherHandler.builder("", "")
            .register_p2_im_message_receive_v1(self._on_lark_sdk_message)
            .build()
        )

        # Silently absorb events we don't have a handler for.  Feishu auto-
        # subscribes a PersonalAgent app to many event types (reactions,
        # read receipts, recalls, member changes…) that EvoScientist doesn't
        # care about.  Without this wrapper, ``_do_without_validation``
        # raises ``EventException("processor not found, type: ...")``,
        # which lark-oapi's WS client (ws/client.py) catches and turns into
        # an HTTP 500 reply on the WebSocket frame — Feishu then marks the
        # event as failed and retries it.  This is especially noisy because
        # our own ``_send_ack_reaction`` triggers ``im.message.reaction.
        # created_v1`` on every inbound message, causing a feedback loop.
        from lark_oapi.core.exception import EventException

        _original_dispatch = handler._do_without_validation

        def _silent_dispatch(payload: bytes):
            try:
                return _original_dispatch(payload)
            except EventException as exc:
                if "processor not found" in str(exc):
                    logger.debug("Feishu: ignored unsubscribed event (%s)", exc)
                    return None
                raise

        handler._do_without_validation = _silent_dispatch

        ws_client = lark.ws.Client(
            self.config.app_id,
            self.config.app_secret,
            event_handler=handler,
            log_level=lark.LogLevel.WARNING,
        )

        def _run_ws():
            # Root cause: lark_oapi.ws.client stores the event loop in a
            # *module-level* variable at import time.  When imported from
            # the main thread this captures the main loop (which has
            # nest_asyncio patches).  The SDK then calls
            # loop.run_until_complete() from THIS thread on that *main*
            # loop, causing cross-thread task-tracking conflicts:
            #   RuntimeError: Leaving task … does not match the current task
            #   AttributeError: 'NoneType' object has no attribute 'select'
            #
            # Fix: create a fresh event loop for this thread and replace
            # the module-level ``loop`` variable so the SDK uses an
            # isolated loop with no cross-thread interaction.
            import lark_oapi.ws.client as _ws_mod

            fresh_loop = asyncio.new_event_loop()
            asyncio.set_event_loop(fresh_loop)
            _ws_mod.loop = fresh_loop

            try:
                ws_client.start()
            except Exception:
                logger.exception(
                    "Feishu WebSocket SDK thread exited unexpectedly. "
                    "The channel will no longer receive messages. "
                    "Check app_id/app_secret and connection limits."
                )

        self._lark_ws_thread = threading.Thread(target=_run_ws, daemon=True)
        self._lark_ws_thread.start()

        logger.info("Feishu channel started (WebSocket long connection mode)")

    def _on_lark_sdk_message(self, data) -> None:
        """Sync callback invoked in lark-oapi SDK thread.

        Converts the SDK event object to a dict and puts it on a
        thread-safe queue.  The ``_consume_ws_events`` task on the main
        asyncio loop picks it up — no asyncio cross-thread calls needed,
        avoiding the nest_asyncio + Python 3.11 contextvars conflict.
        """
        try:
            event = data.event
            msg = event.message
            sender = event.sender

            # Rebuild mentions list from SDK objects
            mentions_list = []
            if msg.mentions:
                for m in msg.mentions:
                    mention_dict: dict[str, Any] = {"key": m.key, "id": {}}
                    if m.id:
                        mention_dict["id"] = {
                            "open_id": getattr(m.id, "open_id", ""),
                            "user_id": getattr(m.id, "user_id", ""),
                        }
                    mentions_list.append(mention_dict)

            event_dict = {
                "sender": {
                    "sender_id": {
                        "open_id": sender.sender_id.open_id if sender.sender_id else "",
                        "user_id": getattr(sender.sender_id, "user_id", "")
                        if sender.sender_id
                        else "",
                    },
                    "sender_type": sender.sender_type or "",
                },
                "message": {
                    "chat_id": msg.chat_id or "",
                    "message_type": msg.message_type or "",
                    "message_id": msg.message_id or "",
                    "chat_type": msg.chat_type or "",
                    "content": msg.content or "{}",
                    "create_time": msg.create_time or "",
                    "mentions": mentions_list,
                },
            }

            self._ws_event_queue.put(event_dict)
        except Exception:
            logger.exception("Feishu SDK message handler error")

    async def _consume_ws_events(self) -> None:
        """Main-loop task that drains the thread-safe event queue."""
        while self._running:
            try:
                event_dict = self._ws_event_queue.get_nowait()
                try:
                    await self._on_message(event_dict)
                except Exception:
                    logger.exception("Feishu WS event processing error")
            except queue.Empty:
                await asyncio.sleep(0.05)

    async def _cleanup(self) -> None:
        if self.config.subscription_mode == "websocket":
            if self._ws_consumer_task:
                self._ws_consumer_task.cancel()
                try:
                    await self._ws_consumer_task
                except asyncio.CancelledError:
                    pass
                self._ws_consumer_task = None
            if self._http_client:
                await self._http_client.aclose()
                self._http_client = None
            # Daemon thread exits with the process; no explicit stop needed
            self._lark_ws_thread = None
            self._main_loop = None
            self._ws_event_queue = None
        else:
            await self._stop_webhook_server()
        self._access_token = None
        logger.info("Feishu channel stopped")

    # ── Token helpers (adapt old API to mixin) ────────────────────

    async def _ensure_token(self) -> str:
        """Return a valid access token, refreshing if needed."""
        return await TokenMixin._ensure_token(self)

    # ── Send (template method overrides) ──────────────────────────

    async def _feishu_send(self, url: str, body: dict, headers: dict) -> bool:
        """POST to Feishu API and return True if code==0."""
        try:
            resp = await self._http_client.post(url, json=body, headers=headers)
            return resp.json().get("code") == 0
        except Exception as e:
            logger.warning(f"Feishu send error: {e}")
            return False

    async def _send_chunk(
        self,
        chat_id,
        formatted_text,
        raw_text,
        reply_to,
        metadata,
    ):
        token = await self._ensure_token()
        headers = {"Authorization": f"Bearer {token}"}
        post_content = _markdown_to_feishu_post(raw_text)

        # If reply_to is set, try the reply API first
        if reply_to:
            reply_url = (
                f"{self.config.feishu_domain}/open-apis/im/v1/messages/{reply_to}/reply"
            )
            if post_content is not None:
                body = {"msg_type": "post", "content": json.dumps(post_content)}
            else:
                body = {
                    "msg_type": "text",
                    "content": json.dumps({"text": formatted_text}),
                }
            if await self._feishu_send(reply_url, body, headers):
                return

        # Normal send (non-reply or reply fallback)
        url = (
            f"{self.config.feishu_domain}"
            f"/open-apis/im/v1/messages?receive_id_type=chat_id"
        )

        # Try post format first
        if post_content is not None:
            body = {
                "receive_id": chat_id,
                "msg_type": "post",
                "content": json.dumps(post_content),
            }
            if await self._feishu_send(url, body, headers):
                return

        # Fallback: plain text
        body = {
            "receive_id": chat_id,
            "msg_type": "text",
            "content": json.dumps({"text": formatted_text}),
        }
        if not await self._feishu_send(url, body, headers):
            raise RuntimeError("Feishu send failed")

    # ── Media helpers ──────────────────────────────────────────────

    _IMAGE_EXTENSIONS: ClassVar[set[str]] = {
        ".jpg",
        ".jpeg",
        ".png",
        ".gif",
        ".bmp",
        ".webp",
    }

    async def _download_media(
        self,
        message_id: str,
        file_key: str,
        msg_type: str,
    ) -> str | None:
        """Download an image or file attachment from Feishu.

        Returns the local file path on success, or None on failure.
        """
        token = await self._ensure_token()
        resource_type = "image" if msg_type == "image" else "file"
        url = (
            f"{self.config.feishu_domain}"
            f"/open-apis/im/v1/messages/{message_id}"
            f"/resources/{file_key}?type={resource_type}"
        )
        headers = {"Authorization": f"Bearer {token}"}
        try:
            resp = await self._http_client.get(url, headers=headers, timeout=30)
            if resp.status_code != 200:
                logger.warning(f"Feishu media download failed: HTTP {resp.status_code}")
                return None

            # Check attachment size before writing to disk
            cl = resp.headers.get("content-length")
            if cl:
                try:
                    too_large = self._check_attachment_size(int(cl), file_key)
                    if too_large:
                        logger.warning(too_large)
                        return None
                except (ValueError, TypeError):
                    pass
            from ..base import MAX_ATTACHMENT_BYTES

            if len(resp.content) > MAX_ATTACHMENT_BYTES:
                logger.warning(f"Feishu media too large: {len(resp.content)} bytes")
                return None

            # Determine extension from Content-Type or default
            content_type = resp.headers.get("content-type", "")
            ext_map = {
                "image/jpeg": ".jpg",
                "image/png": ".png",
                "image/gif": ".gif",
                "image/webp": ".webp",
                "image/bmp": ".bmp",
            }
            ext = ext_map.get(content_type, ".bin")
            local_path = self._media_path(f"feishu_{message_id}_{file_key}{ext}")
            local_path.write_bytes(resp.content)
            return str(local_path)
        except Exception as e:
            logger.warning(f"Failed to download Feishu media: {e}")
            return None

    async def _upload_feishu_resource(
        self,
        url: str,
        headers: dict,
        file_path: str,
        field_name: str,
        extra_data: dict,
    ) -> dict | None:
        """Upload a file to Feishu API. Returns response data or None on failure."""
        with open(file_path, "rb") as f:
            resp = await self._http_client.post(
                url,
                headers=headers,
                data=extra_data,
                files={field_name: (Path(file_path).name, f)},
            )
        data = resp.json()
        if data.get("code") != 0:
            logger.error(f"Feishu upload failed: {data.get('msg')}")
            return None
        return data["data"]

    async def _send_media_impl(
        self,
        recipient: str,
        file_path: str,
        caption: str = "",
        metadata: dict | None = None,
    ) -> bool:
        """Send a media file through Feishu."""
        token = await self._ensure_token()
        headers = {"Authorization": f"Bearer {token}"}
        chat_id = self._resolve_media_chat_id(recipient, metadata)

        path = Path(file_path)
        ext = path.suffix.lower()
        is_image = ext in self._IMAGE_EXTENSIONS

        send_url = (
            f"{self.config.feishu_domain}"
            f"/open-apis/im/v1/messages?receive_id_type=chat_id"
        )

        if is_image:
            upload_url = f"{self.config.feishu_domain}/open-apis/im/v1/images"
            data = await self._upload_feishu_resource(
                upload_url,
                headers,
                file_path,
                "image",
                {"image_type": "message"},
            )
            if not data:
                return False
            body = {
                "receive_id": chat_id,
                "msg_type": "image",
                "content": json.dumps({"image_key": data["image_key"]}),
            }
        else:
            upload_url = f"{self.config.feishu_domain}/open-apis/im/v1/files"
            data = await self._upload_feishu_resource(
                upload_url,
                headers,
                file_path,
                "file",
                {"file_type": "stream", "file_name": path.name},
            )
            if not data:
                return False
            body = {
                "receive_id": chat_id,
                "msg_type": "file",
                "content": json.dumps({"file_key": data["file_key"]}),
            }

        if not await self._feishu_send(send_url, body, headers):
            return False

        # Send caption as a separate text message if provided
        if caption:
            cap_body = {
                "receive_id": chat_id,
                "msg_type": "text",
                "content": json.dumps({"text": caption}),
            }
            await self._feishu_send(send_url, cap_body, headers)

        return True

    # ── ACK reaction ───────────────────────────────────────────────

    async def _send_ack_reaction(
        self, chat_id: str, message_id: str, emoji: str = "THUMBSUP"
    ) -> None:
        """Send an acknowledgment reaction via Feishu Open API."""
        try:
            token = await self._ensure_token()
            url = f"{self.config.feishu_domain}/open-apis/im/v1/messages/{message_id}/reactions"
            await self._http_client.post(
                url,
                json={"reaction_type": {"emoji_type": emoji}},
                headers={"Authorization": f"Bearer {token}"},
            )
        except Exception as e:
            logger.debug(f"Feishu ack reaction failed: {e}")

    async def _remove_ack_reaction(
        self, chat_id: str, message_id: str, emoji: str = "THUMBSUP"
    ) -> None:
        """Remove ACK reaction via Feishu Open API.

        Feishu's DELETE /reactions endpoint requires the reaction_id, which
        we don't track. No-op for now.
        """
        pass

    # ── Mention stripping ─────────────────────────────────────────

    def _strip_mention(self, text: str) -> str:
        """Strip bot @mention placeholders from Feishu text.

        In Feishu v2 events the text contains placeholders like ``@_user_1``
        for each mention. ``_mention_names`` caches the placeholder keys that
        belong to the bot (identified during ``_on_message``).
        """
        result = text
        for key in self._mention_names:
            result = result.replace(key, "")
        # Clean up extra whitespace left behind
        return re.sub(r"  +", " ", result).strip()

    # ── Event decryption ─────────────────────────────────────────

    def _decrypt_event(self, encrypted: str) -> dict:
        """Decrypt a Feishu encrypted event payload (AES-256-CBC).

        Feishu encryption spec:
          key   = SHA256(encrypt_key)
          data  = base64_decode(encrypted)
          iv    = data[:16]
          plain = AES_CBC_decrypt(data[16:], key, iv)  # PKCS7 padded
        """
        from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

        key = hashlib.sha256(self.config.encrypt_key.encode()).digest()
        data = base64.b64decode(encrypted)
        iv, ciphertext = data[:16], data[16:]
        cipher = Cipher(algorithms.AES(key), modes.CBC(iv))
        decryptor = cipher.decryptor()
        padded = decryptor.update(ciphertext) + decryptor.finalize()
        # Remove PKCS7 padding
        pad_len = padded[-1]
        plaintext = padded[:-pad_len].decode()
        return json.loads(plaintext)

    # ── Webhook event handler ─────────────────────────────────────

    async def _handle_event(self, request) -> web.Response:
        """Handle POST /webhook/event from Feishu."""
        from aiohttp import web

        try:
            body = await request.json()
        except Exception:
            return web.Response(status=400)

        # When encryption is configured the inbound POST MUST carry an
        # ``encrypt`` field. A plaintext body used to skip decryption and
        # reach the agent directly, defeating the encryption setup (issue
        # #392). Treat a missing ``encrypt`` field on an
        # encryption-configured channel as an authentication failure.
        if self.config.encrypt_key:
            if not isinstance(body, dict) or "encrypt" not in body:
                logger.warning(
                    "Feishu event rejected: encrypt_key is configured but the "
                    "body has no 'encrypt' field (possible signature bypass)"
                )
                return web.Response(status=403)
            try:
                body = self._decrypt_event(body["encrypt"])
            except Exception:
                logger.exception("Feishu event decryption failed")
                return web.Response(status=400)

        # ── URL verification challenge ──
        if body.get("type") == "url_verification":
            challenge = body.get("challenge", "")
            return web.json_response({"challenge": challenge})

        # ── v2 event schema ──
        schema = body.get("schema")
        if schema == "2.0":
            header = body.get("header", {})

            # Verify token if configured
            if self.config.verification_token:
                token = header.get("token", "")
                if token != self.config.verification_token:
                    logger.warning("Feishu event token mismatch")
                    return web.Response(status=403)

            event_type = header.get("event_type", "")
            logger.info(f"Feishu v2 event received: {event_type}")
            if event_type == "im.message.receive_v1":
                try:
                    await self._on_message(body.get("event", {}))
                except Exception:
                    logger.exception("Feishu _on_message failed")

        # ── v1 event schema (legacy) ──
        elif "event" in body:
            if self.config.verification_token:
                token = body.get("token", "")
                if token != self.config.verification_token:
                    logger.warning("Feishu event token mismatch (v1)")
                    return web.Response(status=403)

            event = body["event"]
            msg_type = event.get("type", "")
            logger.info(f"Feishu v1 event received: type={msg_type}")
            if msg_type == "message":
                try:
                    await self._on_message_v1(event)
                except Exception:
                    logger.exception("Feishu _on_message_v1 failed")
        else:
            logger.info(f"Feishu event ignored: schema={schema}")

        return web.Response(status=200)

    async def _on_message(self, event: dict) -> None:
        """Handle im.message.receive_v1 event (v2 schema)."""
        sender_info = event.get("sender", {})
        sender_id_info = sender_info.get("sender_id", {})
        sender_id = sender_id_info.get("open_id") or sender_id_info.get("user_id") or ""
        sender_type = sender_info.get("sender_type", "")

        # Skip bot's own messages
        if sender_type == "app":
            return

        message = event.get("message", {})
        chat_id = message.get("chat_id", "")
        msg_type = message.get("message_type", "")
        message_id = message.get("message_id", "")

        # In group chats, detect mention status for centralized gating
        chat_type = message.get("chat_type", "")
        is_group = chat_type == "group"
        was_mentioned = True
        if is_group:
            mentions = message.get("mentions", [])
            was_mentioned = bool(mentions)
            # Cache bot mention keys — bot mentions have empty user IDs
            bot_keys = []
            for m in mentions:
                m_id = m.get("id", {})
                # Bot/app mentions have no open_id / user_id
                if not m_id.get("open_id") and not m_id.get("user_id"):
                    key = m.get("key", "")
                    if key:
                        bot_keys.append(key)
            if bot_keys:
                self._mention_names = bot_keys

        # Parse content JSON
        content_str = message.get("content", "{}")
        try:
            content_data = json.loads(content_str)
        except json.JSONDecodeError:
            content_data = {}

        text = ""
        annotations: list[str] = []
        media_paths: list[str] = []

        if msg_type == "text":
            text = content_data.get("text", "")
        elif msg_type == "post":
            text = self._extract_post_text(content_data)
        elif msg_type == "image" and self.config.include_attachments:
            image_key = content_data.get("image_key", "")
            if image_key:
                local = await self._download_media(message_id, image_key, "image")
                if local:
                    media_paths.append(local)
                    annotations.append(f"[attachment: {local}]")
                else:
                    annotations.append("[image message - download failed]")
            else:
                annotations.append("[image message]")
        elif msg_type == "file" and self.config.include_attachments:
            file_key = content_data.get("file_key", "")
            file_name = content_data.get("file_name", "unknown")
            if file_key:
                local = await self._download_media(message_id, file_key, "file")
                if local:
                    media_paths.append(local)
                    annotations.append(f"[attachment: {local}]")
                else:
                    annotations.append(f"[file: {file_name} - download failed]")
            else:
                annotations.append(f"[file message: {file_name}]")
        elif msg_type in ("audio", "media") and self.config.include_attachments:
            # Feishu audio messages are voice recordings
            media_label = "voice" if msg_type == "audio" else msg_type
            file_key = content_data.get("file_key", "")
            if file_key:
                local = await self._download_media(message_id, file_key, "file")
                if local:
                    media_paths.append(local)
                    annotations.append(f"[{media_label}: {local}]")
                else:
                    annotations.append(f"[{media_label} message - download failed]")
            else:
                annotations.append(f"[{media_label} message]")
        elif msg_type == "sticker":
            sticker_key = content_data.get("file_key", "")
            if sticker_key and self.config.include_attachments:
                local = await self._download_media(message_id, sticker_key, "image")
                if local:
                    media_paths.append(local)
                    annotations.append(f"[sticker: {local}]")
                else:
                    annotations.append("[sticker message]")
            else:
                annotations.append("[sticker message]")
        else:
            text = f"[{msg_type} message]"

        if not text and not media_paths and not annotations:
            return

        # Parse timestamp (milliseconds)
        create_time = message.get("create_time", "")
        try:
            timestamp = (
                datetime.fromtimestamp(int(create_time) / 1000)
                if create_time
                else datetime.now()
            )
        except (ValueError, TypeError, OSError):
            timestamp = datetime.now()

        await self._enqueue_raw(
            RawIncoming(
                sender_id=sender_id,
                chat_id=chat_id,
                text=text,
                media_files=media_paths,
                content_annotations=annotations,
                timestamp=timestamp,
                message_id=message_id,
                metadata={
                    "chat_id": chat_id,
                    "chat_type": message.get("chat_type", ""),
                },
                is_group=is_group,
                was_mentioned=was_mentioned,
            )
        )

    async def _on_message_v1(self, event: dict) -> None:
        """Handle v1 schema message event (legacy)."""
        sender_id = event.get("open_id", "")
        if not sender_id:
            return

        # Detect group and mention status for centralized gating
        chat_type = event.get("chat_type", "")
        is_group = chat_type == "group"
        was_mentioned = True
        if is_group:
            text_without_at = event.get("text_without_at_bot", "")
            was_mentioned = bool(text_without_at)

        text = event.get("text_without_at_bot", "") or event.get("text", "")
        if not text:
            return

        chat_id = event.get("open_chat_id", "")
        message_id = event.get("open_message_id", "")

        await self._enqueue_raw(
            RawIncoming(
                sender_id=sender_id,
                chat_id=chat_id,
                text=text,
                timestamp=datetime.now(),
                message_id=message_id,
                metadata={
                    "chat_id": chat_id,
                    "chat_type": event.get("chat_type", ""),
                },
                is_group=is_group,
                was_mentioned=was_mentioned,
            )
        )

    @staticmethod
    def _extract_post_text(content: dict) -> str:
        """Extract plain text from Feishu post (rich text) content."""
        parts: list[str] = []
        # Post content has locale keys like "zh_cn", "en_us"
        for locale_key in ("zh_cn", "en_us", "ja_jp"):
            locale_content = content.get(locale_key)
            if locale_content:
                title = locale_content.get("title", "")
                if title:
                    parts.append(title)
                for paragraph in locale_content.get("content", []):
                    line_parts: list[str] = []
                    for element in paragraph:
                        tag = element.get("tag", "")
                        if tag == "text":
                            line_parts.append(element.get("text", ""))
                        elif tag == "a":
                            line_parts.append(element.get("text", ""))
                        elif tag == "at":
                            # Skip @mentions of the bot
                            pass
                    line = "".join(line_parts).strip()
                    if line:
                        parts.append(line)
                break  # Use first available locale
        return "\n".join(parts)