add slack and wechat

EvoScientist/channels/__init__.py

@@ -1,7 +1,7 @@
 """Communication channels for EvoScientist.
 
 This module provides an extensible interface for different messaging channels
-(iMessage, Telegram, Discord) to communicate with the EvoScientist agent.
+(iMessage, Telegram, Discord, Slack, WeChat) to communicate with the EvoScientist agent.
 """
 
 from .base import Channel, RawIncoming, IncomingMessage, OutgoingMessage, chunk_text

EvoScientist/channels/capabilities.py

@@ -114,7 +114,7 @@ SLACK = ChannelCapabilities(
     streaming=False,
     threading=True,
     reactions=True,
-    typing=True,
+    typing=False,  # no native typing API; workaround via post+delete "..." message
     media_send=True,
     media_receive=True,
     voice=False,  # no distinct voice message type in Slack bot API

EvoScientist/channels/slack/__init__.py

@@ -0,0 +1,20 @@
+from .channel import SlackChannel, SlackConfig
+from ..channel_manager import register_channel, _parse_csv
+
+__all__ = ["SlackChannel", "SlackConfig"]
+
+
+def create_from_config(config) -> SlackChannel:
+    allowed = _parse_csv(config.slack_allowed_senders)
+    channels = _parse_csv(config.slack_allowed_channels)
+    proxy = config.slack_proxy if config.slack_proxy else None
+    return SlackChannel(SlackConfig(
+        bot_token=config.slack_bot_token,
+        app_token=config.slack_app_token,
+        allowed_senders=allowed,
+        allowed_channels=channels,
+        proxy=proxy,
+    ))
+
+
+register_channel("slack", create_from_config)

EvoScientist/channels/slack/channel.py

@@ -0,0 +1,291 @@
+"""Slack channel implementation using slack-sdk Socket Mode."""
+
+import asyncio
+import logging
+from dataclasses import dataclass
+from datetime import datetime
+
+from ..base import Channel, RawIncoming, ChannelError
+from ..capabilities import SLACK as SLACK_CAPS
+from ..config import BaseChannelConfig
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class SlackConfig(BaseChannelConfig):
+    bot_token: str = ""
+    app_token: str = ""
+    text_chunk_limit: int = 4096
+
+
+class SlackChannel(Channel):
+    """Slack channel using slack-sdk Socket Mode."""
+
+    name = "slack"
+
+    capabilities = SLACK_CAPS
+    _ready_attrs = ("_web_client",)
+    _mention_pattern = r"<@{bot_id}>\s*"
+
+    def __init__(self, config: SlackConfig):
+        super().__init__(config)
+        self._socket_client = None
+        self._web_client = None
+        self._typing_message_ts: dict[str, str] = {}
+
+    async def start(self) -> None:
+        try:
+            from slack_sdk.web.async_client import AsyncWebClient
+            from slack_sdk.socket_mode.aiohttp import SocketModeClient
+            from slack_sdk.socket_mode.request import SocketModeRequest
+            from slack_sdk.socket_mode.response import SocketModeResponse
+        except ImportError:
+            raise ChannelError(
+                "slack-sdk or aiohttp not installed. "
+                "Install with: pip install evoscientist[slack]"
+            )
+
+        if not self.config.bot_token:
+            raise ChannelError("Slack bot token is required")
+        if not self.config.app_token:
+            raise ChannelError(
+                "Slack app token is required for Socket Mode "
+                "(starts with xapp-)"
+            )
+
+        self._web_client = AsyncWebClient(
+            token=self.config.bot_token,
+            proxy=self._get_proxy(),
+        )
+
+        # Get bot user ID for filtering own messages
+        try:
+            auth = await asyncio.wait_for(
+                self._web_client.auth_test(), timeout=15,
+            )
+            self._bot_user_id = auth["user_id"]
+        except asyncio.TimeoutError:
+            raise ChannelError(
+                "Slack auth_test timed out — check network and bot token"
+            )
+        except Exception as e:
+            raise ChannelError(f"Failed to authenticate Slack bot: {e}")
+
+        self._socket_client = SocketModeClient(
+            app_token=self.config.app_token,
+            web_client=self._web_client,
+        )
+
+        async def _event_handler(
+            client: SocketModeClient,
+            req: SocketModeRequest,
+        ) -> None:
+            # Acknowledge immediately
+            resp = SocketModeResponse(envelope_id=req.envelope_id)
+            await client.send_socket_mode_response(resp)
+
+            logger.debug(f"Slack socket event: type={req.type}")
+
+            if req.type == "events_api":
+                event = req.payload.get("event", {})
+                event_type = event.get("type", "")
+                if event_type == "message" and "subtype" not in event:
+                    is_dm = event.get("channel_type") == "im"
+                    await self._on_message(
+                        event, is_group=not is_dm, was_mentioned=is_dm,
+                    )
+                elif event_type == "app_mention":
+                    await self._on_message(
+                        event, is_group=True, was_mentioned=True,
+                    )
+
+        self._socket_client.socket_mode_request_listeners.append(
+            _event_handler
+        )
+        try:
+            await asyncio.wait_for(
+                self._socket_client.connect(), timeout=30,
+            )
+        except asyncio.TimeoutError:
+            raise ChannelError(
+                "Slack Socket Mode connection timed out — "
+                "check app token (must start with xapp-) and "
+                "ensure Socket Mode is enabled in your Slack app settings"
+            )
+        self._running = True
+        logger.info("Slack channel started (Socket Mode)")
+
+    async def _cleanup(self) -> None:
+        if self._socket_client:
+            await self._socket_client.close()
+            logger.info("Slack channel stopped")
+
+    # ── Typing indicator (override base) ────────────────────────────
+
+    async def _send_typing_action(self, chat_id: str) -> None:
+        """Send typing indicator via Slack.
+
+        Slack's Web API and Socket Mode do not expose a dedicated
+        typing-indicator endpoint for bot tokens. We approximate
+        the experience by posting a short-lived status message that
+        is deleted once the real reply is sent (handled by
+        ``stop_typing``).  When the status post fails we silently
+        fall back to no indicator.
+        """
+        if not self._web_client:
+            return
+        try:
+            resp = await self._web_client.chat_postMessage(
+                channel=chat_id,
+                text="\u2026",  # "…" ellipsis as minimal typing hint
+            )
+            ts = resp.get("ts")
+            if ts:
+                self._typing_message_ts[chat_id] = ts
+        except Exception:
+            pass
+
+    async def stop_typing(self, chat_id: str) -> None:
+        """Cancel typing loop and clean up the status message."""
+        # Delete the ephemeral "…" message if we posted one
+        ts = self._typing_message_ts.pop(chat_id, None)
+        if ts and self._web_client:
+            try:
+                await self._web_client.chat_delete(channel=chat_id, ts=ts)
+            except Exception:
+                pass
+        await super().stop_typing(chat_id)
+
+    # ── Send (template method overrides) ──────────────────────────
+
+
+    async def _send_chunk(self, chat_id, formatted_text, raw_text, reply_to, metadata):
+        kwargs = dict(channel=chat_id)
+        # Always route to thread if thread_ts is present in metadata,
+        # not just for the first chunk (reply_to is only set for chunk 0).
+        if metadata:
+            thread_ts = metadata.get("thread_ts")
+            if thread_ts:
+                kwargs["thread_ts"] = thread_ts
+
+        async def _send(text):
+            await self._web_client.chat_postMessage(text=text, **kwargs)
+
+        await self._send_with_format_fallback(_send, formatted_text, raw_text)
+
+    async def _send_media_impl(
+        self,
+        recipient: str,
+        file_path: str,
+        caption: str = "",
+        metadata: dict | None = None,
+    ) -> bool:
+        """Send a media file through Slack."""
+        channel_id = self._resolve_media_chat_id(recipient, metadata)
+        await self._web_client.files_upload_v2(
+            channel=channel_id,
+            file=file_path,
+            initial_comment=caption or None,
+        )
+        return True
+
+    def _get_bot_identifier(self) -> str | None:
+        return getattr(self, "_bot_user_id", None)
+
+    # ── ACK Reactions ───────────────────────────────────────────────
+
+    async def _send_ack_reaction(self, chat_id: str, message_id: str, emoji: str = "eyes") -> None:
+        """Add an emoji reaction to acknowledge receipt."""
+        if self._web_client and message_id:
+            try:
+                await self._web_client.reactions_add(
+                    channel=chat_id, timestamp=message_id, name=emoji,
+                )
+            except Exception as e:
+                logger.debug(f"Slack ACK reaction failed: {e}")
+
+    async def _remove_ack_reaction(self, chat_id: str, message_id: str, emoji: str = "eyes") -> None:
+        """Remove the ACK reaction after replying."""
+        if self._web_client and message_id:
+            try:
+                await self._web_client.reactions_remove(
+                    channel=chat_id, timestamp=message_id, name=emoji,
+                )
+            except Exception as e:
+                logger.debug(f"Slack remove ACK reaction failed: {e}")
+
+    async def _on_message(
+        self,
+        event: dict,
+        *,
+        is_group: bool = False,
+        was_mentioned: bool = True,
+    ) -> None:
+        """Handle an incoming Slack message event."""
+        user_id = event.get("user", "")
+
+        # Skip bot's own messages
+        if user_id == getattr(self, "_bot_user_id", None):
+            logger.debug("Skipping own bot message")
+            return
+
+        # Skip bot messages (e.g. from other bots)
+        if event.get("bot_id"):
+            logger.debug(f"Skipping bot message from bot_id={event.get('bot_id')}")
+            return
+
+        channel_id = event.get("channel", "")
+
+        text = event.get("text", "")
+
+        annotations: list[str] = []
+        media_paths: list[str] = []
+
+        # Handle file attachments
+        if self.config.include_attachments:
+            files = event.get("files", [])
+            for file_info in files:
+                file_size = file_info.get("size", 0)
+                filename = file_info.get("name", "unknown")
+
+                url = file_info.get("url_private_download") or file_info.get(
+                    "url_private"
+                )
+                if url and self._web_client:
+                    headers = {
+                        "Authorization": f"Bearer {self.config.bot_token}"
+                    }
+                    local_path, annotation = await self._download_attachment(
+                        url, f"{file_info.get('id', 'unknown')}_{filename}",
+                        headers=headers,
+                        file_size=file_size,
+                    )
+                    if local_path:
+                        media_paths.append(local_path)
+                    if annotation:
+                        annotations.append(annotation)
+
+        ts = event.get("ts", "")
+        thread_ts = event.get("thread_ts") or ts
+        try:
+            timestamp = datetime.fromtimestamp(float(ts)) if ts else datetime.now()
+        except (ValueError, TypeError):
+            timestamp = datetime.now()
+
+        await self._enqueue_raw(RawIncoming(
+            sender_id=user_id,
+            chat_id=channel_id,
+            text=text,
+            media_files=media_paths,
+            content_annotations=annotations,
+            timestamp=timestamp,
+            message_id=ts,
+            metadata={"chat_id": channel_id, "thread_ts": thread_ts},
+            is_group=is_group,
+            was_mentioned=was_mentioned,
+        ))
+        logger.info(
+            f"Slack message queued: sender={user_id}, "
+            f"channel={channel_id}, content={text[:50]}"
+        )

EvoScientist/channels/slack/probe.py

@@ -0,0 +1,48 @@
+"""Slack bot token validation."""
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+async def validate_slack_tokens(
+    bot_token: str,
+    app_token: str | None = None,
+    proxy: str | None = None,
+) -> tuple[bool, str]:
+    """Validate Slack bot token via the auth.test API.
+
+    Optionally checks the app-level token format (must start with ``xapp-``).
+
+    Returns:
+        Tuple of (is_valid, message).
+    """
+    if not bot_token:
+        return False, "No bot token provided"
+
+    try:
+        import httpx
+    except ImportError:
+        return False, "httpx not installed"
+
+    # Validate bot token via auth.test
+    url = "https://slack.com/api/auth.test"
+    headers = {"Authorization": f"Bearer {bot_token}"}
+    try:
+        async with httpx.AsyncClient(proxy=proxy) as client:
+            resp = await client.post(url, headers=headers, timeout=10)
+        data = resp.json()
+        if not data.get("ok"):
+            error = data.get("error", "unknown error")
+            return False, f"Invalid bot token: {error}"
+        bot_name = data.get("user", "unknown")
+        team = data.get("team", "unknown")
+    except Exception as e:
+        return False, f"Error: {e}"
+
+    # Optionally validate app token format
+    if app_token:
+        if not app_token.startswith("xapp-"):
+            return False, "App token must start with 'xapp-'"
+
+    return True, f"Bot: {bot_name} (team: {team})"

EvoScientist/channels/slack/serve.py

@@ -0,0 +1,99 @@
+"""Slack channel server.
+
+Standalone script to run the Slack channel with CLI options.
+
+Usage:
+    python -m EvoScientist.channels.slack.serve --bot-token TOKEN --app-token TOKEN [OPTIONS]
+
+Examples:
+    # Allow all senders (default)
+    python -m EvoScientist.channels.slack.serve --bot-token xoxb-... --app-token xapp-...
+
+    # Only allow specific senders and channels
+    python -m EvoScientist.channels.slack.serve --bot-token xoxb-... --app-token xapp-... --allow U123 --allow-channel C456
+
+    # With proxy, agent and thinking
+    python -m EvoScientist.channels.slack.serve --bot-token xoxb-... --app-token xapp-... --proxy http://proxy:8080 --agent --thinking
+"""
+
+import argparse
+import logging
+
+from .channel import SlackChannel, SlackConfig
+from ..bus import MessageBus
+from ..standalone import run_standalone
+
+logging.basicConfig(
+    level=logging.DEBUG,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    datefmt="%H:%M:%S",
+)
+logger = logging.getLogger(__name__)
+
+
+def parse_args():
+    """Parse command line arguments."""
+    parser = argparse.ArgumentParser(
+        description="Slack channel server",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument(
+        "--bot-token",
+        required=True,
+        help="Slack bot token (xoxb-...)",
+    )
+    parser.add_argument(
+        "--app-token",
+        required=True,
+        help="Slack app-level token for Socket Mode (xapp-...)",
+    )
+    parser.add_argument(
+        "--allow",
+        action="append",
+        dest="allowed_senders",
+        help="Allowed sender (Slack user ID). Can be used multiple times.",
+    )
+    parser.add_argument(
+        "--allow-channel",
+        action="append",
+        dest="allowed_channels",
+        help="Allowed channel ID. Can be used multiple times.",
+    )
+    parser.add_argument(
+        "--proxy",
+        help="HTTP proxy URL for Slack API requests",
+    )
+    parser.add_argument(
+        "--agent",
+        action="store_true",
+        help="Use EvoScientist agent as handler (default: echo)",
+    )
+    parser.add_argument(
+        "--thinking",
+        action="store_true",
+        help="Send thinking content as intermediate messages (requires --agent)",
+    )
+    return parser.parse_args()
+
+
+def main():
+    """Entry point."""
+    args = parse_args()
+
+    config = SlackConfig(
+        bot_token=args.bot_token,
+        app_token=args.app_token,
+        allowed_senders=set(args.allowed_senders) if args.allowed_senders else None,
+        allowed_channels=set(args.allowed_channels) if args.allowed_channels else None,
+        proxy=args.proxy,
+    )
+
+    send_thinking = args.thinking and args.agent
+    bus = MessageBus()
+    channel = SlackChannel(config)
+
+    run_standalone(channel, bus, use_agent=args.agent, send_thinking=send_thinking)
+
+
+if __name__ == "__main__":
+    main()

EvoScientist/channels/wechat/__init__.py

@@ -0,0 +1,69 @@
+"""WeChat channel implementations for EvoScientist.
+
+Supports multiple WeChat backends:
+  - **wecom**: 企业微信应用 (WeCom / WeChat Work) via official API
+    — Most stable, pure HTTP, no third-party dependencies
+  - **wechatmp**: 微信公众号 (WeChat Official Account) via official API
+    — Pure HTTP webhook, suitable for public-facing bots
+
+Both backends use httpx (already a core dependency) and receive messages
+via HTTP webhook, send replies via REST API.
+
+Usage in config:
+    channel_enabled = "wechat"
+    wechat_backend = "wecom"       # or "wechatmp"
+
+    # WeCom settings
+    wechat_wecom_corp_id = "..."
+    wechat_wecom_agent_id = "..."
+    wechat_wecom_secret = "..."
+    wechat_wecom_token = "..."
+    wechat_wecom_encoding_aes_key = "..."
+    wechat_webhook_port = 9001
+
+    # OR: Official Account settings
+    wechat_mp_app_id = "..."
+    wechat_mp_app_secret = "..."
+    wechat_mp_token = "..."
+    wechat_mp_encoding_aes_key = "..."
+    wechat_webhook_port = 9001
+"""
+
+from .channel import WeChatChannel, WeComConfig, WeChatMPConfig
+from ..channel_manager import register_channel, _parse_csv
+
+__all__ = ["WeChatChannel", "WeComConfig", "WeChatMPConfig"]
+
+
+def create_from_config(config) -> WeChatChannel:
+    backend = config.wechat_backend or "wecom"
+    allowed = _parse_csv(config.wechat_allowed_senders)
+    proxy = config.wechat_proxy if config.wechat_proxy else None
+    port = int(config.wechat_webhook_port or 9001)
+
+    if backend == "wechatmp":
+        mp_config = WeChatMPConfig(
+            app_id=config.wechat_mp_app_id,
+            app_secret=config.wechat_mp_app_secret,
+            token=config.wechat_mp_token,
+            encoding_aes_key=config.wechat_mp_encoding_aes_key,
+            webhook_port=port,
+            allowed_senders=allowed,
+            proxy=proxy,
+        )
+        return WeChatChannel(mp_config, backend="wechatmp")
+    else:
+        wecom_config = WeComConfig(
+            corp_id=config.wechat_wecom_corp_id,
+            agent_id=config.wechat_wecom_agent_id,
+            secret=config.wechat_wecom_secret,
+            token=config.wechat_wecom_token,
+            encoding_aes_key=config.wechat_wecom_encoding_aes_key,
+            webhook_port=port,
+            allowed_senders=allowed,
+            proxy=proxy,
+        )
+        return WeChatChannel(wecom_config, backend="wecom")
+
+
+register_channel("wechat", create_from_config)

EvoScientist/channels/wechat/channel.py

@@ -0,0 +1,805 @@
+"""WeChat channel implementation.
+
+Supports two backends via a unified Channel interface:
+
+1. **wecom** (企业微信应用): Corporate WeChat official API
+   - Receives messages via HTTP callback (XML + optional AES encryption)
+   - Sends replies via REST API (POST /cgi-bin/message/send)
+   - Supports text, image, file, markdown messages
+   - Token auto-refresh with 2-hour TTL
+
+2. **wechatmp** (微信公众号): WeChat Official Account API
+   - Receives messages via HTTP callback (XML + optional AES encryption)
+   - Sends replies via REST API (POST /cgi-bin/message/custom/send)
+   - Supports text, image, news messages
+
+Both backends use httpx (already a core dependency) and aiohttp for
+webhook server — matching the Feishu channel pattern.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import logging
+import re
+import time
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from aiohttp import web
+
+from ..mixins import WebhookMixin, TokenMixin
+from ..base import Channel, RawIncoming, ChannelError
+from ..capabilities import WECHAT as WECHAT_CAPS
+from ..config import BaseChannelConfig
+
+logger = logging.getLogger(__name__)
+
+
+# ── Markdown → plain text (fallback for WeChat text messages) ────
+
+def _strip_markdown(text: str) -> str:
+    """Strip Markdown formatting for plain-text WeChat messages."""
+    # Remove code blocks
+    text = re.sub(r"```[\s\S]*?```", lambda m: m.group(0).strip("`").strip(), text)
+    # Remove inline code
+    text = re.sub(r"`([^`]+)`", r"\1", text)
+    # Remove bold
+    text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)
+    # Remove italic
+    text = re.sub(r"(?<!\w)_([^_]+?)_(?!\w)", r"\1", text)
+    # Remove strikethrough
+    text = re.sub(r"~~(.+?)~~", r"\1", text)
+    # Convert links
+    text = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", r"\1(\2)", text)
+    # Remove heading markers
+    text = re.sub(r"^#{1,6}\s+", "", text, flags=re.MULTILINE)
+    # Convert list items
+    text = re.sub(r"^[\-\*]\s+", "• ", text, flags=re.MULTILINE)
+    return text
+
+
+# ── Config dataclasses ───────────────────────────────────────────
+
+@dataclass
+class WeComConfig(BaseChannelConfig):
+    """Configuration for WeCom (企业微信) backend."""
+    corp_id: str = ""
+    agent_id: str = ""
+    secret: str = ""
+    token: str = ""
+    encoding_aes_key: str = ""
+    webhook_port: int = 9001
+
+
+@dataclass
+class WeChatMPConfig(BaseChannelConfig):
+    """Configuration for WeChat Official Account (公众号) backend."""
+    app_id: str = ""
+    app_secret: str = ""
+    token: str = ""
+    encoding_aes_key: str = ""
+    webhook_port: int = 9001
+
+
+# ── Unified WeChat Channel ───────────────────────────────────────
+
+class WeChatChannel(Channel, WebhookMixin, TokenMixin):
+    capabilities = WECHAT_CAPS
+    """Unified WeChat channel supporting WeCom and Official Account backends.
+
+    Architecture follows the same pattern as FeishuChannel:
+    - HTTP webhook server (aiohttp) for inbound messages
+    - REST API calls (httpx) for outbound messages
+    - Token auto-refresh
+    """
+
+    name = "wechat"
+    _typing_interval: float = 5.0  # WeChat has no typing API, but keep for interface
+    _ready_attrs = ("_http_client", "_access_token")
+    _rate_limit_patterns = ("45009", "frequency", "freq")
+    _rate_limit_delay = 2.0
+    _mention_pattern = r"@\S+\s*"
+    _mention_strip_count = 1
+
+    def __init__(
+        self,
+        config: WeComConfig | WeChatMPConfig,
+        backend: str = "wecom",
+    ):
+        super().__init__(config)
+        self._backend = backend
+        self._access_token: str | None = None
+        self._token_expires: float = 0
+        self._runner = None
+        self._site = None
+        self._http_client = None
+        self._crypto = None  # WeChatCrypto instance (optional)
+
+    # ── Lifecycle ─────────────────────────────────────────────────
+
+    def _webhook_routes(self) -> list[tuple[str, str, Any]]:
+        """Return HTTP routes for the shared webhook server."""
+        return [
+            ("GET", "/wechat/callback", self._handle_verify),
+            ("POST", "/wechat/callback", self._handle_message),
+        ]
+
+    async def start(self) -> None:
+        try:
+            from aiohttp import web
+            import httpx  # noqa: F401
+        except ImportError:
+            raise ChannelError(
+                "aiohttp or httpx not installed. "
+                "Install with: pip install aiohttp httpx"
+            )
+
+        self._validate_config()
+
+        import httpx
+        self._http_client = httpx.AsyncClient(
+            timeout=15,
+            proxy=self._get_proxy(),
+        )
+
+        # Set up message encryption if configured
+        if self.config.encoding_aes_key and self.config.token:
+            from .crypto import WeChatCrypto
+            app_id = self._get_app_id()
+            self._crypto = WeChatCrypto(
+                token=self.config.token,
+                encoding_aes_key=self.config.encoding_aes_key,
+                app_id=app_id,
+            )
+
+        # Verify credentials by fetching initial token
+        await self._refresh_token()
+
+        if not getattr(self, "_shared_webhook_server", None):
+            app = web.Application()
+            app.router.add_get("/wechat/callback", self._handle_verify)
+            app.router.add_post("/wechat/callback", self._handle_message)
+
+            self._runner = web.AppRunner(app)
+            await self._runner.setup()
+            self._site = web.TCPSite(
+                self._runner, "0.0.0.0", self.config.webhook_port,
+            )
+            await self._site.start()
+
+        self._running = True
+        logger.info(
+            f"WeChat channel started "
+            f"(backend={self._backend}, "
+            f"webhook on port {self.config.webhook_port})"
+        )
+
+    async def _cleanup(self) -> None:
+        if self._site:
+            await self._site.stop()
+        if self._runner:
+            await self._runner.cleanup()
+        if self._http_client:
+            await self._http_client.aclose()
+            self._http_client = None
+        self._access_token = None
+        logger.info("WeChat channel stopped")
+
+    def _validate_config(self) -> None:
+        """Validate required config fields based on backend."""
+        if self._backend == "wecom":
+            cfg = self.config
+            if not cfg.corp_id:
+                raise ChannelError("WeCom corp_id is required")
+            if not cfg.secret:
+                raise ChannelError("WeCom secret is required")
+            if not cfg.agent_id:
+                raise ChannelError("WeCom agent_id is required")
+        elif self._backend == "wechatmp":
+            cfg = self.config
+            if not cfg.app_id:
+                raise ChannelError("WeChat MP app_id is required")
+            if not cfg.app_secret:
+                raise ChannelError("WeChat MP app_secret is required")
+
+    def _get_app_id(self) -> str:
+        """Return the app identifier for crypto operations."""
+        if self._backend == "wecom":
+            return self.config.corp_id
+        return self.config.app_id
+
+    # ── Token management ──────────────────────────────────────────
+
+    async def _refresh_token(self) -> None:
+        """Fetch or refresh the access_token."""
+        if self._backend == "wecom":
+            url = (
+                f"https://qyapi.weixin.qq.com/cgi-bin/gettoken"
+                f"?corpid={self.config.corp_id}"
+                f"&corpsecret={self.config.secret}"
+            )
+        else:
+            url = (
+                f"https://api.weixin.qq.com/cgi-bin/token"
+                f"?grant_type=client_credential"
+                f"&appid={self.config.app_id}"
+                f"&secret={self.config.app_secret}"
+            )
+
+        try:
+            resp = await self._http_client.get(url)
+            data = resp.json()
+        except Exception as e:
+            if not self._running:
+                raise ChannelError(f"Failed to get WeChat access token: {e}")
+            raise RuntimeError(f"Failed to get WeChat access token: {e}")
+
+        if data.get("errcode", 0) != 0:
+            err_msg = (
+                f"WeChat auth error ({data.get('errcode')}): "
+                f"{data.get('errmsg', 'unknown')}"
+            )
+            if not self._running:
+                raise ChannelError(err_msg)
+            raise RuntimeError(err_msg)
+
+        self._access_token = data["access_token"]
+        expire = data.get("expires_in", 7200)
+        # Refresh 5 minutes before expiry
+        self._token_expires = time.monotonic() + expire - 300
+        logger.debug(f"WeChat token refreshed, expires in {expire}s")
+
+    async def _ensure_token(self) -> str:
+        """Return a valid access token, refreshing if needed."""
+        if not self._access_token or time.monotonic() >= self._token_expires:
+            await self._refresh_token()
+        return self._access_token
+
+    # ── Signature verification (GET callback) ─────────────────────
+
+    async def _handle_verify(self, request) -> "web.Response":
+        """Handle GET /wechat/callback for URL verification.
+
+        WeChat/WeCom sends: msg_signature, timestamp, nonce, echostr
+        We decrypt echostr (encrypted mode) or verify signature (plain mode)
+        and return the plain echostr.
+        """
+        from aiohttp import web
+
+        signature = request.query.get("msg_signature") or request.query.get("signature", "")
+        timestamp = request.query.get("timestamp", "")
+        nonce = request.query.get("nonce", "")
+        echostr = request.query.get("echostr", "")
+
+        logger.info(f"Verify request received: timestamp={timestamp}")
+
+        if not echostr:
+            return web.Response(status=400, text="missing echostr")
+
+        # Encrypted mode: WeCom sends msg_signature and encrypted echostr
+        if self._crypto and request.query.get("msg_signature"):
+            # Verify signature first
+            sig_ok = self._crypto.verify_signature(signature, timestamp, nonce, echostr)
+            if not sig_ok:
+                logger.warning("WeChat verify: signature mismatch")
+            # Try to decrypt regardless — the decrypted echostr must be returned
+            try:
+                plain_echostr, _ = self._crypto.decrypt(echostr)
+                logger.info("WeChat verify: echostr decrypted successfully")
+                return web.Response(text=plain_echostr)
+            except Exception as e:
+                logger.error(f"WeChat verify: echostr decrypt failed: {e}")
+                return web.Response(status=500)
+        else:
+            # Plain mode verification
+            token = self.config.token
+            if token:
+                parts = sorted([token, timestamp, nonce])
+                expected = hashlib.sha1("".join(parts).encode()).hexdigest()
+                if expected != signature:
+                    logger.warning("WeChat verify: signature mismatch (plain)")
+                    return web.Response(status=403)
+            return web.Response(text=echostr)
+
+    # ── Inbound message handling (POST callback) ──────────────────
+
+    async def _handle_message(self, request) -> "web.Response":
+        """Handle POST /wechat/callback for incoming messages."""
+        from aiohttp import web
+        from .crypto import parse_xml
+
+        try:
+            body = await request.text()
+        except Exception:
+            return web.Response(status=400)
+
+        logger.info(f"WeChat callback POST received, body length={len(body)}")
+        xml_data = parse_xml(body)
+
+        # If encrypted, decrypt first
+        encrypt = xml_data.get("Encrypt", "")
+        if encrypt and self._crypto:
+            signature = request.query.get("msg_signature", "")
+            timestamp = request.query.get("timestamp", "")
+            nonce = request.query.get("nonce", "")
+
+            if not self._crypto.verify_signature(signature, timestamp, nonce, encrypt):
+                logger.warning("WeChat message signature mismatch")
+                return web.Response(status=403)
+
+            try:
+                decrypted_xml, from_id = self._crypto.decrypt(encrypt)
+                xml_data = parse_xml(decrypted_xml)
+            except Exception as e:
+                logger.error(f"WeChat decrypt failed: {e}")
+                return web.Response(status=500)
+
+        # Process message asynchronously — WeCom requires a response within
+        # 5 seconds, but media downloads can take much longer.  Return
+        # "success" immediately and handle the message in the background.
+        asyncio.create_task(self._safe_process_message(xml_data))
+
+        return web.Response(text="success")
+
+    async def _safe_process_message(self, xml_data: dict[str, str]) -> None:
+        """Wrapper that catches exceptions so fire-and-forget tasks don't leak."""
+        try:
+            await self._process_message(xml_data)
+        except Exception:
+            logger.exception("Error processing WeChat message")
+
+    async def _process_message(self, xml_data: dict[str, str]) -> None:
+        """Process a parsed XML message from WeChat/WeCom callback."""
+        msg_type = xml_data.get("MsgType", "")
+        from_user = xml_data.get("FromUserName", "")
+        to_user = xml_data.get("ToUserName", "")
+        content = xml_data.get("Content", "")
+        msg_id = xml_data.get("MsgId", "")
+        create_time = xml_data.get("CreateTime", "")
+
+        logger.info(f"WeChat message received: type={msg_type}, from={from_user}, id={msg_id}, keys={list(xml_data.keys())}")
+
+        if not from_user:
+            return
+
+        # Determine chat_id
+        # For WeCom: FromUserName is the user's UserID
+        # For MP: FromUserName is the user's OpenID
+        chat_id = from_user
+
+        # Group chat detection
+        is_group = False
+        was_mentioned = True  # Default: treat as mentioned (DMs)
+
+        # WeCom group detection: ChatId field indicates a group message
+        if self._backend == "wecom":
+            group_chat_id = xml_data.get("ChatId", "")
+            if group_chat_id:
+                is_group = True
+                chat_id = group_chat_id
+                # WeCom sets MsgType=event with Event=sys when bot is @mentioned,
+                # but for text messages we check the XML AtUserList field
+                at_user_list = xml_data.get("AtUserList", "")
+                was_mentioned = bool(at_user_list)
+
+        # Handle different message types
+        text = ""
+        annotations: list[str] = []
+        media_paths: list[str] = []
+
+        if msg_type == "text":
+            text = content
+        elif msg_type == "image":
+            pic_url = xml_data.get("PicUrl", "")
+            media_id = xml_data.get("MediaId", "")
+            if pic_url:
+                local, ann = await self._download_attachment(
+                    pic_url, f"wechat_{msg_id}.jpg",
+                )
+                if local:
+                    media_paths.append(local)
+                if ann:
+                    annotations.append(ann)
+            elif media_id:
+                local, ann = await self._download_wechat_media(
+                    media_id, f"wechat_image_{msg_id}",
+                )
+                if local:
+                    media_paths.append(local)
+                if ann:
+                    annotations.append(ann)
+            else:
+                annotations.append("[image: no download source]")
+        elif msg_type == "voice":
+            recognition = xml_data.get("Recognition", "")
+            media_id = xml_data.get("MediaId", "")
+            if media_id:
+                local, ann = await self._download_wechat_media(media_id, f"wechat_voice_{msg_id}")
+                if local:
+                    media_paths.append(local)
+                if ann:
+                    ann = ann.replace("[attachment:", "[voice:")
+                    annotations.append(ann)
+            if recognition:
+                text = f"[语音识别] {recognition}"
+            elif not media_paths:
+                annotations.append("[voice message]")
+        elif msg_type in ("video", "shortvideo"):
+            media_id = xml_data.get("MediaId", "")
+            if media_id:
+                local, ann = await self._download_wechat_media(media_id, f"wechat_{msg_type}_{msg_id}")
+                if local:
+                    media_paths.append(local)
+                if ann:
+                    annotations.append(ann)
+            if not media_paths:
+                annotations.append(f"[{msg_type} message]")
+        elif msg_type == "location":
+            label = xml_data.get("Label", "")
+            lat = xml_data.get("Location_X", "")
+            lon = xml_data.get("Location_Y", "")
+            text = f"[位置] {label} ({lat}, {lon})"
+        elif msg_type == "file":
+            media_id = xml_data.get("MediaId", "")
+            file_name = xml_data.get("FileName", "") or xml_data.get("Title", f"wechat_file_{msg_id}")
+            logger.info(f"WeChat file message: name={file_name}, media_id={media_id!r}, keys={list(xml_data.keys())}")
+            if media_id:
+                local, ann = await self._download_wechat_media(
+                    media_id, f"wechat_file_{msg_id}_{file_name}",
+                )
+                logger.info(f"WeChat file download result: local={local}, ann={ann}")
+                if local:
+                    media_paths.append(local)
+                if ann:
+                    annotations.append(ann)
+            if not media_paths:
+                annotations.append(f"[file: {file_name}]")
+        elif msg_type == "link":
+            title = xml_data.get("Title", "")
+            description = xml_data.get("Description", "")
+            url = xml_data.get("Url", "")
+            text = f"[链接] {title}\n{description}\n{url}"
+        elif msg_type == "event":
+            event_type = xml_data.get("Event", "")
+            if event_type == "subscribe":
+                text = "[用户关注]"
+            elif event_type == "unsubscribe":
+                logger.info(f"User {from_user} unsubscribed")
+                return  # Don't process
+            elif event_type == "CLICK":
+                event_key = xml_data.get("EventKey", "")
+                text = f"[菜单点击] {event_key}"
+            elif event_type in ("LOCATION", "VIEW"):
+                # Periodic location reports and menu-link clicks — ignore
+                return
+            else:
+                logger.debug(f"Ignoring WeChat event: {event_type}")
+                return
+        else:
+            text = f"[{msg_type} message]"
+
+        if not text and not media_paths and not annotations:
+            return
+
+        # Parse timestamp
+        try:
+            timestamp = datetime.fromtimestamp(
+                int(create_time)
+            ) if create_time else datetime.now()
+        except (ValueError, TypeError, OSError):
+            timestamp = datetime.now()
+
+        await self._enqueue_raw(RawIncoming(
+            sender_id=from_user,
+            chat_id=chat_id,
+            text=text,
+            media_files=media_paths,
+            content_annotations=annotations,
+            timestamp=timestamp,
+            message_id=msg_id,
+            is_group=is_group,
+            was_mentioned=was_mentioned,
+            metadata={
+                "chat_id": chat_id,
+                "to_user": to_user,
+                "backend": self._backend,
+            },
+        ))
+
+    # ── Send (template method overrides) ──────────────────────────
+
+    def _format_chunk(self, text: str) -> str:
+        """WeCom uses markdown formatter; MP uses plain text."""
+        if self._backend == "wecom":
+            return self._formatter.format(text)  # markdown profile
+        return _strip_markdown(text)
+
+    async def _send_chunk(
+        self, chat_id, formatted_text, raw_text, reply_to, metadata,
+    ):
+        token = await self._ensure_token()
+
+        if self._backend == "wecom":
+            # Group chat: use appchat/send endpoint
+            if chat_id.startswith("wr"):
+                try:
+                    await self._wecom_send_group_markdown(token, chat_id, raw_text)
+                    return
+                except Exception:
+                    pass
+                await self._wecom_send_group_text(token, chat_id, raw_text)
+            else:
+                # DM: Try markdown first, fall back to plain text
+                try:
+                    await self._wecom_send_markdown(token, chat_id, raw_text)
+                    return
+                except Exception:
+                    pass
+                await self._wecom_send_text(token, chat_id, raw_text)
+        else:
+            await self._mp_send_text(token, chat_id, raw_text)
+
+    # ── WeCom send ────────────────────────────────────────────────
+
+    async def _wecom_send_text(
+        self, token: str, user_id: str, text: str,
+    ) -> None:
+        """Send a text message via WeCom API."""
+        url = (
+            f"https://qyapi.weixin.qq.com/cgi-bin/message/send"
+            f"?access_token={token}"
+        )
+        body = {
+            "touser": user_id,
+            "msgtype": "text",
+            "agentid": int(self.config.agent_id),
+            "text": {"content": _strip_markdown(text)},
+        }
+        await self._post_api(url, body)
+
+    async def _wecom_send_markdown(
+        self, token: str, user_id: str, text: str,
+    ) -> None:
+        """Send a markdown message via WeCom API.
+
+        Note: WeCom markdown only supports a subset of Markdown
+        (no code blocks, no images). Falls back to text if the
+        message is too complex.
+        """
+        url = (
+            f"https://qyapi.weixin.qq.com/cgi-bin/message/send"
+            f"?access_token={token}"
+        )
+        body = {
+            "touser": user_id,
+            "msgtype": "markdown",
+            "agentid": int(self.config.agent_id),
+            "markdown": {"content": text},
+        }
+        await self._post_api(url, body)
+
+    # ── WeCom group send ────────────────────────────────────────────
+
+    async def _wecom_send_group_text(
+        self, token: str, chatid: str, text: str,
+    ) -> None:
+        """Send a text message to a WeCom group chat."""
+        url = (
+            f"https://qyapi.weixin.qq.com/cgi-bin/appchat/send"
+            f"?access_token={token}"
+        )
+        body = {
+            "chatid": chatid,
+            "msgtype": "text",
+            "text": {"content": _strip_markdown(text)},
+        }
+        await self._post_api(url, body)
+
+    async def _wecom_send_group_markdown(
+        self, token: str, chatid: str, text: str,
+    ) -> None:
+        """Send a markdown message to a WeCom group chat."""
+        url = (
+            f"https://qyapi.weixin.qq.com/cgi-bin/appchat/send"
+            f"?access_token={token}"
+        )
+        body = {
+            "chatid": chatid,
+            "msgtype": "markdown",
+            "markdown": {"content": text},
+        }
+        await self._post_api(url, body)
+
+    # ── MP send ───────────────────────────────────────────────────
+
+    async def _mp_send_text(
+        self, token: str, openid: str, text: str,
+    ) -> None:
+        """Send a text message via WeChat MP customer service API."""
+        url = (
+            f"https://api.weixin.qq.com/cgi-bin/message/custom/send"
+            f"?access_token={token}"
+        )
+        body = {
+            "touser": openid,
+            "msgtype": "text",
+            "text": {"content": _strip_markdown(text)},
+        }
+        await self._post_api(url, body)
+
+    # ── Media send ────────────────────────────────────────────────
+
+    async def _send_media_impl(
+        self,
+        recipient: str,
+        file_path: str,
+        caption: str = "",
+        metadata: dict | None = None,
+    ) -> bool:
+        """Send a media file via WeChat/WeCom."""
+        token = await self._ensure_token()
+        chat_id = self._resolve_media_chat_id(recipient, metadata)
+
+        # Upload media to get media_id
+        media_id = await self._upload_media(token, file_path)
+        if not media_id:
+            return False
+
+        path = Path(file_path)
+        ext = path.suffix.lower()
+        is_image = ext in {".jpg", ".jpeg", ".png", ".gif", ".bmp"}
+
+        if self._backend == "wecom":
+            msg_type = "image" if is_image else "file"
+            # Group chat: use appchat/send endpoint
+            if chat_id.startswith("wr"):
+                url = (
+                    f"https://qyapi.weixin.qq.com/cgi-bin/appchat/send"
+                    f"?access_token={token}"
+                )
+                body = {
+                    "chatid": chat_id,
+                    "msgtype": msg_type,
+                    msg_type: {"media_id": media_id},
+                }
+            else:
+                url = (
+                    f"https://qyapi.weixin.qq.com/cgi-bin/message/send"
+                    f"?access_token={token}"
+                )
+                body = {
+                    "touser": chat_id,
+                    "msgtype": msg_type,
+                    "agentid": int(self.config.agent_id),
+                    msg_type: {"media_id": media_id},
+                }
+        else:
+            url = (
+                f"https://api.weixin.qq.com/cgi-bin/message/custom/send"
+                f"?access_token={token}"
+            )
+            msg_type = "image" if is_image else "file"  # MP only supports image
+            if not is_image:
+                # MP doesn't support file via customer service API;
+                # send caption as text instead
+                if caption:
+                    await self._mp_send_text(token, chat_id, f"[文件] {path.name}\n{caption}")
+                return True
+            body = {
+                "touser": chat_id,
+                "msgtype": "image",
+                "image": {"media_id": media_id},
+            }
+
+        await self._post_api(url, body)
+
+        # Send caption separately if provided
+        if caption:
+            if self._backend == "wecom":
+                if chat_id.startswith("wr"):
+                    await self._wecom_send_group_text(token, chat_id, caption)
+                else:
+                    await self._wecom_send_text(token, chat_id, caption)
+            else:
+                await self._mp_send_text(token, chat_id, caption)
+
+        return True
+
+    async def _upload_media(
+        self, token: str, file_path: str,
+    ) -> str | None:
+        """Upload a media file and return the media_id."""
+        path = Path(file_path)
+        ext = path.suffix.lower()
+        is_image = ext in {".jpg", ".jpeg", ".png", ".gif", ".bmp"}
+        media_type = "image" if is_image else "file"
+
+        if self._backend == "wecom":
+            url = (
+                f"https://qyapi.weixin.qq.com/cgi-bin/media/upload"
+                f"?access_token={token}&type={media_type}"
+            )
+        else:
+            url = (
+                f"https://api.weixin.qq.com/cgi-bin/media/upload"
+                f"?access_token={token}&type={media_type}"
+            )
+
+        try:
+            with open(file_path, "rb") as f:
+                resp = await self._http_client.post(
+                    url,
+                    files={"media": (path.name, f)},
+                )
+            data = resp.json()
+            if data.get("errcode", 0) != 0 and "media_id" not in data:
+                logger.error(
+                    f"WeChat media upload failed: {data.get('errmsg')}"
+                )
+                return None
+            return data.get("media_id")
+        except Exception as e:
+            logger.error(f"WeChat media upload error: {e}")
+            return None
+
+    # ── Media download helper ────────────────────────────────────
+
+    async def _download_wechat_media(
+        self, media_id: str, filename: str,
+    ) -> tuple[str | None, str | None]:
+        """Download media by media_id via WeChat/WeCom media API."""
+        token = await self._ensure_token()
+        if self._backend == "wecom":
+            url = f"https://qyapi.weixin.qq.com/cgi-bin/media/get?access_token={token}&media_id={media_id}"
+        else:
+            url = f"https://api.weixin.qq.com/cgi-bin/media/get?access_token={token}&media_id={media_id}"
+        return await self._download_attachment(url, filename)
+
+    # ── Shared API helper ─────────────────────────────────────────
+
+    async def _post_api(self, url: str, body: dict) -> dict:
+        """POST to WeChat/WeCom API, check errcode, return response."""
+        try:
+            resp = await self._http_client.post(url, json=body)
+            data = resp.json()
+        except Exception as e:
+            raise RuntimeError(f"WeChat API error: {e}")
+
+        errcode = data.get("errcode", 0)
+        if errcode != 0:
+            errmsg = data.get("errmsg", "unknown")
+            # Token expired — refresh and retry once
+            if errcode in (40014, 42001):
+                logger.warning("WeChat token expired, refreshing...")
+                await self._refresh_token()
+                token = self._access_token
+                # Replace token in URL
+                if "access_token=" in url:
+                    url = re.sub(
+                        r"access_token=[^&]+",
+                        f"access_token={token}",
+                        url,
+                    )
+                    resp = await self._http_client.post(url, json=body)
+                    data = resp.json()
+                    if data.get("errcode", 0) != 0:
+                        raise RuntimeError(
+                            f"WeChat API error after retry: "
+                            f"{data.get('errmsg')}"
+                        )
+                    return data
+            else:
+                raise RuntimeError(
+                    f"WeChat API error ({errcode}): {errmsg}"
+                )
+
+        return data
+
+    # _send_typing_action: inherited no-op (WeChat has no typing API)
+
+

EvoScientist/channels/wechat/crypto.py

@@ -0,0 +1,187 @@
+"""WeChat / WeCom crypto helpers.
+
+Implements the message encryption/decryption protocol used by both
+WeCom (企业微信) and WeChat Official Account (公众号) callback APIs.
+
+The protocol uses AES-256-CBC with a key derived from the EncodingAESKey
+(base64-encoded 43-char string → 32-byte AES key).
+
+References:
+  - WeCom: https://developer.work.weixin.qq.com/document/path/90930
+  - MP: https://developers.weixin.qq.com/doc/offiaccount/Message_Management/Message_Encryption_and_Decryption_Instructions.html
+"""
+
+import base64
+import hashlib
+import struct
+import time
+import xml.etree.ElementTree as ET
+
+# Crypto imports — all from the Python standard library + pycryptodome
+# (but we'll use a pure-Python fallback if not available)
+try:
+    from Crypto.Cipher import AES
+    _HAS_PYCRYPTO = True
+except ImportError:
+    _HAS_PYCRYPTO = False
+
+
+def _pkcs7_pad(data: bytes, block_size: int = 32) -> bytes:
+    """PKCS#7 padding."""
+    pad_len = block_size - (len(data) % block_size)
+    return data + bytes([pad_len]) * pad_len
+
+
+def _pkcs7_unpad(data: bytes) -> bytes:
+    """PKCS#7 unpadding."""
+    pad_len = data[-1]
+    if pad_len < 1 or pad_len > 32:
+        return data
+    return data[:-pad_len]
+
+
+def _aes_decrypt(key: bytes, iv: bytes, ciphertext: bytes) -> bytes:
+    """AES-256-CBC decryption."""
+    if _HAS_PYCRYPTO:
+        cipher = AES.new(key, AES.MODE_CBC, iv)
+        return cipher.decrypt(ciphertext)
+    else:
+        # Pure-Python AES fallback (slower but no C deps)
+        # We'll try pyaes as a fallback
+        try:
+            import pyaes
+            decrypter = pyaes.Decrypter(
+                pyaes.AESModeOfOperationCBC(key, iv=iv)
+            )
+            decrypted = decrypter.feed(ciphertext)
+            decrypted += decrypter.feed()
+            return decrypted
+        except ImportError:
+            raise ImportError(
+                "WeChat message decryption requires pycryptodome or pyaes. "
+                "Install with: pip install pycryptodome"
+            )
+
+
+def _aes_encrypt(key: bytes, iv: bytes, plaintext: bytes) -> bytes:
+    """AES-256-CBC encryption."""
+    if _HAS_PYCRYPTO:
+        cipher = AES.new(key, AES.MODE_CBC, iv)
+        return cipher.encrypt(plaintext)
+    else:
+        try:
+            import pyaes
+            encrypter = pyaes.Encrypter(
+                pyaes.AESModeOfOperationCBC(key, iv=iv)
+            )
+            encrypted = encrypter.feed(plaintext)
+            encrypted += encrypter.feed()
+            return encrypted
+        except ImportError:
+            raise ImportError(
+                "WeChat message encryption requires pycryptodome or pyaes. "
+                "Install with: pip install pycryptodome"
+            )
+
+
+class WeChatCrypto:
+    """Handles WeChat/WeCom message encryption and decryption.
+
+    Parameters
+    ----------
+    token:
+        The Token configured in the WeChat/WeCom callback URL settings.
+    encoding_aes_key:
+        The 43-character EncodingAESKey (base64-encoded).
+    app_id:
+        The AppID (for MP) or CorpID (for WeCom).
+    """
+
+    def __init__(self, token: str, encoding_aes_key: str, app_id: str):
+        self.token = token
+        self.app_id = app_id
+        # Decode the AES key: EncodingAESKey + "=" → base64 decode → 32 bytes
+        self.aes_key = base64.b64decode(encoding_aes_key + "=")
+        # IV is the first 16 bytes of the key
+        self.iv = self.aes_key[:16]
+
+    def verify_signature(
+        self, signature: str, timestamp: str, nonce: str,
+        encrypt: str = "",
+    ) -> bool:
+        """Verify the callback signature.
+
+        For plain-mode verification (no encryption), *encrypt* can be empty.
+        """
+        parts = sorted([self.token, timestamp, nonce] + ([encrypt] if encrypt else []))
+        sha1 = hashlib.sha1("".join(parts).encode()).hexdigest()
+        return sha1 == signature
+
+    def decrypt(self, encrypt: str) -> tuple[str, str]:
+        """Decrypt an encrypted message.
+
+        Returns ``(xml_content, from_app_id)`` tuple.
+        """
+        ciphertext = base64.b64decode(encrypt)
+        plaintext = _aes_decrypt(self.aes_key, self.iv, ciphertext)
+        plaintext = _pkcs7_unpad(plaintext)
+
+        # plaintext layout:
+        # 16 bytes random + 4 bytes msg_len (big-endian) + msg + app_id
+        msg_len = struct.unpack("!I", plaintext[16:20])[0]
+        msg = plaintext[20:20 + msg_len].decode("utf-8")
+        from_app_id = plaintext[20 + msg_len:].decode("utf-8")
+        return msg, from_app_id
+
+    def encrypt(self, reply_msg: str) -> str:
+        """Encrypt a reply message.
+
+        Returns the base64-encoded ciphertext.
+        """
+        msg_bytes = reply_msg.encode("utf-8")
+        app_id_bytes = self.app_id.encode("utf-8")
+
+        # Random 16 bytes + msg_len (4 bytes big-endian) + msg + app_id
+        import os
+        random_bytes = os.urandom(16)
+        msg_len = struct.pack("!I", len(msg_bytes))
+        plaintext = random_bytes + msg_len + msg_bytes + app_id_bytes
+        plaintext = _pkcs7_pad(plaintext)
+
+        ciphertext = _aes_encrypt(self.aes_key, self.iv, plaintext)
+        return base64.b64encode(ciphertext).decode("utf-8")
+
+    def generate_signature(
+        self, encrypt: str, timestamp: str, nonce: str,
+    ) -> str:
+        """Generate the msg_signature for an encrypted reply."""
+        parts = sorted([self.token, timestamp, nonce, encrypt])
+        return hashlib.sha1("".join(parts).encode()).hexdigest()
+
+    def wrap_encrypted_reply(self, reply_msg: str) -> str:
+        """Encrypt a reply and wrap it in the XML envelope.
+
+        Returns the full XML string to return in the HTTP response.
+        """
+        encrypt = self.encrypt(reply_msg)
+        timestamp = str(int(time.time()))
+        nonce = hashlib.md5(str(time.time()).encode()).hexdigest()[:10]
+        signature = self.generate_signature(encrypt, timestamp, nonce)
+
+        return (
+            f"<xml>"
+            f"<Encrypt><![CDATA[{encrypt}]]></Encrypt>"
+            f"<MsgSignature><![CDATA[{signature}]]></MsgSignature>"
+            f"<TimeStamp>{timestamp}</TimeStamp>"
+            f"<Nonce><![CDATA[{nonce}]]></Nonce>"
+            f"</xml>"
+        )
+
+
+def parse_xml(xml_str: str) -> dict[str, str]:
+    """Parse a WeChat callback XML into a flat dict."""
+    root = ET.fromstring(xml_str)
+    result = {}
+    for child in root:
+        result[child.tag] = child.text or ""
+    return result

EvoScientist/channels/wechat/probe.py

@@ -0,0 +1,72 @@
+"""WeChat/WeCom credential validation."""
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+async def validate_wecom(
+    corp_id: str,
+    secret: str,
+    proxy: str | None = None,
+) -> tuple[bool, str]:
+    """Validate WeCom credentials by fetching an access token.
+
+    Returns:
+        Tuple of (is_valid, message).
+    """
+    if not corp_id or not secret:
+        return False, "corp_id and secret are required"
+
+    try:
+        import httpx
+    except ImportError:
+        return False, "httpx not installed"
+
+    url = (
+        f"https://qyapi.weixin.qq.com/cgi-bin/gettoken"
+        f"?corpid={corp_id}&corpsecret={secret}"
+    )
+    try:
+        async with httpx.AsyncClient(proxy=proxy) as client:
+            resp = await client.get(url, timeout=10)
+        data = resp.json()
+        if data.get("errcode", 0) == 0:
+            return True, "WeCom credentials valid"
+        return False, f"Error ({data.get('errcode')}): {data.get('errmsg')}"
+    except Exception as e:
+        return False, f"Error: {e}"
+
+
+async def validate_wechat_mp(
+    app_id: str,
+    app_secret: str,
+    proxy: str | None = None,
+) -> tuple[bool, str]:
+    """Validate WeChat Official Account credentials.
+
+    Returns:
+        Tuple of (is_valid, message).
+    """
+    if not app_id or not app_secret:
+        return False, "app_id and app_secret are required"
+
+    try:
+        import httpx
+    except ImportError:
+        return False, "httpx not installed"
+
+    url = (
+        f"https://api.weixin.qq.com/cgi-bin/token"
+        f"?grant_type=client_credential"
+        f"&appid={app_id}&secret={app_secret}"
+    )
+    try:
+        async with httpx.AsyncClient(proxy=proxy) as client:
+            resp = await client.get(url, timeout=10)
+        data = resp.json()
+        if "access_token" in data:
+            return True, "WeChat MP credentials valid"
+        return False, f"Error ({data.get('errcode')}): {data.get('errmsg')}"
+    except Exception as e:
+        return False, f"Error: {e}"

EvoScientist/channels/wechat/serve.py

@@ -0,0 +1,139 @@
+"""WeChat channel server.
+
+Standalone script to run the WeChat channel with CLI options.
+
+Usage:
+    # WeCom (企业微信应用)
+    python -m EvoScientist.channels.wechat.serve \\
+        --backend wecom \\
+        --corp-id CORP_ID \\
+        --agent-id AGENT_ID \\
+        --secret SECRET \\
+        --token TOKEN \\
+        --aes-key AES_KEY
+
+    # WeChat Official Account (公众号)
+    python -m EvoScientist.channels.wechat.serve \\
+        --backend wechatmp \\
+        --app-id APP_ID \\
+        --app-secret APP_SECRET \\
+        --token TOKEN \\
+        --aes-key AES_KEY
+
+Options:
+    --port PORT          Webhook listen port (default: 9001)
+    --allow USER_ID      Allowed sender (repeatable)
+    --agent              Use EvoScientist agent as handler
+    --thinking           Send thinking content as intermediate messages
+"""
+
+import argparse
+import logging
+
+from .channel import WeChatChannel, WeComConfig, WeChatMPConfig
+from ..bus import MessageBus
+from ..standalone import run_standalone
+
+logging.basicConfig(
+    level=logging.DEBUG,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    datefmt="%H:%M:%S",
+)
+logger = logging.getLogger(__name__)
+
+
+def parse_args():
+    """Parse command line arguments."""
+    parser = argparse.ArgumentParser(
+        description="WeChat channel server",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument(
+        "--backend",
+        choices=["wecom", "wechatmp"],
+        default="wecom",
+        help="WeChat backend type (default: wecom)",
+    )
+    parser.add_argument("--port", type=int, default=9001, help="Webhook port")
+    parser.add_argument(
+        "--allow",
+        action="append",
+        dest="allowed_senders",
+        help="Allowed sender ID (repeatable)",
+    )
+    parser.add_argument(
+        "--allow-channel",
+        action="append",
+        dest="allowed_channels",
+        help="Allowed channel ID. Can be used multiple times.",
+    )
+    parser.add_argument(
+        "--agent",
+        action="store_true",
+        help="Use EvoScientist agent as handler",
+    )
+    parser.add_argument(
+        "--thinking",
+        action="store_true",
+        help="Send thinking content (requires --agent)",
+    )
+
+    # WeCom settings
+    wecom = parser.add_argument_group("WeCom (企业微信)")
+    wecom.add_argument("--corp-id", default="", help="WeCom Corp ID")
+    wecom.add_argument("--agent-id", default="", help="WeCom Agent ID")
+    wecom.add_argument("--secret", default="", help="WeCom Secret")
+
+    # MP settings
+    mp = parser.add_argument_group("WeChat Official Account (公众号)")
+    mp.add_argument("--app-id", default="", help="MP App ID")
+    mp.add_argument("--app-secret", default="", help="MP App Secret")
+
+    # Shared settings
+    parser.add_argument("--token", default="", help="Callback verification token")
+    parser.add_argument("--aes-key", default="", help="EncodingAESKey")
+    parser.add_argument("--proxy", default="", help="HTTP proxy URL")
+
+    return parser.parse_args()
+
+
+def main():
+    """Entry point."""
+    args = parse_args()
+    allowed = set(args.allowed_senders) if args.allowed_senders else None
+    allowed_channels = set(args.allowed_channels) if args.allowed_channels else None
+    proxy = args.proxy or None
+
+    if args.backend == "wecom":
+        config = WeComConfig(
+            corp_id=args.corp_id,
+            agent_id=args.agent_id,
+            secret=args.secret,
+            token=args.token,
+            encoding_aes_key=args.aes_key,
+            webhook_port=args.port,
+            allowed_senders=allowed,
+            allowed_channels=allowed_channels,
+            proxy=proxy,
+        )
+    else:
+        config = WeChatMPConfig(
+            app_id=args.app_id,
+            app_secret=args.app_secret,
+            token=args.token,
+            encoding_aes_key=args.aes_key,
+            webhook_port=args.port,
+            allowed_senders=allowed,
+            allowed_channels=allowed_channels,
+            proxy=proxy,
+        )
+
+    send_thinking = args.thinking and args.agent
+    bus = MessageBus()
+    channel = WeChatChannel(config, backend=args.backend)
+
+    run_standalone(channel, bus, use_agent=args.agent, send_thinking=send_thinking)
+
+
+if __name__ == "__main__":
+    main()

EvoScientist/channels/wechat/verify_server.py

@@ -0,0 +1,175 @@
+"""WeChat callback verification server.
+
+Provides a lightweight temporary HTTP server that handles the WeChat/WeCom
+URL verification handshake during onboarding. This solves the chicken-and-egg
+problem: WeChat requires a live server to verify the callback URL before
+saving, but the main EvoScientist service isn't running during onboard.
+
+Usage:
+    server = VerifyServer(port, token, encoding_aes_key, corp_id)
+    await server.start()
+    # ... user clicks "Save" in WeCom admin console ...
+    # ... server auto-responds to the verification GET request ...
+    await server.wait_for_verify(timeout=120)
+    await server.stop()
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from aiohttp import web
+
+logger = logging.getLogger(__name__)
+
+
+class VerifyServer:
+    """Temporary HTTP server for WeChat/WeCom callback URL verification.
+
+    Handles the GET verification request (signature + echostr) and
+    signals when verification succeeds.
+    """
+
+    def __init__(
+        self,
+        port: int,
+        token: str,
+        encoding_aes_key: str = "",
+        app_id: str = "",
+    ):
+        self.port = port
+        self.token = token
+        self._crypto = None
+        self._runner = None
+        self._site = None
+        self._verified = asyncio.Event()
+
+        if encoding_aes_key and token and app_id:
+            from .crypto import WeChatCrypto
+            self._crypto = WeChatCrypto(
+                token=token,
+                encoding_aes_key=encoding_aes_key,
+                app_id=app_id,
+            )
+
+    async def start(self) -> None:
+        """Start the verification server."""
+        from aiohttp import web
+
+        app = web.Application()
+        app.router.add_get("/wechat/callback", self._handle)
+        # Also handle POST in case WeCom sends a POST for some reason
+        app.router.add_post("/wechat/callback", self._handle_post)
+
+        self._runner = web.AppRunner(app)
+        await self._runner.setup()
+        self._site = web.TCPSite(self._runner, "0.0.0.0", self.port)
+        await self._site.start()
+        logger.info(f"Verify server listening on port {self.port}")
+
+    async def stop(self) -> None:
+        """Stop the verification server."""
+        if self._site:
+            await self._site.stop()
+        if self._runner:
+            await self._runner.cleanup()
+        self._site = None
+        self._runner = None
+
+    async def wait_for_verify(self, timeout: float = 120) -> bool:
+        """Wait for verification to succeed.
+
+        Returns True if verified within timeout, False otherwise.
+        """
+        try:
+            await asyncio.wait_for(self._verified.wait(), timeout=timeout)
+            return True
+        except asyncio.TimeoutError:
+            return False
+
+    @property
+    def is_verified(self) -> bool:
+        return self._verified.is_set()
+
+    async def _handle(self, request) -> web.Response:
+        """Handle GET verification request.
+
+        During onboarding we use a lenient approach:
+        1. Try strict crypto verification (encrypted mode)
+        2. Try strict plain-mode signature check
+        3. If both fail, fall back to decrypting echostr without
+           signature check (WeCom requires the decrypted echostr)
+        4. Last resort: echo back raw echostr
+
+        This ensures the callback URL can be saved even if Token/AESKey
+        have minor issues, while still attempting proper verification.
+        """
+        from aiohttp import web
+
+        signature = (
+            request.query.get("msg_signature")
+            or request.query.get("signature", "")
+        )
+        timestamp = request.query.get("timestamp", "")
+        nonce = request.query.get("nonce", "")
+        echostr = request.query.get("echostr", "")
+
+        logger.info(
+            f"Verify request: msg_signature={signature[:16]}... "
+            f"timestamp={timestamp} nonce={nonce} "
+            f"echostr={echostr[:32]}..."
+        )
+
+        if not echostr:
+            return web.Response(status=400, text="missing echostr")
+
+        # Attempt 1: Encrypted mode with full signature verification
+        if self._crypto and request.query.get("msg_signature"):
+            sig_ok = self._crypto.verify_signature(
+                signature, timestamp, nonce, echostr,
+            )
+            if sig_ok:
+                try:
+                    plain_echostr, _ = self._crypto.decrypt(echostr)
+                    self._verified.set()
+                    logger.info("✓ Verified (encrypted, signature OK)")
+                    return web.Response(text=plain_echostr)
+                except Exception as e:
+                    logger.warning(f"Signature OK but decrypt failed: {e}")
+            else:
+                logger.warning("Signature mismatch, trying decrypt anyway...")
+
+            # Attempt 2: Try decrypt without signature check
+            # (WeCom requires the decrypted echostr to be returned)
+            try:
+                plain_echostr, _ = self._crypto.decrypt(echostr)
+                self._verified.set()
+                logger.info("✓ Verified (decrypted, signature skipped)")
+                return web.Response(text=plain_echostr)
+            except Exception as e:
+                logger.warning(f"Decrypt also failed: {e}")
+
+        # Attempt 3: Plain mode signature check
+        if self.token:
+            parts = sorted([self.token, timestamp, nonce])
+            expected = hashlib.sha1("".join(parts).encode()).hexdigest()
+            if expected == signature:
+                self._verified.set()
+                logger.info("✓ Verified (plain mode)")
+                return web.Response(text=echostr)
+
+        # Attempt 4: Last resort — just echo back the echostr
+        # This won't work for encrypted mode (WeCom expects decrypted),
+        # but works for plain mode with wrong token.
+        logger.warning("All verification methods failed, echoing raw echostr")
+        self._verified.set()
+        return web.Response(text=echostr)
+
+    async def _handle_post(self, request) -> web.Response:
+        """Handle POST — just acknowledge during verification phase."""
+        from aiohttp import web
+        return web.Response(text="success")

EvoScientist/config/onboard.py

@@ -1317,6 +1317,8 @@ def _step_channels(config: EvoScientistConfig) -> dict[str, object]:
     _CHANNELS = [
         ("telegram",  "Telegram",  [("telegram_bot_token", "Bot token (from @BotFather)")]),
         ("discord",   "Discord",   [("discord_bot_token", "Bot token")]),
+        ("slack",     "Slack",     [("slack_bot_token", "Bot token (xoxb-...)"), ("slack_app_token", "App token for Socket Mode (xapp-...)")]),
+        ("wechat",    "WeChat",    [("wechat_wecom_corp_id", "WeCom Corp ID"), ("wechat_wecom_agent_id", "WeCom Agent ID"), ("wechat_wecom_secret", "WeCom Secret")]),
         ("imessage",  "iMessage",  []),  # handled via _setup_imessage()
     ]
 
@@ -1475,6 +1477,29 @@ def _probe_channel(
                 _val("discord_bot_token"),
                 _val("discord_proxy") or None,
             )
+        elif ch_name == "slack":
+            from ..channels.slack.probe import validate_slack_tokens
+            return await validate_slack_tokens(
+                _val("slack_bot_token"),
+                _val("slack_app_token") or None,
+                _val("slack_proxy") or None,
+            )
+        elif ch_name == "wechat":
+            backend = _val("wechat_backend", "wecom")
+            if backend == "wechatmp":
+                from ..channels.wechat.probe import validate_wechat_mp
+                return await validate_wechat_mp(
+                    _val("wechat_mp_app_id"),
+                    _val("wechat_mp_app_secret"),
+                    _val("wechat_proxy") or None,
+                )
+            else:
+                from ..channels.wechat.probe import validate_wecom
+                return await validate_wecom(
+                    _val("wechat_wecom_corp_id"),
+                    _val("wechat_wecom_secret"),
+                    _val("wechat_proxy") or None,
+                )
         else:
             return True, "No probe available"
 

EvoScientist/config/settings.py

@@ -85,7 +85,7 @@ class EvoScientistConfig:
     show_thinking: bool = True
 
     # Channel Settings
-    channel_enabled: str = ""  # "imessage" | "telegram" | "discord" | "" (comma-separated for multiple)
+    channel_enabled: str = ""  # "imessage" | "telegram" | "discord" | "slack" | "wechat" | "" (comma-separated for multiple)
     channel_send_thinking: bool = True  # forward thinking to any channel
     require_mention: str = "group"  # "always" | "group" | "off"
     text_chunk_limit: int = 0  # 0 = use capability default

tests/test_slack_channel.py

@@ -0,0 +1,87 @@
+"""Tests for Slack channel implementation."""
+
+import asyncio
+
+import pytest
+
+from EvoScientist.channels.slack.channel import SlackChannel, SlackConfig
+from EvoScientist.channels.base import ChannelError
+
+
+def _run(coro):
+    """Run an async coroutine safely, creating a fresh event loop."""
+    loop = asyncio.new_event_loop()
+    try:
+        return loop.run_until_complete(coro)
+    finally:
+        loop.close()
+
+
+class TestSlackConfig:
+    def test_default_values(self):
+        config = SlackConfig()
+        assert config.bot_token == ""
+        assert config.app_token == ""
+        assert config.allowed_senders is None
+        assert config.allowed_channels is None
+        assert config.text_chunk_limit == 4096
+
+    def test_custom_values(self):
+        config = SlackConfig(
+            bot_token="xoxb-test",
+            app_token="xapp-test",
+            allowed_senders={"U123"},
+            allowed_channels={"C456"},
+            text_chunk_limit=2000,
+        )
+        assert config.bot_token == "xoxb-test"
+        assert config.app_token == "xapp-test"
+        assert config.allowed_senders == {"U123"}
+        assert config.allowed_channels == {"C456"}
+        assert config.text_chunk_limit == 2000
+
+
+class TestSlackChannel:
+    def test_init(self):
+        config = SlackConfig(bot_token="xoxb-test", app_token="xapp-test")
+        channel = SlackChannel(config)
+        assert channel.config is config
+        assert channel._running is False
+
+    def test_start_raises_without_bot_token(self):
+        config = SlackConfig(bot_token="", app_token="xapp-test")
+        channel = SlackChannel(config)
+        with pytest.raises(ChannelError, match="bot token"):
+            _run(channel.start())
+
+    def test_start_raises_without_app_token(self):
+        config = SlackConfig(bot_token="xoxb-test", app_token="")
+        channel = SlackChannel(config)
+        with pytest.raises(ChannelError, match="app token"):
+            _run(channel.start())
+
+    def test_stop_when_not_running(self):
+        config = SlackConfig(bot_token="xoxb-test", app_token="xapp-test")
+        channel = SlackChannel(config)
+        _run(channel.stop())
+
+    def test_send_returns_false_without_client(self):
+        from EvoScientist.channels.base import OutboundMessage
+
+        config = SlackConfig(bot_token="xoxb-test", app_token="xapp-test")
+        channel = SlackChannel(config)
+        msg = OutboundMessage(
+            channel="slack",
+            chat_id="C123",
+            content="hello",
+            metadata={"chat_id": "C123"},
+        )
+        result = _run(channel.send(msg))
+        assert result is False
+
+
+class TestSlackChannelRegistration:
+    def test_slack_registered(self):
+        from EvoScientist.channels.channel_manager import available_channels
+        channels = available_channels()
+        assert "slack" in channels

tests/test_wechat_channel.py

@@ -0,0 +1,458 @@
+"""Tests for WeChat channel implementation."""
+
+import asyncio
+import hashlib
+import time
+import xml.etree.ElementTree as ET
+
+import pytest
+
+from EvoScientist.channels.wechat.channel import (
+    WeChatChannel,
+    WeComConfig,
+    WeChatMPConfig,
+    _strip_markdown,
+)
+from EvoScientist.channels.wechat.crypto import (
+    WeChatCrypto,
+    parse_xml,
+    _pkcs7_pad,
+    _pkcs7_unpad,
+)
+from EvoScientist.channels.base import ChannelError
+
+
+def _run(coro):
+    """Run an async coroutine safely, creating a fresh event loop."""
+    loop = asyncio.new_event_loop()
+    try:
+        return loop.run_until_complete(coro)
+    finally:
+        loop.close()
+
+
+# ── Config tests ──────────────────────────────────────────────────
+
+class TestWeComConfig:
+    def test_default_values(self):
+        config = WeComConfig()
+        assert config.corp_id == ""
+        assert config.agent_id == ""
+        assert config.secret == ""
+        assert config.webhook_port == 9001
+        assert config.allowed_senders is None
+        assert config.text_chunk_limit == 4096
+
+    def test_custom_values(self):
+        config = WeComConfig(
+            corp_id="corp123",
+            agent_id="1000001",
+            secret="my-secret",
+            token="my-token",
+            encoding_aes_key="a" * 43,
+            webhook_port=8080,
+            allowed_senders={"user1", "user2"},
+        )
+        assert config.corp_id == "corp123"
+        assert config.agent_id == "1000001"
+        assert config.allowed_senders == {"user1", "user2"}
+        assert config.webhook_port == 8080
+
+
+class TestWeChatMPConfig:
+    def test_default_values(self):
+        config = WeChatMPConfig()
+        assert config.app_id == ""
+        assert config.app_secret == ""
+        assert config.webhook_port == 9001
+
+    def test_custom_values(self):
+        config = WeChatMPConfig(
+            app_id="wx1234",
+            app_secret="secret",
+            token="mp-token",
+        )
+        assert config.app_id == "wx1234"
+
+
+# ── Channel init / lifecycle tests ────────────────────────────────
+
+class TestWeChatChannelInit:
+    def test_wecom_init(self):
+        config = WeComConfig(corp_id="corp", agent_id="1", secret="s")
+        channel = WeChatChannel(config, backend="wecom")
+        assert channel.name == "wechat"
+        assert channel._backend == "wecom"
+        assert channel._running is False
+
+    def test_mp_init(self):
+        config = WeChatMPConfig(app_id="wx", app_secret="s")
+        channel = WeChatChannel(config, backend="wechatmp")
+        assert channel._backend == "wechatmp"
+
+    def test_start_raises_without_corp_id(self):
+        config = WeComConfig(corp_id="", agent_id="1", secret="s")
+        channel = WeChatChannel(config, backend="wecom")
+        with pytest.raises(ChannelError, match="corp_id"):
+            _run(channel.start())
+
+    def test_start_raises_without_secret(self):
+        config = WeComConfig(corp_id="corp", agent_id="1", secret="")
+        channel = WeChatChannel(config, backend="wecom")
+        with pytest.raises(ChannelError, match="secret"):
+            _run(channel.start())
+
+    def test_start_raises_without_agent_id(self):
+        config = WeComConfig(corp_id="corp", agent_id="", secret="s")
+        channel = WeChatChannel(config, backend="wecom")
+        with pytest.raises(ChannelError, match="agent_id"):
+            _run(channel.start())
+
+    def test_start_raises_mp_without_app_id(self):
+        config = WeChatMPConfig(app_id="", app_secret="s")
+        channel = WeChatChannel(config, backend="wechatmp")
+        with pytest.raises(ChannelError, match="app_id"):
+            _run(channel.start())
+
+    def test_stop_when_not_running(self):
+        config = WeComConfig(corp_id="c", agent_id="1", secret="s")
+        channel = WeChatChannel(config, backend="wecom")
+        _run(channel.stop())  # Should not raise
+
+    def test_send_returns_false_without_client(self):
+        from EvoScientist.channels.base import OutboundMessage
+
+        config = WeComConfig(corp_id="c", agent_id="1", secret="s")
+        channel = WeChatChannel(config, backend="wecom")
+        msg = OutboundMessage(
+            channel="wechat",
+            chat_id="user1",
+            content="hello",
+            metadata={"chat_id": "user1"},
+        )
+        result = _run(channel.send(msg))
+        assert result is False
+
+
+# ── Markdown stripping tests ──────────────────────────────────────
+
+class TestStripMarkdown:
+    def test_plain_text(self):
+        assert _strip_markdown("hello world") == "hello world"
+
+    def test_bold(self):
+        assert _strip_markdown("**bold**") == "bold"
+
+    def test_italic(self):
+        assert _strip_markdown("_italic_") == "italic"
+
+    def test_code(self):
+        assert _strip_markdown("`code`") == "code"
+
+    def test_link(self):
+        result = _strip_markdown("[text](https://example.com)")
+        assert "text" in result
+        assert "https://example.com" in result
+
+    def test_heading(self):
+        assert _strip_markdown("## Title").strip() == "Title"
+
+    def test_list_items(self):
+        result = _strip_markdown("- item1\n- item2")
+        assert "• item1" in result
+        assert "• item2" in result
+
+    def test_strikethrough(self):
+        assert _strip_markdown("~~deleted~~") == "deleted"
+
+    def test_code_block(self):
+        text = "```python\nprint('hi')\n```"
+        result = _strip_markdown(text)
+        assert "print('hi')" in result
+
+
+# ── XML parsing tests ─────────────────────────────────────────────
+
+class TestParseXml:
+    def test_basic_text_message(self):
+        xml = (
+            "<xml>"
+            "<MsgType><![CDATA[text]]></MsgType>"
+            "<Content><![CDATA[hello]]></Content>"
+            "<FromUserName><![CDATA[user123]]></FromUserName>"
+            "<ToUserName><![CDATA[bot]]></ToUserName>"
+            "<MsgId>1234</MsgId>"
+            "<CreateTime>1700000000</CreateTime>"
+            "</xml>"
+        )
+        data = parse_xml(xml)
+        assert data["MsgType"] == "text"
+        assert data["Content"] == "hello"
+        assert data["FromUserName"] == "user123"
+        assert data["MsgId"] == "1234"
+
+    def test_image_message(self):
+        xml = (
+            "<xml>"
+            "<MsgType><![CDATA[image]]></MsgType>"
+            "<PicUrl><![CDATA[https://example.com/img.jpg]]></PicUrl>"
+            "<MediaId><![CDATA[media_123]]></MediaId>"
+            "<FromUserName><![CDATA[user1]]></FromUserName>"
+            "</xml>"
+        )
+        data = parse_xml(xml)
+        assert data["MsgType"] == "image"
+        assert data["PicUrl"] == "https://example.com/img.jpg"
+
+    def test_event_message(self):
+        xml = (
+            "<xml>"
+            "<MsgType><![CDATA[event]]></MsgType>"
+            "<Event><![CDATA[subscribe]]></Event>"
+            "<FromUserName><![CDATA[user1]]></FromUserName>"
+            "</xml>"
+        )
+        data = parse_xml(xml)
+        assert data["MsgType"] == "event"
+        assert data["Event"] == "subscribe"
+
+
+# ── Crypto tests ──────────────────────────────────────────────────
+
+class TestPKCS7:
+    def test_pad_unpad_roundtrip(self):
+        data = b"hello"
+        padded = _pkcs7_pad(data)
+        assert len(padded) % 32 == 0
+        assert _pkcs7_unpad(padded) == data
+
+    def test_pad_block_aligned(self):
+        data = b"x" * 32
+        padded = _pkcs7_pad(data)
+        assert len(padded) == 64  # full padding block added
+        assert _pkcs7_unpad(padded) == data
+
+
+class TestWeChatCrypto:
+    """Test the encryption/decryption roundtrip.
+
+    Uses a deterministic 43-char EncodingAESKey.
+    """
+
+    # Skip encryption tests when no crypto backend is available
+    _has_crypto = False
+    try:
+        from Crypto.Cipher import AES as _aes  # noqa: F401
+        _has_crypto = True
+    except ImportError:
+        try:
+            import pyaes as _pyaes  # noqa: F401
+            _has_crypto = True
+        except ImportError:
+            pass
+    pytestmark = pytest.mark.skipif(
+        not _has_crypto,
+        reason="pycryptodome or pyaes required for encryption tests",
+    )
+
+    @pytest.fixture
+    def crypto(self):
+        # 43 base64 chars → 32 bytes AES key
+        key = "abcdefghijklmnopqrstuvwxyz0123456789ABCDEFG"
+        return WeChatCrypto(
+            token="test_token",
+            encoding_aes_key=key,
+            app_id="wx_test_app",
+        )
+
+    def test_encrypt_decrypt_roundtrip(self, crypto):
+        msg = "<xml><Content>Hello WeChat!</Content></xml>"
+        encrypted = crypto.encrypt(msg)
+        decrypted, app_id = crypto.decrypt(encrypted)
+        assert decrypted == msg
+        assert app_id == "wx_test_app"
+
+    def test_verify_signature(self, crypto):
+        timestamp = "1609459200"
+        nonce = "abc123"
+        parts = sorted([crypto.token, timestamp, nonce])
+        expected = hashlib.sha1("".join(parts).encode()).hexdigest()
+        assert crypto.verify_signature(expected, timestamp, nonce)
+        assert not crypto.verify_signature("wrong", timestamp, nonce)
+
+    def test_verify_signature_with_encrypt(self, crypto):
+        timestamp = "1609459200"
+        nonce = "abc123"
+        encrypt = "some_encrypted_data"
+        parts = sorted([crypto.token, timestamp, nonce, encrypt])
+        expected = hashlib.sha1("".join(parts).encode()).hexdigest()
+        assert crypto.verify_signature(expected, timestamp, nonce, encrypt)
+
+    def test_generate_signature(self, crypto):
+        encrypt = "test_encrypted"
+        timestamp = "1609459200"
+        nonce = "abc"
+        sig = crypto.generate_signature(encrypt, timestamp, nonce)
+        parts = sorted([crypto.token, timestamp, nonce, encrypt])
+        expected = hashlib.sha1("".join(parts).encode()).hexdigest()
+        assert sig == expected
+
+    def test_wrap_encrypted_reply(self, crypto):
+        msg = "<xml><Content>Reply</Content></xml>"
+        xml_reply = crypto.wrap_encrypted_reply(msg)
+        assert "<Encrypt>" in xml_reply
+        assert "<MsgSignature>" in xml_reply
+        assert "<TimeStamp>" in xml_reply
+        assert "<Nonce>" in xml_reply
+
+        # Parse and verify the encrypted content decrypts back
+        root = ET.fromstring(xml_reply)
+        encrypt = root.find("Encrypt").text
+        decrypted, app_id = crypto.decrypt(encrypt)
+        assert decrypted == msg
+
+
+# ── Message processing tests ──────────────────────────────────────
+
+class TestMessageProcessing:
+    """Test the _process_message method with various XML payloads."""
+
+    def _make_channel(self):
+        config = WeComConfig(
+            corp_id="corp", agent_id="1", secret="s",
+        )
+        return WeChatChannel(config, backend="wecom")
+
+    def test_text_message_queued(self):
+        channel = self._make_channel()
+
+        async def _test():
+            await channel._process_message({
+                "MsgType": "text",
+                "Content": "Hello!",
+                "FromUserName": "user1",
+                "ToUserName": "bot",
+                "MsgId": "100",
+                "CreateTime": str(int(time.time())),
+            })
+            # Check message was enqueued
+            assert not channel._queue.empty()
+            msg = await asyncio.wait_for(channel._queue.get(), timeout=1.0)
+            assert msg.content == "Hello!"
+            assert msg.sender_id == "user1"
+            assert msg.channel == "wechat"
+
+        _run(_test())
+
+    def test_location_message(self):
+        channel = self._make_channel()
+
+        async def _test():
+            await channel._process_message({
+                "MsgType": "location",
+                "Location_X": "39.9",
+                "Location_Y": "116.4",
+                "Label": "Beijing",
+                "FromUserName": "user1",
+                "ToUserName": "bot",
+                "MsgId": "101",
+                "CreateTime": str(int(time.time())),
+            })
+            msg = await asyncio.wait_for(channel._queue.get(), timeout=1.0)
+            assert "Beijing" in msg.content
+            assert "39.9" in msg.content
+
+        _run(_test())
+
+    def test_voice_recognition(self):
+        channel = self._make_channel()
+
+        async def _test():
+            await channel._process_message({
+                "MsgType": "voice",
+                "Recognition": "你好世界",
+                "FromUserName": "user1",
+                "ToUserName": "bot",
+                "MsgId": "102",
+                "CreateTime": str(int(time.time())),
+            })
+            msg = await asyncio.wait_for(channel._queue.get(), timeout=1.0)
+            assert "你好世界" in msg.content
+
+        _run(_test())
+
+    def test_link_message(self):
+        channel = self._make_channel()
+
+        async def _test():
+            await channel._process_message({
+                "MsgType": "link",
+                "Title": "Test Link",
+                "Description": "A description",
+                "Url": "https://example.com",
+                "FromUserName": "user1",
+                "ToUserName": "bot",
+                "MsgId": "103",
+                "CreateTime": str(int(time.time())),
+            })
+            msg = await asyncio.wait_for(channel._queue.get(), timeout=1.0)
+            assert "Test Link" in msg.content
+            assert "https://example.com" in msg.content
+
+        _run(_test())
+
+    def test_subscribe_event(self):
+        channel = self._make_channel()
+
+        async def _test():
+            await channel._process_message({
+                "MsgType": "event",
+                "Event": "subscribe",
+                "FromUserName": "user1",
+                "ToUserName": "bot",
+                "MsgId": "",
+                "CreateTime": str(int(time.time())),
+            })
+            msg = await asyncio.wait_for(channel._queue.get(), timeout=1.0)
+            assert "关注" in msg.content
+
+        _run(_test())
+
+    def test_unsubscribe_ignored(self):
+        channel = self._make_channel()
+
+        async def _test():
+            await channel._process_message({
+                "MsgType": "event",
+                "Event": "unsubscribe",
+                "FromUserName": "user1",
+                "ToUserName": "bot",
+                "MsgId": "",
+                "CreateTime": str(int(time.time())),
+            })
+            assert channel._queue.empty()
+
+        _run(_test())
+
+    def test_empty_message_ignored(self):
+        channel = self._make_channel()
+
+        async def _test():
+            await channel._process_message({
+                "MsgType": "text",
+                "Content": "",
+                "FromUserName": "",
+                "ToUserName": "bot",
+            })
+            assert channel._queue.empty()
+
+        _run(_test())
+
+
+# ── Registration test ─────────────────────────────────────────────
+
+class TestChannelRegistration:
+    def test_wechat_registered(self):
+        from EvoScientist.channels.channel_manager import available_channels
+        channels = available_channels()
+        assert "wechat" in channels