refactor: unify message ingress forwarding

app/application/messaging/ingress.py

@@ -0,0 +1,115 @@
+"""消息渠道回环进入宿主消息 API 的统一适配边界。"""
+
+from collections.abc import Callable, Mapping
+from typing import Any
+from urllib.parse import urlencode
+
+from app.adapters.network.http import AsyncRequestUtils, RequestUtils
+from app.runtime.log import logger
+from app.runtime.settings import RuntimeSettingsCompat
+
+settings = RuntimeSettingsCompat()
+
+BackgroundSubmitter = Callable[..., object]
+
+
+def build_message_ingress_url(source: str | None) -> str:
+    """按当前运行配置构造安全编码的本地消息入口 URL。"""
+    query = {"token": settings.API_TOKEN}
+    if source:
+        query["source"] = source
+    return f"http://127.0.0.1:{settings.PORT}/api/v1/message?{urlencode(query)}"
+
+
+def forward_message_to_host(
+    payload: Mapping[str, Any],
+    source: str | None,
+    *,
+    timeout: float = 15,
+) -> bool:
+    """同步转发渠道 payload，统一判断本地入口是否确认接收。"""
+    response = None
+    try:
+        response = RequestUtils(timeout=timeout).post_res(
+            build_message_ingress_url(source),
+            json=dict(payload),
+        )
+        if response is None:
+            logger.error(f"转发渠道消息到本地入口失败：source={source or '-'} - 无响应")
+            return False
+        if response.status_code >= 400:
+            logger.error(
+                "转发渠道消息到本地入口失败："
+                f"source={source or '-'} - HTTP {response.status_code}"
+            )
+            return False
+        return True
+    except Exception as error:
+        logger.error(f"转发渠道消息到本地入口失败：source={source or '-'} - {error}")
+        return False
+    finally:
+        close = getattr(response, "close", None)
+        if callable(close):
+            try:
+                close()
+            except Exception as error:
+                logger.debug(
+                    f"释放本地消息入口响应失败：source={source or '-'} - {error}"
+                )
+
+
+async def async_forward_message_to_host(
+    payload: Mapping[str, Any],
+    source: str | None,
+    *,
+    timeout: float = 15,
+) -> bool:
+    """异步转发渠道 payload，供自有事件循环的消息 SDK 复用同一确认语义。"""
+    response = None
+    try:
+        response = await AsyncRequestUtils(timeout=timeout).post_res(
+            build_message_ingress_url(source),
+            json=dict(payload),
+        )
+        if response is None:
+            logger.error(f"转发渠道消息到本地入口失败：source={source or '-'} - 无响应")
+            return False
+        if response.status_code >= 400:
+            logger.error(
+                "转发渠道消息到本地入口失败："
+                f"source={source or '-'} - HTTP {response.status_code}"
+            )
+            return False
+        return True
+    except Exception as error:
+        logger.error(f"转发渠道消息到本地入口失败：source={source or '-'} - {error}")
+        return False
+    finally:
+        if response is not None:
+            try:
+                await response.aclose()
+            except Exception as error:
+                logger.debug(
+                    f"释放本地消息入口响应失败：source={source or '-'} - {error}"
+                )
+
+
+def submit_message_to_host(
+    payload: Mapping[str, Any],
+    source: str | None,
+    *,
+    submit: BackgroundSubmitter,
+    timeout: float = 15,
+) -> bool:
+    """把同步回环转发提交给调用方注入的受管后台执行器。"""
+    try:
+        submit(
+            forward_message_to_host,
+            dict(payload),
+            source,
+            timeout=timeout,
+        )
+    except Exception as error:
+        logger.error(f"提交渠道消息转发任务失败：source={source or '-'} - {error}")
+        return False
+    return True

app/modules/discord/discord.py

@@ -3,16 +3,15 @@ import re
 import threading
 from pathlib import Path
 from typing import Optional, List, Dict, Any, Tuple, Union
-from urllib.parse import quote
 
 import discord
 from discord import app_commands
-import httpx
 from fastapi.concurrency import run_in_threadpool
 
 from app.runtime.settings import RuntimeSettingsCompat
 
 settings = RuntimeSettingsCompat()
+from app.application.messaging.ingress import async_forward_message_to_host
 from app.domain.context import MediaInfo, Context
 from app.domain.metainfo import MetaInfo
 from app.runtime.log import logger
