feat(alipay): 新增支付宝渠道插件完整实现

实现了支付宝生活号渠道的完整功能，包括消息接收回调、发送、安全验证、去重、配对绑定、流式回复支持等完整能力。

backend/package/yuxi/channel/extensions/alipay/__init__.py

@@ -0,0 +1,440 @@
+import logging
+
+from yuxi.channel.capabilities import ChannelCapabilities
+from yuxi.channel.context import ChannelContext
+from yuxi.channel.extensions.alipay.config import AlipayConfig
+from yuxi.channel.extensions.alipay.dedup import AlipayMessageDeduplicator
+from yuxi.channel.extensions.alipay.errors import AlipayError, classify_alipay_error
+from yuxi.channel.extensions.alipay.format import remove_markdown
+from yuxi.channel.extensions.alipay.gateway import AlipayGateway
+from yuxi.channel.extensions.alipay.media import AlipayMedia
+from yuxi.channel.extensions.alipay.monitor import convert_alipay_event
+from yuxi.channel.extensions.alipay.outbound import AlipayOutbound
+from yuxi.channel.extensions.alipay.pairing import AlipayPairing
+from yuxi.channel.extensions.alipay.security import AlipaySecurity
+from yuxi.channel.extensions.alipay.status import AlipayStatus
+from yuxi.channel.extensions.alipay.streaming import AlipayStreaming
+from yuxi.channel.extensions.alipay import webhook as alipay_webhook
+from yuxi.channel.extensions.base import BaseChannelPlugin
+from yuxi.channel.message.models import PeerKind
+from yuxi.channel.protocols import SessionResolution
+from yuxi.channel.plugins.registry import ChannelPluginRegistry
+
+_logger = logging.getLogger(__name__)
+
+
+class AlipayPlugin(BaseChannelPlugin):
+    id = "alipay"
+    name = "支付宝"
+    order = 50
+    label = "支付宝 (生活号+)"
+    aliases = ["alipay", "支付宝", "生活号"]
+    resolve_reply_to_mode = "off"
+
+    def __init__(self):
+        super().__init__()
+        self._config_adapter = AlipayConfig()
+        self._gateway: AlipayGateway | None = None
+        self._outbound: AlipayOutbound | None = None
+        self._status: AlipayStatus | None = None
+        self._security = AlipaySecurity()
+        self._pairing = AlipayPairing()
+        self._deduplicator = AlipayMessageDeduplicator()
+        self._streaming = AlipayStreaming()
+        self._media: AlipayMedia | None = None
+        self._remove_markdown: bool = True
+
+    @property
+    def capabilities(self) -> ChannelCapabilities:
+        return ChannelCapabilities(
+            chat_types=["direct"],
+            message_types=["text", "image"],
+            reactions=False,
+            typing_indicator=False,
+            threads=False,
+            edit=False,
+            unsend=True,
+            reply=False,
+            media=True,
+            effects=False,
+            native_commands=False,
+            polls=False,
+            group_management=False,
+            streaming=True,
+            streaming_mode="block",
+            block_streaming=True,
+            block_streaming_chunk_min_chars=50,
+            block_streaming_chunk_max_chars=500,
+            block_streaming_chunk_break_preference="double_newline_second_paragraph",
+            block_streaming_coalesce_idle_ms=1200,
+            tts=None,
+        )
+
+    def list_account_ids(self, config: dict | None = None) -> list[str]:
+        return self._config_adapter.list_account_ids(config)
+
+    async def resolve_account(self, account_id: str = "default") -> dict:
+        account = self._config_adapter.resolve_account(account_id)
+        return {
+            "account_id": account.account_id,
+            "app_id": account.app_id,
+            "app_private_key": account.app_private_key,
+            "alipay_public_key": account.alipay_public_key,
+            "aes_key": account.aes_key,
+            "mode": account.mode.value,
+            "gateway_url": account.gateway_url,
+            "name": account.name,
+            "enabled": account.enabled,
+            "dm_policy": account.dm_policy.value,
+            "allow_from": account.allow_from,
+            "remove_markdown": account.remove_markdown,
+            "subscribe_msg": account.subscribe_msg,
+        }
+
+    def is_configured(self, account: dict | None = None) -> bool:
+        return self._config_adapter.is_configured(account)
+
+    def is_enabled(self, account: dict | None = None, config: dict | None = None) -> bool:
+        return self._config_adapter.is_enabled(account, config)
+
+    def disabled_reason(self, account: dict | None = None, config: dict | None = None) -> str:
+        return self._config_adapter.disabled_reason(account, config)
+
+    def unconfigured_reason(self, account: dict | None = None, config: dict | None = None) -> str:
+        return self._config_adapter.unconfigured_reason(account, config)
+
+    def describe_account(self, account: dict | None = None, config: dict | None = None) -> dict:
+        return self._config_adapter.describe_account(account, config)
+
+    def resolve_allow_from(self, config: dict | None = None, account_id: str | None = None) -> list[str] | None:
+        return self._config_adapter.resolve_allow_from(config or {}, account_id)
+
+    def format_allow_from(self, config: dict, account_id: str | None, allow_from: list) -> list[str]:
+        return self._config_adapter.format_allow_from(config, account_id, allow_from)
+
+    def has_configured_state(self, config: dict | None = None) -> bool:
+        return self._config_adapter.has_configured_state(config)
+
+    def has_persisted_auth_state(self, config: dict | None = None) -> bool:
+        return self._config_adapter.has_persisted_auth_state(config)
+
+    def default_account_id(self, config: dict | None = None) -> str:
+        return self._config_adapter.default_account_id(config)
+
+    def config_schema(self) -> dict:
+        return self._config_adapter.config_schema()
+
+    def collect_warnings(
+        self, config: dict | None = None, account_id: str | None = None, account: dict | None = None
+    ) -> list[str]:
+        alipay_account = self._config_adapter.resolve_account(account_id or "default")
+        return self._security.collect_warnings(config, account_id, alipay_account)
+
+    async def start(self, ctx) -> object:
+        self._gateway = AlipayGateway()
+        self._outbound = AlipayOutbound(self._gateway)
+        self._status = AlipayStatus(self._gateway)
+        self._media = AlipayMedia(self._gateway)
+
+        account = self._config_adapter.resolve_account()
+        self._remove_markdown = account.remove_markdown
+        alipay_webhook.init_webhook(
+            self._config_adapter,
+            self._deduplicator,
+            self._security,
+            self._pairing,
+            self._outbound,
+        )
+        return {"running": True, "account_id": account.account_id, "mode": "webhook"}
+
+    async def stop(self, ctx) -> None:
+        if self._gateway:
+            await self._gateway.close()
+            self._gateway = None
+        self._outbound = None
+        self._status = None
+        self._media = None
+
+    async def on_config_changed(self, prev_cfg: dict, next_cfg: dict, account_id: str) -> None:
+        if prev_cfg != next_cfg:
+            _logger.info("Alipay config changed, reloading...")
+            ctx = ChannelContext(channel_type="alipay", account_id=account_id, config=next_cfg)
+            await self.stop(ctx)
+            await self.start(ctx)
+
+    async def send_text(
+        self,
+        target_id: str,
+        content: str,
+        *,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> None:
+        if not content or not self._outbound:
+            return
+
+        account = self._config_adapter.resolve_account(account_id or "default")
+        if self._remove_markdown:
+            content = remove_markdown(content)
+
+        await self._outbound.send_text(
+            target_id=target_id,
+            content=content,
+            account=account,
+            reply_to_id=reply_to_id,
+            thread_id=thread_id,
+        )
+
+    async def send_media(
+        self,
+        target_id: str,
+        media_url: str,
+        media_type: str,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+    ) -> None:
+        if not media_url or not self._outbound:
+            return
+
+        account = self._config_adapter.resolve_account()
+        if self._remove_markdown:
+            media_url = remove_markdown(media_url)
+        await self._outbound.send_media(
+            target_id=target_id,
+            media_url=media_url,
+            media_type=media_type,
+            account=account,
+            reply_to_id=reply_to_id,
+            thread_id=thread_id,
+        )
+
+    async def probe(self, account: dict | None = None) -> bool:
+        if self._status:
+            alipay_account = self._config_adapter.resolve_account()
+            result = await self._status.probe(alipay_account)
+            return result.ok
+        return False
+
+    def build_summary(self, snapshot: object | None = None) -> dict:
+        if self._status:
+            account = self._config_adapter.resolve_account()
+            return self._status.build_summary(account)
+        return {}
+
+    async def check_ready(self, account_id: str | None = None) -> bool:
+        if self._status:
+            account = self._config_adapter.resolve_account(account_id or "default")
+            return await self._status.check_ready(account)
+        return False
+
+    def resolve_dm_policy(self) -> dict:
+        account = self._config_adapter.resolve_account()
+        return self._security.resolve_dm_policy(account)
+
+    async def check_allowlist(self, peer_id: str, channel_type: str) -> bool:
+        account = self._config_adapter.resolve_account()
+        return self._security.check_allowlist(peer_id, channel_type, account)
+
+    async def generate_code(self, peer_id: str) -> str:
+        code = self._pairing.generate_code(peer_id)
+        return code or ""
+
+    async def verify_code(self, peer_id: str, code: str) -> bool:
+        return self._pairing.verify_code(peer_id, code)
+
+    async def send_template(
+        self,
+        target_id: str,
+        template_id: str,
+        context: dict,
+        url: str | None = None,
+        account_id: str | None = None,
+    ) -> None:
+        if not self._outbound:
+            return
+        account = self._config_adapter.resolve_account(account_id or "default")
+        await self._outbound.send_template(
+            account=account,
+            to_user_id=target_id,
+            template_id=template_id,
+            context=context,
+            url=url,
+        )
+
+    async def recall_message(self, msg_id: str, account_id: str | None = None) -> None:
+        if not self._outbound:
+            return
+        account = self._config_adapter.resolve_account(account_id or "default")
+        await self._outbound.recall_message(account=account, msg_id=msg_id)
+
+    async def query_followers(self, next_token: str = "", account_id: str | None = None) -> dict:
+        if not self._outbound:
+            return {}
+        account = self._config_adapter.resolve_account(account_id or "default")
+        result = await self._outbound.query_followers(account=account, next_token=next_token)
+        return result.result or {}
+
+    @property
+    def streaming_mode(self) -> str:
+        return self._streaming.streaming_mode
+
+    @property
+    def preview_stream_throttle_ms(self) -> int:
+        return self._streaming.preview_stream_throttle_ms
+
+    @property
+    def preview_min_initial_chars(self) -> int:
+        return self._streaming.preview_min_initial_chars
+
+    @property
+    def block_streaming_enabled(self) -> bool:
+        return self._streaming.block_streaming_enabled
+
+    @property
+    def block_streaming_break(self) -> str:
+        return self._streaming.block_streaming_break
+
+    @property
+    def block_streaming_chunk_min_chars(self) -> int:
+        return self._streaming.block_streaming_chunk_min_chars
+
+    @property
+    def block_streaming_chunk_max_chars(self) -> int:
+        return self._streaming.block_streaming_chunk_max_chars
+
+    @property
+    def block_streaming_chunk_break_preference(self) -> str:
+        return self._streaming.block_streaming_chunk_break_preference
+
+    @property
+    def block_streaming_coalesce_defaults(self) -> dict:
+        return self._streaming.block_streaming_coalesce_defaults
+
+    def create_draft_stream_session(self, target_id: str) -> dict:
+        return self._streaming.create_draft_stream_session(target_id)
+
+    def create_block_chunker(self) -> dict:
+        return self._streaming.create_block_chunker()
+
+    def parse_explicit_target(self, content: str) -> str | None:
+        return None
+
+    def resolve_session(self, msg):
+        if hasattr(msg, "sender") and hasattr(msg.sender, "kind"):
+            if msg.sender.kind == PeerKind.DIRECT:
+                return SessionResolution(kind="direct", conversation_id=msg.sender.id)
+        gid = msg.group.id if hasattr(msg, "group") and msg.group and msg.group.id else "unknown"
+        return SessionResolution(kind="group", conversation_id=gid)
+
+    # ── ErrorHandlingProtocol ────────────────────────────
+
+    def classify_error(self, error: BaseException):
+        if isinstance(error, AlipayError):
+            code = error.code
+            sub_code = error.sub_code
+            error_code, _ = classify_alipay_error(code, sub_code)
+            from yuxi.channel.errors import ErrorSeverity, ClassifiedError
+
+            severity_map = {
+                "sign_failed": ErrorSeverity.FATAL,
+                "verify_failed": ErrorSeverity.FATAL,
+                "auth_failed": ErrorSeverity.FORBIDDEN,
+                "rate_limited": ErrorSeverity.RATE_LIMITED,
+                "network_error": ErrorSeverity.NETWORK,
+            }
+            return ClassifiedError(
+                severity=severity_map.get(error_code, ErrorSeverity.RETRYABLE),
+                error_message=str(error),
+            )
+        return super().classify_error(error)
+
+    def is_retryable(self, error: BaseException) -> bool:
+        if isinstance(error, AlipayError):
+            return error.retryable
+        return super().is_retryable(error)
+
+    # ── InboundHandlerProtocol ───────────────────────────
+
+    async def handle_raw_event(self, event: dict, account: dict) -> object | None:
+        from yuxi.channel.extensions.alipay.types import AlipayAccount as _AlipayAccount, AlipayInboundEvent
+
+        alipay_account = _AlipayAccount(
+            account_id=account.get("account_id", "default"),
+            app_id=account.get("app_id", ""),
+        )
+        inbound = AlipayInboundEvent(
+            msg_type=event.get("msg_type", event.get("MsgType", "")),
+            event_type=event.get("event_type", event.get("EventType")),
+            from_user_id=event.get("from_user_id", event.get("FromAlipayUserId", "")),
+            from_user_name=event.get("from_user_name", event.get("FromAlipayUserName")),
+            create_time=int(event.get("create_time", event.get("CreateTime", 0))),
+            msg_id=event.get("msg_id", event.get("MsgId", "")),
+            app_id=event.get("app_id", event.get("AppId", "")),
+            biz_content=event.get("biz_content", {}),
+            raw=event,
+        )
+        return convert_alipay_event(inbound, alipay_account)
+
+    def parse_to_unified(self, raw_event: dict, account_id: str) -> object | None:
+        from yuxi.channel.extensions.alipay.types import AlipayAccount as _AlipayAccount, AlipayInboundEvent
+
+        account = _AlipayAccount(account_id=account_id)
+        inbound = AlipayInboundEvent(
+            msg_type=raw_event.get("msg_type", raw_event.get("MsgType", "")),
+            event_type=raw_event.get("event_type", raw_event.get("EventType")),
+            from_user_id=raw_event.get("from_user_id", raw_event.get("FromAlipayUserId", "")),
+            from_user_name=raw_event.get("from_user_name", raw_event.get("FromAlipayUserName")),
+            create_time=int(raw_event.get("create_time", raw_event.get("CreateTime", 0))),
+            msg_id=raw_event.get("msg_id", raw_event.get("MsgId", "")),
+            app_id=raw_event.get("app_id", raw_event.get("AppId", "")),
+            biz_content=raw_event.get("biz_content", {}),
+            raw=raw_event,
+        )
+        return convert_alipay_event(inbound, account)
+
+    # ── DedupeProtocol ───────────────────────────────────
+
+    def is_duplicate(self, key: str) -> bool:
+        return self._deduplicator.is_duplicate(key)
+
+    def mark_seen(self, key: str) -> None:
+        pass
+
+    @property
+    def ttl_seconds(self) -> int:
+        from yuxi.channel.extensions.alipay.constants import ALIPAY_DEDUPE_TTL_SECONDS
+
+        return ALIPAY_DEDUPE_TTL_SECONDS
+
+    @property
+    def max_entries(self) -> int:
+        from yuxi.channel.extensions.alipay.constants import ALIPAY_DEDUPE_MAX_SIZE
+
+        return ALIPAY_DEDUPE_MAX_SIZE
+
+    # ── FormatProtocol ───────────────────────────────────
+
+    def sanitize_text(self, text: str, payload: object | None = None) -> str:
+        if self._remove_markdown:
+            return remove_markdown(text)
+        return text
+
+    # ── AgentPromptProtocol ──────────────────────────────
+
+    def build_system_prompt(self, context) -> str | None:
+        return (
+            "当前渠道为支付宝生活号+，仅支持文本和图片消息。\n"
+            "支付宝消息格式为纯文本（不支持 Markdown），回复内容会自动去除 Markdown 符号。\n"
+            "每条文本消息长度不超过 2048 字符，每天向每个用户最多发送 100 条消息。\n"
+        )
+
+    def build_context_note(self, context) -> str:
+        return "请用简洁的纯文本回复用户，避免使用 Markdown 格式。"
+
+    @property
+    def channel_format_instructions(self) -> str:
+        return "支付宝生活号+ 纯文本渠道：不使用 Markdown 格式、不使用代码块、简洁直接回复。文本上限 2048 字符。"
+
+
+alipay_plugin = ChannelPluginRegistry.register(AlipayPlugin())

backend/package/yuxi/channel/extensions/alipay/config.py

@@ -0,0 +1,169 @@
+import os
+
+from yuxi.channel.extensions.alipay.types import AlipayAccount, AlipayDmPolicy, AlipayMode
+
+
+class AlipayConfig:
+    ENV_MAP = {
+        "app_id": "ALIPAY_APP_ID",
+        "app_private_key": "ALIPAY_PRIVATE_KEY",
+        "alipay_public_key": "ALIPAY_PUBLIC_KEY",
+        "aes_key": "ALIPAY_AES_KEY",
+        "mode": "ALIPAY_MODE",
+        "dm_policy": "ALIPAY_DM_POLICY",
+        "subscribe_msg": "ALIPAY_SUBSCRIBE_MSG",
+        "remove_markdown": "ALIPAY_REMOVE_MD",
+    }
+
+    def list_account_ids(self, config: dict | None = None) -> list[str]:
+        if self._resolve_app_id():
+            return ["default"]
+        return []
+
+    def resolve_account(self, account_id: str = "default") -> AlipayAccount:
+        app_id = self._resolve_app_id()
+        return AlipayAccount(
+            account_id=account_id,
+            app_id=app_id or "",
+            app_private_key=self._env_or_config("app_private_key") or "",
+            alipay_public_key=self._env_or_config("alipay_public_key") or "",
+            aes_key=self._env_or_config("aes_key"),
+            mode=AlipayMode(self._env_or_config("mode") or "production"),
+            gateway_url=(
+                "https://openapi.alipaydev.com/gateway.do"
+                if self._env_or_config("mode") == "sandbox"
+                else "https://openapi.alipay.com/gateway.do"
+            ),
+            name=self._env_or_config("name") or "",
+            enabled=self._env_or_config("enabled") != "false",
+            dm_policy=AlipayDmPolicy(self._env_or_config("dm_policy") or "open"),
+            allow_from=[],
+            remove_markdown=self._env_or_config("remove_markdown") != "false",
+            subscribe_msg=self._env_or_config("subscribe_msg") or "",
+        )
+
+    def is_configured(self, account: dict | None = None) -> bool:
+        return bool(
+            self._resolve_app_id()
+            and self._env_or_config("app_private_key")
+            and self._env_or_config("alipay_public_key")
+        )
+
+    def _resolve_app_id(self) -> str | None:
+        return self._env_or_config("app_id")
+
+    @staticmethod
+    def _env_or_config(key: str) -> str | None:
+        env_key = AlipayConfig.ENV_MAP.get(key, "")
+        if env_key:
+            val = os.getenv(env_key)
+            if val:
+                return val
+        return None
+
+    def config_schema(self) -> dict:
+        return {
+            "$schema": "https://json-schema.org/draft-07/schema#",
+            "type": "object",
+            "title": "支付宝渠道配置",
+            "properties": {
+                "app_id": {
+                    "type": "string",
+                    "title": "AppID",
+                    "description": "支付宝开放平台应用 AppID",
+                },
+                "app_private_key": {
+                    "type": "string",
+                    "title": "应用私钥",
+                    "x-ui-password": True,
+                    "description": "RSA2 应用私钥 (PEM 格式)",
+                },
+                "alipay_public_key": {
+                    "type": "string",
+                    "title": "支付宝公钥",
+                    "description": "支付宝公钥 (PEM 格式)",
+                },
+                "aes_key": {
+                    "type": "string",
+                    "title": "AES 解密密钥",
+                    "x-ui-password": True,
+                    "description": "AES 密钥（可选，16 字符）",
+                },
+                "mode": {
+                    "type": "string",
+                    "enum": ["production", "sandbox"],
+                    "default": "production",
+                    "title": "运行模式",
+                },
+                "dm_policy": {
+                    "type": "string",
+                    "enum": ["open", "pairing", "allowlist", "disabled"],
+                    "default": "open",
+                    "title": "DM 策略",
+                },
+                "subscribe_msg": {
+                    "type": "string",
+                    "title": "关注欢迎语",
+                    "description": "用户关注生活号后自动回复的文本",
+                },
+                "remove_markdown": {
+                    "type": "boolean",
+                    "default": True,
+                    "title": "移除 Markdown 格式",
+                    "description": "自动移除 AI 回复中的 Markdown 符号",
+                },
+            },
+            "required": ["app_id", "app_private_key", "alipay_public_key"],
+        }
+
+    def is_enabled(self, account: dict | None = None, config: dict | None = None) -> bool:
+        if account is None:
+            return self._env_or_config("enabled") != "false"
+        return account.get("enabled", True)
+
+    def disabled_reason(self, account: dict | None = None, config: dict | None = None) -> str:
+        if not account:
+            return "未配置账户"
+        if not account.get("enabled", True):
+            return "账户已被禁用"
+        return ""
+
+    def unconfigured_reason(self, account: dict | None = None, config: dict | None = None) -> str:
+        if not account:
+            return "未配置账户"
+        missing = []
+        if not account.get("app_id"):
+            missing.append("AppId")
+        if not account.get("app_private_key"):
+            missing.append("应用私钥")
+        if not account.get("alipay_public_key"):
+            missing.append("支付宝公钥")
+        return f"缺少: {', '.join(missing)}" if missing else ""
+
+    def describe_account(self, account: dict | None = None, config: dict | None = None) -> dict:
+        if not account:
+            return {"account_id": ""}
+        return {
+            "account_id": account.get("account_id", "default"),
+            "app_id": account.get("app_id", ""),
+            "name": account.get("name", ""),
+            "mode": account.get("mode", "production"),
+            "dm_policy": account.get("dm_policy", "open"),
+            "enabled": account.get("enabled", True),
+        }
+
+    def resolve_allow_from(self, config: dict, account_id: str | None = None) -> list[str] | None:
+        account = self.resolve_account(account_id or "default")
+        return account.allow_from or None
+
+    def format_allow_from(self, config: dict, account_id: str | None, allow_from: list) -> list[str]:
+        return [str(item) for item in allow_from]
+
+    def has_configured_state(self, config: dict | None = None) -> bool:
+        return self.is_configured()
+
+    def has_persisted_auth_state(self, config: dict | None = None) -> bool:
+        return bool(self._resolve_app_id() and self._env_or_config("app_private_key"))
+
+    def default_account_id(self, config: dict | None = None) -> str:
+        return "default"

backend/package/yuxi/channel/extensions/alipay/constants.py

@@ -0,0 +1,36 @@
+ALIPAY_TEXT_LIMIT = 2048
+ALIPAY_IMAGE_TEXT_ARTICLE_LIMIT = 6
+ALIPAY_DAILY_MSG_LIMIT_PER_USER = 100
+ALIPAY_MSG_RATE_LIMIT_PER_SECOND = 1
+ALIPAY_INTERACTION_WINDOW_HOURS = 48
+ALIPAY_WEBHOOK_RESPONSE_TIMEOUT_SECONDS = 5
+ALIPAY_RETRY_MAX_INTERVALS = [15, 30, 60, 120, 180]
+
+ALIPAY_GATEWAY_PRODUCTION = "https://openapi.alipay.com/gateway.do"
+ALIPAY_GATEWAY_SANDBOX = "https://openapi.alipaydev.com/gateway.do"
+ALIPAY_OAUTH_AUTHORIZE_PRODUCTION = "https://openauth.alipay.com/oauth2/publicAppAuthorize.htm"
+ALIPAY_OAUTH_AUTHORIZE_SANDBOX = "https://openapi.alipaydev.com/oauth2/publicAppAuthorize.htm"
+
+ALIPAY_SIGN_TYPE = "RSA2"
+ALIPAY_FORMAT = "JSON"
+ALIPAY_CHARSET = "utf-8"
+ALIPAY_VERSION = "1.0"
+
+ALIPAY_SUCCESS_CODE = "10000"
+ALIPAY_APP_ID_PREFIX = "2021"
+
+ALIPAY_APP_ID_PATTERN = r"^\d{16,32}$"
+ALIPAY_USER_ID_PREFIX = "2088"
+
+ALIPAY_DEDUPE_TTL_SECONDS = 300
+ALIPAY_DEDUPE_MAX_SIZE = 10000
+
+ALIPAY_RETRY_MAX_ATTEMPTS = 3
+ALIPAY_RETRY_BASE_DELAY = 1.0
+ALIPAY_RETRY_MAX_DELAY = 30.0
+
+ALIPAY_PAIRING_CODE_TTL_SECONDS = 300
+ALIPAY_PAIRING_CODE_LENGTH = 6
+
+ALIPAY_BLOCK_STREAMING_CHUNK_MIN_CHARS = 50
+ALIPAY_BLOCK_STREAMING_CHUNK_MAX_CHARS = 500

backend/package/yuxi/channel/extensions/alipay/crypto.py

@@ -0,0 +1,82 @@
+import base64
+from urllib.parse import quote
+
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import padding
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+
+
+class AlipayAESCrypto:
+    def __init__(self, aes_key: str):
+        raw = base64.b64decode(aes_key + "=" * (4 - len(aes_key) % 4))
+        self._key = raw[:16]
+
+    def decrypt(self, encrypted: str | bytes) -> str:
+        from cryptography.hazmat.primitives import padding as sym_padding
+
+        data = base64.b64decode(encrypted)
+        iv = self._key
+        cipher = Cipher(algorithms.AES(self._key), modes.CBC(iv))
+        decryptor = cipher.decryptor()
+        padded = decryptor.update(data) + decryptor.finalize()
+        unpadder = sym_padding.PKCS7(128).unpadder()
+        plaintext = unpadder.update(padded) + unpadder.finalize()
+        return plaintext.decode("utf-8")
+
+    def encrypt(self, plaintext: str) -> str:
+        from cryptography.hazmat.primitives import padding as sym_padding
+
+        iv = self._key
+        padder = sym_padding.PKCS7(128).padder()
+        padded = padder.update(plaintext.encode("utf-8")) + padder.finalize()
+        cipher = Cipher(algorithms.AES(self._key), modes.CBC(iv))
+        encryptor = cipher.encryptor()
+        encrypted = encryptor.update(padded) + encryptor.finalize()
+        return base64.b64encode(encrypted).decode("utf-8")
+
+
+class AlipayCrypto:
+    def __init__(self, app_private_key_pem: str, alipay_public_key_pem: str):
+        self._private_key = serialization.load_pem_private_key(
+            app_private_key_pem.encode("utf-8"),
+            password=None,
+            backend=default_backend(),
+        )
+        self._public_key = serialization.load_pem_public_key(
+            alipay_public_key_pem.encode("utf-8"),
+            backend=default_backend(),
+        )
+
+    def sign(self, params: dict) -> str:
+        content = self._build_sign_string(params)
+        signature = self._private_key.sign(
+            content.encode("utf-8"),
+            padding.PKCS1v15(),
+            hashes.SHA256(),
+        )
+        return base64.b64encode(signature).decode("utf-8")
+
+    def verify(self, params: dict, signature: str) -> bool:
+        content = self._build_sign_string(params)
+        try:
+            self._public_key.verify(
+                base64.b64decode(signature),
+                content.encode("utf-8"),
+                padding.PKCS1v15(),
+                hashes.SHA256(),
+            )
+            return True
+        except Exception:
+            return False
+
+    @staticmethod
+    def _build_sign_string(params: dict) -> str:
+        filtered = {k: v for k, v in params.items() if k not in ("sign", "sign_type") and v is not None}
+        sorted_items = sorted(filtered.items(), key=lambda x: x[0])
+        parts = []
+        for k, v in sorted_items:
+            encoded_key = quote(str(k), safe="~")
+            encoded_value = quote(str(v), safe="~")
+            parts.append(f"{encoded_key}={encoded_value}")
+        return "&".join(parts)

backend/package/yuxi/channel/extensions/alipay/dedup.py

@@ -0,0 +1,28 @@
+import time
+
+from yuxi.channel.extensions.alipay.constants import (
+    ALIPAY_DEDUPE_MAX_SIZE,
+    ALIPAY_DEDUPE_TTL_SECONDS,
+)
+
+
+class AlipayMessageDeduplicator:
+    def __init__(self):
+        self._cache: dict[str, float] = {}
+
+    def is_duplicate(self, msg_id: str) -> bool:
+        if not msg_id:
+            return False
+        now = time.time()
+        if msg_id in self._cache:
+            if now - self._cache[msg_id] < ALIPAY_DEDUPE_TTL_SECONDS:
+                return True
+        self._cache[msg_id] = now
+        self._evict_expired(now)
+        return False
+
+    def _evict_expired(self, now: float):
+        if len(self._cache) > ALIPAY_DEDUPE_MAX_SIZE:
+            expired = [k for k, v in self._cache.items() if now - v > ALIPAY_DEDUPE_TTL_SECONDS]
+            for k in expired:
+                del self._cache[k]

backend/package/yuxi/channel/extensions/alipay/errors.py

@@ -0,0 +1,65 @@
+from enum import StrEnum
+
+
+class AlipayErrorCode(StrEnum):
+    SIGN_FAILED = "sign_failed"
+    VERIFY_FAILED = "verify_failed"
+    AUTH_FAILED = "auth_failed"
+    RATE_LIMITED = "rate_limited"
+    OUT_OF_WINDOW = "out_of_window"
+    DAILY_LIMIT_EXCEEDED = "daily_limit_exceeded"
+    USER_NOT_FOLLOWED = "user_not_followed"
+    MSG_TOO_LONG = "msg_too_long"
+    NETWORK_ERROR = "network_error"
+    BUSINESS_ERROR = "business_error"
+    UNKNOWN = "unknown"
+
+
+class AlipayError(Exception):
+    def __init__(
+        self,
+        code: AlipayErrorCode,
+        message: str = "",
+        sub_code: str = "",
+        sub_msg: str = "",
+        retryable: bool = False,
+        retry_after: float | None = None,
+    ):
+        self.code = code
+        self.message = message
+        self.sub_code = sub_code
+        self.sub_msg = sub_msg
+        self.retryable = retryable
+        self.retry_after = retry_after
+        super().__init__(f"[{code}] {message}" + (f" | {sub_code}: {sub_msg}" if sub_code else ""))
+
+
+_ERROR_CLASSIFICATION: dict[str, tuple[AlipayErrorCode, bool]] = {
+    "40001": (AlipayErrorCode.SIGN_FAILED, False),
+    "40002": (AlipayErrorCode.VERIFY_FAILED, False),
+    "40004": (AlipayErrorCode.BUSINESS_ERROR, False),
+    "40006": (AlipayErrorCode.AUTH_FAILED, False),
+    "20000": (AlipayErrorCode.NETWORK_ERROR, True),
+    "20001": (AlipayErrorCode.AUTH_FAILED, False),
+}
+
+_SUB_CODE_CLASSIFICATION: dict[str, tuple[AlipayErrorCode, bool]] = {
+    "isv.invalid-signature": (AlipayErrorCode.SIGN_FAILED, False),
+    "isv.invalid-app-id": (AlipayErrorCode.VERIFY_FAILED, False),
+    "isv.message-count-per-day-isv-error": (AlipayErrorCode.DAILY_LIMIT_EXCEEDED, False),
+    "isv.invalid-auth-token": (AlipayErrorCode.AUTH_FAILED, False),
+    "isv.business-fail": (AlipayErrorCode.USER_NOT_FOLLOWED, False),
+}
+
+
+def classify_alipay_error(code: str, sub_code: str = "") -> tuple[AlipayErrorCode, bool]:
+    if sub_code and sub_code in _SUB_CODE_CLASSIFICATION:
+        return _SUB_CODE_CLASSIFICATION[sub_code]
+    if code in _ERROR_CLASSIFICATION:
+        return _ERROR_CLASSIFICATION[code]
+    return (AlipayErrorCode.UNKNOWN, False)
+
+
+def is_retryable_error(code: str, sub_code: str = "") -> bool:
+    _, retryable = classify_alipay_error(code, sub_code)
+    return retryable

backend/package/yuxi/channel/extensions/alipay/format.py

@@ -0,0 +1,34 @@
+import re
+
+
+def remove_markdown(text: str) -> str:
+    text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)
+    text = re.sub(r"\*(.+?)\*", r"\1", text)
+    text = re.sub(r"__(.+?)__", r"\1", text)
+    text = re.sub(r"_(.+?)_", r"\1", text)
+    text = re.sub(r"~~(.+?)~~", r"\1", text)
+    text = re.sub(r"```[\s\S]*?```", "", text)
+    text = re.sub(r"`{1,3}[^`]*`{1,3}", "", text)
+    text = re.sub(r"\[([^\]]*)\]\([^)]*\)", r"\1", text)
+    text = re.sub(r"!\[.*?\]\(.*?\)", "[图片]", text)
+    text = re.sub(r"^#+ (.*?)$", r"\1", text, flags=re.MULTILINE)
+    text = re.sub(r"^[*-] (.*?)$", r"· \1", text, flags=re.MULTILINE)
+    text = re.sub(r"^> (.*?)$", r"  \1", text, flags=re.MULTILINE)
+    text = re.sub(r"\n{3,}", "\n\n", text)
+    return text.strip()
+
+
+def split_utf8_safe(text: str, max_len: int) -> list[str]:
+    if len(text) <= max_len:
+        return [text]
+    chunks = []
+    while text:
+        if len(text) <= max_len:
+            chunks.append(text)
+            break
+        split_at = text.rfind("\n", 0, max_len)
+        if split_at == -1 or split_at < max_len // 2:
+            split_at = max_len
+        chunks.append(text[:split_at])
+        text = text[split_at:].lstrip()
+    return chunks

