From a67285e6a2095aaf26aec4b92615a40fa4d5c327 Mon Sep 17 00:00:00 2001 From: chengyongru <2755839590@qq.com> Date: Fri, 19 Jun 2026 00:41:40 +0800 Subject: [PATCH] fix: use async CLI Apps catalog refresh Replace the manual thread-based catalog refresh with an asyncio task and async HTTP catalog fetches so the Settings route stays within the async WebUI model. --- nanobot/apps/cli/service.py | 42 +++++++++++++++++ nanobot/webui/cli_apps_api.py | 49 ++++++++------------ nanobot/webui/settings_routes.py | 11 +++-- tests/channels/test_websocket_http_routes.py | 13 +++--- tests/webui/test_cli_apps_api.py | 29 ++++++++---- 5 files changed, 96 insertions(+), 48 deletions(-) diff --git a/nanobot/apps/cli/service.py b/nanobot/apps/cli/service.py index f366a15b..ef4cb409 100644 --- a/nanobot/apps/cli/service.py +++ b/nanobot/apps/cli/service.py @@ -461,6 +461,48 @@ class CliAppManager: _write_json(cache_path, {"_cached_at": _now(), "data": fetched}) return fetched + async def _fetch_registry_async( + self, + url: str, + cache_path: Path, + *, + force_refresh: bool = False, + ) -> dict[str, Any]: + data, cached_at = self._cached_registry(cache_path) + if ( + not force_refresh + and data is not None + and _now() - cached_at < self.runtime.catalog_ttl_seconds + ): + return data + + try: + async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client: + response = await client.get(url) + response.raise_for_status() + fetched = response.json() + if not isinstance(fetched, dict): + raise ValueError("registry response must be an object") + except Exception: + if data is not None: + return data + raise + + _write_json(cache_path, {"_cached_at": _now(), "data": fetched}) + return fetched + + async def refresh_catalog_cache(self, *, force_refresh: bool = False) -> None: + for source, url, _raw_base, required in _CATALOG_SOURCES: + try: + await self._fetch_registry_async( + url, + self._cache_path(source), + force_refresh=force_refresh, + ) + except Exception: + if required: + raise + def catalog( self, *, diff --git a/nanobot/webui/cli_apps_api.py b/nanobot/webui/cli_apps_api.py index c0a4bf0a..d9eeb8cc 100644 --- a/nanobot/webui/cli_apps_api.py +++ b/nanobot/webui/cli_apps_api.py @@ -2,8 +2,8 @@ from __future__ import annotations +import asyncio import re -import threading import time from typing import Any @@ -22,37 +22,26 @@ _CLI_APP_ATTACHMENT_KEYS = ( "brand_color", ) _CATALOG_REFRESH_RETRY_SECONDS = 60.0 -_catalog_refresh_lock = threading.Lock() -_catalog_refresh_running = False +_catalog_refresh_task: asyncio.Task[None] | None = None _catalog_refresh_last_started = 0.0 -def _start_catalog_refresh() -> bool: - global _catalog_refresh_last_started, _catalog_refresh_running +async def _refresh_catalog(manager: CliAppManager) -> None: + try: + await manager.refresh_catalog_cache(force_refresh=True) + except Exception: + pass + + +def _start_catalog_refresh(manager: CliAppManager) -> bool: + global _catalog_refresh_last_started, _catalog_refresh_task now = time.monotonic() - with _catalog_refresh_lock: - if _catalog_refresh_running: - return True - if now - _catalog_refresh_last_started < _CATALOG_REFRESH_RETRY_SECONDS: - return False - _catalog_refresh_running = True - _catalog_refresh_last_started = now - - def refresh() -> None: - global _catalog_refresh_running - try: - _manager().catalog(force_refresh=True) - except Exception: - pass - finally: - with _catalog_refresh_lock: - _catalog_refresh_running = False - - threading.Thread( - target=refresh, - name="nanobot-cli-app-catalog-refresh", - daemon=True, - ).start() + if _catalog_refresh_task is not None and not _catalog_refresh_task.done(): + return True + if now - _catalog_refresh_last_started < _CATALOG_REFRESH_RETRY_SECONDS: + return False + _catalog_refresh_last_started = now + _catalog_refresh_task = asyncio.create_task(_refresh_catalog(manager)) return True @@ -108,14 +97,14 @@ def _manager() -> CliAppManager: ) -def cli_apps_payload(*, installed_only: bool = False) -> dict[str, Any]: +async def cli_apps_payload(*, installed_only: bool = False) -> dict[str, Any]: manager = _manager() if installed_only: return manager.installed_payload() payload = manager.payload(cache_only=True) refresh_pending = False if not manager.catalog_cache_fresh(include_optional=True): - refresh_pending = _start_catalog_refresh() + refresh_pending = _start_catalog_refresh(manager) if not payload["apps"]: installed = manager.installed_payload() if installed["apps"]: diff --git a/nanobot/webui/settings_routes.py b/nanobot/webui/settings_routes.py index 128871ce..3d04bfb9 100644 --- a/nanobot/webui/settings_routes.py +++ b/nanobot/webui/settings_routes.py @@ -8,6 +8,7 @@ request mapping and response shaping. from __future__ import annotations import asyncio +import inspect import json from collections.abc import Callable from typing import Any @@ -309,10 +310,12 @@ class WebUISettingsRouter: "yes", } try: - if installed_only: - payload = await asyncio.to_thread(cli_apps_payload, installed_only=True) - else: - payload = await asyncio.to_thread(cli_apps_payload) + payload_result = ( + cli_apps_payload(installed_only=True) + if installed_only + else cli_apps_payload() + ) + payload = await payload_result if inspect.isawaitable(payload_result) else payload_result except Exception: self.logger.exception("failed to load CLI Apps payload") return self._error_response(500, "failed to load CLI Apps") diff --git a/tests/channels/test_websocket_http_routes.py b/tests/channels/test_websocket_http_routes.py index b23d1097..f29ba737 100644 --- a/tests/channels/test_websocket_http_routes.py +++ b/tests/channels/test_websocket_http_routes.py @@ -5,8 +5,8 @@ import functools import json import random import socket -import threading import time +from contextlib import suppress from pathlib import Path from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -484,12 +484,13 @@ async def test_cli_apps_catalog_does_not_block_other_webui_http_routes( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: - entered = threading.Event() - release = threading.Event() + entered = asyncio.Event() + release = asyncio.Event() - def slow_payload() -> dict[str, Any]: + async def slow_payload() -> dict[str, Any]: entered.set() - release.wait(2.0) + with suppress(asyncio.TimeoutError): + await asyncio.wait_for(release.wait(), 2.0) return {"apps": [], "installed_count": 0, "catalog_updated_at": None} monkeypatch.setattr("nanobot.webui.settings_routes.cli_apps_payload", slow_payload) @@ -505,7 +506,7 @@ async def test_cli_apps_catalog_does_not_block_other_webui_http_routes( catalog_task = asyncio.create_task( _http_get("http://127.0.0.1:29935/api/settings/cli-apps", headers=auth) ) - assert await asyncio.to_thread(entered.wait, 2.0) + assert await asyncio.wait_for(entered.wait(), 2.0) assert time.perf_counter() - started < 1.0 workspaces_started = time.perf_counter() diff --git a/tests/webui/test_cli_apps_api.py b/tests/webui/test_cli_apps_api.py index 41df402d..4ab0a096 100644 --- a/tests/webui/test_cli_apps_api.py +++ b/tests/webui/test_cli_apps_api.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio from typing import Any from nanobot.webui import cli_apps_api @@ -60,9 +61,13 @@ def test_cli_apps_payload_uses_cache_and_marks_refresh_pending(monkeypatch) -> N manager = _FakeManager(fresh=False) refreshes = [] monkeypatch.setattr(cli_apps_api, "_manager", lambda: manager) - monkeypatch.setattr(cli_apps_api, "_start_catalog_refresh", lambda: refreshes.append(True) or True) + monkeypatch.setattr( + cli_apps_api, + "_start_catalog_refresh", + lambda _manager: refreshes.append(True) or True, + ) - payload = cli_apps_api.cli_apps_payload() + payload = asyncio.run(cli_apps_api.cli_apps_payload()) assert manager.payload_calls == [True] assert manager.fresh_checks == [True] @@ -95,9 +100,13 @@ def test_cli_apps_payload_skips_refresh_when_cache_is_fresh(monkeypatch) -> None ) refreshes = [] monkeypatch.setattr(cli_apps_api, "_manager", lambda: manager) - monkeypatch.setattr(cli_apps_api, "_start_catalog_refresh", lambda: refreshes.append(True) or True) + monkeypatch.setattr( + cli_apps_api, + "_start_catalog_refresh", + lambda _manager: refreshes.append(True) or True, + ) - payload = cli_apps_api.cli_apps_payload() + payload = asyncio.run(cli_apps_api.cli_apps_payload()) assert manager.payload_calls == [True] assert manager.fresh_checks == [True] @@ -131,9 +140,13 @@ def test_cli_apps_payload_refreshes_when_optional_cache_is_stale(monkeypatch) -> ) refreshes = [] monkeypatch.setattr(cli_apps_api, "_manager", lambda: manager) - monkeypatch.setattr(cli_apps_api, "_start_catalog_refresh", lambda: refreshes.append(True) or True) + monkeypatch.setattr( + cli_apps_api, + "_start_catalog_refresh", + lambda _manager: refreshes.append(True) or True, + ) - payload = cli_apps_api.cli_apps_payload() + payload = asyncio.run(cli_apps_api.cli_apps_payload()) assert manager.payload_calls == [True] assert manager.fresh_checks == [True] @@ -145,9 +158,9 @@ def test_cli_apps_payload_refreshes_when_optional_cache_is_stale(monkeypatch) -> def test_cli_apps_payload_reports_not_pending_when_refresh_is_throttled(monkeypatch) -> None: manager = _FakeManager(fresh=False) monkeypatch.setattr(cli_apps_api, "_manager", lambda: manager) - monkeypatch.setattr(cli_apps_api, "_start_catalog_refresh", lambda: False) + monkeypatch.setattr(cli_apps_api, "_start_catalog_refresh", lambda _manager: False) - payload = cli_apps_api.cli_apps_payload() + payload = asyncio.run(cli_apps_api.cli_apps_payload()) assert manager.payload_calls == [True] assert manager.fresh_checks == [True]