@@ -60,17 +59,10 @@ class Discord:
         self._token = DISCORD_BOT_TOKEN
         self._guild_id = self._to_int(DISCORD_GUILD_ID)
         self._channel_id = self._to_int(DISCORD_CHANNEL_ID)
+        self._config_name = kwargs.get("name")
         logger.debug(
             f"[Discord] 解析后的 ID: _guild_id={self._guild_id}, _channel_id={self._channel_id}"
         )
-        base_ds_url = f"http://127.0.0.1:{settings.PORT}/api/v1/message/"
-        self._ds_url = f"{base_ds_url}?token={settings.API_TOKEN}"
-        if kwargs.get("name"):
-            # URL encode the source name to handle special characters in config names
-            encoded_name = quote(kwargs.get("name"), safe="")
-            self._ds_url = f"{self._ds_url}&source={encoded_name}"
-        logger.debug(f"[Discord] 消息回调 URL: {self._ds_url}")
-
         intents = discord.Intents.default()
         intents.message_content = True
         intents.messages = True
@@ -1313,13 +1305,10 @@ class Discord:
         return content
 
     async def _post_to_ds(self, payload: Dict[str, Any]) -> None:
-        try:
-            proxy = None
-            if settings.PROXY:
-                proxy = settings.PROXY.get("https") or settings.PROXY.get("http")
-            async with httpx.AsyncClient(
-                timeout=10, verify=False, proxy=proxy
-            ) as client:
-                await client.post(self._ds_url, json=payload)
-        except Exception as err:
-            logger.error(f"转发 Discord 消息失败：{err}")
+        """把 Discord 事件异步转交统一消息入口。"""
+        if not await async_forward_message_to_host(
+            payload,
+            self._config_name,
+            timeout=10,
+        ):
+            logger.error("转发 Discord 消息失败")

app/modules/feishu/feishu.py