backend/package/yuxi/channel/extensions/alipay/gateway.py

@@ -0,0 +1,124 @@
+import asyncio
+import json
+import logging
+from datetime import datetime
+
+import httpx
+
+from yuxi.channel.extensions.alipay.constants import (
+    ALIPAY_CHARSET,
+    ALIPAY_FORMAT,
+    ALIPAY_RETRY_BASE_DELAY,
+    ALIPAY_RETRY_MAX_ATTEMPTS,
+    ALIPAY_SIGN_TYPE,
+    ALIPAY_SUCCESS_CODE,
+    ALIPAY_VERSION,
+)
+from yuxi.channel.extensions.alipay.crypto import AlipayCrypto
+from yuxi.channel.extensions.alipay.errors import (
+    AlipayError,
+    AlipayErrorCode,
+    classify_alipay_error,
+)
+from yuxi.channel.extensions.alipay.types import AlipayAccount
+
+logger = logging.getLogger(__name__)
+
+
+class AlipayGateway:
+    def __init__(self, crypto: AlipayCrypto | None = None):
+        self._crypto = crypto
+        self._clients: dict[str, httpx.AsyncClient] = {}
+        self._lock = asyncio.Lock()
+
+    async def _get_client(self, account: AlipayAccount) -> httpx.AsyncClient:
+        if account.account_id not in self._clients:
+            self._clients[account.account_id] = httpx.AsyncClient(
+                base_url=account.gateway_url,
+                timeout=httpx.Timeout(10.0),
+                limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
+            )
+        return self._clients[account.account_id]
+
+    async def close(self):
+        for client in self._clients.values():
+            await client.aclose()
+        self._clients.clear()
+
+    async def request(
+        self,
+        account: AlipayAccount,
+        method: str,
+        biz_content: dict | None = None,
+    ) -> dict:
+        crypto = self._crypto or AlipayCrypto(account.app_private_key, account.alipay_public_key)
+        params = {
+            "app_id": account.app_id,
+            "method": method,
+            "format": ALIPAY_FORMAT,
+            "charset": ALIPAY_CHARSET,
+            "sign_type": ALIPAY_SIGN_TYPE,
+            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+            "version": ALIPAY_VERSION,
+        }
+        if biz_content:
+            params["biz_content"] = json.dumps(biz_content, ensure_ascii=False)
+
+        params["sign"] = crypto.sign(params)
+        client = await self._get_client(account)
+
+        last_error = None
+        for attempt in range(ALIPAY_RETRY_MAX_ATTEMPTS):
+            try:
+                resp = await client.post("", data=params)
+                resp.raise_for_status()
+                data = resp.json()
+
+                response_key = next((k for k in data if k.endswith("_response")), None)
+                if not response_key:
+                    raise AlipayError(
+                        code=AlipayErrorCode.UNKNOWN,
+                        message="无法识别的响应格式",
+                    )
+
+                body = data[response_key]
+                code = body.get("code", "")
+
+                if code != ALIPAY_SUCCESS_CODE:
+                    sub_code = body.get("sub_code", "")
+                    error_code, retryable = classify_alipay_error(code, sub_code)
+                    err = AlipayError(
+                        code=error_code,
+                        message=body.get("msg", ""),
+                        sub_code=sub_code,
+                        sub_msg=body.get("sub_msg", ""),
+                        retryable=retryable,
+                    )
+                    if retryable and attempt < ALIPAY_RETRY_MAX_ATTEMPTS - 1:
+                        last_error = err
+                        delay = ALIPAY_RETRY_BASE_DELAY * (2**attempt)
+                        await asyncio.sleep(delay)
+                        continue
+                    raise err
+
+                return body
+
+            except (httpx.ConnectError, httpx.TimeoutException, httpx.NetworkError) as e:
+                if attempt < ALIPAY_RETRY_MAX_ATTEMPTS - 1:
+                    last_error = AlipayError(
+                        code=AlipayErrorCode.NETWORK_ERROR,
+                        message=str(e),
+                        retryable=True,
+                    )
+                    delay = ALIPAY_RETRY_BASE_DELAY * (2**attempt)
+                    await asyncio.sleep(delay)
+                    continue
+                raise AlipayError(
+                    code=AlipayErrorCode.NETWORK_ERROR,
+                    message=str(e),
+                )
+
+        raise last_error or AlipayError(
+            code=AlipayErrorCode.UNKNOWN,
+            message="请求失败，已达最大重试次数",
+        )

