fix: 协调本地插件与绑定仓库的版本选择 (#6482)

* fix(plugin): reconcile local and bound payload versions

* fix(plugin): preserve bound source on updates

* fix(plugin): normalize source reconciliation ids

* fix(plugin): defer local payload activation

* fix(plugin): centralize startup source admission

* fix(plugin): derive sync mode from admitted source

app/api/endpoints/plugin.py

@@ -669,7 +669,7 @@ async def install(
     """
     result = await get_plugin_install_service().install(
         plugin_id=plugin_id,
-        repo_url=None,
+        repo_url=repo_url or None,
         release_version=release_version,
         force=bool(force),
         explicit_source=False,

app/application/plugin/catalog.py

@@ -11,6 +11,7 @@ from app.application.plugin.identity import (
     PluginBindingBasis,
     PluginIdentity,
     TrustedPluginSourceType,
+    normalize_physical_plugin_id,
 )
 from app.schemas.plugin import Plugin, PluginSourceBindingStatus
 
@@ -274,13 +275,16 @@ class PluginCatalogService:
         """按代际、来源顺序和版本合并插件目录。"""
         all_plugins = list(higher_plugins)
         higher_keys = {
-            f"{plugin.id}{plugin.plugin_version}"
+            (normalize_physical_plugin_id(plugin.id), plugin.plugin_version)
             for plugin in higher_plugins
         }
         all_plugins.extend(
             plugin
             for plugin in base_plugins
-            if f"{plugin.id}{plugin.plugin_version}" not in higher_keys
+            if (
+                normalize_physical_plugin_id(plugin.id),
+                plugin.plugin_version,
+            ) not in higher_keys
         )
 
         def repo_order(plugin: Any) -> int:
@@ -293,7 +297,10 @@ class PluginCatalogService:
 
         deduplicated = {}
         for plugin in sorted(all_plugins, key=repo_order):
-            key = f"{plugin.id}{plugin.plugin_version}"
+            key = (
+                normalize_physical_plugin_id(plugin.id),
+                plugin.plugin_version,
+            )
             exists = deduplicated.get(key)
             if not exists or (
                 self._is_local_repo(exists.repo_url)
@@ -303,7 +310,8 @@ class PluginCatalogService:
 
         result_by_id = {}
         for plugin in sorted(deduplicated.values(), key=repo_order):
-            exists = result_by_id.get(plugin.id)
+            normalized_id = normalize_physical_plugin_id(plugin.id)
+            exists = result_by_id.get(normalized_id)
             if not exists \
                     or self._version_compare(
                         plugin.plugin_version,
@@ -315,7 +323,7 @@ class PluginCatalogService:
                         and self._is_local_repo(exists.repo_url)
                         and not self._is_local_repo(plugin.repo_url)
                     ):
-                result_by_id[plugin.id] = plugin
+                result_by_id[normalized_id] = plugin
         return list(result_by_id.values())
 
     def merge_by_source(
@@ -326,14 +334,22 @@ class PluginCatalogService:
     ) -> list[Any]:
         """每个仓库保留同一插件的最高兼容版本，供来源准入继续决策。"""
         higher_keys = {
-            (plugin.repo_url, plugin.id, plugin.plugin_version)
+            (
+                plugin.repo_url,
+                normalize_physical_plugin_id(plugin.id),
+                plugin.plugin_version,
+            )
             for plugin in higher_plugins
         }
         all_plugins = list(higher_plugins)
         all_plugins.extend(
             plugin
             for plugin in base_plugins
-            if (plugin.repo_url, plugin.id, plugin.plugin_version) not in higher_keys
+            if (
+                plugin.repo_url,
+                normalize_physical_plugin_id(plugin.id),
+                plugin.plugin_version,
+            ) not in higher_keys
         )
 
         def repo_order(plugin: Any) -> int:
@@ -341,9 +357,9 @@ class PluginCatalogService:
                 return markets.index(plugin.repo_url)
             return len(markets)
 
-        result_by_source: dict[tuple[Optional[str], Optional[str]], Any] = {}
+        result_by_source: dict[tuple[Optional[str], str], Any] = {}
         for plugin in sorted(all_plugins, key=repo_order):
-            key = (plugin.repo_url, plugin.id)
+            key = (plugin.repo_url, normalize_physical_plugin_id(plugin.id))
             exists = result_by_source.get(key)
             if not exists or self._version_compare(
                 plugin.plugin_version,

app/application/plugin/gateway.py

@@ -117,11 +117,15 @@ class PluginInstallGateway:
                     raise PluginSourceAdmissionError(
                         message or "插件包与当前 MoviePilot 版本不兼容"
                     )
+                # 本地同步是最终准入候选的执行属性，不能由调用前的 URL 推断。
                 return await self.__executor.execute(
                     admission=admission,
                     release_version=release_version,
                     force=force,
-                    local_sync=local_sync,
+                    local_sync=isinstance(
+                        admission.candidate,
+                        PluginLocalCandidate,
+                    ),
                 )
         except (TypeError, ValueError, PluginSourceAdmissionError) as error:
             return PluginInstallResult(

app/application/plugin/identity.py

@@ -11,7 +11,7 @@ from typing import Protocol
 
 from app.application.plugin.declaration import PluginDeclaredMetadata
 
-_PLUGIN_ID_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9]{0,127}$")
+_PLUGIN_ID_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,127}$")
 _ONLINE_SOURCE_KEY_PATTERN = re.compile(
     r"^github:[a-z0-9](?:[a-z0-9-]{0,38})/"
     r"[a-z0-9._-]{1,100}$"
@@ -59,9 +59,9 @@ class PluginIdentityConflictError(RuntimeError):
 
 
 def normalize_physical_plugin_id(plugin_id: str) -> str:
-    """校验物理插件 ID，并返回大小写无关的数据库身份键。"""
+    """校验可安全持久化的物理插件 ID，并返回大小写无关身份键。"""
     if plugin_id != plugin_id.strip() or not _PLUGIN_ID_PATTERN.fullmatch(plugin_id):
-        raise ValueError("插件 ID 必须以字母开头且只能包含 ASCII 字母或数字")
+        raise ValueError("插件 ID 必须以字母开头且只能包含 ASCII 字母、数字或下划线")
     return plugin_id.lower()
 
 

app/application/plugin/source.py

@@ -553,13 +553,13 @@ def select_plugin_candidate(
     allow_source_change: bool = False,
 ) -> PluginSelection:
     """
-    按允许来源、运行代际和同源版本选择一个插件载荷。
+    按在线绑定、本地候选、运行代际和版本选择一个插件载荷。
 
     :param inventory: 本轮市场读取快照
     :param plugin_id: 要选择的物理插件 ID
     :param generations: 调用方按优先级传入的代际顺序
     :param identity: 已安装插件来源身份；为空表示未安装
-    :param local_candidates: 可选的本地载荷候选，优先于在线候选
+    :param local_candidates: 可选的本地载荷候选；已有在线绑定时参与代际和版本比较
     :param requested_source_key: 调用方提供的规范在线来源；非显式调用不能绕过本地载荷
     :param explicit_source: 本次调用是否代表管理员明确选源
     :param allow_source_change: 是否是带 revision 的显式换源命令
@@ -572,6 +572,7 @@ def select_plugin_candidate(
         if requested_source_key is not None
         else None
     )
+    local_selection: PluginSelection | None = None
     if requested_source is None or not (explicit_source or allow_source_change):
         local_selection = _select_local_candidate(
             inventory,
@@ -580,11 +581,16 @@ def select_plugin_candidate(
             generation_order=generation_order,
             local_candidates=local_candidates,
         )
-        if local_selection is not None:
+        if (
+            local_selection is not None
+            and local_selection.status is PluginSelectionStatus.INCOMPLETE
+        ):
             return local_selection
 
     online = inventory.candidates_for(plugin_id)
     if not online:
+        if local_selection is not None:
+            return local_selection
         return PluginSelection(
             status=PluginSelectionStatus.UNAVAILABLE,
             reason=f"没有找到插件 {plugin_id} 的可用安装包",
@@ -627,35 +633,25 @@ def select_plugin_candidate(
                 ),
             )
     if allowed_source is not None:
-        source_type, source_key = allowed_source
+        return _select_bound_source_candidate(
-        online = tuple(
+            online=online,
-            candidate
+            allowed_source=allowed_source,
-            for candidate in online
+            local_selection=local_selection,
-            if candidate.source_type is source_type and candidate.source_key == source_key
+            identity=identity,
-        )
+            generation_order=generation_order,
-        if not online:
-            return PluginSelection(
-                status=PluginSelectionStatus.UNAVAILABLE,
-                reason="已绑定仓库中暂无可用插件包",
-            )
-        selected_online = _select_best(online, generation_order)
-        if selected_online is None:
-            return PluginSelection(
-                status=PluginSelectionStatus.UNAVAILABLE,
-                reason="已绑定仓库没有适用于当前 MoviePilot 版本的插件包",
-            )
-        return PluginSelection(
-            status=PluginSelectionStatus.SELECTED,
-            candidate=selected_online,
-            reason="已使用绑定仓库中的插件包",
         )
 
     if identity is not None:
+        if local_selection is not None:
+            return local_selection
         return PluginSelection(
             status=PluginSelectionStatus.INCOMPLETE,
             reason="当前插件尚未绑定仓库",
         )
 
+    if local_selection is not None:
+        return local_selection
+
     source_pairs = {(candidate.source_type, candidate.source_key) for candidate in online}
     if len(source_pairs) > 1:
         return PluginSelection(
@@ -683,6 +679,102 @@ def select_plugin_candidate(
     )
 
 
+def _select_bound_source_candidate(
+    *,
+    online: tuple[PluginMarketCandidate, ...],
+    allowed_source: tuple[TrustedPluginSourceType, str],
+    local_selection: PluginSelection | None,
+    identity: PluginIdentity | None,
+    generation_order: tuple[str, ...],
+) -> PluginSelection:
+    """只在已绑定仓库范围内选取在线候选，并与可用本地载荷协调。"""
+    if identity is None:
+        raise PluginSourceSelectionError("已绑定来源选择缺少插件身份")
+    source_type, source_key = allowed_source
+    allowed_online = tuple(
+        candidate
+        for candidate in online
+        if candidate.source_type is source_type and candidate.source_key == source_key
+    )
+    if not allowed_online:
+        if (
+            local_selection is not None
+            and local_selection.status is PluginSelectionStatus.SELECTED
+        ):
+            return local_selection
+        return PluginSelection(
+            status=PluginSelectionStatus.UNAVAILABLE,
+            reason="已绑定仓库中暂无可用插件包",
+        )
+    selected_online = _select_best(allowed_online, generation_order)
+    if selected_online is None:
+        if (
+            local_selection is not None
+            and local_selection.status is PluginSelectionStatus.SELECTED
+        ):
+            return local_selection
+        return PluginSelection(
+            status=PluginSelectionStatus.UNAVAILABLE,
+            reason="已绑定仓库没有适用于当前 MoviePilot 版本的插件包",
+        )
+    return _select_bound_or_local_candidate(
+        local_selection=local_selection,
+        online_candidate=selected_online,
+        identity=identity,
+        generation_order=generation_order,
+    )
+
+
+def _select_bound_or_local_candidate(
+    *,
+    local_selection: PluginSelection | None,
+    online_candidate: Candidate,
+    identity: PluginIdentity,
+    generation_order: tuple[str, ...],
+) -> PluginSelection:
+    """在已绑定在线候选和本地候选之间选择代际、版本更高的载荷。"""
+    if (
+        local_selection is None
+        or local_selection.status is not PluginSelectionStatus.SELECTED
+        or local_selection.candidate is None
+    ):
+        return PluginSelection(
+            status=PluginSelectionStatus.SELECTED,
+            candidate=online_candidate,
+            reason="已使用绑定仓库中的插件包",
+        )
+
+    local_candidate = local_selection.candidate
+    local_generation = generation_order.index(local_candidate.package_generation)
+    online_generation = generation_order.index(online_candidate.package_generation)
+    if local_generation < online_generation:
+        return local_selection
+    if online_generation < local_generation:
+        return PluginSelection(
+            status=PluginSelectionStatus.SELECTED,
+            candidate=online_candidate,
+            reason="绑定仓库提供了更高代际的插件包",
+        )
+
+    local_version = local_candidate.plugin_version or "0"
+    online_version = online_candidate.plugin_version or "0"
+    if compare_version(local_version, ">", online_version):
+        return local_selection
+    if compare_version(online_version, ">", local_version):
+        return PluginSelection(
+            status=PluginSelectionStatus.SELECTED,
+            candidate=online_candidate,
+            reason="绑定仓库提供了更高版本的插件包",
+        )
+    if identity.payload_source_type is PluginPayloadSourceType.LOCAL:
+        return local_selection
+    return PluginSelection(
+        status=PluginSelectionStatus.SELECTED,
+        candidate=online_candidate,
+        reason="当前继续使用绑定仓库中的插件包",
+    )
+
+
 def list_effective_online_candidates(
     inventory: CandidateInventory,
     *,

app/runtime/extensions/plugin/monitor.py

@@ -207,7 +207,7 @@ class PluginChangeMonitor:
                 else None
             )
             if runtime_plugin_id:
-                last_sync_time = self._recent_sync.get(runtime_plugin_id)
+                last_sync_time = self._recent_sync.get(runtime_plugin_id.lower())
                 if last_sync_time and time.time() - last_sync_time < 2:
                     continue
                 plugins_to_reload.add(runtime_plugin_id)

app/runtime/extensions/plugin/sync.py

@@ -44,10 +44,17 @@ class PluginSyncService:
         if self._frozen():
             return []
 
-        installed = self._installed_plugins()
+        installed = {
+            plugin_id.lower()
+            for plugin_id in self._installed_plugins()
+        }
         online = self._online_plugins()
         local = self._local_plugins()
-        local_plugin_ids = {plugin.id.lower() for plugin in local}
+        local_plugin_ids = {
+            plugin.id.lower()
+            for plugin in local
+        }
+        deferred_plugin_ids = installed & local_plugin_ids
         restore_plugin_ids = {
             plugin_id.lower()
             for plugin_id in (online_restore_plugins or set())
@@ -56,9 +63,10 @@ class PluginSyncService:
         targets = [
             plugin
             for plugin in candidates
-            if plugin.id in installed
+            if plugin.id.lower() in installed
             and (
-                plugin.id.lower() in restore_plugin_ids
+                plugin.id.lower() in deferred_plugin_ids
+                or plugin.id.lower() in restore_plugin_ids
                 or (
                     plugin.system_version_compatible is not False
                     and not self._plugin_exists(plugin.id, plugin.plugin_version)
@@ -71,6 +79,7 @@ class PluginSyncService:
         self._logger.info("开始安装第三方插件...")
         synced: list[str] = []
         failed: list[str] = []
+        failed_deferred: list[str] = []
 
         def install_one(plugin: Any) -> None:
             """安装一个插件并记录结果。"""
@@ -84,13 +93,14 @@ class PluginSyncService:
             elapsed = time.time() - started
             if state:
                 self._logger.info(
-                    f"插件 {plugin.plugin_name} 安装成功，版本：{plugin.plugin_version}，"
+                    f"插件 {plugin.plugin_name} 同步成功，耗时：{elapsed:.2f} 秒"
-                    f"耗时：{elapsed:.2f} 秒"
                 )
                 synced.append(plugin.id)
             else:
+                if plugin.id.lower() in deferred_plugin_ids:
+                    failed_deferred.append(plugin.id)
                 self._logger.error(
-                    f"插件 {plugin.plugin_name} v{plugin.plugin_version} 安装失败："
+                    f"插件 {plugin.plugin_name} 同步失败："
                     f"{message}，耗时：{elapsed:.2f} 秒"
                 )
                 failed.append(plugin.id)
@@ -102,6 +112,8 @@ class PluginSyncService:
                 try:
                     future.result()
                 except Exception as error:  # noqa: BLE001
+                    if plugin.id.lower() in deferred_plugin_ids:
+                        failed_deferred.append(plugin.id)
                     self._logger.error(
                         f"插件 {plugin.plugin_name} 安装过程中出现异常: {error}"
                     )
@@ -109,6 +121,11 @@ class PluginSyncService:
         self._logger.info(
             f"第三方插件安装完成，成功：{len(synced)} 个，失败：{len(failed)} 个"
         )
+        if failed_deferred:
+            raise RuntimeError(
+                "延后激活的插件同步未完成："
+                f"{', '.join(sorted(set(failed_deferred)))}"
+            )
         return synced
 
 
@@ -133,7 +150,12 @@ class LocalPluginSyncService:
 
     def sync(self, plugin_id: str, candidate: Optional[dict] = None) -> bool:
         """同步已安装且兼容的本地插件，成功后记录短时事件抑制标记。"""
-        if plugin_id not in self._installed_plugins():
+        normalized_plugin_id = plugin_id.lower()
+        installed = {
+            installed_id.lower()
+            for installed_id in self._installed_plugins()
+        }
+        if normalized_plugin_id not in installed:
             self._logger.info(f"本地插件 {plugin_id} 尚未安装，跳过自动同步和热重载")
             return False
         candidate = candidate or self._candidate(plugin_id)
@@ -160,7 +182,7 @@ class LocalPluginSyncService:
             if not state:
                 self._logger.error(f"同步本地插件 {plugin_id} 失败：{message}")
                 return False
-            self._recent_sync[plugin_id] = time.time()
+            self._recent_sync[normalized_plugin_id] = time.time()
             self._logger.info(f"已同步本地插件 {plugin_id}")
             return True
         except Exception as error:

app/startup/initializers/plugins.py

@@ -453,6 +453,8 @@ async def _sync_plugins_admitted(
         ),
         "插件同步到本地",
     )
+    if sync_result is None:
+        return False
     dependency_result = await (
         plugin_manager.async_install_plugin_missing_dependencies_with_status()
     )
@@ -471,7 +473,7 @@ async def _sync_plugins_admitted(
         lambda: _activate_ready_plugins(
             plugin_manager,
             classification.ready,
-            sync_result or [],
+            sync_result,
             previous_statuses,
         ),
         "插件运行态激活",
@@ -502,14 +504,18 @@ def _activate_ready_plugins(
 ) -> list[str]:
     """在线程池中完成插件导入和初始化，避免阻塞 Web 事件循环。"""
     running_ids = set(plugin_manager.running_plugins)
-    synced = set(synced_ids)
+    synced = {
+        _plugin_source_id(plugin_manager, plugin_id)
+        for plugin_id in synced_ids
+    }
     changed_ids: list[str] = []
     for plugin_id in ready_ids:
+        source_id = _plugin_source_id(plugin_manager, plugin_id)
         dependency_recovered = (
             previous_statuses.get(plugin_id)
             is PluginRuntimeStatus.DEPENDENCY_PENDING
         )
-        if plugin_id in running_ids and (plugin_id in synced or dependency_recovered):
+        if plugin_id in running_ids and (source_id in synced or dependency_recovered):
             plugin_manager.reload_plugin(plugin_id)
             changed_ids.append(plugin_id)
             continue
@@ -519,6 +525,35 @@ def _activate_ready_plugins(
     return changed_ids
 
 
+def _plugin_source_id(plugin_manager: PluginManager, plugin_id: str) -> str:
+    """把物理插件和虚拟实例归一到同一个源码身份。"""
+    source_id = plugin_manager.get_plugin_source_id(plugin_id)
+    try:
+        return normalize_physical_plugin_id(source_id)
+    except ValueError:
+        return source_id.lower()
+
+
+def _local_plugin_sources(plugin_manager: PluginManager) -> set[str]:
+    """返回安装清单中存在本地仓候选的物理插件身份。"""
+    installed = {
+        normalize_physical_plugin_id(plugin_id)
+        for plugin_id in (
+            get_configured_system_config().get(SystemConfigKey.UserInstalledPlugins)
+            or []
+        )
+    }
+    candidates: set[str] = set()
+    for plugin in plugin_manager.get_local_repo_plugins():
+        try:
+            source_id = normalize_physical_plugin_id(plugin.id)
+        except ValueError:
+            continue
+        if source_id in installed:
+            candidates.add(source_id)
+    return candidates
+
+
 async def quiesce_plugins(timeout: float = 240.0) -> bool:
     """封口插件变更并停用 handler，保留超时 Future 的运行所有权。"""
     plugin_manager = PluginManager.get_existing_instance()
@@ -579,13 +614,20 @@ def init_plugins():
     classification = plugin_manager.classify_plugins()
     plugin_manager.apply_plugin_dependency_classification(classification)
     plugin_manager.set_plugin_settling(True)
-    for plugin_id in classification.ready:
+    deferred_sources = _local_plugin_sources(plugin_manager)
+    immediate_ready = [
+        plugin_id
+        for plugin_id in classification.ready
+        if _plugin_source_id(plugin_manager, plugin_id) not in deferred_sources
+    ]
+    for plugin_id in immediate_ready:
         plugin_manager.start(plugin_id)
     register_plugin_api()
     plugin_manager.start_monitor(reopen=True)
     logger.info(
-        "插件启动分类：立即加载=%s，等待依赖=%s，等待源码=%s",
+        "插件启动分类：立即加载=%s，等待本地同步=%s，等待依赖=%s，等待源码=%s",
-        len(classification.ready),
+        len(immediate_ready),
+        len(classification.ready) - len(immediate_ready),
         len(classification.missing_dependencies),
         len(classification.missing_source),
     )

docs/architecture-optimization-checklist.md

@@ -79,7 +79,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 | 长方法 | 281 个超过 80 行 | 67 个超过 150 行，23 个超过 250 行；大量是私有方法 |
 | 全量 mypy 历史债务 | 11,983 / 601 文件 | strict frontier 当前只覆盖 41 个文件，且 ratchet 已新增 2 个错误 |
 | Ruff 历史诊断 | 934 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
-| 覆盖率低水位 | Application 78.22%，Domain 79.29% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
+| 覆盖率低水位 | Application 78.24%，Domain 79.29% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 
 ### 3.3 热点文件
 

tests/fixtures/architecture/coverage-baseline.json

@@ -1,8 +1,8 @@
 {
   "application": {
-    "covered_lines": 9620,
+    "covered_lines": 9649,
-    "percent": 78.22,
+    "percent": 78.24,
-    "statements": 12298
+    "statements": 12333
   },
   "domain": {
     "covered_lines": 3392,

tests/test_plugin_catalog_service.py

@@ -63,6 +63,36 @@ def test_merge_prefers_newer_version_and_remote_source():
     assert result == [new_remote]
 
 
+def test_merge_treats_plugin_id_casing_as_one_physical_plugin():
+    """同一物理插件的市场与本地 ID 大小写差异不能产生两个目录条目。"""
+    service = _service()
+    online = _plugin("DownloadCenter", "3.2.1", "https://market-a")
+    local = _plugin("downloadcenter", "3.3.2", "local://DownloadCenter")
+
+    result = service.merge(
+        [online, local],
+        [],
+        ["https://market-a"],
+    )
+
+    assert result == [local]
+
+
+def test_merge_by_source_treats_plugin_id_casing_as_one_physical_plugin():
+    """同一仓库跨代际大小写不一致时只保留最高版本候选。"""
+    service = _service()
+    old = _plugin("DownloadCenter", "3.2.1", "https://market-a")
+    new = _plugin("downloadcenter", "3.3.2", "https://market-a")
+
+    result = service.merge_by_source(
+        [new],
+        [old],
+        ["https://market-a"],
+    )
+
+    assert result == [new]
+
+
 def test_load_maps_market_entries_with_installed_snapshot():
     """单市场读取只获取一次已安装快照并按索引顺序映射 DTO。"""
     mapper = Mock(side_effect=lambda plugin_id, *_args: plugin_id)
@@ -94,7 +124,8 @@ async def test_async_collect_isolates_failure_and_completes_progress():
         if market == "https://market-a" and package_version == "v3":
             raise RuntimeError("unavailable")
         version = "2.0.0" if package_version else "1.0.0"
-        return [_plugin(market, version, market)]
+        plugin_id = "MarketA" if market.endswith("market-a") else "MarketB"
+        return [_plugin(plugin_id, version, market)]
 
     result = await service.async_collect(
         markets=["https://market-a", "https://market-b"],
@@ -105,8 +136,8 @@ async def test_async_collect_isolates_failure_and_completes_progress():
     )
 
     assert {plugin.id for plugin in result} == {
-        "https://market-a",
+        "MarketA",
-        "https://market-b",
+        "MarketB",
     }
     error.assert_called_once()
     assert progress.call_args_list[0].kwargs["value"] == 0

tests/test_plugin_external_install_boundary.py

@@ -214,10 +214,10 @@ async def test_external_async_helper_preserves_failure_tuple_on_gateway_error(
 
 
 @pytest.mark.asyncio
-async def test_http_install_does_not_treat_repo_url_as_explicit_source(
+async def test_http_install_forwards_repo_url_without_granting_source_authority(
     monkeypatch,
 ) -> None:
-    """旧 GET 安装入口不能把兼容参数误当成管理员明确选源。"""
+    """普通更新保留绑定仓库提示，但不能把它升级为选源授权。"""
     gateway = Mock()
     gateway.install = AsyncMock(
         return_value=SimpleNamespace(success=True, message="")
@@ -239,7 +239,7 @@ async def test_http_install_does_not_treat_repo_url_as_explicit_source(
     assert result.success is True
     gateway.install.assert_awaited_once_with(
         plugin_id="DemoPlugin",
-        repo_url=None,
+        repo_url=REPO_URL,
         release_version="1.2.3",
         force=False,
         explicit_source=False,

tests/test_plugin_install_gateway.py

@@ -76,6 +76,105 @@ async def test_gateway_freezes_admission_before_executing_transaction() -> None:
     assert admission.expected_revision is None
 
 
+@pytest.mark.asyncio
+async def test_local_only_requires_explicit_online_binding() -> None:
+    """本地专属身份即使发现唯一在线来源，也只能由管理员显式绑定。"""
+    online = PluginMarketCandidate(
+        plugin_id="DemoPlugin",
+        source_key="github:jxxghp/moviepilot-plugins",
+        source_type=TrustedPluginSourceType.OFFICIAL,
+        repo_url=REPO_URL,
+        package_generation="v3",
+        plugin_version="9.0.0",
+        dto={"v3": True},
+    )
+    local = PluginLocalCandidate(
+        plugin_id="DemoPlugin",
+        repo_url="local://DemoPlugin?version=v3",
+        package_generation="v3",
+        plugin_version="1.0.0",
+        dto={"v3": True},
+    )
+    identity = PluginIdentity(
+        plugin_id="DemoPlugin",
+        normalized_plugin_id="demoplugin",
+        trusted_source_type=TrustedPluginSourceType.UNKNOWN,
+        trusted_source_key=None,
+        binding_basis=PluginBindingBasis.LOCAL_ONLY,
+        payload_source_type=PluginPayloadSourceType.LOCAL,
+        payload_source_key=None,
+        declared_version="1.0.0",
+        package_generation="v3",
+        declared_metadata=PluginDeclaredMetadata.from_package(
+            {"name": "Demo local", "v3": True},
+            declaration_version="1.0.0",
+            manifest_matches_payload=True,
+        ),
+        payload_receipt="sha256:" + "1" * 64,
+        revision=2,
+        created_at=NOW,
+        updated_at=NOW,
+        bound_at=None,
+        payload_applied_at=NOW,
+    )
+    executor = AsyncMock()
+    executor.execute.return_value = type(
+        "Result",
+        (),
+        {"success": True, "message": ""},
+    )()
+    gateway = PluginInstallGateway(
+        inventory=AsyncMock(
+            return_value=CandidateInventory(
+                (MarketRead.present(REPO_URL, (online,)),),
+                (local,),
+                local_read=LocalCandidateRead.present((local,)),
+            )
+        ),
+        identity=AsyncMock(return_value=identity),
+        candidate_compatibility=lambda _candidate: (True, ""),
+        executor=executor,
+        clock=lambda: NOW,
+    )
+
+    automatic = await gateway.install(
+        plugin_id="DemoPlugin",
+        repo_url=None,
+        package_version="v3",
+    )
+
+    assert automatic.success is True
+    automatic_admission = executor.execute.await_args.kwargs["admission"]
+    assert automatic_admission.candidate is local
+    assert automatic_admission.binding_basis is PluginBindingBasis.LOCAL_ONLY
+    assert automatic_admission.trusted_source_key is None
+
+    hinted = await gateway.install(
+        plugin_id="DemoPlugin",
+        repo_url=REPO_URL,
+        package_version="v3",
+    )
+
+    assert hinted.success is True
+    hinted_admission = executor.execute.await_args.kwargs["admission"]
+    assert hinted_admission.candidate is local
+    assert hinted_admission.binding_basis is PluginBindingBasis.LOCAL_ONLY
+    assert hinted_admission.trusted_source_key is None
+
+    explicit = await gateway.install(
+        plugin_id="DemoPlugin",
+        repo_url=REPO_URL,
+        package_version="v3",
+        explicit_source=True,
+    )
+
+    assert explicit.success is True
+    explicit_admission = executor.execute.await_args.kwargs["admission"]
+    assert explicit_admission.candidate is online
+    assert explicit_admission.binding_basis is PluginBindingBasis.EXPLICIT_INSTALL
+    assert explicit_admission.trusted_source_key == online.source_key
+
+
 @pytest.mark.asyncio
 async def test_gateway_rejects_source_conflict_before_package_execution() -> None:
     """来源准入失败时不进入文件和数据库事务。"""

tests/test_plugin_monitor_lifecycle.py

@@ -79,6 +79,8 @@ def test_init_plugins_starts_monitor_after_runtime_and_routes(monkeypatch) -> No
     manager.reopen_plugins.side_effect = lambda: order.append("reopen") or True
     manager.start.side_effect = lambda plugin_id: order.append(f"plugin:{plugin_id}")
     manager.start_monitor.side_effect = lambda **_kwargs: order.append("monitor")
+    manager.get_local_repo_plugins.return_value = []
+    manager.get_plugin_source_id.side_effect = lambda plugin_id: plugin_id
     monkeypatch.setattr(
         plugins_initializer,
         "configure_plugin_services",
@@ -105,6 +107,37 @@ def test_init_plugins_starts_monitor_after_runtime_and_routes(monkeypatch) -> No
     manager.start_monitor.assert_called_once_with(reopen=True)
 
 
+def test_init_plugins_defers_installed_local_candidate_until_sync(monkeypatch) -> None:
+    """本地仓候选不得在源码协调前先启动运行目录中的旧载荷。"""
+    order: list[str] = []
+    manager = MagicMock()
+    manager.classify_plugins.return_value = PluginDependencyClassification(
+        ready=("DownloadCenter", "OnlineOnly"),
+        missing_dependencies=(),
+        missing_source=(),
+    )
+    manager.reopen_plugins.return_value = True
+    manager.get_local_repo_plugins.return_value = [
+        SimpleNamespace(id="downloadcenter", plugin_version="3.3.2")
+    ]
+    manager.get_plugin_source_id.side_effect = lambda plugin_id: plugin_id
+    manager.start.side_effect = lambda plugin_id: order.append(plugin_id)
+    config = MagicMock()
+    config.get.return_value = ["DownloadCenter", "OnlineOnly"]
+    monkeypatch.setattr(plugins_initializer, "configure_plugin_services", lambda: None)
+    monkeypatch.setattr(plugins_initializer, "PluginManager", lambda: manager)
+    monkeypatch.setattr(
+        plugins_initializer,
+        "get_configured_system_config",
+        lambda: config,
+    )
+    monkeypatch.setattr(plugins_initializer, "register_plugin_api", MagicMock())
+
+    plugins_initializer.init_plugins()
+
+    assert order == ["OnlineOnly"]
+
+
 def test_plugin_manager_projects_dependency_classification_to_runtime_status() -> None:
     """真实管理器按分类字段写入三类启动状态，避免测试替身掩盖字段漂移。"""
     _reset_plugin_manager()
@@ -196,9 +229,58 @@ def _patch_sync_plugins(monkeypatch, manager: MagicMock) -> MagicMock:
         return_value=dependency_result,
     )
     manager.get_plugin_runtime_statuses.return_value = {}
+    manager.get_local_repo_plugins.return_value = []
+    manager.get_plugin_source_id.side_effect = lambda plugin_id: plugin_id
     return register
 
 
+@pytest.mark.asyncio
+async def test_sync_plugins_installs_local_payload_before_first_activation(
+    monkeypatch,
+) -> None:
+    """本地高版本写入完成后才能首次激活对应插件。"""
+    order: list[str] = []
+    manager = MagicMock()
+    manager.sync.side_effect = lambda *_args, **_kwargs: (
+        order.append("install:downloadcenter") or ["downloadcenter"]
+    )
+    manager.async_install_plugin_missing_dependencies_with_status.return_value = (
+        PluginDependencyInstallResult(missing=[], success=True)
+    )
+    manager.classify_plugins.return_value = PluginDependencyClassification(
+        ready=("DownloadCenter",),
+        missing_dependencies=(),
+        missing_source=(),
+    )
+    manager.running_plugins = {}
+    manager.start.side_effect = lambda plugin_id: order.append(f"start:{plugin_id}")
+    register = _patch_sync_plugins(monkeypatch, manager)
+
+    assert await plugins_initializer.sync_plugins() is True
+
+    assert order == ["install:downloadcenter", "start:DownloadCenter"]
+    register.assert_called_once_with("DownloadCenter")
+
+
+@pytest.mark.asyncio
+async def test_sync_plugins_does_not_activate_after_package_sync_failure(
+    monkeypatch,
+) -> None:
+    """包同步失败后不得继续激活启动阶段主动延后的旧载荷。"""
+    manager = MagicMock()
+    _patch_sync_plugins(monkeypatch, manager)
+    monkeypatch.setattr(
+        plugins_initializer,
+        "execute_task",
+        AsyncMock(return_value=None),
+    )
+
+    assert await plugins_initializer.sync_plugins() is False
+
+    manager.async_install_plugin_missing_dependencies_with_status.assert_not_awaited()
+    manager.start.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_sync_plugins_rejects_before_configuring_mutable_services(
     monkeypatch,

tests/test_plugin_source_policy.py

@@ -1,10 +1,12 @@
 """插件候选事实与来源选择策略测试。"""
 
+from app.application.plugin.declaration import PluginDeclaredMetadata
 from app.application.plugin.identity import (
     PluginBindingBasis,
     PluginIdentity,
     PluginPayloadSourceType,
     TrustedPluginSourceType,
+    normalize_physical_plugin_id,
 )
 from app.application.plugin.source import (
     CandidateInventory,
@@ -21,6 +23,11 @@ THIRD_PARTY_SOURCE = "github:example/moviepilot-plugins"
 OTHER_SOURCE = "github:other/moviepilot-plugins"
 
 
+def test_plugin_identity_normalization_keeps_existing_underscore_ids() -> None:
+    """来源身份必须兼容市场中已经存在的下划线插件 ID。"""
+    assert normalize_physical_plugin_id("Nullbr_Search") == "nullbr_search"
+
+
 def _online(
     source_key: str,
     *,
@@ -47,7 +54,13 @@ def _inventory(*reads: MarketRead, local=()) -> CandidateInventory:
     return CandidateInventory(tuple(reads), tuple(local))
 
 
-def _identity(source_type: TrustedPluginSourceType, source_key: str) -> PluginIdentity:
+def _identity(
+    source_type: TrustedPluginSourceType,
+    source_key: str,
+    *,
+    payload_source_type: PluginPayloadSourceType = PluginPayloadSourceType.UNKNOWN,
+    payload_version: str = "1.0.0",
+) -> PluginIdentity:
     """构造已绑定在线来源身份。"""
     from datetime import datetime, timezone
 
@@ -60,17 +73,77 @@ def _identity(source_type: TrustedPluginSourceType, source_key: str) -> PluginId
         binding_basis=PluginBindingBasis.OFFICIAL_DEFAULT
         if source_type is TrustedPluginSourceType.OFFICIAL
         else PluginBindingBasis.TOFU,
-        payload_source_type=PluginPayloadSourceType.UNKNOWN,
+        payload_source_type=payload_source_type,
-        payload_source_key=None,
+        payload_source_key=(
-        declared_version=None,
+            source_key
-        package_generation=None,
+            if payload_source_type in {
-        declared_metadata=None,
+                PluginPayloadSourceType.OFFICIAL,
-        payload_receipt=None,
+                PluginPayloadSourceType.THIRD_PARTY,
+            }
+            else None
+        ),
+        declared_version=(
+            payload_version
+            if payload_source_type is not PluginPayloadSourceType.UNKNOWN
+            else None
+        ),
+        package_generation=(
+            "v3"
+            if payload_source_type is not PluginPayloadSourceType.UNKNOWN
+            else None
+        ),
+        declared_metadata=(
+            PluginDeclaredMetadata.from_package(
+                {"name": "Demo", "v3": True},
+                declaration_version=payload_version,
+                manifest_matches_payload=True,
+            )
+            if payload_source_type is not PluginPayloadSourceType.UNKNOWN
+            else None
+        ),
+        payload_receipt=(
+            "sha256:" + "1" * 64
+            if payload_source_type is not PluginPayloadSourceType.UNKNOWN
+            else None
+        ),
         revision=1,
         created_at=now,
         updated_at=now,
         bound_at=now,
-        payload_applied_at=None,
+        payload_applied_at=(
+            now
+            if payload_source_type is not PluginPayloadSourceType.UNKNOWN
+            else None
+        ),
+    )
+
+
+def _local_only_identity(version: str = "3.0.0") -> PluginIdentity:
+    """构造只能由用户显式绑定在线仓库的本地身份。"""
+    from datetime import datetime, timezone
+
+    now = datetime(2026, 8, 25, tzinfo=timezone.utc)
+    return PluginIdentity(
+        plugin_id="DemoPlugin",
+        normalized_plugin_id="demoplugin",
+        trusted_source_type=TrustedPluginSourceType.UNKNOWN,
+        trusted_source_key=None,
+        binding_basis=PluginBindingBasis.LOCAL_ONLY,
+        payload_source_type=PluginPayloadSourceType.LOCAL,
+        payload_source_key=None,
+        declared_version=version,
+        package_generation="v3",
+        declared_metadata=PluginDeclaredMetadata.from_package(
+            {"name": "Demo local", "v3": True},
+            declaration_version=version,
+            manifest_matches_payload=True,
+        ),
+        payload_receipt="sha256:" + "2" * 64,
+        revision=1,
+        created_at=now,
+        updated_at=now,
+        bound_at=None,
+        payload_applied_at=now,
     )
 
 
@@ -232,6 +305,128 @@ def test_non_explicit_source_hint_cannot_bypass_local_state() -> None:
     assert failed_local_read.status is PluginSelectionStatus.INCOMPLETE
 
 
+def test_bound_online_and_local_candidates_choose_higher_version() -> None:
+    """已绑定在线来源与本地候选并存时，插件版本决定实际载荷。"""
+    local = PluginLocalCandidate(
+        plugin_id="DemoPlugin",
+        repo_url="local://DemoPlugin?version=v3",
+        package_generation="v3",
+        plugin_version="3.0.0",
+    )
+    identity = _identity(
+        TrustedPluginSourceType.THIRD_PARTY,
+        THIRD_PARTY_SOURCE,
+    )
+
+    online_higher = select_plugin_candidate(
+        _inventory(
+            MarketRead.present(
+                "market-a",
+                (_online(THIRD_PARTY_SOURCE, version="3.2.0"),),
+            ),
+            local=(local,),
+        ),
+        plugin_id="DemoPlugin",
+        generations=("v3", "v2", "v1"),
+        identity=identity,
+    )
+    local_higher = select_plugin_candidate(
+        _inventory(
+            MarketRead.present(
+                "market-a",
+                (_online(THIRD_PARTY_SOURCE, version="2.9.0"),),
+            ),
+            local=(local,),
+        ),
+        plugin_id="DemoPlugin",
+        generations=("v3", "v2", "v1"),
+        identity=identity,
+    )
+
+    assert isinstance(online_higher.candidate, PluginMarketCandidate)
+    assert online_higher.candidate.plugin_version == "3.2.0"
+    assert local_higher.candidate is local
+
+
+def test_equal_bound_and_local_versions_keep_current_payload_source() -> None:
+    """相同版本保持当前载荷来源，避免每次启动在本地与在线之间切换。"""
+    local = PluginLocalCandidate(
+        plugin_id="DemoPlugin",
+        repo_url="local://DemoPlugin?version=v3",
+        package_generation="v3",
+        plugin_version="3.0.0",
+    )
+    inventory = _inventory(
+        MarketRead.present(
+            "market-a",
+            (_online(THIRD_PARTY_SOURCE, version="3.0.0"),),
+        ),
+        local=(local,),
+    )
+
+    local_current = select_plugin_candidate(
+        inventory,
+        plugin_id="DemoPlugin",
+        generations=("v3", "v2", "v1"),
+        identity=_identity(
+            TrustedPluginSourceType.THIRD_PARTY,
+            THIRD_PARTY_SOURCE,
+            payload_source_type=PluginPayloadSourceType.LOCAL,
+            payload_version="3.0.0",
+        ),
+    )
+    online_current = select_plugin_candidate(
+        inventory,
+        plugin_id="DemoPlugin",
+        generations=("v3", "v2", "v1"),
+        identity=_identity(
+            TrustedPluginSourceType.THIRD_PARTY,
+            THIRD_PARTY_SOURCE,
+            payload_source_type=PluginPayloadSourceType.THIRD_PARTY,
+            payload_version="3.0.0",
+        ),
+    )
+
+    assert local_current.candidate is local
+    assert isinstance(online_current.candidate, PluginMarketCandidate)
+
+
+def test_local_only_identity_never_uses_online_candidate_implicitly() -> None:
+    """本地专属身份只能由用户显式确认后建立在线绑定。"""
+    local = PluginLocalCandidate(
+        plugin_id="DemoPlugin",
+        repo_url="local://DemoPlugin?version=v3",
+        package_generation="v3",
+        plugin_version="3.0.0",
+    )
+    inventory = _inventory(
+        MarketRead.present(
+            "market-a",
+            (_online(THIRD_PARTY_SOURCE, version="9.0.0"),),
+        ),
+        local=(local,),
+    )
+
+    automatic = select_plugin_candidate(
+        inventory,
+        plugin_id="DemoPlugin",
+        generations=("v3", "v2", "v1"),
+        identity=_local_only_identity(),
+    )
+    explicit = select_plugin_candidate(
+        inventory,
+        plugin_id="DemoPlugin",
+        generations=("v3", "v2", "v1"),
+        identity=_local_only_identity(),
+        requested_source_key=THIRD_PARTY_SOURCE,
+        explicit_source=True,
+    )
+
+    assert automatic.candidate is local
+    assert isinstance(explicit.candidate, PluginMarketCandidate)
+    assert explicit.candidate.plugin_version == "9.0.0"
+
+
 def test_uninstalled_unique_and_multiple_sources_are_distinct() -> None:
     """未安装插件允许完整快照中的唯一来源，多来源必须返回冲突。"""
     unique = select_plugin_candidate(

tests/test_plugin_sync_service.py

@@ -6,7 +6,9 @@ from types import SimpleNamespace
 from unittest.mock import AsyncMock, Mock
 
 import pytest
+from packaging.version import Version
 
+from app.application.plugin.catalog import PluginCatalogService
 from app.application.plugin.declaration import PluginDeclaredMetadata
 from app.application.plugin.gateway import PluginInstallGateway
 from app.application.plugin.identity import (
@@ -19,16 +21,35 @@ from app.application.plugin.install import PluginInstallResult
 from app.application.plugin.lifecycle import plugin_lifecycle
 from app.application.plugin.source import (
     CandidateInventory,
+    LocalCandidateRead,
     MarketRead,
+    PluginLocalCandidate,
     PluginMarketCandidate,
 )
 from app.runtime.config import global_vars
-from app.runtime.extensions.plugin.sync import PluginSyncService
+from app.runtime.extensions.plugin.sync import LocalPluginSyncService, PluginSyncService
 from app.startup.initializers import plugins as plugins_initializer
 
 REPO_URL = "https://github.com/jxxghp/MoviePilot-Plugins"
 
 
+def _merge_catalog_plugins(higher, base, markets):
+    """通过生产目录服务合并启动同步候选。"""
+    service = PluginCatalogService(
+        market_loader=Mock(return_value={}),
+        async_market_loader=AsyncMock(return_value={}),
+        installed_plugins_provider=Mock(return_value=[]),
+        plugin_mapper=Mock(),
+        is_local_repo=lambda value: str(value).startswith("local://"),
+        version_compare=lambda left, operator, right: (
+            operator == ">" and Version(left) > Version(right)
+        ),
+        warning=Mock(),
+        error=Mock(),
+    )
+    return service.merge(higher, base, markets)
+
+
 def test_market_sync_keeps_install_rollback_enabled() -> None:
     """自动更新插件时保留旧版本，失败后可由安装器恢复。"""
     plugin = SimpleNamespace(
@@ -81,8 +102,8 @@ def test_market_sync_restores_trusted_online_payload_after_local_source_removed(
     install.assert_called_once_with(plugin.id, None, False, None)
 
 
-def test_market_sync_keeps_active_local_payload_when_candidate_still_exists() -> None:
+def test_market_sync_reconciles_existing_local_candidate_through_gateway() -> None:
-    """本地候选仍存在时，不应被启动在线恢复覆盖。"""
+    """本地候选对应插件已延后激活，必须经 Gateway 协调后再启动。"""
     online = SimpleNamespace(
         id="DemoPlugin",
         repo_url=REPO_URL,
@@ -109,8 +130,357 @@ def test_market_sync_keeps_active_local_payload_when_candidate_still_exists() ->
         log=Mock(),
     )
 
-    assert service.sync(online_restore_plugins={"demoplugin"}) == []
+    assert service.sync(online_restore_plugins={"demoplugin"}) == [online.id]
-    install.assert_not_called()
+    install.assert_called_once_with(online.id, None, False, None)
+
+
+def test_market_sync_defers_source_selection_to_gateway() -> None:
+    """启动目录只能决定同步目标，不能把本地候选升级为选源授权。"""
+    local = SimpleNamespace(
+        id="DemoPlugin",
+        repo_url="local://DemoPlugin?path=/private/plugins&version=v3",
+        plugin_name="Demo Local",
+        plugin_version="3.3.2",
+        system_version_compatible=True,
+    )
+    install = Mock(return_value=(True, ""))
+    service = PluginSyncService(
+        frozen=lambda: False,
+        installed_plugins=lambda: [local.id],
+        online_plugins=lambda: [],
+        local_plugins=lambda: [local],
+        merge_plugins=lambda items, *_args: items,
+        plugin_exists=lambda *_args: False,
+        install=install,
+        log=Mock(),
+    )
+
+    assert service.sync() == [local.id]
+    install.assert_called_once_with(local.id, None, False, None)
+
+
+def test_market_sync_reports_local_install_failure() -> None:
+    """本地载荷安装失败必须阻止启动编排继续激活旧代码。"""
+    local = SimpleNamespace(
+        id="DemoPlugin",
+        repo_url="local://DemoPlugin?path=/private/plugins&version=v3",
+        plugin_name="Demo Local",
+        plugin_version="3.3.2",
+        system_version_compatible=True,
+    )
+    service = PluginSyncService(
+        frozen=lambda: False,
+        installed_plugins=lambda: [local.id],
+        online_plugins=lambda: [],
+        local_plugins=lambda: [local],
+        merge_plugins=lambda items, *_args: items,
+        plugin_exists=lambda *_args: False,
+        install=Mock(return_value=(False, "copy failed")),
+        log=Mock(),
+    )
+
+    with pytest.raises(
+        RuntimeError,
+        match="延后激活的插件同步未完成：DemoPlugin",
+    ):
+        service.sync()
+
+
+def test_local_sync_matches_installed_plugin_id_case_insensitively() -> None:
+    """本地热同步应把大小写不同的索引 ID 识别为同一已安装插件。"""
+    candidate = {
+        "repo_url": "local://downloadcenter?package_version=v3",
+        "package_version": "v3",
+        "compatible": True,
+    }
+    system = Mock()
+    system.install_plugin.return_value = (True, "")
+    recent_sync: dict[str, float] = {}
+    service = LocalPluginSyncService(
+        installed_plugins=lambda: ["DownloadCenter"],
+        candidate=Mock(return_value=candidate),
+        system=lambda: system,
+        recent_sync=recent_sync,
+        log=Mock(),
+    )
+
+    assert service.sync("DownloadCenter", candidate)
+    system.install_plugin.assert_called_once_with(
+        plugin_id="DownloadCenter",
+        repo_url=candidate["repo_url"],
+        package_version="v3",
+        force=True,
+        local_sync=True,
+        explicit_source=True,
+    )
+    assert "downloadcenter" in recent_sync
+
+
+@pytest.mark.asyncio
+async def test_market_sync_preserves_generation_priority_through_gateway(
+    monkeypatch,
+) -> None:
+    """目录中的高版本 V2 不能绕过 Gateway 覆盖绑定仓库的 V3。"""
+    online = PluginMarketCandidate(
+        plugin_id="DownloadCenter",
+        source_key="github:jxxghp/moviepilot-plugins",
+        source_type=TrustedPluginSourceType.OFFICIAL,
+        repo_url=REPO_URL,
+        package_generation="v3",
+        plugin_version="1.0.0",
+        dto={"v3": True},
+    )
+    local = PluginLocalCandidate(
+        plugin_id="downloadcenter",
+        repo_url="local://downloadcenter?path=/private/plugins&version=v2",
+        package_generation="v2",
+        plugin_version="9.0.0",
+        dto={"v3": True},
+    )
+    identity = PluginIdentity(
+        plugin_id="DownloadCenter",
+        normalized_plugin_id="downloadcenter",
+        trusted_source_type=TrustedPluginSourceType.OFFICIAL,
+        trusted_source_key=online.source_key,
+        binding_basis=PluginBindingBasis.OFFICIAL_DEFAULT,
+        payload_source_type=PluginPayloadSourceType.OFFICIAL,
+        payload_source_key=online.source_key,
+        declared_version=online.plugin_version,
+        package_generation="v3",
+        declared_metadata=PluginDeclaredMetadata.from_package(
+            {"name": "Demo", "v3": True},
+            declaration_version=online.plugin_version,
+            manifest_matches_payload=True,
+        ),
+        payload_receipt="sha256:" + "2" * 64,
+        revision=2,
+        created_at=datetime(2026, 8, 25, 12, 0, tzinfo=timezone.utc),
+        updated_at=datetime(2026, 8, 25, 12, 0, tzinfo=timezone.utc),
+        bound_at=datetime(2026, 8, 25, 12, 0, tzinfo=timezone.utc),
+        payload_applied_at=datetime(2026, 8, 25, 12, 0, tzinfo=timezone.utc),
+    )
+    inventory = CandidateInventory(
+        (MarketRead.present(REPO_URL, (online,)),),
+        (local,),
+        local_read=LocalCandidateRead.present((local,)),
+    )
+    executor = AsyncMock()
+    executor.execute.return_value = PluginInstallResult(success=True)
+    gateway = PluginInstallGateway(
+        inventory=AsyncMock(return_value=inventory),
+        identity=AsyncMock(return_value=identity),
+        candidate_compatibility=lambda _candidate: (True, ""),
+        executor=executor,
+        clock=lambda: datetime(2026, 8, 25, 12, 0, tzinfo=timezone.utc),
+    )
+    monkeypatch.setattr(
+        global_vars,
+        "CURRENT_EVENT_LOOP",
+        asyncio.get_running_loop(),
+    )
+
+    def install(
+        plugin_id: str,
+        repo_url: str | None,
+        force: bool,
+        startup_token: object | None,
+    ) -> tuple[bool, str]:
+        """按生产兼容入口让 Gateway 完成最终来源准入。"""
+        return plugins_initializer._run_plugin_install_sync(
+            gateway,
+            plugin_id=plugin_id,
+            repo_url=repo_url or "",
+            package_version="v3",
+            release_version=None,
+            force=force,
+            local_sync=True,
+            explicit_source=False,
+            startup_token=startup_token,
+        )
+
+    merged_local = SimpleNamespace(
+        id=local.plugin_id,
+        repo_url=local.repo_url,
+        plugin_name="Demo Local",
+        plugin_version=local.plugin_version,
+        system_version_compatible=True,
+    )
+    merged_online = SimpleNamespace(
+        id=online.plugin_id,
+        repo_url=online.repo_url,
+        plugin_name="Download Center",
+        plugin_version=online.plugin_version,
+        system_version_compatible=True,
+    )
+    service = PluginSyncService(
+        frozen=lambda: False,
+        installed_plugins=lambda: [online.plugin_id],
+        online_plugins=lambda: [merged_online],
+        local_plugins=lambda: [merged_local],
+        merge_plugins=_merge_catalog_plugins,
+        plugin_exists=lambda *_args: False,
+        install=install,
+        log=Mock(),
+    )
+
+    async with plugin_lifecycle.hold_startup() as startup_token:
+        synced = await asyncio.wait_for(
+            asyncio.to_thread(service.sync, startup_token),
+            timeout=2,
+        )
+
+    assert synced == [local.plugin_id]
+    executor.execute.assert_awaited_once()
+    assert executor.execute.await_args.kwargs["local_sync"] is False
+    admission = executor.execute.await_args.kwargs["admission"]
+    assert admission.candidate is online
+    assert admission.trusted_source_key == online.source_key
+
+
+@pytest.mark.asyncio
+async def test_market_sync_blocks_activation_when_gateway_selected_local_fails(
+    monkeypatch,
+) -> None:
+    """目录预选在线候选时，Gateway 改选本地失败仍必须阻止旧载荷激活。"""
+    competing_repo_url = "https://github.com/example/MoviePilot-Plugins"
+    official = PluginMarketCandidate(
+        plugin_id="DemoPlugin",
+        source_key="github:jxxghp/moviepilot-plugins",
+        source_type=TrustedPluginSourceType.OFFICIAL,
+        repo_url=REPO_URL,
+        package_generation="v3",
+        plugin_version="1.0.0",
+        dto={"v3": True},
+    )
+    competing = PluginMarketCandidate(
+        plugin_id=official.plugin_id,
+        source_key="github:example/moviepilot-plugins",
+        source_type=TrustedPluginSourceType.THIRD_PARTY,
+        repo_url=competing_repo_url,
+        package_generation="v3",
+        plugin_version="9.0.0",
+        dto={"v3": True},
+    )
+    local = PluginLocalCandidate(
+        plugin_id=official.plugin_id,
+        repo_url="local://DemoPlugin?path=/private/plugins&version=v3",
+        package_generation="v3",
+        plugin_version="2.0.0",
+        dto={"v3": True},
+    )
+    identity = PluginIdentity(
+        plugin_id=official.plugin_id,
+        normalized_plugin_id="demoplugin",
+        trusted_source_type=TrustedPluginSourceType.OFFICIAL,
+        trusted_source_key=official.source_key,
+        binding_basis=PluginBindingBasis.OFFICIAL_DEFAULT,
+        payload_source_type=PluginPayloadSourceType.OFFICIAL,
+        payload_source_key=official.source_key,
+        declared_version=official.plugin_version,
+        package_generation="v3",
+        declared_metadata=PluginDeclaredMetadata.from_package(
+            {"name": "Demo", "v3": True},
+            declaration_version=official.plugin_version,
+            manifest_matches_payload=True,
+        ),
+        payload_receipt="sha256:" + "3" * 64,
+        revision=2,
+        created_at=datetime(2026, 8, 25, 12, 0, tzinfo=timezone.utc),
+        updated_at=datetime(2026, 8, 25, 12, 0, tzinfo=timezone.utc),
+        bound_at=datetime(2026, 8, 25, 12, 0, tzinfo=timezone.utc),
+        payload_applied_at=datetime(2026, 8, 25, 12, 0, tzinfo=timezone.utc),
+    )
+    inventory = CandidateInventory(
+        (
+            MarketRead.present(REPO_URL, (official,)),
+            MarketRead.present(competing_repo_url, (competing,)),
+        ),
+        (local,),
+        local_read=LocalCandidateRead.present((local,)),
+    )
+    executor = AsyncMock(
+        **{"execute.return_value": PluginInstallResult(
+            success=False,
+            message="copy failed",
+        )}
+    )
+    gateway = PluginInstallGateway(
+        inventory=AsyncMock(return_value=inventory),
+        identity=AsyncMock(return_value=identity),
+        candidate_compatibility=lambda _candidate: (True, ""),
+        executor=executor,
+        clock=lambda: datetime(2026, 8, 25, 12, 0, tzinfo=timezone.utc),
+    )
+    monkeypatch.setattr(
+        global_vars,
+        "CURRENT_EVENT_LOOP",
+        asyncio.get_running_loop(),
+    )
+
+    def install(
+        plugin_id: str,
+        repo_url: str | None,
+        force: bool,
+        startup_token: object | None,
+    ) -> tuple[bool, str]:
+        return plugins_initializer._run_plugin_install_sync(
+            gateway,
+            plugin_id=plugin_id,
+            repo_url=repo_url or "",
+            package_version="v3",
+            release_version=None,
+            force=force,
+            local_sync=False,
+            explicit_source=False,
+            startup_token=startup_token,
+        )
+
+    merged_official = SimpleNamespace(
+        id=official.plugin_id,
+        repo_url=official.repo_url,
+        plugin_name="Demo Official",
+        plugin_version=official.plugin_version,
+        system_version_compatible=True,
+    )
+    merged_competing = SimpleNamespace(
+        id=competing.plugin_id,
+        repo_url=competing.repo_url,
+        plugin_name="Demo Competing",
+        plugin_version=competing.plugin_version,
+        system_version_compatible=True,
+    )
+    merged_local = SimpleNamespace(
+        id=local.plugin_id,
+        repo_url=local.repo_url,
+        plugin_name="Demo Local",
+        plugin_version=local.plugin_version,
+        system_version_compatible=True,
+    )
+    service = PluginSyncService(
+        frozen=lambda: False,
+        installed_plugins=lambda: [official.plugin_id],
+        online_plugins=lambda: [merged_official, merged_competing],
+        local_plugins=lambda: [merged_local],
+        merge_plugins=_merge_catalog_plugins,
+        plugin_exists=lambda *_args: False,
+        install=install,
+        log=Mock(),
+    )
+
+    async with plugin_lifecycle.hold_startup() as startup_token:
+        with pytest.raises(
+            RuntimeError,
+            match="延后激活的插件同步未完成：DemoPlugin",
+        ):
+            await asyncio.wait_for(
+                asyncio.to_thread(service.sync, startup_token),
+                timeout=2,
+            )
+
+    executor.execute.assert_awaited_once()
+    assert executor.execute.await_args.kwargs["local_sync"] is True
+    admission = executor.execute.await_args.kwargs["admission"]
+    assert admission.candidate is local
 
 
 @pytest.mark.asyncio