@@ -53,6 +53,7 @@ from lark_oapi.event.callback.model.p2_card_action_trigger import (
 from app.runtime.settings import RuntimeSettingsCompat
 
 settings = RuntimeSettingsCompat()
+from app.application.messaging.ingress import submit_message_to_host
 from app.domain.context import Context, MediaInfo
 from app.application.security.user import get_configured_user_channel_lookup
 from app.application.messaging.agent import matches_channel_admin
@@ -61,6 +62,7 @@ from app.schemas.message import IncomingMessage
 from app.schemas.message import Message
 from app.schemas.types import NotificationChannel, MessageType
 from app.adapters.network.http import RequestUtils
+from app.runtime.thread import ThreadHelper
 
 
 class UserOper:
@@ -295,19 +297,13 @@ class Feishu:
             ws_client._service_id = ""
             ws_client._lock.release()
 
-    def _forward_to_message_chain(self, payload: dict) -> None:
+    def _forward_to_message_chain(self, payload: dict) -> bool:
         """将飞书入站消息转发到统一消息入口，复用现有交互主链。"""
-
-        def _run() -> None:
-            try:
-                RequestUtils(timeout=15).post_res(
-                    f"http://127.0.0.1:{settings.PORT}/api/v1/message?token={settings.API_TOKEN}&source={self._name}",
-                    json=payload,
-                )
-            except Exception as err:
-                logger.error(f"飞书转发消息失败：{err}")
-
-        threading.Thread(target=_run, daemon=True).start()
+        return submit_message_to_host(
+            payload,
+            self._name,
+            submit=ThreadHelper().submit,
+        )
 
     @staticmethod
     def _parse_message_content(message) -> Tuple[

app/modules/qqbot/qqbot.py

@@ -15,9 +15,11 @@ from app.runtime.cache import FileCache
 from app.runtime.settings import RuntimeSettingsCompat
 
 settings = RuntimeSettingsCompat()
+from app.application.messaging.ingress import submit_message_to_host
 from app.domain.context import MediaInfo, Context
 from app.domain.metainfo import MetaInfo
 from app.runtime.log import logger
+from app.runtime.thread import ThreadHelper
 from app.modules.qqbot.api import (
     get_access_token,
     get_gateway_url,
@@ -103,20 +105,13 @@ class QQBot:
         except Exception as e:
             logger.debug(f"QQ Bot 保存 known_targets 失败: {e}")
 
-    def _forward_to_message_chain(self, payload: dict) -> None:
-        """直接调用消息链处理，避免 HTTP 开销"""
-
-        def _run():
-            try:
-                # 回调
-                RequestUtils(timeout=15).post_res(
-                    f"http://127.0.0.1:{settings.PORT}/api/v1/message?token={settings.API_TOKEN}&source={self._config_name}",
-                    json=payload
-                )
-            except Exception as e:
-                logger.error(f"QQ Bot 转发消息失败: {e}")
-
-        threading.Thread(target=_run, daemon=True).start()
+    def _forward_to_message_chain(self, payload: dict) -> bool:
+        """通过受管线程池把 QQ Bot 入站 payload 转交统一消息入口。"""
+        return submit_message_to_host(
+            payload,
+            self._config_name,
+            submit=ThreadHelper().submit,
+        )
 
     def _on_gateway_message(self, payload: dict) -> None:
         """Gateway 收到消息时转发至 MP 消息链，并记录发送者用于广播"""

app/modules/slack/slack.py

@@ -3,9 +3,7 @@ import re
 from threading import Lock
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
-from urllib.parse import quote
 
-import requests
 from slack_bolt import App
 from slack_bolt.adapter.socket_mode import SocketModeHandler
 from slack_sdk import WebClient
@@ -13,6 +11,7 @@ from slack_sdk import WebClient
 from app.runtime.settings import RuntimeSettingsCompat
 
 settings = RuntimeSettingsCompat()
+from app.application.messaging.ingress import forward_message_to_host
 from app.domain.context import MediaInfo, Context
 from app.domain.metainfo import MetaInfo
 from app.runtime.log import logger
@@ -27,7 +26,6 @@ class Slack:
 
     _client: WebClient = None
     _service: SocketModeHandler = None
-    _ds_url = f"http://127.0.0.1:{settings.PORT}/api/v1/message?token={settings.API_TOKEN}"
     _channel = ""
     _oauth_token = ""
     _MAX_SLASH_COMMANDS = 50
@@ -65,6 +63,7 @@ class Slack:
         self._client = slack_app.client
         self._channel = SLACK_CHANNEL
         self._oauth_token = SLACK_OAUTH_TOKEN
+        self._config_name = kwargs.get("name")
         self._app_id = (SLACK_APP_ID or "").strip()
         self._command_request_url = (SLACK_COMMAND_REQUEST_URL or "").strip()
         self._manifest_client = (
@@ -74,41 +73,30 @@ class Slack:
         )
         self._registered_command_names: set[str] = set()
 
-        # 标记消息来源
-        if kwargs.get("name"):
-            # URL encode the source name to handle special characters
-            encoded_name = quote(kwargs.get('name'), safe='')
-            self._ds_url = f"{self._ds_url}&source={encoded_name}"
-
         # 注册消息响应
         @slack_app.event("message")
         def slack_message(message):
-            with requests.post(self._ds_url, json=message, timeout=10) as local_res:
-                logger.debug("message: %s processed, response is: %s" % (message, local_res.text))
+            self._forward_to_message_chain(message, timeout=10)
 
         @slack_app.action(re.compile(r"actionId-.*"))
         def slack_action(ack, body):
             ack()
-            with requests.post(self._ds_url, json=body, timeout=60) as local_res:
-                logger.debug("message: %s processed, response is: %s" % (body, local_res.text))
+            self._forward_to_message_chain(body, timeout=60)
 
         @slack_app.event("app_mention")
         def slack_mention(say, body):
             say(f"收到，请稍等... <@{body.get('event', {}).get('user')}>")
-            with requests.post(self._ds_url, json=body, timeout=10) as local_res:
-                logger.debug("message: %s processed, response is: %s" % (body, local_res.text))
+            self._forward_to_message_chain(body, timeout=10)
 
         @slack_app.shortcut(re.compile(r"/*"))
         def slack_shortcut(ack, body):
             ack()
-            with requests.post(self._ds_url, json=body, timeout=10) as local_res:
-                logger.debug("message: %s processed, response is: %s" % (body, local_res.text))
+            self._forward_to_message_chain(body, timeout=10)
 
         @slack_app.command(re.compile(r"/*"))
         def slack_command(ack, body):
             ack()
-            with requests.post(self._ds_url, json=body, timeout=10) as local_res:
-                logger.debug("message: %s processed, response is: %s" % (body, local_res.text))
+            self._forward_to_message_chain(body, timeout=10)
 
         # 启动服务
         try:
@@ -121,6 +109,14 @@ class Slack:
         except Exception as err:
             logger.error("Slack消息接收服务启动失败: %s" % str(err))
 
+    def _forward_to_message_chain(self, payload: dict, *, timeout: float) -> bool:
+        """把 Slack SDK 回调同步转交统一消息入口。"""
+        return forward_message_to_host(
+            payload,
+            self._config_name,
+            timeout=timeout,
+        )
+
     def stop(self):
         if self._service:
             try:

app/modules/telegram/telegram.py

@@ -6,7 +6,7 @@ import threading
 import time
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional, Union
-from urllib.parse import urljoin, quote
+from urllib.parse import urljoin
 
 from app.modules.telegram.compat import ensure_urllib3_header_param_compat
 
@@ -42,6 +42,7 @@ settings = RuntimeSettingsCompat()
 from app.domain.context import MediaInfo, Context  # noqa: E402
 from app.domain.metainfo import MetaInfo  # noqa: E402
 from app.application.image import ImageHelper  # noqa: E402
+from app.application.messaging.ingress import forward_message_to_host  # noqa: E402
 from app.runtime.thread import ThreadHelper  # noqa: E402
 from app.runtime.log import logger  # noqa: E402
 from app.runtime.execution import retry  # noqa: E402
@@ -74,9 +75,6 @@ class Telegram:
     Telegram 消息客户端，负责发送、编辑、接收和转发 Telegram 消息。
     """
 
-    _ds_url = (
-        f"http://127.0.0.1:{settings.PORT}/api/v1/message?token={settings.API_TOKEN}"
-    )
     _bot: TeleBot = None
     _callback_handlers: Dict[str, Callable] = {}  # 存储回调处理器
     _user_chat_mapping: Dict[
@@ -137,11 +135,7 @@ class Telegram:
                 logger.error(f"获取bot信息失败: {e}")
                 self._bot_username = None
 
-            # 标记渠道来源
-            if kwargs.get("name"):
-                # URL encode the source name to handle special characters
-                encoded_name = quote(kwargs.get("name"), safe="")
-                self._ds_url = f"{self._ds_url}&source={encoded_name}"
+            self._config_name = kwargs.get("name")
 
             @_bot.message_handler(commands=["start", "help"])
             def send_welcome(message):
@@ -164,10 +158,7 @@ class Telegram:
                     if not payload:
                         logger.warn("Telegram消息序列化失败，跳过转发")
                         return
-                    response = RequestUtils(timeout=15).post_res(
-                        self._ds_url, json=payload
-                    )
-                    if not response or response.status_code >= 400:
+                    if not self._forward_to_message_chain(payload):
                         logger.warn("Telegram消息转发失败")
 
             @_bot.callback_query_handler(func=lambda call: True)
@@ -208,10 +199,7 @@ class Telegram:
                     _bot.answer_callback_query(call.id)
 
                     # 发送给主程序处理
-                    response = RequestUtils(timeout=15).post_res(
-                        self._ds_url, json=callback_json
-                    )
-                    if not response or response.status_code >= 400:
+                    if not self._forward_to_message_chain(callback_json):
                         logger.warn("Telegram按钮回调转发失败")
 
                 except Exception as err:
@@ -232,6 +220,10 @@ class Telegram:
             self._polling_thread.start()
             logger.info("Telegram消息接收服务启动")
 
+    def _forward_to_message_chain(self, payload: dict) -> bool:
+        """把 Telegram SDK 回调同步转交统一消息入口。"""
+        return forward_message_to_host(payload, self._config_name)
+
     @property
     def bot(self):
         """

app/modules/wechat/wechatbot.py

@@ -15,10 +15,12 @@ from app.runtime.cache import FileCache
 from app.runtime.settings import RuntimeSettingsCompat
 
 settings = RuntimeSettingsCompat()
+from app.application.messaging.ingress import submit_message_to_host
 from app.domain.context import MediaInfo, Context
 from app.domain.metainfo import MetaInfo
 from app.application.messaging.agent import matches_channel_admin
 from app.runtime.log import logger
+from app.runtime.thread import ThreadHelper
 from app.schemas.message import IncomingMessage
 from app.schemas.types import NotificationChannel
 from app.adapters.network.http import RequestUtils
@@ -34,7 +36,6 @@ class WeChatBot:
     """
 
     _default_ws_url = "wss://openws.work.weixin.qq.com"
-    _ds_url = f"http://127.0.0.1:{settings.PORT}/api/v1/message?token={settings.API_TOKEN}"
     _heartbeat_interval = 30
     _ack_timeout = 10
 
@@ -512,18 +513,13 @@ class WeChatBot:
         )
         self._forward_to_message_chain(payload)
 
-    def _forward_to_message_chain(self, payload: dict) -> None:
-        def _run():
-            try:
-                # 回调
-                RequestUtils(timeout=15).post_res(
-                    f"http://127.0.0.1:{settings.PORT}/api/v1/message?token={settings.API_TOKEN}&source={self._config_name}",
-                    json=payload
-                )
-            except Exception as err:
-                logger.error(f"企业微信智能机器人转发消息失败：{err}")
-
-        threading.Thread(target=_run, daemon=True).start()
+    def _forward_to_message_chain(self, payload: dict) -> bool:
+        """通过受管线程池把企业微信 payload 转交统一消息入口。"""
+        return submit_message_to_host(
+            payload,
+            self._config_name,
+            submit=ThreadHelper().submit,
+        )
 
     @staticmethod
     def _normalize_target(userid: Optional[str], default_chat_id: Optional[str]) -> Tuple[Optional[str], int]:

app/modules/wechatclawbot/wechatclawbot.py

@@ -20,6 +20,7 @@ from app.runtime.cache import FileCache
 from app.runtime.settings import RuntimeSettingsCompat
 
 settings = RuntimeSettingsCompat()
+from app.application.messaging.ingress import forward_message_to_host
 from app.domain.context import Context, MediaInfo
 from app.domain.metainfo import MetaInfo
 from app.runtime.log import logger
@@ -1477,9 +1478,6 @@ class WechatClawBot:
         self._stop_event = threading.Event()
         self._poll_thread: Optional[threading.Thread] = None
         self._state = self._load_state()
-        self._message_endpoint = (
-            f"http://127.0.0.1:{settings.PORT}/api/v1/message?token={settings.API_TOKEN}&source={quote(self._config_name, safe='')}"
-        )
         if self._state.get("bot_token") and self._auto_start_polling:
             self._start_polling()
 
@@ -1751,33 +1749,18 @@ class WechatClawBot:
                         username=message.username,
                         context_token=message.context_token,
                     )
-                    response = None
                     try:
-                        response = RequestUtils(timeout=15).post_res(
-                            self._message_endpoint,
-                            json=message.to_message_payload(),
-                        )
-                        if response is None:
-                            logger.error(
-                                f"转发微信 ClawBot 消息失败：message_id={message.message_id}, "
-                                "本地消息入口无响应"
-                            )
-                        elif response.status_code != 200:
+                        if not self._forward_to_message_chain(
+                            message.to_message_payload()
+                        ):
                             logger.error(
                                 "转发微信 ClawBot 消息失败："
-                                f"message_id={message.message_id}, status={response.status_code}, "
-                                f"body={self._short_text(response.text)}"
+                                f"message_id={message.message_id}"
                             )
                     except Exception as err:
                         logger.error(
                             f"转发微信 ClawBot 消息失败：message_id={message.message_id}, error={err}"
                         )
-                    finally:
-                        if response is not None:
-                            try:
-                                response.close()
-                            except Exception:
-                                pass
                 consecutive_failures = 0
             except Exception as err:
                 consecutive_failures += 1
@@ -1789,6 +1772,10 @@ class WechatClawBot:
                     break
                 self._stop_event.wait(delay)
 
+    def _forward_to_message_chain(self, payload: dict) -> bool:
+        """把 WeChatClawBot 轮询消息同步转交统一消息入口。"""
+        return forward_message_to_host(payload, self._config_name)
+
     def _build_known_targets(self) -> List[Dict[str, Any]]:
         known_targets = self._state.get("known_targets") or {}
         items = []