backend/package/yuxi/channel/extensions/alipay/media.py

@@ -0,0 +1,42 @@
+import json
+import logging
+from datetime import datetime
+
+import httpx
+
+from yuxi.channel.extensions.alipay.types import AlipayAccount
+
+logger = logging.getLogger(__name__)
+
+ALIPAY_FILE_GATEWAY_PRODUCTION = "https://openfile.alipay.com/chat/multimedia.do"
+ALIPAY_FILE_GATEWAY_SANDBOX = "https://openfile.alipaydev.com/chat/multimedia.do"
+
+
+class AlipayMedia:
+    def __init__(self, gateway=None):
+        self._gateway = gateway
+
+    async def download(self, account: AlipayAccount, media_id: str) -> bytes | None:
+        gateway_url = (
+            ALIPAY_FILE_GATEWAY_SANDBOX if "sandbox" in account.gateway_url else ALIPAY_FILE_GATEWAY_PRODUCTION
+        )
+        try:
+            async with httpx.AsyncClient(timeout=httpx.Timeout(15.0)) as client:
+                resp = await client.get(
+                    gateway_url,
+                    params={
+                        "app_id": account.app_id,
+                        "method": "alipay.mobile.public.multimedia.download",
+                        "format": "JSON",
+                        "charset": "utf-8",
+                        "sign_type": "RSA2",
+                        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                        "version": "1.0",
+                        "biz_content": json.dumps({"media_id": media_id}),
+                    },
+                )
+                resp.raise_for_status()
+                return resp.content
+        except Exception:
+            logger.exception("Failed to download media %s", media_id)
+            return None

