Merge remote-tracking branch 'origin/v3' into pr-6401-resolve

# Conflicts:
#	app/startup/modules_initializer.py
#	tests/test_configuration_ports.py

app/adapters/network/browser.py

@@ -1118,6 +1118,7 @@ class PlaywrightHelper:
         :param headless: 是否无头模式
         :param timeout: 超时时间
         """
+        timeout = timeout or 60
         source = None
         # 如果配置为 FlareSolverr，则直接调用获取页面源码
         if self.__browser_emulation() == "flaresolverr":
@@ -1140,10 +1141,33 @@ class PlaywrightHelper:
                 if cookies:
                     page.set_extra_http_headers({"cookie": cookies})
 
-                page.goto(url)
-                page.wait_for_load_state("networkidle", timeout=timeout * 1000)
+                page.goto(url, wait_until="load", timeout=timeout * 1000)
 
-                source = page.content()
+                # 修复: 部分站点(如 Cloudflare 质询页)会持续轮询请求,
+                # 导致 networkidle 永不触发而超时。改为等待页面加载完成后
+                # 轮询检查标题, 直到不再停留在质询/加载页。
+                challenge_titles = ("just a moment", "请稍候", "loading")
+                deadline = time.time() + timeout
+                while time.time() < deadline:
+                    try:
+                        current_title = (page.title() or "").strip().lower()
+                    except Exception:
+                        current_title = ""
+                    if current_title and not any(
+                            t in current_title for t in challenge_titles):
+                        break
+                    time.sleep(2)
+
+                # 页面跳转中 content() 可能失败, 重试几次
+                source = None
+                for _attempt in range(5):
+                    try:
+                        source = page.content()
+                        if source:
+                            break
+                    except Exception:
+                        source = None
+                        time.sleep(2)
 
             except Exception as e:
                 logger.error(f"获取网页源码失败: {str(e)}")

app/chain/__init__.py

@@ -613,6 +613,29 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
             "search_torrents", site=site, keyword=keyword, mtype=mtype, page=page
         )
 
+    def search_plugin_torrents(
+            self,
+            keyword: str,
+            mtype: Optional[MediaType] = None,
+            page: Optional[int] = 0,
+    ) -> List[TorrentInfo]:
+        """仅搜索插件提供的资源源，避免依赖或重复绑定站点索引器。"""
+        return self._module_dispatcher.execute_plugin_modules(
+            "search_torrents", None, site={}, keyword=keyword, mtype=mtype, page=page
+        ) or []
+
+    def search_site_torrents(
+            self,
+            site: dict,
+            keyword: str,
+            mtype: Optional[MediaType] = None,
+            page: Optional[int] = 0,
+    ) -> List[TorrentInfo]:
+        """仅搜索指定站点索引器；插件资源源由搜索链统一调用一次。"""
+        return self._module_dispatcher.execute_system_modules(
+            "search_torrents", None, site=site, keyword=keyword, mtype=mtype, page=page
+        ) or []
+
     def search_subtitles(
             self,
             site: dict,
@@ -649,6 +672,31 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin,
             "async_search_torrents", site=site, keyword=keyword, mtype=mtype, page=page
         )
 
+    async def async_search_plugin_torrents(
+            self,
+            keyword: str,
+            mtype: Optional[MediaType] = None,
+            page: Optional[int] = 0,
+    ) -> List[TorrentInfo]:
+        """异步搜索插件提供的资源源。"""
+        return await self._module_dispatcher.async_execute_plugin_modules(
+            "async_search_torrents", None,
+            site={}, keyword=keyword, mtype=mtype, page=page
+        ) or []
+
+    async def async_search_site_torrents(
+            self,
+            site: dict,
+            keyword: str,
+            mtype: Optional[MediaType] = None,
+            page: Optional[int] = 0,
+    ) -> List[TorrentInfo]:
+        """异步搜索指定站点索引器。"""
+        return await self._module_dispatcher.async_execute_system_modules(
+            "async_search_torrents", None,
+            site=site, keyword=keyword, mtype=mtype, page=page
+        ) or []
+
     async def async_search_subtitles(
             self,
             site: dict,

app/chain/search.py

@@ -2321,9 +2321,15 @@ class SearchChain(ChainBase):
             # 检查站点索引开关
             if not sites or indexer.get("id") in sites:
                 indexer_sites.append(indexer)
+
+        plugin_results = self.search_plugin_torrents(
+            keyword=keyword,
+            mtype=mediainfo.type if mediainfo else mtype,
+            page=page,
+        )
         if not indexer_sites:
-            logger.warn('未开启任何有效站点，无法搜索资源')
-            return []
+            logger.info(f'未开启有效站点，插件资源源返回 {len(plugin_results)} 条资源')
+            return plugin_results
 
         # 开始进度
         progress = ProgressHelper(ProgressKey.Search)
@@ -2339,7 +2345,7 @@ class SearchChain(ChainBase):
         progress.update(value=0,
                         text=f"开始搜索，共 {len(indexer_sites)} 个站点，{len(search_pages)} 页 ...")
         # 结果集
-        results = []
+        results = list(plugin_results)
         # 同一站点按页顺序抓取，避免空页后仍继续请求该站点的后续页。
         max_workers = min(
             len(indexer_sites),
@@ -2356,13 +2362,13 @@ class SearchChain(ChainBase):
                 search_keyword = mediainfo.imdb_id if area == "imdbid" and mediainfo else keyword
                 if area == "imdbid":
                     # 搜索IMDBID
-                    task = executor.submit(self.search_torrents, site=site,
+                    task = executor.submit(self.search_site_torrents, site=site,
                                            keyword=search_keyword,
                                            mtype=mediainfo.type if mediainfo else mtype,
                                            page=search_page)
                 else:
                     # 搜索标题
-                    task = executor.submit(self.search_torrents, site=site,
+                    task = executor.submit(self.search_site_torrents, site=site,
                                            keyword=search_keyword,
                                            mtype=mediainfo.type if mediainfo else mtype,
                                            page=search_page)
@@ -2438,9 +2444,15 @@ class SearchChain(ChainBase):
             # 检查站点索引开关
             if not sites or indexer.get("id") in sites:
                 indexer_sites.append(indexer)
+
+        plugin_results = await self.async_search_plugin_torrents(
+            keyword=keyword,
+            mtype=mediainfo.type if mediainfo else mtype,
+            page=page,
+        )
         if not indexer_sites:
-            logger.warn('未开启任何有效站点，无法搜索资源')
-            return []
+            logger.info(f'未开启有效站点，插件资源源返回 {len(plugin_results)} 条资源')
+            return plugin_results
 
         # 开始进度（异步后端，避免同步 Redis 在事件循环上阻塞）
         progress = AsyncProgressHelper(ProgressKey.Search)
@@ -2456,7 +2468,7 @@ class SearchChain(ChainBase):
         await progress.update(value=0,
                               text=f"开始搜索，共 {len(indexer_sites)} 个站点，{len(search_pages)} 页 ...")
         # 结果集
-        results = []
+        results = list(plugin_results)
         semaphore = asyncio.Semaphore(
             self.runtime_config.search_threadpool_size or total_num
         )
@@ -2468,12 +2480,12 @@ class SearchChain(ChainBase):
             async with semaphore:
                 if area == "imdbid":
                     # 搜索IMDBID
-                    return await self.async_search_torrents(site=site,
+                    return await self.async_search_site_torrents(site=site,
                                                             keyword=mediainfo.imdb_id if mediainfo else None,
                                                             mtype=mediainfo.type if mediainfo else mtype,
                                                             page=search_page)
                 # 搜索标题
-                return await self.async_search_torrents(site=site,
+                return await self.async_search_site_torrents(site=site,
                                                         keyword=keyword,
                                                         mtype=mediainfo.type if mediainfo else mtype,
                                                         page=search_page)
@@ -2562,16 +2574,37 @@ class SearchChain(ChainBase):
         for indexer in await SitesHelper().async_get_indexers():
             if not sites or indexer.get("id") in sites:
                 indexer_sites.append(indexer)
+
+        plugin_results = await self.async_search_plugin_torrents(
+            keyword=keyword,
+            mtype=mediainfo.type if mediainfo else mtype,
+            page=page,
+        )
+        if plugin_results:
+            yield {
+                "type": "append",
+                "stage": "searching",
+                "value": 100 if not indexer_sites else 0,
+                "text": f"插件资源源返回 {len(plugin_results)} 条资源",
+                "items": plugin_results,
+                "site": "插件资源源",
+                "site_id": None,
+                "page": page,
+                "finished": 0,
+                "total": len(indexer_sites),
+                "total_items": len(plugin_results),
+            }
         if not indexer_sites:
-            logger.warn('未开启任何有效站点，无法搜索资源')
+            logger.info(f'未开启有效站点，插件资源源返回 {len(plugin_results)} 条资源')
             yield {
                 "type": "done",
                 "stage": "searching",
                 "value": 100,
-                "text": "未开启任何有效站点，无法搜索资源",
+                "text": f"搜索完成，共 {len(plugin_results)} 条资源",
                 "items": [],
                 "finished": 0,
-                "total": 0
+                "total": 0,
+                "total_items": len(plugin_results),
             }
             return
 
@@ -2604,12 +2637,12 @@ class SearchChain(ChainBase):
             """
             async with semaphore:
                 if area == "imdbid":
