diff --git a/app/adapters/external/plugin/client.py b/app/adapters/external/plugin/client.py index d7450bf92..c261fa1a2 100644 --- a/app/adapters/external/plugin/client.py +++ b/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]: diff --git a/tests/test_plugin_market_index_policy.py b/tests/test_plugin_market_index_policy.py index 06129a45a..03cd0b93b 100644 --- a/tests/test_plugin_market_index_policy.py +++ b/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: """同一仓库和代际的并发索引读取只能产生一次出站请求。"""