Merge pull request #6656 from InfinityPacer/pr/plugin-source-binding-v3

修复 V3 插件本地来源优先级与更新索引刷新

app/adapters/external/plugin/client.py

@@ -489,8 +489,7 @@ class PluginMarketTransport(metaclass=WeakSingleton):
         candidates: PluginIndex = {}
         for repo_order, repo_path in enumerate(self.get_local_repo_paths()):
             if not repo_path.exists() or not repo_path.is_dir():
-                logger.warn(f"本地插件仓库目录不存在或不可读：{repo_path}")
-                continue
+                raise RuntimeError(f"本地插件仓库目录不存在或不可读：{repo_path}")
 
             package_candidates = []
             if get_runtime_setting('VERSION_FLAG'):

app/api/endpoints/plugin.py

@@ -248,6 +248,7 @@ async def runtime_status(
         _SchemaPluginRuntimeStatus.READY,
     }
     failed = {
+        _SchemaPluginRuntimeStatus.SYNC_FAILED,
         _SchemaPluginRuntimeStatus.BLOCKED_BY_POLICY,
         _SchemaPluginRuntimeStatus.LOAD_FAILED,
     }

app/application/plugin/catalog.py

@@ -379,8 +379,8 @@ class PluginCatalogService:
                     ) \
                     or (
                         plugin.plugin_version == exists.plugin_version
-                        and self._is_local_repo(exists.repo_url)
-                        and not self._is_local_repo(plugin.repo_url)
+                        and not self._is_local_repo(exists.repo_url)
+                        and self._is_local_repo(plugin.repo_url)
                     ):
                 result_by_id[normalized_id] = plugin
         return list(result_by_id.values())

app/application/plugin/gateway.py

@@ -94,7 +94,9 @@ class PluginInstallGateway:
     ) -> PluginInstallResult:
         """读取冻结库存并执行一次不能绕过来源身份的插件写入。"""
         try:
-            inventory = await self.__inventory(force)
+            # 安装 ``force`` 只控制载荷覆盖；远程市场索引由市场手动刷新和
+            # 定时缓存任务维护，不能让单个插件更新放大全部仓库请求。
+            inventory = await self.__inventory(False)
             async with plugin_lifecycle.hold(plugin_id, startup_token):
                 identity = await self.__identity(plugin_id)
                 admission = admit_plugin_install(

app/application/plugin/release.py

@@ -7,7 +7,7 @@ from dataclasses import dataclass
 from typing import Any, cast
 
 from app.application.plugin.catalog import apply_declared_metadata_fallback
-from app.application.plugin.identity import PluginIdentity
+from app.application.plugin.identity import PluginIdentity, PluginPayloadSourceType
 from app.schemas.plugin import Plugin
 
 MarketPluginLoader = Callable[[str, str | None, bool], Awaitable[list[Plugin] | None]]
@@ -77,16 +77,26 @@ class PluginReleaseService:
             (plugin for plugin in self._local_repo_plugins() if plugin.id == plugin_id),
             None,
         )
-        if local_plugin is not None:
-            return _merge_market_metadata(installed_plugin, local_plugin)
+        if identity is not None and identity.payload_source_type is PluginPayloadSourceType.LOCAL:
+            return (
+                _merge_market_metadata(installed_plugin, local_plugin)
+                if local_plugin is not None
+                else installed_plugin
+            )
 
         repo_url = _trusted_repo_url(identity)
-        if repo_url is None:
-            return installed_plugin
-        market_plugin = await self._market_plugin(plugin_id, repo_url, force)
+        if repo_url is not None:
+            market_plugin = await self._market_plugin(plugin_id, repo_url, force)
+            return (
+                _merge_market_metadata(installed_plugin, market_plugin)
+                if market_plugin is not None
+                else installed_plugin
+            )
+
+        # 尚未建立来源身份的本地插件继续使用本地仓库元数据，兼容首次迁移前的运行态。
         return (
-            _merge_market_metadata(installed_plugin, market_plugin)
-            if market_plugin is not None
+            _merge_market_metadata(installed_plugin, local_plugin)
+            if local_plugin is not None
             else installed_plugin
         )
 