backend/package/yuxi/channel/extensions/alipay/monitor.py

@@ -0,0 +1,97 @@
+from datetime import datetime, UTC
+
+from yuxi.channel.extensions.alipay.types import AlipayAccount, AlipayInboundEvent
+from yuxi.channel.message.models import MessageType, PeerInfo, PeerKind, UnifiedMessage
+
+
+def convert_alipay_event(event: AlipayInboundEvent, account: AlipayAccount) -> UnifiedMessage | None:
+    msg_type = _map_msg_type(event)
+
+    if event.msg_type == "event":
+        if event.event_type in ("follow", "enter"):
+            return _build_event_message(event, account, msg_type)
+        return None
+
+    return UnifiedMessage(
+        msg_id=event.msg_id,
+        channel_type="alipay",
+        account_id=account.account_id,
+        content=_extract_content(event),
+        sender=PeerInfo(
+            kind=PeerKind.DIRECT,
+            id=event.from_user_id,
+            display_name=event.from_user_name or event.from_user_id,
+        ),
+        message_type=msg_type,
+        media_urls=_extract_media_urls(event),
+        timestamp=datetime.fromtimestamp(event.create_time / 1000, tz=UTC),
+        raw_payload=event.raw,
+        metadata={
+            "FromAlipayUserId": event.from_user_id,
+            "AppId": event.app_id,
+            "MsgType": event.msg_type,
+            "EventType": event.event_type or "",
+        },
+        body_for_agent=_extract_content(event),
+    )
+
+
+def _map_msg_type(event: AlipayInboundEvent) -> MessageType:
+    if event.msg_type == "text":
+        return MessageType.TEXT
+    if event.msg_type == "image":
+        return MessageType.IMAGE
+    return MessageType.EVENT
+
+
+def _extract_content(event: AlipayInboundEvent) -> str:
+    if event.msg_type == "text":
+        return (event.biz_content or {}).get("content", "")
+    if event.msg_type == "image":
+        return "[图片消息]"
+    if event.msg_type == "event":
+        event_names = {
+            "follow": "关注了生活号",
+            "unfollow": "取消关注",
+            "enter": "进入了生活号",
+            "click": f"点击了菜单: {(event.biz_content or {}).get('action_name', '')}",
+        }
+        return event_names.get(event.event_type or "", f"事件: {event.event_type}")
+    return ""
+
+
+def _extract_media_urls(event: AlipayInboundEvent) -> list[str]:
+    if event.msg_type == "image":
+        biz = event.biz_content or {}
+        urls = []
+        if biz.get("pic_url"):
+            urls.append(biz["pic_url"])
+        if biz.get("media_id"):
+            urls.append(f"alipay://media/{biz['media_id']}")
+        return urls
+    return []
+
+
+def _build_event_message(event: AlipayInboundEvent, account: AlipayAccount, msg_type: MessageType) -> UnifiedMessage:
+    return UnifiedMessage(
+        msg_id=event.msg_id,
+        channel_type="alipay",
+        account_id=account.account_id,
+        content=_extract_content(event),
+        sender=PeerInfo(
+            kind=PeerKind.DIRECT,
+            id=event.from_user_id,
+            display_name=event.from_user_name or event.from_user_id,
+        ),
+        message_type=msg_type,
+        media_urls=[],
+        timestamp=datetime.fromtimestamp(event.create_time / 1000, tz=UTC),
+        raw_payload=event.raw,
+        metadata={
+            "FromAlipayUserId": event.from_user_id,
+            "AppId": event.app_id,
+            "MsgType": event.msg_type,
+            "EventType": event.event_type or "",
+        },
+        body_for_agent=_extract_content(event),
+    )