-                    site_result = await self.async_search_torrents(site=site,
+                    site_result = await self.async_search_site_torrents(site=site,
                                                                    keyword=mediainfo.imdb_id if mediainfo else None,
                                                                    mtype=mediainfo.type if mediainfo else mtype,
                                                                    page=search_page)
                 else:
-                    site_result = await self.async_search_torrents(site=site,
+                    site_result = await self.async_search_site_torrents(site=site,
                                                                    keyword=keyword,
                                                                    mtype=mediainfo.type if mediainfo else mtype,
                                                                    page=search_page)
@@ -2629,7 +2662,7 @@ class SearchChain(ChainBase):
         for site in indexer_sites:
             submit_site_page(site=site, page_index=0)
 
-        results_count = 0
+        results_count = len(plugin_results)
         try:
             while tasks:
                 if global_vars.is_system_stopped:

app/modules/indexer/__init__.py

@@ -634,12 +634,14 @@ class IndexerModule(_ModuleBase):
         :return: 用户数据
         """
 
-        def __get_site_obj() -> Optional[SiteParserBase]:
+        def __get_site_obj(schema_value: Optional[str] = None) -> Optional[SiteParserBase]:
             """
             获取站点解析器
+            :param schema_value: 指定 schema, 默认取站点声明的 schema
             """
+            schema_value = schema_value or site.get("schema")
             for site_schema in self._site_schemas:
-                if site_schema.schema and site_schema.schema.value == site.get("schema"):
+                if site_schema.schema and site_schema.schema.value == schema_value:
                     return site_schema(
                         site_name=site.get("name"),
                         url=site.get("url"),
@@ -651,6 +653,7 @@ class IndexerModule(_ModuleBase):
                         api_url=site.get("api_url"))
             return None
 
+        # 按站点声明的 schema 获取解析器
         site_obj = __get_site_obj()
         if not site_obj:
             if not site.get("public"):
@@ -662,6 +665,28 @@ class IndexerModule(_ModuleBase):
             logger.info(f"站点 {site.get('name')} 开始以 {site.get('schema')} 模型解析数据...")
             site_obj.parse()
             logger.debug(f"站点 {site.get('name')} 数据解析完成")
+            # 站点声明的 schema 解析失败(userid 为空)时, 自动尝试其他解析器,
+            # 兼容资源文件 schema 标注错误/变种站点的场景
+            if not site_obj.userid and not site.get("public"):
+                tried = {site.get("schema")}
+                for site_schema in self._site_schemas:
+                    if not site_schema.schema or site_schema.schema.value in tried:
+                        continue
+                    tried.add(site_schema.schema.value)
+                    logger.info(f"站点 {site.get('name')} schema {site.get('schema')} 解析失败, "
+                                f"尝试 {site_schema.schema.value} 模型...")
+                    alt_obj = __get_site_obj(site_schema.schema.value)
+                    if not alt_obj:
+                        continue
+                    try:
+                        alt_obj.parse()
+                    except Exception as e:
+                        logger.error(f"站点 {site.get('name')} 以 {site_schema.schema.value} 解析失败: {str(e)}")
+                        continue
+                    if alt_obj.userid:
+                        site_obj = alt_obj
+                        logger.info(f"站点 {site.get('name')} 改用 {site_schema.schema.value} 模型解析成功")
+                        break
             return SiteUserData(
                 domain=site_rules.extract_domain(site.get("url")),
                 userid=site_obj.userid,

app/modules/indexer/parser/gazelle.py

@@ -93,6 +93,27 @@ class GazelleSiteUserInfo(SiteParserBase):
                     '//div[contains(@class, "box_userinfo_stats")]//li[contains(text(), "加入时间")]/span/text()')
                 if join_at_text:
                     self.join_at = time_tools.normalize_datetime(join_at_text[0].strip())
+
+            # 兼容部分 Gazelle 站点(如 JPopsuki)以文本形式展示上传/下载:
+            # <li>Uploaded: 77.44 GB</li> / <li>Downloaded: 8.51 GB</li>
+            if not self.upload:
+                upload_text = html.xpath(
+                    '//li[starts-with(normalize-space(text()), "Uploaded:")]')
+                if upload_text:
+                    size_match = re.search(
+                        r"([\d.,]+\s*[GMKT]?i?B)", upload_text[0].xpath("string(.)"), re.I)
+                    if size_match:
+                        self.upload = size_tools.parse_size(size_match.group(1))
+            if not self.download:
+                download_text = html.xpath(
+                    '//li[starts-with(normalize-space(text()), "Downloaded:")]')
+                if download_text:
+                    size_match = re.search(
+                        r"([\d.,]+\s*[GMKT]?i?B)", download_text[0].xpath("string(.)"), re.I)
+                    if size_match:
+                        self.download = size_tools.parse_size(size_match.group(1))
+            if not self.ratio and self.upload and self.download:
+                self.ratio = round(self.upload / self.download, 3)
         finally:
             if html is not None:
                 del html

app/modules/indexer/parser/nexus_project.py

@@ -1,6 +1,8 @@
 # -*- coding: utf-8 -*-
 import re
 
+from lxml import etree
+
 from app.modules.indexer.parser import SiteSchema
 from app.modules.indexer.parser.nexus_php import NexusPhpSiteUserInfo
 
@@ -11,9 +13,45 @@ class NexusProjectSiteUserInfo(NexusPhpSiteUserInfo):
     def _parse_site_page(self, html_text: str):
         html_text = self._prepare_html_text(html_text)
 
-        user_detail = re.search(r"userdetails.php\?id=(\d+)", html_text)
+        user_detail = re.search(r"userdetails\.php\?id=(\d+)", html_text)
         if user_detail and user_detail.group().strip():
             self._user_detail_page = user_detail.group().strip().lstrip('/')
             self.userid = user_detail.group(1)
+        else:
+            # 兼容部分 NexusProject 变种站点(如 star-space.net)的
+            # p_user/user_detail.php?uid= 用户定位格式
+            user_detail = re.search(r"user_detail\.php\?uid=(\d+)", html_text)
+            if user_detail and user_detail.group().strip():
+                self._user_detail_page = user_detail.group().strip().lstrip('/')
+                self.userid = user_detail.group(1)
+                uname = re.search(
+                    r"user_detail\.php\?uid=\d+[^>]*?>\s*<[^>]*>([^<]+)<", html_text)
+                if uname:
+                    self.username = uname.group(1).strip()
 
         self._torrent_seeding_page = f"viewusertorrents.php?id={self.userid}&show=seeding"
+
+    def _parse_user_traffic_info(self, html_text):
+        # 兼容部分 NexusProject 变种站点(如 star-space.net)以
+        # span#user_info / span#user_info_no_hover 文本展示流量:
+        # "上传：445.11 G" / "下载：29.61 G"(单位无 B 后缀)
+        try:
+            html = etree.HTML(html_text)
+            if html is not None:
+                body_text = " ".join(html.xpath(
+                    '//span[@id="user_info"]//text() | //span[@id="user_info_no_hover"]//text()'))
+                size_match = re.search(r"上传[：:]\s*([\d.,]+)\s*([GMKT]?i?B?)", body_text, re.I)
+                if size_match:
+                    self.upload = self.num_filesize(
+                        f"{size_match.group(1).replace(',', '')} {size_match.group(2).upper()}B")
+                size_match = re.search(r"下载[：:]\s*([\d.,]+)\s*([GMKT]?i?B?)", body_text, re.I)
+                if size_match:
+                    self.download = self.num_filesize(
+                        f"{size_match.group(1).replace(',', '')} {size_match.group(2).upper()}B")
+                if self.upload and self.download:
+                    self.ratio = round(self.upload / self.download, 3)
+                if self.upload or self.download:
+                    return
+        except Exception:
+            pass
+        super()._parse_user_traffic_info(html_text)

docker/update.sh

@@ -286,7 +286,12 @@ function stage_runtime_payload() {
     else
         mkdir -p "${stage_plugin_dir}" || return 1
     fi
-    rm -f "${stage_plugin_dir}/__init__.py"
+    # 保留 app.plugins 兼容入口；V1/V2 插件仍从这里导入 _PluginBase。
+    # 删除后 app.plugins 会退化为 namespace package，旧插件会在启动时全部导入失败。
+    if [ ! -f "${stage_plugin_dir}/__init__.py" ]; then
+        ERROR "插件运行目录缺少 app.plugins 兼容入口"
+        return 1
+    fi
 
     resource_source_dir="$(existing_resource_dir)"
     mkdir -p "${stage_resource_dir}" || return 1

docs/refactor/backend-architecture-next-stage.md

@@ -8,6 +8,65 @@
 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md`
 > 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；插件仓适配、Outbox 外围扩展和 Model 查询兼容面仍按风险切片推进。
 
+## 当前复核结论（2026-08-23）
+
+本节是本轮全面复核后的当前事实源。本文后续的阶段实施记录保留历史审计证据，
+其中的数量和判断以当时审计提交为准，不能直接当作当前未完成项。
+
+### 总体判断
+
+当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**：
+
+- 继续采用单进程控制面是正确选择，不建议现在拆成微服务；插件、调度器、工作流、事件和数据库共享进程内状态，拆分会放大部署、事务和兼容成本。
+- `foundation/domain/runtime/adapters/application/chain/api/startup` 的职责方向基本成立；宿主架构基线、复杂度 ratchet、异步阻塞 ratchet 当前均通过。
+- 依赖图当前约 `796` 个 Python 模块、`6432` 条内部导入边；唯一非平凡 SCC 位于隔离的 TMDB 第三方移植包内部，不应为了指标归零重写。
+- 当前主要风险已经从“目录和依赖失控”转移到运行时协议、后台副作用的可靠性和遗留兼容面。换言之，下一阶段重点应是**语义收口和可验证性**，而不是继续搬文件或机械拆大文件。
+
+综合评价：架构方向可持续，生产可用性较高；可演进性仍处于中等水平。现阶段没有静态审计发现必须立即推倒重来的 P0 架构问题，但存在需要按 P1/P2 计划治理的真实债务。
+
+### P1：需要优先治理的真实债务
+
+1. **后台任务没有统一的所有权和恢复模型。** 当前约有 `50` 个 `create_task`/等价任务创建点，另有 FastAPI `BackgroundTasks`、线程池和 APScheduler 并存。生命周期清单能关闭模块、插件、调度器和 Agent，但 API 层的若干任务集合（如 `app/api/endpoints/agent.py`、`app/api/endpoints/plugin.py`）没有统一注册到 HostRuntime，也没有在 shutdown 阶段统一等待或取消。`app/api/endpoints/webhook.py:32`、`app/api/endpoints/site.py:178` 这类接口会先返回成功，再执行关键副作用；进程崩溃、重启或客户端断开时可能丢失。需要为每类任务明确 owner、取消、等待、重试、幂等和是否 durable，关键业务副作用优先接入已有 Outbox/恢复表。
+2. **动态模块契约仍以 legacy 聚合语义为主。** 当前登记 `212` 个模块方法，其中 `194` 个仍使用 `legacy` aggregation，只有 `14` 个 `first_non_empty`、`4` 个 `ordered_list_merge`。`app/runtime/extensions/module/contracts.py:422-455` 已能登记 family、输入/结果标签和基础签名诊断，但 `193` 个方法没有 required parameters，调度器 `app/runtime/extensions/module/dispatcher.py:109-260` 仍主要依赖运行时反射、返回值形状和短路规则。未知第三方方法保留 legacy fallback 是兼容要求，不应删除；宿主高频能力则应逐族补齐可执行的输入校验、结果校验、超时和错误语义。
+3. **查询侧数据库兼容 ABI 仍未完全收口。** 写事务装饰器已降为 `0`，事务所有权已经明显改善；但 `app/db/models` 仍有 `106` 个 `db_query/async_db_query`（`62` 个同步、`44` 个异步）。这些装饰器会在调用方未传 Session 时隐式创建并关闭会话（见 `app/db/decorators.py:224-298`），查询返回的 ORM 对象仍可能跨层流转，导致事务组合、对象生命周期和懒加载行为需要依赖隐含约定。应按高频业务路径逐步迁移到显式 Query/Repository + 请求/任务级 Session，不宜一次性全仓改写。
+4. **组合根和全局状态仍形成复杂的隐式运行时图。** Singleton 实例、模块级 provider、`configure_*` 注册函数和兼容 Facade 同时存在；它们解决了旧 ABI 和启动顺序问题，但增加测试污染、重复装配、实例身份和初始化顺序风险。`app/startup/lifecycle/__init__.py:161-376` 已有声明式生命周期，`app/startup/modules_initializer.py:505-530` 也有分阶段关闭，但尚未做到所有进程级资源都只通过 typed HostRuntime 访问。后续应以“新代码禁止新增 Service Locator/Singleton 依赖、旧入口有命中观测”为 ratchet。
+
+### P2：中长期可演进性债务
+
+- **大型职责域仍偏重。** 代表性热点包括 `app/chain/subscribe.py`（约 `4141` 行）、`app/chain/transfer.py`（约 `2944` 行）、`app/agent/orchestrator.py`（约 `3535` 行）、`app/agent/llm/provider.py`（约 `3529` 行）、`app/adapters/external/market.py`（约 `2805` 行）和 `app/api/endpoints/agent.py`（约 `2326` 行）。复杂度 ratchet 只保证不超过当前基线，不代表这些文件已经易维护。只有在行为快照、调用命中和事务边界明确后，才值得按用例拆分。
+- **类型门禁覆盖面不足。** `mypy.ini` strict 文件清单目前约 `37` 个文件，Agent、Chain、Module、Adapter 大量代码仍依赖动态类型。应从模块契约、生命周期、Repository/Port 和关键 Chain 返回值开始扩展，而不是直接开启全仓 strict。
+- **Pylint 仍是增量硬门禁。** `.github/workflows/pylint.yml` 对改动 Python 文件执行硬检查，但全仓报告使用 `|| true` 仅作 advisory。该策略适合存量迁移，却没有形成全仓质量趋势约束；应增加按目录和新增问题数的 ratchet。
+- **测试风格存在历史混用。** 当前约 `499` 个测试文件，仍有约 `70` 个 `unittest.TestCase` 文件。它不是生产架构缺陷，但会增加 fixture、状态隔离和异步测试迁移成本，应在触碰相关模块时渐进迁移。
+- **跨仓治理链路尚未完全闭环。** 前端已有 lint、typecheck、分片 Vitest 和构建门禁；插件仓有 V1/V2/V3 索引及版本/依赖检查；资源和 Rust 仓有独立构建发布链路。但插件 CI 本地复核因插件仓环境缺少主仓依赖 `httpx2` 无法完成收集，说明“插件仓测试环境与主仓锁定依赖”的可复现性仍需加强。资源构建通过 PR 同步到 `MoviePilot-Resources`，Rust 发布后自动向主仓发依赖 bump PR，链路合理但仍是多仓异步发布，需保留版本 provenance 和回滚点。
+
+### 已解决、不应重复治理的问题
+
+- 分层依赖和重点禁止边已建立门禁；不要再以“减少目录数量”作为目标。
+- 全功能多 worker 的误导性配置已由 `app/runtime/topology.py` 和 `app/main.py` 拒绝；V3 默认单 worker 的部署事实已经明确。
+- 写事务已由组合根/UoW/Outbox 方向收口，`db_update/async_db_update` 为零；不要重新引入 Model 自动提交。
+- 已具备 correlation ID、`/health/live`、`/health/ready`、模块/事件/调度观测端口和兼容 Facade 命中指标；历史文档中“完全缺少观测能力”的描述已过时。
+- 旧导入路径、SDK 导出、插件 manifest 和 V1/V2/V3 索引均有白名单或版本约束；兼容层应继续保持“薄、可观测、只增不删”，不应为了清理目录直接删除。
+- TMDB 移植包内部 SCC 属于第三方隔离代码，按现状豁免是合理的技术决策。
+
+### 刻意保留的兼容成本
+
+以下内容不是遗漏，而是当前产品 ABI 的有意成本：
+
+1. 未知第三方插件自定义模块方法继续走 `legacy` fallback，不能因宿主契约收口而拒绝加载旧插件。
+2. `PluginManager`、`PluginHelper`、`MoviePilotServerHelper` 等 Facade 继续保留旧公开/私有调用面，并通过 `compat.facade.hit` 统计迁移命中。
+3. `app/runtime/compat` 的精确旧导入映射、`app.sdk._legacy` 薄门面和插件 V1/V2/V3 三代索引继续存在，直到命中数据和发行策略支持删除。
+4. 查询装饰器保留为只读兼容入口，迁移以高频路径和可观测收益为依据，不以“全仓零装饰器”作为短期目标。
+
+### 建议的后续治理顺序
+
+1. **先做后台任务审计与统一登记**：建立 TaskOwner/生命周期协议，区分请求后非关键通知、可重试 Outbox 副作用和必须在请求内完成的业务写入；为断线、崩溃、重复执行和 shutdown 补测试。
+2. **再做模块契约 V2 增量收口**：优先识别调用量最高、影响下载/整理/识别的能力族，补真实参数对象、结果验证、超时预算和 provider 行为快照；legacy fallback 只保留给第三方未知方法。
+3. **随后迁移查询 ABI**：从订阅、历史、消息、用户和站点等高频查询开始，逐步让 Query/Repository 接收显式 Session，并验证 detached 对象、懒加载和事务组合。
+4. **最后扩展类型和复杂度预算**：每次触碰大型职责域时拆一个可回滚垂直切片，同时扩大 mypy strict 清单和 Pylint 新增问题 ratchet；不要为追求行数指标进行无行为收益的拆分。
+5. **跨仓发布以契约为中心**：保持插件索引、前端远程组件、资源版本、Rust wheel 和主仓依赖的 provenance；将插件测试环境固定为主仓 `uv.lock` 可复现安装，避免本地和 CI 依赖漂移。
+
+本轮复核结论：**当前架构不需要推倒重来，真正未完成的是运行时可靠性和协议收口。** 下一轮治理完成上述 P1 后，再评估是否值得继续拆分大型文件或扩大严格类型范围。
+
 ## 1. 结论先行
 
 MoviePilot V3 当前不是“目录混乱、必须推倒重来”的状态。第一阶段治理已经取得实质成果：

scripts/perf/moviepilot_docker_ab.py

@@ -107,7 +107,7 @@ FROM ${MP_SUBSTRATE} AS frozen
 RUN set -eux; \
     mkdir -p /frozen/plugins /frozen/site; \
     cp -a /app/app/plugins/. /frozen/plugins/; \
-    rm -f /frozen/plugins/__init__.py; \
+    test -f /frozen/plugins/__init__.py; \
     rm -rf /frozen/plugins/__pycache__; \
     find /app/app/application/site -maxdepth 1 -type f \
       \( -name 'sites.*.so' -o -name 'user.sites.v3.bin' \) \

tests/test_configuration_ports.py

@@ -20,6 +20,7 @@ from app.application.configuration import (
     get_transfer_retry_config,
 )
 from app.application.security.userconfig import UserConfigurationService
+from app.runtime.settings import RuntimeSettingsCompat, configure_runtime_settings_compat
 
 
 class _InlineDatabaseExecutor:
@@ -71,6 +72,16 @@ def test_runtime_settings_service_hides_mutable_settings_implementation() -> Non
     assert service.get("VALUE") == "final"
 
 
+def test_runtime_settings_compat_delegates_to_concrete_service_backend() -> None:
+    """兼容 Settings 代理委托到真实设置对象时不会在 model_dump 中递归。"""
+    service = RuntimeSettingsService(_MutableSettings())
+    configure_runtime_settings_compat(service)
+
+    assert RuntimeSettingsCompat().model_dump(include={"VALUE"}) == {
+        "VALUE": "before"
+    }
+
+
 def test_system_config_service_supports_separate_reader_and_writer() -> None:
     """应用服务可以分别注入只读与写入适配器。"""
     reader = MagicMock()

tests/test_docker_bootstrap.py

@@ -921,6 +921,9 @@ def test_failed_dependency_sync_does_not_replace_program_files(tmp_path: Path) -
     (live_app / "app" / "application" / "site").mkdir(parents=True)
     live_public.mkdir()
     (live_app / "app" / "old.py").write_text("old", encoding="utf-8")
+    (live_app / "app" / "plugins" / "__init__.py").write_text(
+        "# legacy plugin compatibility entrypoint\n", encoding="utf-8"
+    )
     (live_app / "app" / "plugins" / "plugin.py").write_text("plugin", encoding="utf-8")
     (live_app / "app" / "application" / "site" / "user.sites.v3.bin").write_text(
         "sites", encoding="utf-8"
@@ -931,6 +934,9 @@ def test_failed_dependency_sync_does_not_replace_program_files(tmp_path: Path) -
 
     update_tree = tmp_path / "update" / "App"
     (update_tree / "app" / "plugins").mkdir(parents=True)
+    (update_tree / "app" / "plugins" / "__init__.py").write_text(
+        "# legacy plugin compatibility entrypoint\n", encoding="utf-8"
+    )
     (update_tree / "pyproject.toml").write_text("[project]\n", encoding="utf-8")
     (update_tree / "uv.lock").write_text("version = 1\n", encoding="utf-8")
     (update_tree / "version.py").write_text("FRONTEND_VERSION = 'v3.0.1'\n", encoding="utf-8")

tests/test_docker_payload_contract.py

@@ -77,6 +77,15 @@ def test_dockerfile_assigns_each_payload_to_an_independent_stage() -> None:
     assert "RUN rm -rf /app/frontend-dist" in dockerfile
 
 
+def test_plugin_runtime_updates_preserve_legacy_base_entrypoint() -> None:
+    """更新和性能覆盖镜像必须保留旧插件导入 _PluginBase 所需的兼容入口。"""
+    update_script = _read(ROOT / "docker" / "update.sh")
+    perf_script = _read(ROOT / "scripts" / "perf" / "moviepilot_docker_ab.py")
+
+    assert 'rm -f "${stage_plugin_dir}/__init__.py"' not in update_script
+    assert "rm -f /frozen/plugins/__init__.py" not in perf_script
+
+
 def test_release_workflows_pin_and_record_external_payload_identities() -> None:
     """正式与 Beta 构建都必须以真实制品身份驱动缓存并写入镜像标签。"""
     for workflow_path in (RELEASE_WORKFLOW, BETA_WORKFLOW):

tests/test_search_ai_recommend.py

@@ -55,6 +55,12 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase):
         chain.save_cache = lambda _cache, _filename: None
         chain.remove_cache = lambda _filename: None
         chain.get_search_page_size = IndexerModule.get_search_page_size
+        chain.search_plugin_torrents = lambda **_kwargs: []
+
+        async def no_plugin_results(**_kwargs):
+            return []
+
+        chain.async_search_plugin_torrents = no_plugin_results
         return chain
 
     async def test_start_recommend_task_restores_original_indices(self):
@@ -185,7 +191,7 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase):
                 for index in range(count)
             ]
 
-        chain.search_torrents = search_torrents
+        chain.search_site_torrents = search_torrents
 
         with (
             patch.object(settings, "SEARCH_RESOURCE_PAGES", 4, create=True),
@@ -231,7 +237,7 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase):
                 for index in range(count)
             ]
 
-        chain.search_torrents = search_torrents
+        chain.search_site_torrents = search_torrents
 
         with (
             patch.object(settings, "SEARCH_RESOURCE_PAGES", 3, create=True),
@@ -277,7 +283,7 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase):
                 for index in range(count)
             ]
 
-        chain.search_torrents = search_torrents
+        chain.search_site_torrents = search_torrents
 
         with (
             patch.object(settings, "SEARCH_RESOURCE_PAGES", 3, create=True),
@@ -342,7 +348,7 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase):
                 for index in range(count)
             ]
 
-        chain.async_search_torrents = async_search_torrents
+        chain.async_search_site_torrents = async_search_torrents
 
         with (
             patch.object(settings, "SEARCH_RESOURCE_PAGES", 4, create=True),
@@ -388,7 +394,7 @@ class SearchChainAIRecommendTest(unittest.IsolatedAsyncioTestCase):
                 for index in range(count)
             ]
 
-        chain.async_search_torrents = async_search_torrents
+        chain.async_search_site_torrents = async_search_torrents
 
         with (
             patch.object(settings, "SEARCH_RESOURCE_PAGES", 3, create=True),

tests/test_search_plugin_sources.py

@@ -0,0 +1,124 @@
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+from app.chain.search import SearchChain
+from app.modules.indexer import IndexerModule
+from app.runtime.config import settings
+
+
+def make_chain() -> SearchChain:
+    """构造不触发完整启动流程的搜索链。"""
+    chain = object.__new__(SearchChain)
+    chain.get_search_page_size = IndexerModule.get_search_page_size
+    return chain
+
+
+def test_search_returns_plugin_results_without_indexer_sites():
+    """未配置 PT 站点时，插件资源仍应进入原生资源搜索。"""
+    chain = make_chain()
+    plugin_item = SimpleNamespace(title="Plugin Result", description="")
+    calls = []
+    chain.search_plugin_torrents = lambda **kwargs: calls.append(kwargs) or [plugin_item]
+
+    with (
+        patch("app.chain.search.get_configured_system_config") as system_config_oper,
+        patch("app.chain.search.SitesHelper") as sites_helper,
+    ):
+        system_config_oper.return_value.get.return_value = []
+        sites_helper.return_value.get_indexers.return_value = []
+        results = chain._SearchChain__search_all_sites(keyword="keyword")
+
+    assert results == [plugin_item]
+    assert len(calls) == 1
+    assert calls[0]["keyword"] == "keyword"
+
+
+def test_search_invokes_plugin_once_with_multiple_indexers():
+    """多个 PT 站点不应导致插件资源源被重复搜索。"""
+    chain = make_chain()
+    plugin_calls = []
+    site_calls = []
+    chain.search_plugin_torrents = lambda **kwargs: plugin_calls.append(kwargs) or [
+        SimpleNamespace(title="Plugin Result", description="")
+    ]
+    chain.search_site_torrents = lambda **kwargs: site_calls.append(kwargs) or [
+        SimpleNamespace(title=f"Site {kwargs['site']['id']}", description="")
+    ]
+
+    with (
+        patch.object(settings, "SEARCH_RESOURCE_PAGES", 1, create=True),
+        patch("app.chain.search.get_configured_system_config") as system_config_oper,
+        patch("app.chain.search.SitesHelper") as sites_helper,
+        patch("app.chain.search.ProgressHelper") as progress_helper,
+    ):
+        system_config_oper.return_value.get.return_value = [1, 2]
+        sites_helper.return_value.get_indexers.return_value = [
+            {"id": 1, "name": "站点一"},
+            {"id": 2, "name": "站点二"},
+        ]
+        progress_helper.return_value = SimpleNamespace(
+            start=lambda: None, update=lambda **_kwargs: None, end=lambda: None
+        )
+        results = chain._SearchChain__search_all_sites(keyword="keyword")
+
+    assert len(plugin_calls) == 1
+    assert sorted(call["site"]["id"] for call in site_calls) == [1, 2]
+    assert len(results) == 3
+
+
+def test_async_search_returns_plugin_results_without_indexers():
+    """异步搜索应支持只有插件资源源的部署方式。"""
+    chain = make_chain()
+    plugin_item = SimpleNamespace(title="Plugin Result", description="")
+    calls = []
+
+    async def plugin_search(**kwargs):
+        calls.append(kwargs)
+        return [plugin_item]
+
+    chain.async_search_plugin_torrents = plugin_search
+    async def run_search():
+        with (
+            patch("app.chain.search.get_configured_system_config") as system_config_oper,
+            patch("app.chain.search.SitesHelper") as sites_helper,
+        ):
+            system_config_oper.return_value.get.return_value = []
+            sites_helper.return_value.async_get_indexers = AsyncMock(return_value=[])
+            return await chain._SearchChain__async_search_all_sites(keyword="keyword")
+
+    results = asyncio.run(run_search())
+
+    assert results == [plugin_item]
+    assert len(calls) == 1
+
+
+def test_async_search_stream_emits_plugin_results_once_without_indexers():
+    """流式搜索完成事件不应重复发送插件资源。"""
+    chain = make_chain()
+    plugin_item = SimpleNamespace(title="Plugin Result", description="")
+
+    async def plugin_search(**_kwargs):
+        return [plugin_item]
+
+    chain.async_search_plugin_torrents = plugin_search
+    async def collect_events():
+        with (
+            patch("app.chain.search.get_configured_system_config") as system_config_oper,
+            patch("app.chain.search.SitesHelper") as sites_helper,
+        ):
+            system_config_oper.return_value.get.return_value = []
+            sites_helper.return_value.async_get_indexers = AsyncMock(return_value=[])
+            return [
+                event
+                async for event in chain._SearchChain__async_search_all_sites_stream(
+                    keyword="keyword"
+                )
+            ]
+
+    events = asyncio.run(collect_events())
+
+    assert [event["type"] for event in events] == ["append", "done"]
+    assert events[0]["items"] == [plugin_item]
+    assert events[1]["items"] == []
+    assert events[1]["total_items"] == 1