app/application/plugin/source.py

@@ -126,10 +126,11 @@ class PluginLocalCandidate:
         return None
 
     def public_dict(self) -> dict[str, Any]:
-        """生成本地候选的公共投影，永不暴露仓库路径或原始 metadata。"""
+        """生成本地候选的公共投影，保留管理员配置的仓库路径。"""
         return {
             "plugin_id": self.plugin_id,
             "source_type": PluginPayloadSourceType.LOCAL.value,
+            "repo_url": self.repo_url,
             "package_generation": self.package_generation,
             "plugin_version": self.plugin_version,
         }

app/runtime/extensions/plugin/catalog.py

@@ -143,25 +143,30 @@ class PluginCatalogFacade:
         plugin_class = self._classes().get(plugin_id)
         return getattr(plugin_class, "plugin_version", None)
 
-    def local_repository(self) -> list[Plugin]:
+    def local_repository(self, *, raise_errors: bool = False) -> list[Plugin]:
         """读取本地插件仓候选并映射为目录 DTO。"""
         installed = self._storage().read(SystemConfigKey.UserInstalledPlugins) or []
         try:
             candidates = self._system().local_candidates()
         except Exception as error:  # noqa: BLE001 - 展示失败不能阻断整个插件目录
             self._logger.warning(f"读取本地插件仓候选失败，已跳过本地目录展示：{error}")
+            if raise_errors:
+                raise
             return []
         plugins: list[Plugin] = []
         for plugin_id, info in candidates.items():
             package_version = info.get("package_version")
+            repo_url = info.get("repo_url")
+            if not isinstance(repo_url, str) or not repo_url.startswith("local://"):
+                repo_url = self._system().local_repo_url(
+                    plugin_id,
+                    info.get("repo_path"),
+                    package_version,
+                )
             plugin = self._map_plugin(
                 pid=plugin_id,
                 plugin_info=info,
-                market=self._system().local_repo_url(
-                    plugin_id,
-                    None,
-                    package_version,
-                ),
+                market=repo_url,
                 installed_apps=installed,
                 add_time=0,
                 package_version=package_version,

app/runtime/extensions/plugin/runtime.py

@@ -280,7 +280,9 @@ def build_plugin_runtime(
             SystemConfigKey.UserInstalledPlugins
         ) or [],
         online_plugins=catalog.online,
-        local_plugins=catalog.local_repository,
+        # 启动恢复必须保留本地仓库扫描失败，不能把异常降级为空候选后
+        # 再从在线市场下载覆盖当前载荷。
+        local_plugins=lambda: catalog.local_repository(raise_errors=True),
         merge_plugins=lambda higher, base, _markets: catalog.merge(higher, base),
         plugin_exists=catalog.exists,
         install=lambda plugin_id, repo_url, force, startup_token: environment.system().install_plugin(
@@ -289,6 +291,7 @@ def build_plugin_runtime(
             force=force,
             startup_token=startup_token,
         ),
+        runtime_status_writer=registry.set_runtime_status,
         log=environment.logger,
     )
     def source_plugin_id(plugin_id: str) -> str:

app/runtime/extensions/plugin/sync.py

@@ -7,6 +7,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Any, Callable, Optional
 
 from app.runtime.extensions.plugin.system import PluginSystemServices
+from app.schemas.plugin import PluginRuntimeStatus
 
 
 class PluginSyncService:
@@ -22,6 +23,7 @@ class PluginSyncService:
         merge_plugins: Callable[[list[Any], list[Any], list[Any]], list[Any]],
         plugin_exists: Callable[[str, Optional[str]], bool],
         install: Callable[[str, Optional[str], bool, object | None], tuple[bool, str]],
+        runtime_status_writer: Callable[[str, PluginRuntimeStatus], None] | None = None,
         log: Any,
     ) -> None:
         """保存目录读取、包安装和持久化报告端口。"""
