refactor(notification): 建立通知渠道通用管理契约 channel_manage

- schemas 层新增 NotificationAction 公共动作词汇表（状态/二维码/退出/连通性/缓存迁移）
- 模块实现统一 channel_manage(channel, action, **params)，按渠道名路由，
  非本渠道返回 None；动作语义、表单参数解释与临时参数初始化封闭在模块内
- NotificationChain 收敛为单一 manage_channel 纯透明转发，
  chain 层不再出现任何渠道名与模块特色
- endpoint 路由路径保持前端兼容，内部统一走通用接口
- 新增 6 项契约守护测试，文档记录通用模式：新渠道接入无需改 chain

app/api/endpoints/notification.py

@@ -7,6 +7,7 @@ from app.api.response import ResponseAPIRouter
 from app.chain.notification import NotificationChain
 from app.db.models import User
 from app.api.deps import get_current_active_superuser
+from app.schemas.types import MessageChannel, NotificationAction
 
 router = ResponseAPIRouter()
 
@@ -27,15 +28,17 @@ def wechatclawbot_status(
     _: User = Depends(get_current_active_superuser),
 ):
     """查询微信 ClawBot 登录状态和二维码。"""
-    result = NotificationChain().get_wechatclawbot_status(
+    result = NotificationChain().manage_channel(
+        channel=MessageChannel.WechatClawBot,
+        action=NotificationAction.STATUS,
         source=source,
         fallback_source=fallback_source,
+        refresh_remote=refresh_remote,
+        auto_generate_qrcode=auto_generate_qrcode,
         WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
         WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
         WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
         WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
-        refresh_remote=refresh_remote,
-        auto_generate_qrcode=auto_generate_qrcode,
     )
     return schemas.Response(
         success=bool(result.get("success")),
@@ -59,7 +62,9 @@ def refresh_wechatclawbot_qrcode(
     _: User = Depends(get_current_active_superuser),
 ):
     """刷新微信 ClawBot 二维码。"""
-    result = NotificationChain().refresh_wechatclawbot_qrcode(
+    result = NotificationChain().manage_channel(
+        channel=MessageChannel.WechatClawBot,
+        action=NotificationAction.REFRESH_QRCODE,
         source=source,
         fallback_source=fallback_source,
         WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
@@ -89,7 +94,9 @@ def logout_wechatclawbot(
     _: User = Depends(get_current_active_superuser),
 ):
     """退出微信 ClawBot 登录。"""
-    result = NotificationChain().logout_wechatclawbot(
+    result = NotificationChain().manage_channel(
+        channel=MessageChannel.WechatClawBot,
+        action=NotificationAction.LOGOUT,
         source=source,
         fallback_source=fallback_source,
         WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
@@ -119,7 +126,9 @@ def test_wechatclawbot(
     _: User = Depends(get_current_active_superuser),
 ):
     """测试微信 ClawBot 当前登录态是否可用。"""
-    result = NotificationChain().test_wechatclawbot_connection(
+    result = NotificationChain().manage_channel(
+        channel=MessageChannel.WechatClawBot,
+        action=NotificationAction.TEST_CONNECTION,
         source=source,
         fallback_source=fallback_source,
         WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
@@ -143,10 +152,12 @@ def migrate_wechatclawbot_cache(
     _: User = Depends(get_current_active_superuser),
 ):
     """在通知名称变更时迁移对应的微信 ClawBot 登录缓存。"""
-    success, message = NotificationChain().migrate_wechatclawbot_cache(
+    result = NotificationChain().manage_channel(
+        channel=MessageChannel.WechatClawBot,
+        action=NotificationAction.MIGRATE_CACHE,
         old_name=old_source,
         new_name=new_source,
         cleanup_old=cleanup_old,
         overwrite=overwrite,
     )
-    return schemas.Response(success=success, message=message)
+    return schemas.Response(success=bool(result.get("success")), message=result.get("message"))

app/chain/notification.py

@@ -1,114 +1,30 @@
-from typing import Any, Dict, Tuple
+from typing import Any, Dict
 
 from app.chain import ChainBase
+from app.schemas.types import MessageChannel, NotificationAction
 
 
 class NotificationChain(ChainBase):
     """
-    通知渠道管理链，仅做模块方法名契约的薄分发，渠道连接与能力全部封闭在模块内部
+    通知渠道管理链，仅按渠道名透明转发管理动作到模块
+
+    不包含任何渠道特定逻辑：动作语义、表单参数解释、客户端实例解析与
+    临时参数初始化全部封闭在实现 channel_manage 契约的模块内部
     """
 
-    def get_wechatclawbot_status(
+    def manage_channel(
             self,
-            source=None,
-            fallback_source=None,
-            WECHATCLAWBOT_BASE_URL=None,
-            WECHATCLAWBOT_DEFAULT_TARGET=None,
-            WECHATCLAWBOT_ADMINS=None,
-            WECHATCLAWBOT_POLL_TIMEOUT=None,
-            refresh_remote: bool = True,
-            auto_generate_qrcode: bool = True,
+            channel: MessageChannel,
+            action: NotificationAction,
+            **params: Any,
     ) -> Dict[str, Any]:
-        """查询微信 ClawBot 登录状态与二维码。"""
-        result = self.run_module(
-            "wechatclawbot_status",
-            source=source,
-            fallback_source=fallback_source,
-            WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
-            WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
-            WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
-            WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
-            refresh_remote=refresh_remote,
-            auto_generate_qrcode=auto_generate_qrcode,
-        )
-        return result or {"success": False, "message": "微信 ClawBot 通知未启用或配置尚未保存"}
+        """
+        对指定通知渠道执行管理动作
 
-    def refresh_wechatclawbot_qrcode(
-            self,
-            source=None,
-            fallback_source=None,
-            WECHATCLAWBOT_BASE_URL=None,
-            WECHATCLAWBOT_DEFAULT_TARGET=None,
-            WECHATCLAWBOT_ADMINS=None,
-            WECHATCLAWBOT_POLL_TIMEOUT=None,
-    ) -> Dict[str, Any]:
-        """刷新微信 ClawBot 登录二维码。"""
-        result = self.run_module(
-            "wechatclawbot_refresh_qrcode",
-            source=source,
-            fallback_source=fallback_source,
-            WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
-            WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
-            WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
-            WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
-        )
-        return result or {"success": False, "message": "微信 ClawBot 通知未启用或配置尚未保存"}
-
-    def logout_wechatclawbot(
-            self,
-            source=None,
-            fallback_source=None,
-            WECHATCLAWBOT_BASE_URL=None,
-            WECHATCLAWBOT_DEFAULT_TARGET=None,
-            WECHATCLAWBOT_ADMINS=None,
-            WECHATCLAWBOT_POLL_TIMEOUT=None,
-    ) -> Dict[str, Any]:
-        """退出微信 ClawBot 登录。"""
-        result = self.run_module(
-            "wechatclawbot_logout",
-            source=source,
-            fallback_source=fallback_source,
-            WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
-            WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
-            WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
-            WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
-        )
-        return result or {"success": False, "message": "微信 ClawBot 通知未启用或配置尚未保存"}
-
-    def test_wechatclawbot_connection(
-            self,
-            source=None,
-            fallback_source=None,
-            WECHATCLAWBOT_BASE_URL=None,
-            WECHATCLAWBOT_DEFAULT_TARGET=None,
-            WECHATCLAWBOT_ADMINS=None,
-            WECHATCLAWBOT_POLL_TIMEOUT=None,
-    ) -> Dict[str, Any]:
-        """测试微信 ClawBot 当前登录态是否可用。"""
-        result = self.run_module(
-            "wechatclawbot_test_connection",
-            source=source,
-            fallback_source=fallback_source,
-            WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
-            WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
-            WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
-            WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
-        )
-        return result or {"success": False, "message": "微信 ClawBot 通知未启用或配置尚未保存"}
-
-    def migrate_wechatclawbot_cache(
-            self,
-            old_name: str,
-            new_name: str,
-            cleanup_old: bool = False,
-            overwrite: bool = False,
-    ) -> Tuple[bool, str]:
-        """在通知名称变更时迁移对应的微信 ClawBot 登录缓存。"""
-        result = self.run_module(
-            "wechatclawbot_migrate_cache",
-            old_name=old_name,
-            new_name=new_name,
-            cleanup_old=cleanup_old,
-            overwrite=overwrite,
-        )
-        return result or (False, "微信 ClawBot 通知未启用")
+        :param channel: 渠道标识，用于模块路由
+        :param action: 通用管理动作，具体语义由渠道模块解释
+        :param params: 表单与动作参数，原样透传给模块
+        :return: 统一结构 {"success": bool, "message": ..., ...}
+        """
+        result = self.run_module("channel_manage", channel=channel, action=action, **params)
+        return result or {"success": False, "message": "该通知渠道未启用或不支持此管理动作"}

app/modules/wechatclawbot/__init__.py

@@ -12,7 +12,7 @@ from app.runtime.log import logger
 from app.modules import _MessageBase, _ModuleBase
 from app.modules.wechatclawbot.wechatclawbot import WechatClawBot
 from app.schemas import CommingMessage, Notification
-from app.schemas.types import MessageChannel, ModuleType
+from app.schemas.types import MessageChannel, ModuleType, NotificationAction
 
 
 register_channel_admin_resolver(
@@ -83,22 +83,60 @@ class WechatClawBotModule(_ModuleBase, _MessageBase[WechatClawBot]):
         """初始化模块设置。"""
         pass
 
-    def _resolve_client(
+    def channel_manage(
             self,
-            source: Optional[str] = None,
-            fallback_source: Optional[str] = None,
-            WECHATCLAWBOT_BASE_URL: Optional[str] = None,
-            WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None,
-            WECHATCLAWBOT_ADMINS: Optional[str] = None,
-            WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None,
-    ) -> Tuple[Optional[Any], Optional[str]]:
+            channel: MessageChannel,
+            action: NotificationAction,
+            **params: Any,
+    ) -> Optional[Dict[str, Any]]:
+        """通知渠道通用管理入口，按渠道名路由，仅处理本渠道。
+
+        动作语义与表单参数全部由模块自行解释：优先使用已保存配置实例，
+        无匹配配置时可基于表单参数构造临时实例（未保存配置的扫码预览）。
+        统一返回 {"success": bool, "message": ..., ...} 结构。
+        """
+        if channel != self.get_subtype():
+            return None
+        try:
+            action = NotificationAction(action)
+        except ValueError:
+            return {"success": False, "message": f"不支持的渠道管理动作：{action}"}
+
+        if action == NotificationAction.MIGRATE_CACHE:
+            success, message = WechatClawBot.migrate_cached_state(
+                old_name=params.get("old_name"),
+                new_name=params.get("new_name"),
+                cleanup_old=bool(params.get("cleanup_old")),
+                overwrite=bool(params.get("overwrite")),
+            )
+            return {"success": success, "message": message}
+
+        client, errmsg = self._resolve_client(params)
+        if not client:
+            return {"success": False, "message": errmsg}
+
+        if action == NotificationAction.STATUS:
+            return client.get_status(
+                refresh_remote=bool(params.get("refresh_remote", True)),
+                auto_generate_qrcode=bool(params.get("auto_generate_qrcode", True)),
+            )
+        if action == NotificationAction.REFRESH_QRCODE:
+            return client.refresh_qrcode()
+        if action == NotificationAction.LOGOUT:
+            return client.logout()
+        if action == NotificationAction.TEST_CONNECTION:
+            state, message = client.test_connection()
+            return {"success": state, "message": message}
+        return {"success": False, "message": f"不支持的渠道管理动作：{action.value}"}
+
+    def _resolve_client(self, params: Dict[str, Any]) -> Tuple[Optional[Any], Optional[str]]:
         """解析微信 ClawBot 客户端实例，返回 (客户端, 错误信息)。
 
         优先使用已加载的配置实例，均无配置时退回到基于表单参数的临时客户端，
         用于未保存配置的扫码状态预览。
         """
-        source_name = str(source or "").strip() or None
-        fallback_name = str(fallback_source or "").strip() or None
+        source_name = str(params.get("source") or "").strip() or None
+        fallback_name = str(params.get("fallback_source") or "").strip() or None
 
         candidate_names = []
         for candidate in (fallback_name, source_name):
@@ -117,13 +155,7 @@ class WechatClawBotModule(_ModuleBase, _MessageBase[WechatClawBot]):
             if client:
                 return client, None
 
-        temp_client = self._build_temp_client(
-            source=source_name or fallback_name,
-            WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
-            WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
-            WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
-            WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
-        )
+        temp_client = self._build_temp_client(params)
         if temp_client:
             return temp_client, None
 
@@ -131,136 +163,20 @@ class WechatClawBotModule(_ModuleBase, _MessageBase[WechatClawBot]):
             return None, f"未找到名为 {source_name} 的微信 ClawBot 通知配置"
         return None, "微信 ClawBot 通知未启用或配置尚未保存，请先保存并启用当前渠道"
 
-    def wechatclawbot_status(
-            self,
-            source: Optional[str] = None,
-            fallback_source: Optional[str] = None,
-            WECHATCLAWBOT_BASE_URL: Optional[str] = None,
-            WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None,
-            WECHATCLAWBOT_ADMINS: Optional[str] = None,
-            WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None,
-            refresh_remote: bool = True,
-            auto_generate_qrcode: bool = True,
-    ) -> Dict[str, Any]:
-        """查询微信 ClawBot 登录状态与二维码，实例解析全部封闭在模块内部。"""
-        client, errmsg = self._resolve_client(
-            source=source,
-            fallback_source=fallback_source,
-            WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
-            WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
-            WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
-            WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
-        )
-        if not client:
-            return {"success": False, "message": errmsg}
-        return client.get_status(
-            refresh_remote=refresh_remote,
-            auto_generate_qrcode=auto_generate_qrcode,
-        )
-
-    def wechatclawbot_refresh_qrcode(
-            self,
-            source: Optional[str] = None,
-            fallback_source: Optional[str] = None,
-            WECHATCLAWBOT_BASE_URL: Optional[str] = None,
-            WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None,
-            WECHATCLAWBOT_ADMINS: Optional[str] = None,
-            WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None,
-    ) -> Dict[str, Any]:
-        """刷新微信 ClawBot 登录二维码。"""
-        client, errmsg = self._resolve_client(
-            source=source,
-            fallback_source=fallback_source,
-            WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
-            WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
-            WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
-            WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
-        )
-        if not client:
-            return {"success": False, "message": errmsg}
-        return client.refresh_qrcode()
-
-    def wechatclawbot_logout(
-            self,
-            source: Optional[str] = None,
-            fallback_source: Optional[str] = None,
-            WECHATCLAWBOT_BASE_URL: Optional[str] = None,
-            WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None,
-            WECHATCLAWBOT_ADMINS: Optional[str] = None,
-            WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None,
-    ) -> Dict[str, Any]:
-        """退出微信 ClawBot 登录。"""
-        client, errmsg = self._resolve_client(
-            source=source,
-            fallback_source=fallback_source,
-            WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
-            WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
-            WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
-            WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
-        )
-        if not client:
-            return {"success": False, "message": errmsg}
-        return client.logout()
-
-    def wechatclawbot_test_connection(
-            self,
-            source: Optional[str] = None,
-            fallback_source: Optional[str] = None,
-            WECHATCLAWBOT_BASE_URL: Optional[str] = None,
-            WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None,
-            WECHATCLAWBOT_ADMINS: Optional[str] = None,
-            WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None,
-    ) -> Dict[str, Any]:
-        """测试微信 ClawBot 当前登录态是否可用。"""
-        client, errmsg = self._resolve_client(
-            source=source,
-            fallback_source=fallback_source,
-            WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
-            WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
-            WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
-            WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
-        )
-        if not client:
-            return {"success": False, "message": errmsg}
-        state, message = client.test_connection()
-        return {"success": state, "message": message}
-
-    def _build_temp_client(
-            self,
-            source: Optional[str] = None,
-            WECHATCLAWBOT_BASE_URL: Optional[str] = None,
-            WECHATCLAWBOT_DEFAULT_TARGET: Optional[str] = None,
-            WECHATCLAWBOT_ADMINS: Optional[str] = None,
-            WECHATCLAWBOT_POLL_TIMEOUT: Optional[int] = None,
-    ):
-        """基于当前表单配置创建一个临时客户端，用于未保存时的扫码状态预览。"""
-        source_name = str(source or "").strip()
+    def _build_temp_client(self, params: Dict[str, Any]) -> Optional[Any]:
+        """基于表单参数创建临时客户端，用于未保存配置时的扫码状态预览。"""
+        source_name = str(params.get("source") or params.get("fallback_source") or "").strip()
         if not source_name:
             return None
         return WechatClawBot(
             name=source_name,
-            WECHATCLAWBOT_BASE_URL=WECHATCLAWBOT_BASE_URL,
-            WECHATCLAWBOT_DEFAULT_TARGET=WECHATCLAWBOT_DEFAULT_TARGET,
-            WECHATCLAWBOT_ADMINS=WECHATCLAWBOT_ADMINS,
-            WECHATCLAWBOT_POLL_TIMEOUT=WECHATCLAWBOT_POLL_TIMEOUT,
+            WECHATCLAWBOT_BASE_URL=params.get("WECHATCLAWBOT_BASE_URL"),
+            WECHATCLAWBOT_DEFAULT_TARGET=params.get("WECHATCLAWBOT_DEFAULT_TARGET"),
+            WECHATCLAWBOT_ADMINS=params.get("WECHATCLAWBOT_ADMINS"),
+            WECHATCLAWBOT_POLL_TIMEOUT=params.get("WECHATCLAWBOT_POLL_TIMEOUT"),
             auto_start_polling=False,
         )
 
-    def wechatclawbot_migrate_cache(
-            self,
-            old_name: str,
-            new_name: str,
-            cleanup_old: bool = False,
-            overwrite: bool = False,
-    ):
-        """在通知名称变更时迁移对应的微信 ClawBot 登录缓存。"""
-        return WechatClawBot.migrate_cached_state(
-            old_name=old_name,
-            new_name=new_name,
-            cleanup_old=cleanup_old,
-            overwrite=overwrite,
-        )
-
     @staticmethod
     def _load_json(body: Any) -> Optional[dict]:
         """将内容解析为 JSON 字典。"""

app/schemas/types.py

@@ -491,6 +491,24 @@ class MessageChannel(Enum):
     QQ = "QQ"
 
 
+class NotificationAction(str, Enum):
+    """
+    通知渠道通用管理动作
+
+    作为渠道管理契约的公共词汇表，具体动作的支持范围与参数语义由渠道模块自行解释
+    """
+    # 查询登录状态与二维码
+    STATUS = "status"
+    # 刷新登录二维码
+    REFRESH_QRCODE = "refresh_qrcode"
+    # 退出登录
+    LOGOUT = "logout"
+    # 测试连通性
+    TEST_CONNECTION = "test_connection"
+    # 迁移渠道名变更前的登录缓存
+    MIGRATE_CACHE = "migrate_cache"
+
+
 # 下载器类型
 class DownloaderType(Enum):
     # Qbittorrent

docs/rules/05-architecture.md

@@ -212,6 +212,15 @@ exceptions and value domains used by both modules and upper layers live in
 method names. The directory remains unchanged because discovery and plugin code
 depend on this established runtime root.
 
+Channels that need login management or temporary-parameter initialization
+follow one generic contract instead of per-channel APIs: modules implement
+`channel_manage(channel, action, **params)`, route by the requested
+`MessageChannel` (returning `None` for other channels), and interpret actions
+from the shared `schemas.types.NotificationAction` vocabulary plus opaque form
+parameters themselves. `NotificationChain.manage_channel` forwards transparently
+and must stay free of any channel-specific names or logic; new channels adopt
+the same contract without touching the chain.
+
 ### DB / Oper layer
 
 SQLAlchemy models stay under `app/db/models/`; the data access classes live in

tests/test_notification_channel_manage.py

@@ -0,0 +1,120 @@
+"""
+通知渠道通用管理契约（channel_manage）守护测试
+
+验证通用模式的三条核心性质：
+1. 按渠道名路由：非本渠道直接返回 None，交由分发机制继续执行其它模块
+2. 动作词汇表由 schemas 契约层统一定义，未支持动作返回统一错误结构
+3. 临时参数初始化封闭在模块内：无已保存配置时可基于表单参数构造临时客户端
+"""
+from types import SimpleNamespace
+
+import pytest
+
+from app.modules.wechatclawbot import WechatClawBotModule
+from app.schemas.types import MessageChannel, NotificationAction
+
+
+@pytest.fixture
+def module():
+    return WechatClawBotModule()
+
+
+def test_channel_manage_routes_only_matching_channel(module):
+    """非本渠道的管理请求返回 None，run_module 分发将继续执行其它模块。"""
+    result = module.channel_manage(
+        channel=MessageChannel.Telegram,
+        action=NotificationAction.STATUS,
+    )
+    assert result is None
+
+
+def test_channel_manage_rejects_unknown_action(module):
+    """动作词汇表之外的请求返回统一错误结构。"""
+    result = module.channel_manage(
+        channel=MessageChannel.WechatClawBot,
+        action="not_an_action",
+    )
+    assert result["success"] is False
+    assert "不支持" in result["message"]
+
+
+def test_channel_manage_requires_saved_config_without_form_params(module, monkeypatch):
+    """无任何配置且未提供表单参数时，返回提示保存配置的错误。"""
+    monkeypatch.setattr(module, "get_instance", lambda name=None: None)
+    result = module.channel_manage(
+        channel=MessageChannel.WechatClawBot,
+        action=NotificationAction.TEST_CONNECTION,
+    )
+    assert result["success"] is False
+    assert "保存" in result["message"]
+
+
+def test_channel_manage_builds_temporary_client_from_form_params(module, monkeypatch):
+    """无已保存配置时基于表单参数构造临时客户端，实现未保存即预览。"""
+    monkeypatch.setattr(module, "get_instance", lambda name=None: None)
+
+    captured = {}
+
+    class _FakeClient:
+        def test_connection(self):
+            captured["called"] = True
+            return False, "未登录，请先扫码完成绑定"
+
+    monkeypatch.setattr(
+        "app.modules.wechatclawbot.WechatClawBot",
+        lambda **kwargs: (captured.update(kwargs=kwargs), _FakeClient())[1],
+    )
+
+    result = module.channel_manage(
+        channel=MessageChannel.WechatClawBot,
+        action=NotificationAction.TEST_CONNECTION,
+        source="预览渠道",
+        WECHATCLAWBOT_BASE_URL="http://127.0.0.1:1",
+    )
+
+    assert captured["called"] is True
+    assert captured["kwargs"]["name"] == "预览渠道"
+    assert captured["kwargs"]["auto_start_polling"] is False
+    assert result["success"] is False
+    assert "未登录" in result["message"]
+
+
+def test_channel_manage_prefers_saved_instance(module, monkeypatch):
+    """存在已保存配置实例时优先使用，不构造临时客户端。"""
+    saved = SimpleNamespace()
+    saved.get_status = lambda refresh_remote, auto_generate_qrcode: {
+        "success": True,
+        "connected": True,
+    }
+    monkeypatch.setattr(module, "get_config", lambda name=None: SimpleNamespace(name="已保存"))
+    monkeypatch.setattr(module, "get_instance", lambda name=None: saved)
+
+    result = module.channel_manage(
+        channel=MessageChannel.WechatClawBot,
+        action=NotificationAction.STATUS,
+        source="已保存",
+    )
+    assert result["success"] is True
+    assert result["connected"] is True
+
+
+def test_channel_manage_migrate_cache_dispatches_without_client(module, monkeypatch):
+    """缓存迁移动作不依赖客户端实例，直接走静态迁移逻辑。"""
+
+    def _fake_migrate(old_name, new_name, cleanup_old, overwrite):
+        assert old_name == "旧名"
+        assert new_name == "新名"
+        return True, "迁移成功"
+
+    monkeypatch.setattr(
+        "app.modules.wechatclawbot.WechatClawBot.migrate_cached_state",
+        staticmethod(_fake_migrate),
+    )
+
+    result = module.channel_manage(
+        channel=MessageChannel.WechatClawBot,
+        action=NotificationAction.MIGRATE_CACHE,
+        old_name="旧名",
+        new_name="新名",
+    )
+    assert result == {"success": True, "message": "迁移成功"}