backend/package/yuxi/channel/extensions/alipay/outbound.py

@@ -0,0 +1,314 @@
+import logging
+import time
+from datetime import date
+
+from yuxi.channel.extensions.alipay.constants import (
+    ALIPAY_DAILY_MSG_LIMIT_PER_USER,
+    ALIPAY_IMAGE_TEXT_ARTICLE_LIMIT,
+    ALIPAY_INTERACTION_WINDOW_HOURS,
+    ALIPAY_TEXT_LIMIT,
+)
+from yuxi.channel.extensions.alipay.errors import AlipayError, AlipayErrorCode
+from yuxi.channel.extensions.alipay.format import split_utf8_safe
+from yuxi.channel.extensions.alipay.gateway import AlipayGateway
+from yuxi.channel.extensions.alipay.types import AlipayAccount, AlipayOutboundResult
+
+logger = logging.getLogger(__name__)
+
+
+class AlipayOutbound:
+    delivery_mode = "direct"
+    chunker_mode = "length"
+    text_chunk_limit = ALIPAY_TEXT_LIMIT
+    poll_max_options = None
+    supports_poll_duration_seconds = False
+    supports_anonymous_polls = False
+    extract_markdown_images = True
+    presentation_capabilities = None
+    delivery_capabilities = None
+
+    def __init__(self, gateway: AlipayGateway):
+        self._gateway = gateway
+        self._daily_counters: dict[str, int] = {}
+        self._last_interaction: dict[str, float] = {}
+
+    def _get_counter_key(self, account_id: str, user_id: str) -> str:
+        today = date.today().isoformat()
+        return f"{account_id}:{user_id}:{today}"
+
+    def _check_daily_limit(self, user_id: str, account_id: str) -> bool:
+        key = self._get_counter_key(account_id, user_id)
+        count = self._daily_counters.get(key, 0)
+        return count < ALIPAY_DAILY_MSG_LIMIT_PER_USER
+
+    def _increment_daily_counter(self, user_id: str, account_id: str):
+        key = self._get_counter_key(account_id, user_id)
+        self._daily_counters[key] = self._daily_counters.get(key, 0) + 1
+        today = date.today().isoformat()
+        stale = [k for k in self._daily_counters if not k.endswith(f":{today}")]
+        for k in stale:
+            del self._daily_counters[k]
+
+    def record_interaction(self, user_id: str, account_id: str):
+        key = f"{account_id}:{user_id}"
+        self._last_interaction[key] = time.time()
+
+    def _check_window(self, user_id: str, account_id: str) -> bool:
+        key = f"{account_id}:{user_id}"
+        last = self._last_interaction.get(key, 0)
+        return (time.time() - last) < ALIPAY_INTERACTION_WINDOW_HOURS * 3600
+
+    async def send_text(
+        self,
+        target_id: str,
+        content: str,
+        account: AlipayAccount,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+    ) -> AlipayOutboundResult:
+        if not account.is_configured():
+            raise AlipayError(code=AlipayErrorCode.UNKNOWN, message="缺少账户配置")
+
+        if not self._check_window(target_id, account.account_id):
+            return AlipayOutboundResult(
+                msg_id=None,
+                success=False,
+                error="超过 48 小时交互窗口，无法发送消息",
+            )
+
+        if not self._check_daily_limit(target_id, account.account_id):
+            return AlipayOutboundResult(
+                msg_id=None,
+                success=False,
+                error="单用户每日消息数已达上限（100条）",
+            )
+
+        if len(content) > ALIPAY_TEXT_LIMIT:
+            content = content[:ALIPAY_TEXT_LIMIT]
+
+        try:
+            result = await self._gateway.request(
+                account=account,
+                method="alipay.open.public.message.custom.send",
+                biz_content={
+                    "to_user_id": target_id,
+                    "msg_type": "text",
+                    "text": {"content": content},
+                },
+            )
+            self._increment_daily_counter(target_id, account.account_id)
+            return AlipayOutboundResult(
+                msg_id=result.get("msg_id"),
+                success=True,
+                result=result,
+            )
+        except AlipayError as e:
+            return AlipayOutboundResult(
+                msg_id=None,
+                success=False,
+                error=str(e),
+            )
+
+    async def send_template(
+        self,
+        account: AlipayAccount,
+        to_user_id: str,
+        template_id: str,
+        context: dict,
+        url: str | None = None,
+    ) -> AlipayOutboundResult:
+        if not account.is_configured():
+            raise AlipayError(code=AlipayErrorCode.UNKNOWN, message="缺少账户配置")
+
+        biz_content: dict = {
+            "to_user_id": to_user_id,
+            "template": {
+                "template_id": template_id,
+                "context": context,
+            },
+        }
+        if url:
+            biz_content["url"] = url
+
+        try:
+            result = await self._gateway.request(
+                account=account,
+                method="alipay.open.public.message.single.send",
+                biz_content=biz_content,
+            )
+            return AlipayOutboundResult(
+                msg_id=result.get("msg_id"),
+                success=True,
+                result=result,
+            )
+        except AlipayError as e:
+            return AlipayOutboundResult(
+                msg_id=None,
+                success=False,
+                error=str(e),
+            )
+
+    async def recall_message(
+        self,
+        account: AlipayAccount,
+        msg_id: str,
+    ) -> AlipayOutboundResult:
+        if not account.is_configured():
+            raise AlipayError(code=AlipayErrorCode.UNKNOWN, message="缺少账户配置")
+
+        try:
+            result = await self._gateway.request(
+                account=account,
+                method="alipay.open.public.life.msg.recall",
+                biz_content={"msg_id": msg_id},
+            )
+            return AlipayOutboundResult(
+                msg_id=msg_id,
+                success=True,
+                result=result,
+            )
+        except AlipayError as e:
+            return AlipayOutboundResult(
+                msg_id=None,
+                success=False,
+                error=str(e),
+            )
+
+    async def query_followers(
+        self,
+        account: AlipayAccount,
+        next_token: str = "",
+    ) -> AlipayOutboundResult:
+        if not account.is_configured():
+            raise AlipayError(code=AlipayErrorCode.UNKNOWN, message="缺少账户配置")
+
+        biz_content: dict = {}
+        if next_token:
+            biz_content["next_token"] = next_token
+
+        try:
+            result = await self._gateway.request(
+                account=account,
+                method="alipay.open.public.follow.batchquery",
+                biz_content=biz_content,
+            )
+            return AlipayOutboundResult(
+                msg_id=None,
+                success=True,
+                result=result,
+            )
+        except AlipayError as e:
+            return AlipayOutboundResult(
+                msg_id=None,
+                success=False,
+                error=str(e),
+            )
+
+    def sanitize_text(self, text: str, payload: dict | None = None) -> str:
+        return text
+
+    def should_skip_plain_text_sanitization(self, payload: dict | None = None) -> bool:
+        return False
+
+    def normalize_payload(self, payload: dict, config: dict, account_id: str | None = None) -> dict:
+        return payload
+
+    def resolve_effective_text_chunk_limit(
+        self, config: dict, account_id: str | None = None, fallback_limit: int | None = None
+    ) -> int:
+        return fallback_limit or ALIPAY_TEXT_LIMIT
+
+    def chunker(self, text: str, limit: int, ctx=None) -> list[str]:
+        return split_utf8_safe(text, limit)
+
+    async def send_payload(self, ctx) -> None:
+        return None
+
+    async def send_poll(self, ctx) -> None:
+        return None
+
+    async def send_media(
+        self,
+        target_id: str,
+        media_url: str,
+        media_type: str,
+        account: AlipayAccount,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+    ) -> AlipayOutboundResult:
+        if not account.is_configured():
+            raise AlipayError(code=AlipayErrorCode.UNKNOWN, message="缺少账户配置")
+
+        if not self._check_window(target_id, account.account_id):
+            return AlipayOutboundResult(
+                msg_id=None,
+                success=False,
+                error="超过 48 小时交互窗口，无法发送消息",
+            )
+
+        try:
+            result = await self._gateway.request(
+                account=account,
+                method="alipay.open.public.message.custom.send",
+                biz_content={
+                    "to_user_id": target_id,
+                    "msg_type": "image-text",
+                    "articles": [
+                        {
+                            "title": "",
+                            "desc": "",
+                            "image_url": media_url,
+                            "url": media_url,
+                        }
+                    ],
+                },
+            )
+            return AlipayOutboundResult(
+                msg_id=result.get("msg_id"),
+                success=True,
+                result=result,
+            )
+        except AlipayError as e:
+            return AlipayOutboundResult(
+                msg_id=None,
+                success=False,
+                error=str(e),
+            )
+
+    async def send_image_text(
+        self,
+        account: AlipayAccount,
+        to_user_id: str,
+        articles: list[dict],
+    ) -> AlipayOutboundResult:
+        if not self._check_window(to_user_id, account.account_id):
+            return AlipayOutboundResult(
+                msg_id=None,
+                success=False,
+                error="超过 48 小时交互窗口，无法发送消息",
+            )
+
+        if len(articles) > ALIPAY_IMAGE_TEXT_ARTICLE_LIMIT:
+            articles = articles[:ALIPAY_IMAGE_TEXT_ARTICLE_LIMIT]
+
+        try:
+            result = await self._gateway.request(
+                account=account,
+                method="alipay.open.public.message.custom.send",
+                biz_content={
+                    "to_user_id": to_user_id,
+                    "msg_type": "image-text",
+                    "articles": articles,
+                },
+            )
+            return AlipayOutboundResult(
+                msg_id=result.get("msg_id"),
+                success=True,
+                result=result,
+            )
+        except AlipayError as e:
+            return AlipayOutboundResult(
+                msg_id=None,
+                success=False,
+                error=str(e),
+            )

