fix(plugin): support BOM and wrapped market indexes

app/adapters/external/plugin/client.py

@@ -50,6 +50,7 @@ PluginPayload = dict[str, Any]
 PluginIndex = dict[str, PluginPayload]
 PluginReleaseList = list[PluginPayload]
 PluginRequestOptions = dict[str, Any]
+PluginIndexTaskResult = tuple[Optional[PluginIndex], Optional[Exception]]
 PluginReleaseTask = asyncio.Task[Optional[PluginReleaseList]]
 
 PLUGIN_INDEX_MAX_BYTES = 1024 * 1024
@@ -215,7 +216,7 @@ class PluginMarketTransport(metaclass=WeakSingleton):
     _index_task_lock = threading.Lock()
     _index_tasks: dict[
         tuple[asyncio.AbstractEventLoop, str, str],
-        asyncio.Task[Optional[PluginIndex]],
+        asyncio.Task[PluginIndexTaskResult],
     ] = {}
     _release_task_lock = threading.Lock()
     _release_tasks: dict[
@@ -760,9 +761,9 @@ class PluginMarketTransport(metaclass=WeakSingleton):
 
     @classmethod
     def __parse_plugin_index_response(cls, content: str) -> Optional[PluginIndex]:
-        """解析并规范化插件索引，仅缓存满足索引级边界的结果。"""
+        """解析并规范化插件索引，兼容 BOM 与 plugins 列表包装格式。"""
         try:
-            payload = json.loads(content)
+            payload = json.loads(content.lstrip("\ufeff"))
         except (ValueError, RecursionError):
             logger.warning("插件包数据解析失败：响应不是有效 JSON")
             return None
@@ -777,9 +778,25 @@ class PluginMarketTransport(metaclass=WeakSingleton):
                 f"插件包 JSON 嵌套超过上限：{PLUGIN_INDEX_MAX_NESTING} 层"
             )
             return None
-        if len(payload) > PLUGIN_INDEX_MAX_ENTRIES:
+        plugin_entries: list[tuple[object, object]]
+        wrapped_plugins = payload.get("plugins")
+        if isinstance(wrapped_plugins, list):
+            plugin_entries = [
+                (
+                    plugin_info.get("id")
+                    if isinstance(plugin_info, dict)
+                    else None,
+                    plugin_info,
+                )
+                for plugin_info in wrapped_plugins
+            ]
+        else:
+            plugin_entries = list(payload.items())
+
+        if len(plugin_entries) > PLUGIN_INDEX_MAX_ENTRIES:
             logger.warning(
-                f"插件包条目超过上限：{len(payload)} > {PLUGIN_INDEX_MAX_ENTRIES}"
+                f"插件包条目超过上限：{len(plugin_entries)} > "
+                f"{PLUGIN_INDEX_MAX_ENTRIES}"
             )
             return None
 
@@ -787,7 +804,7 @@ class PluginMarketTransport(metaclass=WeakSingleton):
         skipped_reasons: Counter[str] = Counter()
         dropped_fields: Counter[str] = Counter()
         skipped_examples: list[str] = []