@@ -32,6 +34,7 @@ class PluginSyncService:
         self._merge_plugins = merge_plugins
         self._plugin_exists = plugin_exists
         self._install = install
+        self._runtime_status_writer = runtime_status_writer
         self._logger = log
 
     def sync(
@@ -48,28 +51,71 @@ class PluginSyncService:
             plugin_id.lower()
             for plugin_id in self._installed_plugins()
         }
-        online = self._online_plugins()
-        local = self._local_plugins()
+        try:
+            local = self._local_plugins()
+        except Exception as error:  # noqa: BLE001 - 本地来源失败不能降级在线
+            if self._runtime_status_writer:
+                for plugin_id in installed:
+                    self._runtime_status_writer(
+                        plugin_id,
+                        PluginRuntimeStatus.SYNC_FAILED,
+                    )
+            self._logger.error(f"读取本地插件仓失败，已停止启动同步：{error}")
+            raise RuntimeError(f"本地插件仓读取失败：{error}") from error
         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())
         } - local_plugin_ids
-        candidates = self._merge_plugins(online + local, [], []) if online or local else []
+        missing_ids = {
+            plugin_id
+            for plugin_id in installed
+            if not self._plugin_exists(plugin_id, None)
+        }
+        # 运行目录已有可用插件且没有在线载荷恢复任务时，不读取整个在线市场。
+        # 版本更新由用户操作触发，启动阶段不应为市场刷新阻塞插件激活。
+        online = (
+            self._online_plugins()
+            if restore_plugin_ids or (missing_ids - local_plugin_ids)
+            else []
+        )
+        # 启动同步只负责恢复缺失载荷；候选来源按本地优先建立索引，避免
+        # 市场目录合并结果把本地候选替换成在线候选。
+        candidates_by_id: dict[str, Any] = {
+            plugin.id.lower(): plugin
+            for plugin in online
+            if getattr(plugin, "id", None)
+        }
+        candidates_by_id.update(
+            {
+                plugin.id.lower(): plugin
+                for plugin in local
+                if getattr(plugin, "id", None)
+            }
+        )
+        candidates = list(candidates_by_id.values())
+        recovery_ids = {
+            plugin.id.lower()
+            for plugin in candidates
+            if plugin.id.lower() in installed
+            and (
+                plugin.id.lower() in restore_plugin_ids
+                or plugin.id.lower() in missing_ids
+            )
+        }
         targets = [
             plugin
             for plugin in candidates
             if plugin.id.lower() in installed
             and (
-                plugin.id.lower() in deferred_plugin_ids
-                or plugin.id.lower() in restore_plugin_ids
+                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)
+                    # 版本差异由用户点击更新处理；启动只判断运行目录是否可用。
+                    and plugin.id.lower() in missing_ids
                 )
             )
         ]
@@ -79,14 +125,18 @@ class PluginSyncService:
         self._logger.info("开始安装第三方插件...")
         synced: list[str] = []
         failed: list[str] = []
-        failed_deferred: list[str] = []
+        failed_recovery: list[str] = []
 
         def install_one(plugin: Any) -> None:
             """安装一个插件并记录结果。"""
             started = time.time()
             state, message = self._install(
                 plugin.id,
-                None,
+                (
+                    getattr(plugin, "repo_url", None)
+                    if str(getattr(plugin, "repo_url", "")).startswith("local://")
+                    else None
+                ),
                 False,
                 startup_token,
             )
@@ -97,8 +147,13 @@ class PluginSyncService:
                 )
                 synced.append(plugin.id)
             else:
-                if plugin.id.lower() in deferred_plugin_ids:
-                    failed_deferred.append(plugin.id)
+                if plugin.id.lower() in recovery_ids:
+                    failed_recovery.append(plugin.id)
+                if self._runtime_status_writer:
+                    self._runtime_status_writer(
+                        plugin.id,
+                        PluginRuntimeStatus.SYNC_FAILED,
+                    )
                 self._logger.error(
                     f"插件 {plugin.plugin_name} 同步失败："
                     f"{message}，耗时：{elapsed:.2f} 秒"
@@ -112,8 +167,13 @@ class PluginSyncService:
                 try:
                     future.result()
                 except Exception as error:  # noqa: BLE001
-                    if plugin.id.lower() in deferred_plugin_ids:
-                        failed_deferred.append(plugin.id)
+                    if plugin.id.lower() in recovery_ids:
+                        failed_recovery.append(plugin.id)
+                    if self._runtime_status_writer:
+                        self._runtime_status_writer(
+                            plugin.id,
+                            PluginRuntimeStatus.SYNC_FAILED,
+                        )
                     self._logger.error(
                         f"插件 {plugin.plugin_name} 安装过程中出现异常: {error}"
                     )
@@ -121,10 +181,10 @@ class PluginSyncService:
         self._logger.info(
             f"第三方插件安装完成，成功：{len(synced)} 个，失败：{len(failed)} 个"
         )
-        if failed_deferred:
+        if failed_recovery:
             raise RuntimeError(
-                "延后激活的插件同步未完成："
-                f"{', '.join(sorted(set(failed_deferred)))}"
+                "插件同步未完成："
+                f"{', '.join(sorted(set(failed_recovery)))}"
             )
         return synced
 

app/schemas/plugin.py

@@ -10,8 +10,9 @@ from app.schemas.common import JsonData
 
 
 class PluginRuntimeStatus(str, _Enum):
-    """插件从源码准备到运行激活的六类状态。"""
+    """插件从来源同步、依赖准备到运行激活的状态。"""
 
+    SYNC_FAILED = "sync_failed"
     SOURCE_MISSING = "source_missing"
     DEPENDENCY_PENDING = "dependency_pending"
     READY = "ready"
@@ -266,14 +267,14 @@ class PluginSourceIdentity(BaseModel):  # type: ignore[misc]
 class PluginSourceCandidate(BaseModel):  # type: ignore[misc]
     """一个可供管理员识别的脱敏插件来源候选。"""
 
-    source_type: Literal["official", "third_party", "local"] = Field(description="来源类型；本地候选不公开路径")
+    source_type: Literal["official", "third_party", "local"] = Field(description="来源类型")
     source_key: Optional[str] = Field(
         default=None,
         description="规范化在线来源键；本地候选为空",
     )
     repo_url: Optional[str] = Field(
         default=None,
-        description="可明确选择的在线仓库地址；本地候选为空",
+        description="可明确选择的在线仓库地址或本地仓库标识",
     )
     package_generation: Literal["v1", "v2", "v3"] = Field(description="当前运行时会采用的插件包代际")
     plugin_version: Optional[str] = Field(

docs/architecture-overview.md

@@ -757,7 +757,7 @@ flowchart LR
 | 指标 | 当前值 |
 |---|---:|
 | Python 模块 | 1010 |
-| 内部导入边 | 8,579 |
+| 内部导入边 | 8,581 |
 | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 53（债务已清零，53 条精确 containment） |

docs/refactor/optimization-checklist.md

@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 1010 / 8,579 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索、整理恢复、Agent 计划、工具视觉、终端生命周期与终端作用域模块的受控依赖 |
+| 宿主 Python 模块 / 内部依赖边 | 1010 / 8,581 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索、整理恢复、Agent 计划、工具视觉、终端生命周期与终端作用域模块的受控依赖 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |

tests/fixtures/architecture/dependency-baseline.json

@@ -1077,8 +1077,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 8579,
-  "edge_sha256": "1ffc77ec803f7ee447d884559e5e6298e5a812aabb3e5d74cec72f61a37c10a8",
+  "edge_count": 8581,
+  "edge_sha256": "7b89a2f04c4b256955e9d17821d7e4ac448c1847e12772aadf5906a520a4e3eb",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -8324,6 +8324,8 @@
     "app.runtime.extensions.plugin.sync -> app.runtime.extensions",
     "app.runtime.extensions.plugin.sync -> app.runtime.extensions.plugin",
     "app.runtime.extensions.plugin.sync -> app.runtime.extensions.plugin.system",
+    "app.runtime.extensions.plugin.sync -> app.schemas",
+    "app.runtime.extensions.plugin.sync -> app.schemas.plugin",
     "app.runtime.extensions.plugin.tools -> app.runtime",
     "app.runtime.extensions.plugin.tools -> app.runtime.extensions",
     "app.runtime.extensions.plugin.tools -> app.runtime.extensions.plugin",

tests/test_plugin_candidate_inventory.py

@@ -222,8 +222,8 @@ def test_local_scan_preserves_absent_present_and_failed_states() -> None:
     assert failed.local_read.error == "local repository unavailable"
 
 
-def test_local_candidates_never_expose_path_in_inventory_projection() -> None:
-    """本地候选可参与库存，但公共投影永不携带本地仓库路径。"""
+def test_local_candidates_keep_source_marker_in_inventory_projection() -> None:
+    """本地候选可参与库存，并保留可供来源选择使用的来源标识。"""
     reader = PluginCandidateInventoryReader(
         market_loader=lambda *_args: {},
         local_candidate_loader=lambda: {
@@ -244,10 +244,10 @@ def test_local_candidates_never_expose_path_in_inventory_projection() -> None:
     assert public["local_candidates"] == [{
         "plugin_id": "LocalPlugin",
         "source_type": "local",
+        "repo_url": "local://LocalPlugin?path=/private/local&version=v3",
         "package_generation": "v3",
         "plugin_version": "3.0.0",
     }]
-    assert "/private/local" not in str(public)
 
 
 def test_invalid_local_candidate_does_not_abort_online_inventory() -> None:

tests/test_plugin_catalog_runtime.py

@@ -141,3 +141,5 @@ def test_local_repository_failure_does_not_break_catalog_projection():
     assert warnings == [
         "读取本地插件仓候选失败，已跳过本地目录展示：invalid local package"
     ]
+    with pytest.raises(RuntimeError, match="invalid local package"):
+        facade.local_repository(raise_errors=True)

tests/test_plugin_endpoint.py

@@ -1,5 +1,6 @@
 import asyncio
 from contextlib import asynccontextmanager, nullcontext
+from dataclasses import replace
 from datetime import datetime, timezone
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -396,6 +397,84 @@ def test_plugin_history_merges_remote_metadata():
     plugin_manager.async_get_plugins_from_market.assert_awaited_once_with(SOURCE_URL, settings.VERSION_FLAG, True)
 
 
+def test_plugin_history_uses_payload_source_when_local_candidate_also_exists():
+    """在线运行载荷不得被同 ID 的本地仓库历史覆盖。"""
+    installed_plugin = schemas.Plugin(
+        id="DemoPlugin",
+        plugin_name="Demo Plugin",
+        plugin_version="1.0.0",
+        installed=True,
+        history={},
+    )
+    local_plugin = schemas.Plugin(
+        id="DemoPlugin",
+        repo_url="local://DemoPlugin?path=/plugins&version=v3",
+        history={"v0.9.0": "本地旧版说明"},
+    )
+    market_plugin = schemas.Plugin(
+        id="DemoPlugin",
+        repo_url=SOURCE_URL,
+        history={"v1.1.0": "在线新版说明"},
+    )
+    plugin_manager = MagicMock()
+    plugin_manager.get_installed_plugins.return_value = [installed_plugin]
+    plugin_manager.get_local_repo_plugins.return_value = [local_plugin]
+    plugin_manager.async_get_plugins_from_market = AsyncMock(return_value=[market_plugin])
+    metadata = PluginDeclaredMetadata.from_package(
+        {"name": "Demo Plugin", "v3": True},
+        declaration_version="1.0.0",
+        manifest_matches_payload=True,
+    )
+    identity = _plugin_identity(metadata=metadata)
+    release_service = _release_service(plugin_manager, persistence=_persistence(identity))
+
+    result = asyncio.run(release_service.history("DemoPlugin", force=True))
+
+    assert result is not None
+    assert result.repo_url == SOURCE_URL
+    assert result.history == {"v1.1.0": "在线新版说明"}
+    plugin_manager.async_get_plugins_from_market.assert_awaited_once_with(SOURCE_URL, settings.VERSION_FLAG, True)
+
+
+def test_plugin_history_uses_local_metadata_for_local_payload():
+    """本地运行载荷继续读取同一本地仓库的历史。"""
+    installed_plugin = schemas.Plugin(
+        id="DemoPlugin",
+        plugin_name="Demo Plugin",
+        plugin_version="1.0.0",
+        installed=True,
+        history={},
+    )
+    local_plugin = schemas.Plugin(
+        id="DemoPlugin",
+        repo_url="local://DemoPlugin?path=/plugins&version=v3",
+        history={"v1.1.0": "本地新版说明"},
+    )
+    plugin_manager = MagicMock()
+    plugin_manager.get_installed_plugins.return_value = [installed_plugin]
+    plugin_manager.get_local_repo_plugins.return_value = [local_plugin]
+    plugin_manager.async_get_plugins_from_market = AsyncMock()
+    metadata = PluginDeclaredMetadata.from_package(
+        {"name": "Demo Plugin", "v3": True},
+        declaration_version="1.0.0",
+        manifest_matches_payload=True,
+    )
+    identity = _plugin_identity(metadata=metadata)
+    identity = replace(
+        identity,
+        payload_source_type=PluginPayloadSourceType.LOCAL,
+        payload_source_key=None,
+    )
+    release_service = _release_service(plugin_manager, persistence=_persistence(identity))
+
+    result = asyncio.run(release_service.history("DemoPlugin", force=True))
+
+    assert result is not None
+    assert result.repo_url == local_plugin.repo_url
+    assert result.history == {"v1.1.0": "本地新版说明"}
+    plugin_manager.async_get_plugins_from_market.assert_not_awaited()
+
+
 def test_plugin_history_falls_back_to_backward_compatible_package():
     """已绑定来源的插件在当前索引缺失时仍读取向后兼容代际的历史。"""
     installed_plugin = schemas.Plugin(

tests/test_plugin_install_admission.py

@@ -322,8 +322,8 @@ def test_first_local_payload_creates_local_only_identity() -> None:
     assert target.trusted_source_type is TrustedPluginSourceType.UNKNOWN
 
 
-def test_sanitized_local_reference_selects_configured_candidate_without_path() -> None:
-    """脱敏本地来源标识仍能选择配置内候选，但公共投影不暴露路径。"""
+def test_local_reference_selects_configured_candidate_with_source_marker() -> None:
+    """本地来源标识仍能选择配置内候选，并保留可回传的来源标识。"""
     local = PluginLocalCandidate(
         plugin_id="DemoPlugin",
         repo_url=(
@@ -350,10 +350,10 @@ def test_sanitized_local_reference_selects_configured_candidate_without_path() -
     assert public == {
         "plugin_id": "DemoPlugin",
         "source_type": "local",
+        "repo_url": "local://DemoPlugin?path=/private/secret/plugins&version=v3",
         "package_generation": "v3",
         "plugin_version": "2.0.0-dev",
     }
-    assert "/private/secret/plugins" not in str(public)
 
 
 def test_legacy_identity_can_bind_explicit_online_source() -> None:

tests/test_plugin_install_gateway.py

@@ -76,6 +76,37 @@ async def test_gateway_freezes_admission_before_executing_transaction() -> None:
     assert admission.expected_revision is None
 
 
+@pytest.mark.asyncio
+async def test_force_install_reuses_cached_candidate_inventory() -> None:
+    """强制覆盖载荷时不得连带强刷全部远程插件仓库。"""
+    inventory = AsyncMock(return_value=_inventory())
+    executor = AsyncMock()
+    executor.execute.return_value = type(
+        "Result",
+        (),
+        {"success": True, "message": ""},
+    )()
+    gateway = PluginInstallGateway(
+        inventory=inventory,
+        identity=AsyncMock(return_value=None),
+        candidate_compatibility=lambda _candidate: (True, ""),
+        executor=executor,
+        clock=lambda: NOW,
+    )
+
+    result = await gateway.install(
+        plugin_id="DemoPlugin",
+        repo_url=REPO_URL,
+        package_version="v3",
+        force=True,
+        explicit_source=True,
+    )
+
+    assert result.success is True
+    inventory.assert_awaited_once_with(False)
+    assert executor.execute.await_args.kwargs["force"] is True
+
+
 @pytest.mark.asyncio
 async def test_local_only_requires_explicit_online_binding() -> None:
     """本地专属身份即使发现唯一在线来源，也只能由管理员显式绑定。"""
@@ -357,8 +388,8 @@ async def test_gateway_checks_compatibility_on_final_trusted_candidate() -> None
 
 
 @pytest.mark.asyncio
-async def test_gateway_source_inspection_preserves_sources_and_hides_local_path() -> None:
-    """来源查询按在线仓归并版本，本地候选只保留类型与版本。"""
+async def test_gateway_source_inspection_preserves_sources_and_local_repo_url() -> None:
+    """来源查询按在线仓归并版本，并保留本地候选仓库标识。"""
     official_v3 = _inventory().online_candidates[0]
     official_v2 = PluginMarketCandidate(
         plugin_id="DemoPlugin",
@@ -414,7 +445,7 @@ async def test_gateway_source_inspection_preserves_sources_and_hides_local_path(
     ]
     assert inspection.online_candidates[0].package_generation == "v3"
     assert inspection.local_candidate is local
-    assert "/private/plugins" not in str(inspection.local_candidate.public_dict())
+    assert inspection.local_candidate.public_dict()["repo_url"] == local.repo_url
 
 
 @pytest.mark.asyncio

tests/test_plugin_source_policy.py

@@ -480,8 +480,8 @@ def test_uninstalled_unique_and_multiple_sources_are_distinct() -> None:
     assert set(conflict.conflict_source_keys) == {THIRD_PARTY_SOURCE, OTHER_SOURCE}
 
 
-def test_official_candidate_is_selectable_and_local_projection_hides_path() -> None:
-    """官方来源可正常选择，本地公共投影不能泄漏仓库路径或 metadata。"""
+def test_official_candidate_is_selectable_and_local_projection_keeps_repo_url() -> None:
+    """官方来源可正常选择，本地公共投影保留可回传的仓库标识。"""
     official = select_plugin_candidate(
         _inventory(
             MarketRead.present(
@@ -518,5 +518,4 @@ def test_official_candidate_is_selectable_and_local_projection_hides_path() -> N
     assert local.source_key is None
     assert local_result.candidate is local
     public = local_result.public_dict()
-    assert "/private/secret/plugins" not in str(public)
-    assert "repo_url" not in public["candidate"]
+    assert public["candidate"]["repo_url"] == local.repo_url

tests/test_plugin_sync_service.py

@@ -28,6 +28,7 @@ from app.application.plugin.source import (
 )
 from app.runtime.config import global_vars
 from app.runtime.extensions.plugin.sync import LocalPluginSyncService, PluginSyncService
+from app.schemas.plugin import PluginRuntimeStatus
 from app.startup.initializers import plugins as plugins_initializer
 
 REPO_URL = "https://github.com/jxxghp/MoviePilot-Plugins"
@@ -102,8 +103,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_reconciles_existing_local_candidate_through_gateway() -> None:
-    """本地候选对应插件已延后激活，必须经 Gateway 协调后再启动。"""
+def test_market_sync_skips_existing_local_candidate() -> None:
+    """运行目录已有可用插件时，启动同步不因本地候选重复安装。"""
     online = SimpleNamespace(
         id="DemoPlugin",
         repo_url=REPO_URL,
@@ -130,8 +131,8 @@ def test_market_sync_reconciles_existing_local_candidate_through_gateway() -> No
         log=Mock(),
     )
 
-    assert service.sync(online_restore_plugins={"demoplugin"}) == [online.id]
-    install.assert_called_once_with(online.id, None, False, None)
+    assert service.sync(online_restore_plugins={"demoplugin"}) == []
+    install.assert_not_called()
 
 
 def test_market_sync_defers_source_selection_to_gateway() -> None:
@@ -156,7 +157,7 @@ def test_market_sync_defers_source_selection_to_gateway() -> None:
     )
 
     assert service.sync() == [local.id]
-    install.assert_called_once_with(local.id, None, False, None)
+    install.assert_called_once_with(local.id, local.repo_url, False, None)
 
 
 def test_market_sync_reports_local_install_failure() -> None:
@@ -179,13 +180,30 @@ def test_market_sync_reports_local_install_failure() -> None:
         log=Mock(),
     )
 
-    with pytest.raises(
-        RuntimeError,
-        match="延后激活的插件同步未完成：DemoPlugin",
-    ):
+    with pytest.raises(RuntimeError, match="插件同步未完成：DemoPlugin"):
         service.sync()
 
 
+def test_market_sync_stops_without_online_fallback_when_local_scan_fails() -> None:
+    """本地仓扫描失败时，启动同步不得退回在线候选。"""
+    statuses: list[tuple[str, object]] = []
+    service = PluginSyncService(
+        frozen=lambda: False,
+        installed_plugins=lambda: ["DemoPlugin"],
+        online_plugins=lambda: pytest.fail("本地扫描失败后不应读取在线候选"),
+        local_plugins=lambda: (_ for _ in ()).throw(RuntimeError("repo unavailable")),
+        merge_plugins=lambda items, *_args: items,
+        plugin_exists=lambda *_args: False,
+        install=Mock(return_value=(True, "")),
+        runtime_status_writer=lambda plugin_id, status: statuses.append((plugin_id, status)),
+        log=Mock(),
+    )
+
+    with pytest.raises(RuntimeError, match="本地插件仓读取失败"):
+        service.sync()
+    assert statuses == [("demoplugin", PluginRuntimeStatus.SYNC_FAILED)]
+
+
 def test_local_sync_matches_installed_plugin_id_case_insensitively() -> None:
     """本地热同步应把大小写不同的索引 ID 识别为同一已安装插件。"""
     candidate = {
@@ -294,7 +312,7 @@ async def test_market_sync_preserves_generation_priority_through_gateway(
             release_version=None,
             force=force,
             local_sync=True,
-            explicit_source=False,
+            explicit_source=True,
             startup_token=startup_token,
         )
 
@@ -331,9 +349,9 @@ async def test_market_sync_preserves_generation_priority_through_gateway(
 
     assert synced == [local.plugin_id]
     executor.execute.assert_awaited_once()
-    assert executor.execute.await_args.kwargs["local_sync"] is False
+    assert executor.execute.await_args.kwargs["local_sync"] is True
     admission = executor.execute.await_args.kwargs["admission"]
-    assert admission.candidate is online
+    assert admission.candidate is local
     assert admission.trusted_source_key == online.source_key
 
 
@@ -468,10 +486,7 @@ async def test_market_sync_blocks_activation_when_gateway_selected_local_fails(
     )
 
     async with plugin_lifecycle.hold_startup() as startup_token:
-        with pytest.raises(
-            RuntimeError,
-            match="延后激活的插件同步未完成：DemoPlugin",
-        ):
+        with pytest.raises(RuntimeError, match="插件同步未完成：DemoPlugin"):
             await asyncio.wait_for(
                 asyncio.to_thread(service.sync, startup_token),
                 timeout=2,