backend/package/yuxi/channel/extensions/alipay/pairing.py

@@ -0,0 +1,32 @@
+import secrets
+import time
+
+from yuxi.channel.extensions.alipay.constants import (
+    ALIPAY_PAIRING_CODE_LENGTH,
+    ALIPAY_PAIRING_CODE_TTL_SECONDS,
+)
+
+
+class AlipayPairing:
+    def __init__(self):
+        self._codes: dict[str, tuple[str, float]] = {}
+
+    def generate_code(self, peer_id: str) -> str:
+        import random
+
+        code = "".join(str(random.randint(0, 9)) for _ in range(ALIPAY_PAIRING_CODE_LENGTH))
+        self._codes[peer_id] = (code, time.time())
+        return code
+
+    def verify_code(self, peer_id: str, code: str) -> bool:
+        entry = self._codes.get(peer_id)
+        if not entry:
+            return False
+        stored_code, created_at = entry
+        if time.time() - created_at > ALIPAY_PAIRING_CODE_TTL_SECONDS:
+            del self._codes[peer_id]
+            return False
+        if not secrets.compare_digest(stored_code, code):
+            return False
+        del self._codes[peer_id]
+        return True

backend/package/yuxi/channel/extensions/alipay/plugin.json

@@ -0,0 +1,29 @@
+{
+    "id": "alipay",
+    "name": "支付宝",
+    "version": "0.1.0",
+    "description": "支付宝生活号+渠道插件，支持 RSA2 签名认证、被动回调消息接收、客服消息发送。覆盖 P0+P1 适配器（meta/capabilities/config/gateway/outbound/status/streaming/security/pairing/agentPrompt）",
+    "author": "ForcePilot Team",
+    "order": 50,
+    "dependencies": ["httpx>=0.27.0", "cryptography>=42.0.0"],
+    "capabilities": {
+        "chat_types": ["direct"],
+        "message_types": ["text", "image"],
+        "streaming": true,
+        "streaming_mode": "block",
+        "block_streaming": true,
+        "reactions": false,
+        "typing_indicator": false,
+        "threads": false,
+        "edit": false,
+        "unsend": true,
+        "reply": false,
+        "media": true,
+        "effects": false,
+        "native_commands": false,
+        "polls": false,
+        "group_management": false
+    },
+    "enabled": false,
+    "python_requires": ">=3.12"
+}