-        for plugin_id, plugin_info in payload.items():
+        for plugin_id, plugin_info in plugin_entries:
             item, reason, dropped = cls.__normalize_plugin_index_entry(
                 plugin_id,
                 plugin_info,
@@ -1162,7 +1179,7 @@ class PluginMarketTransport(metaclass=WeakSingleton):
     def _remove_index_task(
         cls,
         key: tuple[asyncio.AbstractEventLoop, str, str],
-        task: asyncio.Future[Optional[PluginIndex]],
+        task: asyncio.Future[PluginIndexTaskResult],
     ) -> None:
         """异步索引请求结束后释放事件循环和仓库引用。"""
         with cls._index_task_lock:
@@ -1210,6 +1227,17 @@ class PluginMarketTransport(metaclass=WeakSingleton):
             _PLUGIN_INDEX_FETCH_GATE.release()
         return self._resolve_plugin_index_result(response)
 
+    async def _fetch_plugin_index_async_task(
+        self,
+        package_url: str,
+        headers: Optional[dict[str, str]],
+    ) -> PluginIndexTaskResult:
+        """把预期的索引读取失败转成共享任务结果，交由调用方记录仓库失败。"""
+        try:
+            return await self._fetch_plugin_index_async(package_url, headers), None
+        except Exception as error:  # noqa: BLE001 - 结果由读取调用方统一转换
+            return None, error
+
     @cached(maxsize=1024, ttl=1800, skip_none=False)  # type: ignore[misc]
     def get_plugin_index_result(
             self,
@@ -1590,9 +1618,12 @@ class PluginMarketTransport(metaclass=WeakSingleton):
             task = self._index_tasks.get(task_key)
             if task is None:
                 task = cast(
-                    asyncio.Task[Optional[PluginIndex]],
+                    asyncio.Task[PluginIndexTaskResult],
                     get_task_registry().create(
-                        self._fetch_plugin_index_async(package_url, headers),
+                        self._fetch_plugin_index_async_task(
+                            package_url,
+                            headers,
+                        ),
                         owner="plugin.market.index",
                     ),
                 )
@@ -1606,7 +1637,10 @@ class PluginMarketTransport(metaclass=WeakSingleton):
                 task.add_done_callback(on_index_task_done)
 
         # 单个调用方取消等待时不能连带取消共享请求。
-        return await asyncio.shield(task)
+        result, error = await asyncio.shield(task)
+        if error is not None:
+            raise error
+        return result
 
     async def async_get_plugins(self, repo_url: str,
                                 package_version: Optional[str] = None) -> Optional[PluginIndex]:

tests/test_plugin_market_index_policy.py

@@ -17,6 +17,7 @@ from app.adapters.external.plugin.client import (
     _format_request_error,
 )
 from app.domain.plugin import split_plugin_market_repo_urls
+from app.runtime.tasks import TaskRegistry
 
 SYNC_INDEX_REQUEST = "_PluginMarketTransport__request_plugin_index_with_fallback"
 ASYNC_INDEX_REQUEST = "_PluginMarketTransport__async_request_plugin_index_with_fallback"
@@ -472,6 +473,39 @@ def test_plugin_index_entry_count_boundary(entry_count: int, accepted: bool) ->
         assert len(result) == PLUGIN_INDEX_MAX_ENTRIES
 
 
+@pytest.mark.parametrize(
+    ("content", "expected"),
+    [
+        (
+            '\ufeff{"BomPlugin": {"version": "1.0.0"}}',
+            {"BomPlugin": {"version": "1.0.0"}},
+        ),
+        (
+            json.dumps(
+                {
+                    "plugins": [
+                        {
+                            "id": "WrappedPlugin",
+                            "version": "1.2.3",
+                        }
+                    ]
+                }
+            ),
+            {"WrappedPlugin": {"id": "WrappedPlugin", "version": "1.2.3"}},
+        ),
+    ],
+)
+def test_plugin_index_accepts_bom_and_wrapped_entries(
+    content: str,
+    expected: dict,
+) -> None:
+    """插件索引应兼容 UTF-8 BOM 与 plugins 数组包装格式。"""
+    assert (
+        PluginMarketTransport()._resolve_plugin_index_response(200, content)
+        == expected
+    )
+
+
 def test_plugin_index_isolates_bad_entries_and_preserves_unknown_fields() -> None:
     """决策字段非法只隔离该条目，展示字段降级且未知字段保持前向兼容。"""
     history = {str(index): "change" for index in range(513)}
@@ -692,6 +726,42 @@ async def test_async_plugin_index_result_preserves_absent_state(monkeypatch) ->
     assert result is None
 
 
+@pytest.mark.asyncio
+async def test_async_plugin_index_failure_is_not_reported_as_background_task_error(
+    monkeypatch,
+) -> None:
+    """异步索引失败交给市场调用方处理，不应再被任务登记器重复上报。"""
+    helper = PluginMarketTransport()
+    registry = TaskRegistry()
+    reports: list[dict[str, object]] = []
+    loop = asyncio.get_running_loop()
+    previous_handler = loop.get_exception_handler()
+
+    async def request(_url: str, *, headers: dict):
+        return 200, "not-json"
+
+    monkeypatch.setattr(helper, ASYNC_INDEX_REQUEST, request)
+    monkeypatch.setattr(
+        plugin_client_module,
+        "get_task_registry",
+        lambda: registry,
+    )
+    await helper.async_get_plugin_index_result.cache_clear()
+    loop.set_exception_handler(lambda _, context: reports.append(context))
+
+    try:
+        with pytest.raises(RuntimeError, match="插件索引响应格式无效"):
+            await helper.async_get_plugin_index_result(
+                "https://github.com/policy-owner/async-failed-index",
+                "v3",
+            )
+        await asyncio.sleep(0)
+        assert reports == []
+        assert registry.records == ()
+    finally:
+        loop.set_exception_handler(previous_handler)
+
+
 @pytest.mark.asyncio
 async def test_async_plugin_index_requests_are_single_flight(monkeypatch) -> None:
     """同一仓库和代际的并发索引读取只能产生一次出站请求。"""