diff --git a/AGENTS.md b/AGENTS.md index 54c95e223..9965933f2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -67,7 +67,7 @@ The legacy roots have no physical directories in the source tree. Current images | `app/adapters/system/` | 操作系统、文件、进程、标准流、包/资源安装、显示和 Rust 加速适配 | 业务规则、进程重启决策 | `host.py`, `display/`, `stdio.py`, `package.py`, `resource.py`, `rust.py`, `fsproxy.py` | | `app/adapters/external/` | CookieCloud、插件市场、OCR、IP 归属和 MoviePilot Server 等命名外部生态 | 通用 HTTP/DNS/文件机制或可复用领域语义 | `market.py`, `server.py`, `cookiecloud.py`, `ocr.py`, `location.py`, `wechat_crypt.py` | | `app/application/` | 聚焦应用服务、用例命令,以及由用例拥有的持久化 Port/Protocol | SQLAlchemy、Session、Oper 等具体 DB 实现,多领域 Chain 编排、底层通用机制、通用传输协议 | `recognition.py`, `filter.py`, `outbox.py`, `subscription/write.py`, `workflow.py` | -| `app/application/messaging/` | 消息渲染/路由、交互和 Agent 到消息桥接:`interaction.py` 通用交互契约和视图工具;`router.py` 统一交互优先级和回调分发;`site.py`/`subscribe.py`/`skill.py` 对应命令的会话、输入解析和视图;`media.py` 媒体交互状态(业务工作流仍由 `MediaInteractionChain` 执行);`plugin.py` 插件输入接管和插件按钮回调;`agent.py` Agent 选择状态、回调协议和 WebAgent 消息桥接;`message.py` 通知渲染、模板和队列。不作为推荐给插件直接使用的公开 SDK | 认证策略、通用 HTTP、服务发现、仅端点使用的 Web Push 行为 | `message.py`, `interaction.py`, `router.py`, `agent.py` | +| `app/application/messaging/` | 消息渲染/路由、交互和 Agent 到消息桥接:`ingress.py` 统一渠道回环入口;`interaction.py` 通用交互契约和视图工具;`router.py` 统一交互优先级和回调分发;`site.py`/`subscribe.py`/`skill.py` 对应命令的会话、输入解析和视图;`media.py` 媒体交互状态(业务工作流仍由 `MediaInteractionChain` 执行);`plugin.py` 插件输入接管和插件按钮回调;`agent.py` Agent 选择状态、回调协议和 WebAgent 消息桥接;`message.py` 通知渲染、模板和队列。不作为推荐给插件直接使用的公开 SDK | 认证策略、通用 HTTP、服务发现、仅端点使用的 Web Push 行为 | `ingress.py`, `message.py`, `interaction.py`, `router.py`, `agent.py` | | `app/application/security/` | 认证、授权、Cookie、Passkey、OTP/二次认证、路径/URL 安全、SSRF 和签名策略 | 通用 URL 解析、进程运行策略、普通业务校验 | `access.py`, `auth.py`, `cookie.py`, `passkey.py`, `otp.py`, `twofactor.py`, `url.py` | | `app/chain/` | Reusable use-case orchestration across modules, Application services, injected ports, events, and caches; chains reach modules only through `run_module` dispatch on method-name contracts | Transport schemas, backend-specific protocol details, generic primitives, direct Oper/DB imports, direct imports of module internals (classes, exceptions, constants) | `media.py`, `download.py`, `subscribe.py`, `transfer.py` | | `app/db/oper/` | 面向表和持久化值的 SQLAlchemy 数据访问;接收调用方 Session,只查询、暂存或 flush | Application 业务规则、隐式事务所有权、外部副作用 | `subscribe.py`, `site.py`, `workflow.py` | diff --git a/app/application/messaging/ingress.py b/app/application/messaging/ingress.py new file mode 100644 index 000000000..7aa145958 --- /dev/null +++ b/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 diff --git a/app/modules/discord/discord.py b/app/modules/discord/discord.py index cc4a0c2cd..fbf5672dd 100644 --- a/app/modules/discord/discord.py +++ b/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 消息失败") diff --git a/app/modules/feishu/feishu.py b/app/modules/feishu/feishu.py index c3651d831..06f8b6431 100644 --- a/app/modules/feishu/feishu.py +++ b/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[ diff --git a/app/modules/qqbot/qqbot.py b/app/modules/qqbot/qqbot.py index 5c4b84ffa..72cda24db 100644 --- a/app/modules/qqbot/qqbot.py +++ b/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 消息链,并记录发送者用于广播""" diff --git a/app/modules/slack/slack.py b/app/modules/slack/slack.py index 2f88c32f9..a8e280ba8 100644 --- a/app/modules/slack/slack.py +++ b/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: diff --git a/app/modules/telegram/telegram.py b/app/modules/telegram/telegram.py index 96ab59713..5d592398a 100644 --- a/app/modules/telegram/telegram.py +++ b/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): """ diff --git a/app/modules/wechat/wechatbot.py b/app/modules/wechat/wechatbot.py index a970c221c..f7236af79 100644 --- a/app/modules/wechat/wechatbot.py +++ b/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]: diff --git a/app/modules/wechatclawbot/wechatclawbot.py b/app/modules/wechatclawbot/wechatclawbot.py index 01ffbf232..20fb587b3 100644 --- a/app/modules/wechatclawbot/wechatclawbot.py +++ b/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 = [] diff --git a/docs/adr/0007-background-action-reliability.md b/docs/adr/0007-background-action-reliability.md index f738f84ca..571f2c4da 100644 --- a/docs/adr/0007-background-action-reliability.md +++ b/docs/adr/0007-background-action-reliability.md @@ -63,6 +63,9 @@ Event Contract Registry 是 53 个事件的逐项机器清单。下表按相同 调度,不表示执行完成。 - Webhook E0 广播、消息入口和 Seerr 订阅入口均已迁入 lifespan TaskRegistry,具备 owner、停止接收和 有限等待语义;进程崩溃时仍允许丢失,不因此提升为 durable。 +- Slack、Telegram、Discord、飞书、QQBot、企业微信与 WeChatClawBot 的渠道回环统一经 + `application.messaging.ingress` 进入同一个 API/TaskRegistry 主链;需要立即返回 SDK 回调的渠道把同步 + HTTP 交给宿主共享线程池,模块关闭后由线程池生命周期等待,不再创建逐消息 daemon 线程。 - 主仓不再新增或保留裸 FastAPI `BackgroundTasks`;若任务源于已提交的用户数据且不可从数据库重建, 必须提升为 E2,进入 Outbox 或持久任务表。 diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 360d96f58..c0a6acba9 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -176,7 +176,7 @@ flowchart TB | `app/application/` | 读取配置/持久化状态的聚焦应用服务:识别、过滤、通知、RSS、站点、下载器、媒体服务器、存储、整理规则等;同一主题拆成子包 | `recognition.py`、`rules.py`、`rss.py`、`site/`、`subscription/`、`plugin/` | | `app/application/subscription/` | 订阅新增、查询、变更、删除、媒体身份与搜索契约 | `write.py`、`contract.py`、`mutation.py`、`delete.py`、`identity.py`、`search.py` | | `app/application/plugin/` | 插件市场、安装、运行时端口、文件夹操作和动态路由用例;具体 FastAPI 路由适配器在 adapters 层 | `catalog.py`、`install.py`、`runtime.py`、`folders.py`、`routes.py` | -| `app/application/messaging/` | 消息渲染/路由、命令交互会话、插件按钮回调、Agent 消息桥接 | `message.py`、`router.py`、`agent.py` | +| `app/application/messaging/` | 渠道回环入口、消息渲染/路由、命令交互会话、插件按钮回调、Agent 消息桥接 | `ingress.py`、`message.py`、`router.py`、`agent.py` | | `app/application/security/` | 认证、授权、Cookie、Passkey、OTP/二次认证、SSRF 与 URL/路径安全 | `auth.py`、`url.py`、`twofactor.py` | | `app/chain/` | 跨入口复用的用例编排:订阅、搜索、下载、整理、媒体、消息等 Chain | `subscribe.py`、`search.py`、`transfer.py` | | `app/modules/` | 可插拔后端:下载器、媒体服务器、元数据源、消息渠道、索引器、存储 | `qbittorrent/`、`emby/`、`telegram/`、`themoviedb/` | @@ -478,7 +478,7 @@ flowchart LR IT --> Chain ``` -- `app/application/messaging/` 负责消息渲染、模板、队列(`message.py`)、交互会话与视图; +- `app/application/messaging/` 负责渠道回环入口(`ingress.py`)、消息渲染、模板、队列(`message.py`)、交互会话与视图; 业务工作流仍由对应 Chain 执行(如媒体交互的业务部分在 `MediaInteractionChain`)。 - 该包不作为推荐给插件直接使用的公开 SDK。 diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index eaead1554..b48ddffda 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -6,7 +6,7 @@ > 审计范围:宿主后端;排除 `app/plugins/**` 运行时插件副本 > 规范优先级:`AGENTS.md` 与 `docs/rules/` 高于本文 > 相关文档:`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md` -> 实施进度:阶段 0~6 的宿主架构能力已完成收口;API/Application 公共复杂度基线已清零,启动组合根的 SystemConfigOper 构造点已由 14 降至 1;API 进程内后台任务已完成首批统一登记,插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零,插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线;阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义;阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权;2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零,并补齐可执行 fanout 与下载器文件 DTO 边界;阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox,宿主不再调用裸线程统计入口。 +> 实施进度:阶段 0~6 的宿主架构能力已完成收口;API/Application 公共复杂度基线已清零,启动组合根的 SystemConfigOper 构造点已由 14 降至 1;API 进程内后台任务已完成首批统一登记,插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零,插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线;阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义;阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权;2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零,并补齐可执行 fanout 与下载器文件 DTO 边界;阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox,宿主不再调用裸线程统计入口;阶段 4 已统一七种消息渠道的宿主回环与后台执行边界。 ## 当前复核结论(2026-08-24) @@ -15,7 +15,7 @@ ### 长期整改阶段 0:治理门禁恢复(2026-08-23) -- 宿主依赖基线已审查 TaskRegistry、有界后台 owner 与插件变更准入接入后的语义差异:当前为 `805` 个模块、`6503` 条内部导入边,12 组重点禁止边继续全部为 `0`,唯一非平凡 SCC 仍是隔离的 TMDB 移植包。 +- 宿主依赖基线已审查 TaskRegistry、有界后台 owner 与插件变更准入接入后的语义差异:当前为 `806` 个模块、`6528` 条内部导入边,12 组重点禁止边继续全部为 `0`,唯一非平凡 SCC 仍是隔离的 TMDB 移植包。 - 启动性能探针会在隔离生命周期中真实创建并释放 TaskRegistry;normal/safe 组件数分别为 `23`/`11`,CI 只读检查使用稳定的宿主模块集合和生命周期组件顺序,不再把 Python/平台模块数量当作硬合同。 - 官方插件快照覆盖 `plugins.v3`、`plugins.v2` 以及 V3 实际会从 `package.json` 回退加载的 31 个默认实现;`app/plugins/**` 仍只是宿主运行副本,不进入扫描。 - SDK 快照以各模块显式 `__all__` 为公开合同,能够记录赋值别名;`typing`、`__future__` 等实现期导入不再被误冻结,既有数据库备份门面已补精确导出清单。 @@ -92,13 +92,27 @@ - 插件兼容边界保持不变:`MoviePilotServerHelper.sub_reg_async` 与 `sub_done_async` 的类方法、签名和返回值 继续保留给旧插件;仅宿主生产调用清零,因此没有改动插件仓、SDK/Compat 映射或事件 payload。 +### 长期整改阶段 4:消息渠道回环与线程所有权统一(2026-08-24) + +- Slack、Telegram、Discord、飞书、QQBot、企业微信与 WeChatClawBot 原先分别拼接本地消息 API URL、 + 执行 HTTP 并判断响应;其中飞书、QQBot、企业微信还为每条入站消息创建不可追踪的 daemon 线程。现在七者统一复用 + `app.application.messaging.ingress` 的 URL 编码、请求、状态确认、异常日志和响应释放语义。 +- SDK 回调需要同步确认的 Slack/Telegram/WeChatClawBot 保留同步调用,Discord 保留自有事件循环内的 + 异步调用;飞书、QQBot、企业微信继续立即返回,但任务改由现有 `ThreadHelper` 共享执行器承载,应用 + 关闭时由既有线程池生命周期等待,不再产生逐消息游离线程。HTTP 到达后仍由 `/api/v1/message` 的 + TaskRegistry owner 执行消息主链。 +- 新门禁扫描全部宿主 `app/modules/**/*.py`,禁止渠道再次硬编码 `/api/v1/message`;新增渠道必须使用统一 + ingress。source、token 统一通过 query encoder 处理,修复配置名含 `&` 等字符时被拆成额外参数的问题。 +- 兼容边界不变:七个渠道类、模块方法、配置字段、消息 payload、同步/异步 SDK 回调方式和 + `/api/v1/message` HTTP 合同均未改;没有修改插件仓或向 SDK/Compat 新增宿主内部入口。 + ### 总体判断 当前架构总体合理,已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**: - 继续采用单进程控制面是正确选择,不建议现在拆成微服务;插件、调度器、工作流、事件和数据库共享进程内状态,拆分会放大部署、事务和兼容成本。 - `foundation/domain/runtime/adapters/application/chain/api/startup` 的职责方向基本成立;宿主架构基线、复杂度 ratchet、异步阻塞 ratchet 当前均通过。 -- 依赖图当前为 `805` 个 Python 模块、`6503` 条内部导入边;唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部,不应为了指标归零重写。 +- 依赖图当前为 `806` 个 Python 模块、`6528` 条内部导入边;唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部,不应为了指标归零重写。 - 当前主要风险已经从“目录和依赖失控”转移到运行时协议、后台副作用的可靠性和遗留兼容面。换言之,下一阶段重点应是**语义收口和可验证性**,而不是继续搬文件或机械拆大文件。 综合评价:架构方向可持续,生产可用性较高;可演进性仍处于中等水平。现阶段没有静态审计发现必须立即推倒重来的 P0 架构问题,但存在需要按 P1/P2 计划治理的真实债务。 @@ -118,6 +132,8 @@ `finally` 取消并等待清理,断线和 ASGI 取消均不会留下请求级 task。 订阅删除的宿主生产者也已完成 durable 分级:消息交互和远程删除不再调用裸线程统计入口,而是 与业务删除原子暂存事件和统计 intent;旧类方法只作为插件 ABI 保留,不纳入宿主可靠性证明。 + 消息渠道回环也不再各自拥有 URL、HTTP 判定和逐消息线程:七种宿主渠道共享 ingress,三种立即返回 + 渠道交给生命周期持有的共享线程池;这仍是 E0 投递,不宣称跨进程恢复。 2. **Model/Base 的数据库装饰器和隐式会话 ABI 已全部清零。** 查询、写事务和 `legacy_*` 装饰器均为 `0`;所有 Model `db` 参数要求显式 Session,Base CRUD 仅在调用方事务内查询或 stage。可无会话构造的入口统一留在 Oper,经组合根事务执行器运行;插件 SDK 不再导出宿主 Model。后续重点转为减少 ORM 对象跨层流转,并保持 Model 隐式事务零回退。 Oper 内部的执行入口也已统一:最后一处 `AgentTaskOper` 直接 transaction runner 调用已迁入 diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index b45788283..c27e0e338 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -69,7 +69,7 @@ to make the directory tree look symmetrical. | `app/application/plugin/` | Plugin market catalog, installation command, runtime port, folder operations and dynamic-route use cases; filenames remain single words (`catalog.py`, `install.py`, `runtime.py`, `folders.py`, `routes.py`) | | `app/application/server/` | MoviePilot Server reporting and sharing use cases; local data readers and transport callbacks are injected by startup | | `app/application/site/` | Configured site catalog, authentication level and index-resource capability; the generated extension and its data bundle stay together here | -| `app/application/messaging/` | Message rendering/routing, interactions and the Agent-to-message bridge: `interaction.py` shared interaction contracts and view helpers; `router.py` unified interaction priority and callback dispatch; `site.py`/`subscribe.py`/`skill.py` per-command sessions, input parsing and views; `media.py` media interaction state while the business workflow stays in `MediaInteractionChain`; `plugin.py` plugin input capture and plugin button callbacks; `agent.py` agent choice state, callback protocol and WebAgent bridge; `message.py` notification rendering, templates and queue. Not a public SDK recommended for direct plugin use | +| `app/application/messaging/` | Message rendering/routing, interactions and the Agent-to-message bridge: `ingress.py` owns the single channel-to-host loopback boundary; `interaction.py` shared interaction contracts and view helpers; `router.py` unified interaction priority and callback dispatch; `site.py`/`subscribe.py`/`skill.py` per-command sessions, input parsing and views; `media.py` media interaction state while the business workflow stays in `MediaInteractionChain`; `plugin.py` plugin input capture and plugin button callbacks; `agent.py` agent choice state, callback protocol and WebAgent bridge; `message.py` notification rendering, templates and queue. Not a public SDK recommended for direct plugin use | | `app/application/security/` | Authentication, authorization, cookies, passkeys, OTP/two-factor, path/URL safety, SSRF and signing policy | Application services may use domain rules and runtime contracts. They own the diff --git a/mypy.ini b/mypy.ini index 38e528b31..6cbcaee20 100644 --- a/mypy.ini +++ b/mypy.ini @@ -26,6 +26,7 @@ files = app/application/workflow.py, app/application/chain/context.py, app/application/chain/durable_events.py, + app/application/messaging/ingress.py, app/application/subscription/delete.py, app/application/subscription/identity.py, app/application/subscription/mutation.py, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 5f06e98dd..4afdaa287 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6503, - "edge_sha256": "57c326a81dbba07909871a97df1c73416850205c3d385a15afe3a07e3e44c00c", + "edge_count": 6528, + "edge_sha256": "2e30c775e78f578d804f69972ab1348e1e9d0872b00bdbfcffad08398fae68ac", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -2605,6 +2605,12 @@ "app.application.messaging.chat -> app.schemas", "app.application.messaging.chat -> app.schemas.agent", "app.application.messaging.chat -> app.schemas.exception", + "app.application.messaging.ingress -> app.adapters", + "app.application.messaging.ingress -> app.adapters.network", + "app.application.messaging.ingress -> app.adapters.network.http", + "app.application.messaging.ingress -> app.runtime", + "app.application.messaging.ingress -> app.runtime.log", + "app.application.messaging.ingress -> app.runtime.settings", "app.application.messaging.interaction -> app.schemas", "app.application.messaging.interaction -> app.schemas.message", "app.application.messaging.interaction -> app.schemas.notification", @@ -4036,6 +4042,9 @@ "app.modules.discord -> app.schemas.message", "app.modules.discord -> app.schemas.notification", "app.modules.discord -> app.schemas.types", + "app.modules.discord.discord -> app.application", + "app.modules.discord.discord -> app.application.messaging", + "app.modules.discord.discord -> app.application.messaging.ingress", "app.modules.discord.discord -> app.domain", "app.modules.discord.discord -> app.domain.context", "app.modules.discord.discord -> app.domain.metainfo", @@ -4140,6 +4149,7 @@ "app.modules.feishu.feishu -> app.application", "app.modules.feishu.feishu -> app.application.messaging", "app.modules.feishu.feishu -> app.application.messaging.agent", + "app.modules.feishu.feishu -> app.application.messaging.ingress", "app.modules.feishu.feishu -> app.application.security", "app.modules.feishu.feishu -> app.application.security.user", "app.modules.feishu.feishu -> app.domain", @@ -4147,6 +4157,7 @@ "app.modules.feishu.feishu -> app.runtime", "app.modules.feishu.feishu -> app.runtime.log", "app.modules.feishu.feishu -> app.runtime.settings", + "app.modules.feishu.feishu -> app.runtime.thread", "app.modules.feishu.feishu -> app.schemas", "app.modules.feishu.feishu -> app.schemas.message", "app.modules.feishu.feishu -> app.schemas.types", @@ -4865,6 +4876,9 @@ "app.modules.qqbot.qqbot -> app.adapters", "app.modules.qqbot.qqbot -> app.adapters.network", "app.modules.qqbot.qqbot -> app.adapters.network.http", + "app.modules.qqbot.qqbot -> app.application", + "app.modules.qqbot.qqbot -> app.application.messaging", + "app.modules.qqbot.qqbot -> app.application.messaging.ingress", "app.modules.qqbot.qqbot -> app.domain", "app.modules.qqbot.qqbot -> app.domain.context", "app.modules.qqbot.qqbot -> app.domain.metainfo", @@ -4878,6 +4892,7 @@ "app.modules.qqbot.qqbot -> app.runtime.cache", "app.modules.qqbot.qqbot -> app.runtime.log", "app.modules.qqbot.qqbot -> app.runtime.settings", + "app.modules.qqbot.qqbot -> app.runtime.thread", "app.modules.redis -> app.adapters", "app.modules.redis -> app.adapters.cache", "app.modules.redis -> app.adapters.cache.redis", @@ -4922,6 +4937,9 @@ "app.modules.slack.slack -> app.adapters", "app.modules.slack.slack -> app.adapters.network", "app.modules.slack.slack -> app.adapters.network.http", + "app.modules.slack.slack -> app.application", + "app.modules.slack.slack -> app.application.messaging", + "app.modules.slack.slack -> app.application.messaging.ingress", "app.modules.slack.slack -> app.domain", "app.modules.slack.slack -> app.domain.context", "app.modules.slack.slack -> app.domain.metainfo", @@ -4993,6 +5011,8 @@ "app.modules.telegram.telegram -> app.adapters.network.http", "app.modules.telegram.telegram -> app.application", "app.modules.telegram.telegram -> app.application.image", + "app.modules.telegram.telegram -> app.application.messaging", + "app.modules.telegram.telegram -> app.application.messaging.ingress", "app.modules.telegram.telegram -> app.domain", "app.modules.telegram.telegram -> app.domain.context", "app.modules.telegram.telegram -> app.domain.metainfo", @@ -5395,6 +5415,7 @@ "app.modules.wechat.wechatbot -> app.application", "app.modules.wechat.wechatbot -> app.application.messaging", "app.modules.wechat.wechatbot -> app.application.messaging.agent", + "app.modules.wechat.wechatbot -> app.application.messaging.ingress", "app.modules.wechat.wechatbot -> app.domain", "app.modules.wechat.wechatbot -> app.domain.context", "app.modules.wechat.wechatbot -> app.domain.metainfo", @@ -5404,6 +5425,7 @@ "app.modules.wechat.wechatbot -> app.runtime.cache", "app.modules.wechat.wechatbot -> app.runtime.log", "app.modules.wechat.wechatbot -> app.runtime.settings", + "app.modules.wechat.wechatbot -> app.runtime.thread", "app.modules.wechat.wechatbot -> app.schemas", "app.modules.wechat.wechatbot -> app.schemas.message", "app.modules.wechat.wechatbot -> app.schemas.types", @@ -5424,6 +5446,9 @@ "app.modules.wechatclawbot.wechatclawbot -> app.adapters", "app.modules.wechatclawbot.wechatclawbot -> app.adapters.network", "app.modules.wechatclawbot.wechatclawbot -> app.adapters.network.http", + "app.modules.wechatclawbot.wechatclawbot -> app.application", + "app.modules.wechatclawbot.wechatclawbot -> app.application.messaging", + "app.modules.wechatclawbot.wechatclawbot -> app.application.messaging.ingress", "app.modules.wechatclawbot.wechatclawbot -> app.domain", "app.modules.wechatclawbot.wechatclawbot -> app.domain.context", "app.modules.wechatclawbot.wechatclawbot -> app.domain.metainfo", @@ -6520,7 +6545,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 805, + "module_count": 806, "modules": [ "app", "app.adapters", @@ -6791,6 +6816,7 @@ "app.application.messaging", "app.application.messaging.agent", "app.application.messaging.chat", + "app.application.messaging.ingress", "app.application.messaging.interaction", "app.application.messaging.media", "app.application.messaging.message", diff --git a/tests/test_message_ingress.py b/tests/test_message_ingress.py new file mode 100644 index 000000000..c4eed2aac --- /dev/null +++ b/tests/test_message_ingress.py @@ -0,0 +1,224 @@ +"""多消息渠道复用统一宿主回环入口的契约测试。""" + +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock +from urllib.parse import parse_qs, urlparse + +import pytest + +from app.application.messaging import ingress +from app.modules.discord import discord as discord_module +from app.modules.feishu import feishu as feishu_module +from app.modules.qqbot import qqbot as qqbot_module +from app.modules.slack import slack as slack_module +from app.modules.telegram import telegram as telegram_module +from app.modules.wechat import wechatbot as wechat_module +from app.modules.wechatclawbot import wechatclawbot as clawbot_module + + +PROJECT_ROOT = Path(__file__).resolve().parents[1] + + +def test_forward_message_to_host_encodes_source_and_closes_response(monkeypatch): + """统一入口必须安全编码查询参数并释放本地 HTTP 响应。""" + response = SimpleNamespace(status_code=200, close=MagicMock()) + post_res = MagicMock(return_value=response) + request = MagicMock() + request.post_res = post_res + request_factory = MagicMock(return_value=request) + monkeypatch.setattr( + ingress, + "settings", + SimpleNamespace(PORT=3000, API_TOKEN="token value"), + ) + monkeypatch.setattr(ingress, "RequestUtils", request_factory) + + assert ingress.forward_message_to_host( + {"text": "hello"}, + "channel & one", + timeout=9, + ) is True + + request_factory.assert_called_once_with(timeout=9) + url = post_res.call_args.args[0] + assert urlparse(url).path == "/api/v1/message" + assert parse_qs(urlparse(url).query) == { + "token": ["token value"], + "source": ["channel & one"], + } + assert post_res.call_args.kwargs["json"] == {"text": "hello"} + response.close.assert_called_once_with() + + +@pytest.mark.parametrize("status_code", [400, 500]) +def test_forward_message_to_host_rejects_unconfirmed_response( + monkeypatch, + status_code, +): + """本地入口无响应或返回错误状态时不得宣称渠道消息已接收。""" + response = SimpleNamespace(status_code=status_code, close=MagicMock()) + request = MagicMock() + request.post_res.return_value = response + monkeypatch.setattr( + ingress, + "settings", + SimpleNamespace(PORT=3000, API_TOKEN="token"), + ) + monkeypatch.setattr(ingress, "RequestUtils", MagicMock(return_value=request)) + + assert ingress.forward_message_to_host({}, "channel") is False + response.close.assert_called_once_with() + + +@pytest.mark.asyncio +async def test_async_forward_message_to_host_uses_same_contract(monkeypatch): + """自有事件循环的渠道必须复用同一 URL、确认规则和异步资源释放。""" + response = SimpleNamespace(status_code=200, aclose=AsyncMock()) + request = MagicMock() + request.post_res = AsyncMock(return_value=response) + request_factory = MagicMock(return_value=request) + monkeypatch.setattr( + ingress, + "settings", + SimpleNamespace(PORT=3000, API_TOKEN="token value"), + ) + monkeypatch.setattr(ingress, "AsyncRequestUtils", request_factory) + + assert await ingress.async_forward_message_to_host( + {"text": "hello"}, + "discord & one", + timeout=10, + ) is True + + request_factory.assert_called_once_with(timeout=10) + url = request.post_res.await_args.args[0] + assert parse_qs(urlparse(url).query) == { + "token": ["token value"], + "source": ["discord & one"], + } + response.aclose.assert_awaited_once_with() + + +def test_submit_message_to_host_copies_payload_and_reports_admission_failure(): + """异步渠道提交时冻结顶层 payload,执行器拒绝任务则返回 False。""" + submitted = [] + + def submit(function, *args, **kwargs): + """记录受管执行器收到的函数和参数。""" + submitted.append((function, args, kwargs)) + + payload = {"text": "before"} + assert ingress.submit_message_to_host( + payload, + "channel", + submit=submit, + ) is True + payload["text"] = "after" + + assert submitted[0][0] is ingress.forward_message_to_host + assert submitted[0][1] == ({"text": "before"}, "channel") + assert submitted[0][2] == {"timeout": 15} + + def reject(*_args, **_kwargs): + """模拟生命周期关闭后的执行器拒绝新任务。""" + raise RuntimeError("executor closed") + + assert ingress.submit_message_to_host({}, "channel", submit=reject) is False + + +@pytest.mark.parametrize( + ("module", "client_type", "source_attr"), + [ + (feishu_module, feishu_module.Feishu, "_name"), + (qqbot_module, qqbot_module.QQBot, "_config_name"), + (wechat_module, wechat_module.WeChatBot, "_config_name"), + ], +) +def test_threaded_channels_submit_through_managed_executor( + monkeypatch, + module, + client_type, + source_attr, +): + """原裸线程渠道必须把回环任务交给共享 ThreadHelper。""" + calls = [] + executor = SimpleNamespace(submit=lambda *_args, **_kwargs: None) + + def submit_message(payload, source, *, submit, timeout=15): + """记录渠道传给统一提交边界的参数。""" + calls.append((payload, source, submit, timeout)) + return True + + monkeypatch.setattr(module, "ThreadHelper", lambda: executor) + monkeypatch.setattr(module, "submit_message_to_host", submit_message) + client = object.__new__(client_type) + setattr(client, source_attr, "channel-main") + + assert client._forward_to_message_chain({"text": "hello"}) is True + assert calls == [({"text": "hello"}, "channel-main", executor.submit, 15)] + + +@pytest.mark.parametrize( + ("module", "client_type", "source_attr"), + [ + (telegram_module, telegram_module.Telegram, "_config_name"), + (clawbot_module, clawbot_module.WechatClawBot, "_config_name"), + ], +) +def test_sync_channels_forward_through_shared_ingress( + monkeypatch, + module, + client_type, + source_attr, +): + """同步轮询渠道必须复用统一回环请求和确认语义。""" + forward = MagicMock(return_value=True) + monkeypatch.setattr(module, "forward_message_to_host", forward) + client = object.__new__(client_type) + setattr(client, source_attr, "channel-main") + + assert client._forward_to_message_chain({"text": "hello"}) is True + forward.assert_called_once_with({"text": "hello"}, "channel-main") + + +def test_slack_preserves_callback_timeout_through_shared_ingress(monkeypatch): + """Slack action 的历史长超时必须继续传给统一入口。""" + forward = MagicMock(return_value=True) + monkeypatch.setattr(slack_module, "forward_message_to_host", forward) + client = object.__new__(slack_module.Slack) + client._config_name = "slack-main" + + assert client._forward_to_message_chain({"action": "run"}, timeout=60) is True + forward.assert_called_once_with( + {"action": "run"}, + "slack-main", + timeout=60, + ) + + +@pytest.mark.asyncio +async def test_discord_uses_shared_async_ingress(monkeypatch): + """Discord 自有事件循环不得继续维护独立 httpx 回环实现。""" + forward = AsyncMock(return_value=True) + monkeypatch.setattr(discord_module, "async_forward_message_to_host", forward) + client = object.__new__(discord_module.Discord) + client._config_name = "discord-main" + + await client._post_to_ds({"text": "hello"}) + + forward.assert_awaited_once_with( + {"text": "hello"}, + "discord-main", + timeout=10, + ) + + +def test_message_modules_cannot_reimplement_loopback_endpoint(): + """消息模块不得重新拼接宿主 URL,新增渠道必须复用统一 ingress。""" + violations = [] + for path in (PROJECT_ROOT / "app" / "modules").rglob("*.py"): + if "/api/v1/message" in path.read_text(encoding="utf-8-sig"): + violations.append(path.relative_to(PROJECT_ROOT).as_posix()) + + assert violations == []