backend/package/yuxi/channel/extensions/alipay/security.py

@@ -0,0 +1,42 @@
+from yuxi.channel.extensions.alipay.types import AlipayAccount, AlipayDmPolicy
+
+
+class AlipaySecurity:
+    def __init__(self):
+        self._allow_from: list[str] = []
+
+    def resolve_dm_policy(self, account: AlipayAccount | None = None) -> dict:
+        if not account:
+            return {"mode": "disabled", "allow_from": []}
+        return {
+            "mode": account.dm_policy.value,
+            "allow_from": account.allow_from or [],
+        }
+
+    def check_allowlist(self, peer_id: str, channel_type: str = "direct", account: AlipayAccount | None = None) -> bool:
+        if not account:
+            return False
+        if account.dm_policy == AlipayDmPolicy.OPEN:
+            return True
+        if account.dm_policy == AlipayDmPolicy.DISABLED:
+            return False
+        if account.dm_policy == AlipayDmPolicy.ALLOWLIST:
+            return peer_id in (account.allow_from or [])
+        if account.dm_policy == AlipayDmPolicy.PAIRING:
+            return peer_id in self._allow_from
+        return False
+
+    def add_to_allowlist(self, peer_id: str) -> None:
+        if peer_id not in self._allow_from:
+            self._allow_from.append(peer_id)
+
+    def collect_warnings(
+        self, config: dict | None = None, account_id: str | None = None, account: AlipayAccount | None = None
+    ) -> list[str]:
+        warnings = []
+        if account:
+            if account.dm_policy == AlipayDmPolicy.OPEN:
+                warnings.append("DM 策略为 'open'，所有关注者均可发送消息")
+            if account.dm_policy == AlipayDmPolicy.OPEN and not account.allow_from:
+                warnings.append("DM 策略为 'open' 且无白名单限制")
+        return warnings

backend/package/yuxi/channel/extensions/alipay/status.py

@@ -0,0 +1,76 @@
+import logging
+import time
+
+from yuxi.channel.extensions.alipay.gateway import AlipayGateway
+from yuxi.channel.extensions.alipay.types import AlipayAccount, AlipayMode, AlipayProbeResult
+
+logger = logging.getLogger(__name__)
+
+
+class AlipayStatus:
+    def __init__(self, gateway: AlipayGateway):
+        self._gateway = gateway
+
+    async def probe(self, account: AlipayAccount) -> AlipayProbeResult:
+        start = time.monotonic()
+        try:
+            result = await self._gateway.request(
+                account=account,
+                method="alipay.open.public.life.account.get",
+            )
+            latency_ms = (time.monotonic() - start) * 1000
+            return AlipayProbeResult(
+                ok=True,
+                app_id=account.app_id,
+                name=account.name or result.get("name", ""),
+                mode=account.mode,
+                latency_ms=latency_ms,
+            )
+        except Exception as e:
+            latency_ms = (time.monotonic() - start) * 1000
+            return AlipayProbeResult(
+                ok=False,
+                app_id=account.app_id,
+                name=account.name,
+                mode=account.mode,
+                latency_ms=latency_ms,
+                error=str(e),
+            )
+
+    def build_summary(self, account: AlipayAccount | None = None) -> dict:
+        if not account:
+            return {"status": "unconfigured"}
+        configured = account.is_configured()
+        return {
+            "status": "configured" if configured else "unconfigured",
+            "app_id": account.app_id,
+            "name": account.name or account.app_id,
+            "mode": account.mode.value,
+            "dm_policy": account.dm_policy.value,
+            "enabled": account.enabled,
+        }
+
+    async def check_ready(self, account: AlipayAccount | None = None) -> bool:
+        if not account:
+            return False
+        if not account.is_configured():
+            return False
+        if not account.enabled:
+            return False
+        result = await self.probe(account)
+        return result.ok
+
+    def collect_status_issues(self, account: AlipayAccount | None = None) -> list[str]:
+        issues = []
+        if not account:
+            issues.append("未配置支付宝账户")
+            return issues
+        if not account.app_id:
+            issues.append("未设置 AppId")
+        if not account.app_private_key:
+            issues.append("未设置应用私钥")
+        if not account.alipay_public_key:
+            issues.append("未设置支付宝公钥")
+        if account.mode == AlipayMode.SANDBOX:
+            issues.append("当前使用沙箱环境")
+        return issues

backend/package/yuxi/channel/extensions/alipay/streaming.py

@@ -0,0 +1,33 @@
+from yuxi.channel.extensions.alipay.constants import (
+    ALIPAY_BLOCK_STREAMING_CHUNK_MAX_CHARS,
+    ALIPAY_BLOCK_STREAMING_CHUNK_MIN_CHARS,
+)
+
+
+class AlipayStreaming:
+    streaming_mode = "block"
+    preview_stream_throttle_ms = 160
+    preview_min_initial_chars = 18
+
+    block_streaming_enabled = True
+    block_streaming_break = "double_newline"
+    block_streaming_chunk_min_chars = ALIPAY_BLOCK_STREAMING_CHUNK_MIN_CHARS
+    block_streaming_chunk_max_chars = ALIPAY_BLOCK_STREAMING_CHUNK_MAX_CHARS
+    block_streaming_chunk_break_preference = "double_newline_second_paragraph"
+    block_streaming_coalesce_defaults = {
+        "mintime": 1.2,
+        "maxtime": 3.5,
+    }
+
+    ENABLED = True
+    STRATEGY = "block"
+    MIN_CHARS = ALIPAY_BLOCK_STREAMING_CHUNK_MIN_CHARS
+    MAX_INTERVAL_MS = 1000
+    MAX_BLOCKS = 20
+    FINISH_MARK = ""
+
+    def create_draft_stream_session(self, target_id: str) -> dict:
+        return {"target_id": target_id, "mode": "block"}
+
+    def create_block_chunker(self) -> dict:
+        return {"mode": "length", "min_chars": 50, "max_chars": 500}

backend/package/yuxi/channel/extensions/alipay/types.py

@@ -0,0 +1,67 @@
+from dataclasses import dataclass, field
+from enum import StrEnum
+
+
+class AlipayMode(StrEnum):
+    PRODUCTION = "production"
+    SANDBOX = "sandbox"
+
+
+class AlipayDmPolicy(StrEnum):
+    OPEN = "open"
+    PAIRING = "pairing"
+    ALLOWLIST = "allowlist"
+    DISABLED = "disabled"
+
+
+@dataclass
+class AlipayAccount:
+    account_id: str
+    app_id: str = ""
+    app_private_key: str = ""
+    alipay_public_key: str = ""
+    aes_key: str | None = None
+    mode: AlipayMode = AlipayMode.PRODUCTION
+    gateway_url: str = "https://openapi.alipay.com/gateway.do"
+    name: str = ""
+    enabled: bool = True
+    dm_policy: AlipayDmPolicy = AlipayDmPolicy.OPEN
+    allow_from: list[str] = field(default_factory=list)
+    remove_markdown: bool = True
+    subscribe_msg: str = ""
+    streaming_enabled: bool = True
+    block_streaming_chunk_max_chars: int = 500
+
+    def is_configured(self) -> bool:
+        return bool(self.app_id and self.app_private_key and self.alipay_public_key)
+
+
+@dataclass
+class AlipayInboundEvent:
+    msg_type: str
+    event_type: str | None
+    from_user_id: str
+    from_user_name: str | None
+    create_time: int
+    msg_id: str
+    app_id: str
+    biz_content: dict
+    raw: dict
+
+
+@dataclass
+class AlipayOutboundResult:
+    msg_id: str | None
+    success: bool
+    error: str | None = None
+    result: dict | None = None
+
+
+@dataclass
+class AlipayProbeResult:
+    ok: bool
+    app_id: str
+    name: str | None
+    mode: AlipayMode
+    latency_ms: float
+    error: str | None = None

backend/package/yuxi/channel/extensions/alipay/webhook.py

@@ -0,0 +1,258 @@
+import asyncio
+import json
+import logging
+
+from fastapi import APIRouter, Request, Response
+
+from yuxi.channel.extensions.alipay.config import AlipayConfig
+from yuxi.channel.extensions.alipay.crypto import AlipayAESCrypto, AlipayCrypto
+from yuxi.channel.extensions.alipay.dedup import AlipayMessageDeduplicator
+from yuxi.channel.extensions.alipay.format import remove_markdown
+from yuxi.channel.extensions.alipay.monitor import convert_alipay_event
+from yuxi.channel.extensions.alipay.outbound import AlipayOutbound
+from yuxi.channel.extensions.alipay.pairing import AlipayPairing
+from yuxi.channel.extensions.alipay.security import AlipaySecurity
+from yuxi.channel.extensions.alipay.types import AlipayInboundEvent
+from yuxi.channel.runtime.manager import gateway
+from yuxi.channel.message.models import MessageType, PeerInfo, PeerKind, UnifiedMessage
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/channel/alipay", tags=["alipay"])
+
+_webhook_state: dict | None = None
+
+
+def init_webhook(
+    config: AlipayConfig,
+    deduplicator: AlipayMessageDeduplicator,
+    security: AlipaySecurity,
+    pairing: AlipayPairing,
+    outbound: AlipayOutbound | None = None,
+) -> None:
+    global _webhook_state
+    _webhook_state = {
+        "config": config,
+        "deduplicator": deduplicator,
+        "security": security,
+        "pairing": pairing,
+        "outbound": outbound,
+    }
+
+
+def _get_state() -> dict:
+    if _webhook_state is None:
+        raise RuntimeError("Webhook not initialized")
+    return _webhook_state
+
+
+def _build_crypto() -> AlipayCrypto | None:
+    state = _get_state()
+    account = state["config"].resolve_account()
+    if not account.is_configured():
+        return None
+    return AlipayCrypto(
+        app_private_key_pem=account.app_private_key,
+        alipay_public_key_pem=account.alipay_public_key,
+    )
+
+
+@router.get("/callback")
+async def alipay_url_verify(request: Request):
+    params = dict(request.query_params)
+    sign = params.pop("sign", None)
+    params.pop("sign_type", None)
+
+    crypto = _build_crypto()
+    if crypto is None:
+        return Response(content="fail", status_code=503)
+
+    if sign and crypto.verify(params, sign):
+        echostr = params.get("echostr", "")
+        return Response(content=echostr)
+    return Response(content="fail", status_code=400)
+
+
+@router.post("/callback")
+async def alipay_message_callback(request: Request):
+    try:
+        data = await request.json()
+    except (json.JSONDecodeError, UnicodeDecodeError):
+        return Response(content="fail", status_code=400)
+
+    sign = data.pop("sign", None)
+    data.pop("sign_type", None)
+
+    state = _get_state()
+    account = state["config"].resolve_account()
+    if not account.is_configured():
+        return Response(content="fail", status_code=500)
+
+    crypto = _build_crypto()
+    if crypto is None:
+        return Response(content="fail", status_code=500)
+
+    if sign and not crypto.verify(data, sign):
+        logger.warning("支付宝回调签名验证失败")
+        return Response(content="fail", status_code=400)
+
+    encrypt_type = data.get("encrypt_type", "")
+    if encrypt_type == "aes" and account.aes_key:
+        aes = AlipayAESCrypto(account.aes_key)
+        encrypted = data.get("biz_content", "")
+        if isinstance(encrypted, str) and encrypted:
+            try:
+                decrypted = aes.decrypt(encrypted)
+                data = json.loads(decrypted)
+            except Exception:
+                logger.exception("AES decrypt failed")
+                return Response(content="fail", status_code=400)
+
+    msg_type = data.get("msg_type", data.get("MsgType", ""))
+    from_user_id = data.get("from_user_id", data.get("FromAlipayUserId", ""))
+    msg_id = data.get("msg_id", data.get("MsgId", ""))
+
+    biz_content = data.get("biz_content", {})
+    if isinstance(biz_content, str):
+        try:
+            biz_content = json.loads(biz_content)
+        except json.JSONDecodeError:
+            biz_content = {}
+
+    if not msg_id or state["deduplicator"].is_duplicate(msg_id):
+        return Response(content="success")
+
+    event = AlipayInboundEvent(
+        msg_type=msg_type,
+        event_type=data.get("event_type", data.get("EventType")),
+        from_user_id=from_user_id,
+        from_user_name=data.get("from_user_name", data.get("FromAlipayUserName")),
+        create_time=int(data.get("create_time", data.get("CreateTime", 0))),
+        msg_id=msg_id,
+        app_id=data.get("app_id", data.get("AppId", "")),
+        biz_content=biz_content,
+        raw=data,
+    )
+
+    security = state["security"]
+    if not security.check_allowlist(from_user_id, "direct", account):
+        if account.dm_policy.value == "pairing":
+            content = biz_content.get("content", "") if isinstance(biz_content, dict) else ""
+            if content.strip().startswith("配对 "):
+                code_input = content.strip()[3:].strip()
+                if state["pairing"].verify_code(from_user_id, code_input):
+                    return await _handle_pairing_success(from_user_id, security)
+                else:
+                    await _send_alipay_text(from_user_id, "配对码无效或已过期，请重新发送消息获取配对码。")
+            else:
+                code = state["pairing"].generate_code(from_user_id)
+                if code:
+                    await _send_alipay_text(
+                        from_user_id,
+                        f"首次对话需要验证身份，请输入以下配对码：\n\n配对 {code}\n\n（配对码有效期 5 分钟）",
+                    )
+                else:
+                    await _send_alipay_text(from_user_id, "配对请求过于频繁，请稍后再试。")
+            return Response(content="success")
+        return Response(content="success")
+
+    outbound = state.get("outbound")
+    if outbound:
+        outbound.record_interaction(from_user_id, account.account_id)
+
+    if event.msg_type == "event":
+        return await _handle_event(event, account)
+
+    content_str = (event.biz_content or {}).get("content", "")
+    if not content_str:
+        return Response(content="success")
+
+    if account.remove_markdown:
+        content_str = remove_markdown(content_str)
+
+    return await _dispatch_to_agent(event, content_str)
+
+
+async def _handle_event(event: AlipayInboundEvent, account) -> Response:
+    if event.event_type in ("follow", "enter"):
+        unified = convert_alipay_event(event, account)
+        if unified:
+            await _dispatch_unified(unified)
+        welcome = account.subscribe_msg or ""
+        if welcome:
+            await _send_alipay_text(event.from_user_id, welcome)
+            logger.info("Sent welcome message to %s", event.from_user_id)
+    return Response(content="success")
+
+
+async def _dispatch_to_agent(event: AlipayInboundEvent, content_str: str) -> Response:
+    state = _get_state()
+    account = state["config"].resolve_account()
+
+    unified = UnifiedMessage(
+        msg_id=event.msg_id,
+        channel_type="alipay",
+        account_id=account.account_id,
+        content=content_str,
+        message_type=MessageType.TEXT if event.msg_type == "text" else MessageType.IMAGE,
+        sender=PeerInfo(
+            id=event.from_user_id,
+            kind=PeerKind.DIRECT,
+            display_name=event.from_user_name or event.from_user_id,
+        ),
+        raw_payload=event.raw,
+        body_for_agent=content_str,
+        metadata={
+            "FromAlipayUserId": event.from_user_id,
+            "AppId": event.app_id,
+            "MsgType": event.msg_type,
+            "EventType": event.event_type or "",
+        },
+    )
+
+    await _dispatch_unified(unified)
+    return Response(content="success")
+
+
+async def _dispatch_unified(msg: UnifiedMessage) -> None:
+    processor = gateway._processor
+    if processor is None:
+        logger.warning("Message processor not available, cannot dispatch Alipay message")
+        return
+
+    asyncio.create_task(
+        _run_processor(processor, msg),
+        name=f"alipay-dispatch-{msg.sender.id}",
+    )
+
+
+async def _run_processor(processor, msg: UnifiedMessage) -> None:
+    try:
+        await asyncio.wait_for(processor.process(msg), timeout=120.0)
+    except TimeoutError:
+        logger.error("Agent response timeout for alipay user %s", msg.sender.id)
+    except Exception:
+        logger.exception("Failed to process Alipay message for user %s", msg.sender.id)
+
+
+async def _handle_pairing_success(from_user_id: str, security: AlipaySecurity) -> Response:
+    security.add_to_allowlist(from_user_id)
+    await _send_alipay_text(from_user_id, "配对成功！现在可以开始对话了。")
+    return Response(content="success")
+
+
+async def _send_alipay_text(to_user: str, content: str) -> None:
+    state = _get_state()
+    account = state["config"].resolve_account()
+    if not account.is_configured():
+        logger.warning("Alipay account not configured, cannot send text")
+        return
+
+    outbound: AlipayOutbound | None = state.get("outbound")
+    if outbound is None:
+        return
+
+    try:
+        await outbound.send_text(to_user, content, account=account)
+    except Exception:
+        logger.exception("Failed to send Alipay text to %s", to_user)