fix: prevent SharedPreferences deadlocks (#9649)

* fix: avoid blocking shared preference access

* docs: clarify shared preference cache roles

* docs: explain shared preference cache purpose

* docs: link cache rationale to pull request

astrbot/builtin_stars/astrbot/group_chat_context.py

@@ -92,7 +92,7 @@ class GroupChatContext:
         image_caption_prompt: str,
     ) -> str:
         if not image_caption_provider_id:
-            provider = self.context.get_using_provider()
+            provider = await self.context.get_using_provider_async()
         else:
             provider = self.context.get_provider_by_id(image_caption_provider_id)
             if not provider:

astrbot/builtin_stars/astrbot/main.py

@@ -227,7 +227,9 @@ class Main(star.Star):
                         logger.error(e)
 
             if need_active:
-                provider = self.context.get_using_provider(event.unified_msg_origin)
+                provider = await self.context.get_using_provider_async(
+                    event.unified_msg_origin
+                )
                 if not provider:
                     logger.error("未找到任何 LLM 提供商。请先配置。无法主动回复")
                     return

astrbot/builtin_stars/builtin_commands/commands/conversation.py

@@ -161,7 +161,7 @@ class ConversationCommands:
             )
             return
 
-        if not self.context.get_using_provider(umo):
+        if not await self.context.get_using_provider_async(umo):
             message.set_result(
                 MessageEventResult().message(
                     "😕 Cannot find any LLM provider. Configure one first."

astrbot/builtin_stars/builtin_commands/commands/provider.py

@@ -149,7 +149,7 @@ class ProviderCommands:
                 ),
             )
 
-            provider_using = self.context.get_using_provider(umo=umo)
+            provider_using = await self.context.get_using_provider_async(umo=umo)
             for i, d in enumerate(llm_data):
                 line = f"{i + 1}. {d['info']}{d['mark']}"
                 if (
@@ -161,7 +161,7 @@ class ProviderCommands:
 
             if tts_data:
                 parts.append("\n## TTS Providers\n")
-                tts_using = self.context.get_using_tts_provider(umo=umo)
+                tts_using = await self.context.get_using_tts_provider_async(umo=umo)
                 for i, d in enumerate(tts_data):
                     line = f"{i + 1}. {d['info']}{d['mark']}"
                     if tts_using and tts_using.meta().id == d["provider"].meta().id:
@@ -170,7 +170,7 @@ class ProviderCommands:
 
             if stt_data:
                 parts.append("\n## STT Providers\n")
-                stt_using = self.context.get_using_stt_provider(umo=umo)
+                stt_using = await self.context.get_using_stt_provider_async(umo=umo)
                 for i, d in enumerate(stt_data):
                     line = f"{i + 1}. {d['info']}{d['mark']}"
                     if stt_using and stt_using.meta().id == d["provider"].meta().id:

astrbot/core/astr_main_agent.py

@@ -227,10 +227,18 @@ def _set_llm_error_message(event: AstrMessageEvent, message: str) -> None:
     event.set_extra(LLM_ERROR_MESSAGE_EXTRA_KEY, message)
 
 
-def _select_provider(
+async def _select_provider(
     event: AstrMessageEvent, plugin_context: Context
 ) -> Provider | None:
-    """Select chat provider for the event."""
+    """Select the chat provider for an event.
+
+    Args:
+        event: Message event that may contain an explicit provider selection.
+        plugin_context: Plugin context used to resolve configured providers.
+
+    Returns:
+        Selected chat provider, or None if selection fails.
+    """
     sel_provider = event.get_extra("selected_provider")
     if sel_provider and isinstance(sel_provider, str):
         provider = plugin_context.get_provider_by_id(sel_provider)
@@ -252,7 +260,9 @@ def _select_provider(
             return None
         return provider
     try:
-        return plugin_context.get_using_provider(umo=event.unified_msg_origin)
+        return await plugin_context.get_using_provider_async(
+            umo=event.unified_msg_origin
+        )
     except ValueError as exc:
         logger.error("Error occurred while selecting provider: %s", exc)
         _set_llm_error_message(event, f"LLM 请求失败：{exc}")
@@ -916,7 +926,9 @@ async def _process_quote_message(
                 compress_path = None
                 prov = plugin_context.get_provider_by_id(img_cap_prov_id)
                 if prov is None:
-                    prov = plugin_context.get_using_provider(event.unified_msg_origin)
+                    prov = await plugin_context.get_using_provider_async(
+                        event.unified_msg_origin
+                    )
 
                 if prov and isinstance(prov, Provider):
                     path = await image_seg.convert_to_file_path()
@@ -1292,11 +1304,21 @@ def _apply_web_search_citation_prompt(
     req.system_prompt = f"{system_prompt}\n{WEB_SEARCH_CITATION_PROMPT}\n"
 
 
-def _get_compress_provider(
+async def _get_compress_provider(
     config: MainAgentBuildConfig,
     plugin_context: Context,
     event: AstrMessageEvent | None = None,
 ) -> Provider | None:
+    """Resolve the provider used for context compression.
+
+    Args:
+        config: Main agent build configuration.
+        plugin_context: Plugin context used to resolve providers.
+        event: Optional event used for session-specific fallback selection.
+
+    Returns:
+        Compression provider, or None if compression is disabled or unavailable.
+    """
     if config.context_limit_reached_strategy != "llm_compress":
         return None
     if config.llm_compress_provider_id:
@@ -1310,7 +1332,9 @@ def _get_compress_provider(
     # fallback: use current chat provider for this session
     if event:
         try:
-            return plugin_context.get_using_provider(umo=event.unified_msg_origin)
+            return await plugin_context.get_using_provider_async(
+                umo=event.unified_msg_origin
+            )
         except ValueError:
             pass
     return None
@@ -1398,7 +1422,7 @@ async def build_main_agent(
 
     If apply_reset is False, will not call reset on the agent runner.
     """
-    provider = provider or _select_provider(event, plugin_context)
+    provider = provider or await _select_provider(event, plugin_context)
     if provider is None:
         logger.info("未找到任何对话模型（提供商），跳过 LLM 请求处理。")
         if not event.get_extra(LLM_ERROR_MESSAGE_EXTRA_KEY):
@@ -1699,7 +1723,11 @@ async def build_main_agent(
         streaming=config.streaming_response,
         llm_compress_instruction=config.llm_compress_instruction,
         llm_compress_keep_recent_ratio=config.llm_compress_keep_recent_ratio,
-        llm_compress_provider=_get_compress_provider(config, plugin_context, event),
+        llm_compress_provider=await _get_compress_provider(
+            config,
+            plugin_context,
+            event,
+        ),
         truncate_turns=config.dequeue_context_length,
         enforce_max_turns=config.max_context_length,
         tool_schema_mode=config.tool_schema_mode,

astrbot/core/core_lifecycle.py

@@ -172,6 +172,8 @@ class AstrBotCoreLifecycle:
             LogManager.configure_trace_logger(self.astrbot_config)
 
         await self.db.initialize()
+        if sp.db_helper is self.db:
+            await sp.initialize()
 
         await html_renderer.initialize()
 
@@ -404,6 +406,8 @@ class AstrBotCoreLifecycle:
         await self.provider_manager.terminate()
         await self.platform_manager.terminate()
         await self.kb_manager.terminate()
+        if sp.db_helper is self.db:
+            await sp.close()
         self.dashboard_shutdown_event.set()
 
         # 再次遍历curr_tasks等待每个任务真正结束
@@ -427,6 +431,8 @@ class AstrBotCoreLifecycle:
         await self.provider_manager.terminate()
         await self.platform_manager.terminate()
         await self.kb_manager.terminate()
+        if sp.db_helper is self.db:
+            await sp.close()
         self.dashboard_shutdown_event.set()
         threading.Thread(
             target=restart_process,

astrbot/core/db/__init__.py

@@ -577,11 +577,11 @@ class BaseDatabase(abc.ABC):
     @abc.abstractmethod
     async def get_preferences(
         self,
-        scope: str,
+        scope: str | None = None,
         scope_id: str | None = None,
         key: str | None = None,
     ) -> list[Preference]:
-        """Get all preferences for a specific scope ID or key."""
+        """Get preferences, optionally filtered by scope, scope ID, or key."""
         ...
 
     @abc.abstractmethod

astrbot/core/db/sqlite.py

@@ -1379,11 +1379,13 @@ class SQLiteDatabase(BaseDatabase):
             result = await session.execute(query)
             return result.scalar_one_or_none()
 
-    async def get_preferences(self, scope, scope_id=None, key=None):
-        """Get all preferences for a specific scope ID or key."""
+    async def get_preferences(self, scope=None, scope_id=None, key=None):
+        """Get preferences, optionally filtered by scope, scope ID, or key."""
         async with self.get_db() as session:
             session: AsyncSession
-            query = select(Preference).where(Preference.scope == scope)
+            query = select(Preference)
+            if scope is not None:
+                query = query.where(Preference.scope == scope)
             if scope_id is not None:
                 query = query.where(Preference.scope_id == scope_id)
             if key is not None:

astrbot/core/pipeline/preprocess_stage/stage.py

@@ -169,7 +169,9 @@ class PreProcessStage(Stage):
         if self.stt_settings.get("enable", False):
             # TODO: 独立
             ctx = self.plugin_manager.context
-            stt_provider = ctx.get_using_stt_provider(event.unified_msg_origin)
+            stt_provider = await ctx.get_using_stt_provider_async(
+                event.unified_msg_origin
+            )
             if not stt_provider:
                 logger.warning(
                     f"Session {event.unified_msg_origin} has no speech-to-text "

astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py

@@ -298,10 +298,8 @@ class InternalAgentSubStage(Stage):
                         )
 
                         # 获取 TTS Provider
-                        tts_provider = (
-                            self.ctx.plugin_manager.context.get_using_tts_provider(
-                                event.unified_msg_origin
-                            )
+                        tts_provider = await self.ctx.plugin_manager.context.get_using_tts_provider_async(
+                            event.unified_msg_origin
                         )
 
                         if not tts_provider:

astrbot/core/pipeline/result_decorate/stage.py

@@ -267,8 +267,10 @@ class ResultDecorateStage(Stage):
                     result.chain = new_chain
 
             # TTS
-            tts_provider = self.ctx.plugin_manager.context.get_using_tts_provider(
-                event.unified_msg_origin,
+            tts_provider = (
+                await self.ctx.plugin_manager.context.get_using_tts_provider_async(
+                    event.unified_msg_origin,
+                )
             )
 
             should_tts = (

astrbot/core/provider/func_tool_manager.py

@@ -983,6 +983,7 @@ class FunctionToolManager:
         toolset = ToolSet(tools)
         return toolset.google_schema()
 
+    @deprecated(reason="Use deactivate_llm_tool_async() instead.")
     def deactivate_llm_tool(self, name: str) -> bool:
         """停用一个已经注册的函数调用工具。
 
@@ -1012,7 +1013,39 @@ class FunctionToolManager:
             return True
         return False
 
+    async def deactivate_llm_tool_async(self, name: str) -> bool:
+        """Asynchronously deactivate a registered function-calling tool.
+
+        Args:
+            name: Tool name.
+
+        Returns:
+            True when the tool was deactivated, or False when it was not found.
+        """
+        func_tool = self.get_func(name)
+        if func_tool is not None:
+            func_tool.active = False
+
+            inactivated_llm_tools: list = await sp.get_async(
+                "global",
+                "global",
+                "inactivated_llm_tools",
+                [],
+            )
+            if name not in inactivated_llm_tools:
+                inactivated_llm_tools.append(name)
+                await sp.put_async(
+                    "global",
+                    "global",
+                    "inactivated_llm_tools",
+                    inactivated_llm_tools,
+                )
+
+            return True
+        return False
+
     # 因为不想解决循环引用，所以这里直接传入 star_map 先了...
+    @deprecated(reason="Use activate_llm_tool_async() instead.")
     def activate_llm_tool(self, name: str, star_map: dict) -> bool:
         func_tool = self.get_func(name)
         if func_tool is not None:
@@ -1042,6 +1075,47 @@ class FunctionToolManager:
             return True
         return False
 
+    async def activate_llm_tool_async(self, name: str, star_map: dict) -> bool:
+        """Asynchronously activate a registered function-calling tool.
+
+        Args:
+            name: Tool name.
+            star_map: Loaded plugins indexed by module path.
+
+        Returns:
+            True when the tool was activated, or False when it was not found.
+
+        Raises:
+            ValueError: If the plugin that owns the tool is disabled.
+        """
+        func_tool = self.get_func(name)
+        if func_tool is not None:
+            if func_tool.handler_module_path in star_map:
+                if not star_map[func_tool.handler_module_path].activated:
+                    raise ValueError(
+                        f"此函数调用工具所属的插件 {star_map[func_tool.handler_module_path].name} 已被禁用，请先在管理面板启用再激活此工具。",
+                    )
+
+            func_tool.active = True
+
+            inactivated_llm_tools: list = await sp.get_async(
+                "global",
+                "global",
+                "inactivated_llm_tools",
+                [],
+            )
+            if name in inactivated_llm_tools:
+                inactivated_llm_tools.remove(name)
+                await sp.put_async(
+                    "global",
+                    "global",
+                    "inactivated_llm_tools",
+                    inactivated_llm_tools,
+                )
+
+            return True
+        return False
+
     @property
     def mcp_config_path(self):
         data_dir = get_astrbot_data_path()

astrbot/core/provider/manager.py

@@ -215,30 +215,27 @@ class ProviderManager:
         """根据提供商 ID 获取提供商实例"""
         return self.inst_map.get(provider_id)
 
-    def get_using_provider(
-        self, provider_type: ProviderType, umo=None
+    def _resolve_using_provider(
+        self,
+        provider_type: ProviderType,
+        umo: str | None,
+        provider_id: str | None,
     ) -> Providers | None:
-        """获取正在使用的提供商实例。
+        """Resolve a provider preference with configuration fallbacks.
 
         Args:
-            provider_type (ProviderType): 提供商类型。
-            umo (str, optional): 用户会话 ID，用于提供商会话隔离。
+            provider_type: Provider type to resolve.
+            umo: User message origin used to load session configuration.
+            provider_id: Preferred provider ID, if one is configured.
 
         Returns:
-            Provider: 正在使用的提供商实例。
+            Resolved provider instance, or None when the provider type is disabled
+            or no provider is available.
 
+        Raises:
+            ValueError: If provider_type is unsupported.
         """
-        provider = None
-        provider_id = None
-        if umo:
-            provider_id = sp.get(
-                f"provider_perf_{provider_type.value}",
-                None,
-                scope="umo",
-                scope_id=umo,
-            )
-            if provider_id:
-                provider = self.inst_map.get(provider_id)
+        provider = self.inst_map.get(provider_id) if provider_id else None
         if not provider:
             # default setting
             config = self.acm.get_conf(umo)
@@ -280,6 +277,55 @@ class ProviderManager:
 
         return provider
 
+    @deprecated(reason="Use get_using_provider_async() instead.")
+    def get_using_provider(
+        self,
+        provider_type: ProviderType,
+        umo: str | None = None,
+    ) -> Providers | None:
+        """获取正在使用的提供商实例。
+
+        Args:
+            provider_type: 提供商类型。
+            umo: 用户会话 ID，用于提供商会话隔离。
+
+        Returns:
+            正在使用的提供商实例。
+        """
+        provider_id = None
+        if umo:
+            provider_id = sp.get(
+                f"provider_perf_{provider_type.value}",
+                None,
+                scope="umo",
+                scope_id=umo,
+            )
+        return self._resolve_using_provider(provider_type, umo, provider_id)
+
+    async def get_using_provider_async(
+        self,
+        provider_type: ProviderType,
+        umo: str | None = None,
+    ) -> Providers | None:
+        """Asynchronously get the provider currently in use.
+
+        Args:
+            provider_type: Provider type to resolve.
+            umo: User message origin used for session-specific preferences.
+
+        Returns:
+            Provider instance currently in use, or None if unavailable.
+        """
+        provider_id = None
+        if umo:
+            provider_id = await sp.get_async(
+                "umo",
+                umo,
+                f"provider_perf_{provider_type.value}",
+                None,
+            )
+        return self._resolve_using_provider(provider_type, umo, provider_id)
+
     async def initialize(self) -> None:
         # 逐个初始化提供商
         for provider_config in self.providers_config:

astrbot/core/star/context.py

@@ -338,7 +338,7 @@ class Context:
         Raises:
             ProviderNotFoundError: 未找到。
         """
-        prov = self.get_using_provider(umo)
+        prov = await self.get_using_provider_async(umo)
         if not prov:
             raise ProviderNotFoundError("Provider not found")
         return prov.meta().id
@@ -357,6 +357,7 @@ class Context:
         """获取 LLM Tool Manager，其用于管理注册的所有的 Function-calling tools"""
         return self.provider_manager.llm_tools
 
+    @deprecated(reason="Use activate_llm_tool_async() instead.")
     def activate_llm_tool(self, name: str) -> bool:
         """激活一个已经注册的函数调用工具。
 
@@ -371,6 +372,24 @@ class Context:
         """
         return self.provider_manager.llm_tools.activate_llm_tool(name, star_map)
 
+    async def activate_llm_tool_async(self, name: str) -> bool:
+        """Asynchronously activate a registered function-calling tool.
+
+        Args:
+            name: Tool name.
+
+        Returns:
+            True when the tool was activated, or False when it was not found.
+
+        Note:
+            Registered tools are active by default.
+        """
+        return await self.provider_manager.llm_tools.activate_llm_tool_async(
+            name,
+            star_map,
+        )
+
+    @deprecated(reason="Use deactivate_llm_tool_async() instead.")
     def deactivate_llm_tool(self, name: str) -> bool:
         """停用一个已经注册的函数调用工具。
 
@@ -382,6 +401,17 @@ class Context:
         """
         return self.provider_manager.llm_tools.deactivate_llm_tool(name)
 
+    async def deactivate_llm_tool_async(self, name: str) -> bool:
+        """Asynchronously deactivate a registered function-calling tool.
+
+        Args:
+            name: Tool name.
+
+        Returns:
+            True when the tool was deactivated, or False when it was not found.
+        """
+        return await self.provider_manager.llm_tools.deactivate_llm_tool_async(name)
+
     def get_provider_by_id(
         self,
         provider_id: str,
@@ -423,6 +453,7 @@ class Context:
         """获取所有用于 Embedding 任务的 Provider。"""
         return self.provider_manager.embedding_provider_insts
 
+    @deprecated(reason="Use get_using_provider_async() instead.")
     def get_using_provider(self, umo: str | None = None) -> Provider | None:
         """获取当前使用的用于文本生成任务的 LLM Provider(Chat_Completion 类型)。
 
@@ -448,6 +479,34 @@ class Context:
             )
         return prov
 
+    async def get_using_provider_async(
+        self,
+        umo: str | None = None,
+    ) -> Provider | None:
+        """Asynchronously get the current text-generation provider.
+
+        Args:
+            umo: Unified message origin used for session-specific preferences.
+
+        Returns:
+            Current chat provider, or None if no provider is available.
+
+        Raises:
+            ValueError: If the resolved provider is not a chat provider.
+        """
+        prov = await self.provider_manager.get_using_provider_async(
+            provider_type=ProviderType.CHAT_COMPLETION,
+            umo=umo,
+        )
+        if prov is None:
+            return None
+        if not isinstance(prov, Provider):
+            raise ValueError(
+                f"该会话来源的对话模型（提供商）的类型不正确: {type(prov)}"
+            )
+        return prov
+
+    @deprecated(reason="Use get_using_tts_provider_async() instead.")
     def get_using_tts_provider(self, umo: str | None = None) -> TTSProvider | None:
         """获取当前使用的用于 TTS 任务的 Provider。
 
@@ -468,6 +527,30 @@ class Context:
             raise ValueError("返回的 Provider 不是 TTSProvider 类型")
         return prov
 
+    async def get_using_tts_provider_async(
+        self,
+        umo: str | None = None,
+    ) -> TTSProvider | None:
+        """Asynchronously get the current text-to-speech provider.
+
+        Args:
+            umo: Unified message origin used for session-specific preferences.
+
+        Returns:
+            Current TTS provider, or None if no provider is available.
+
+        Raises:
+            ValueError: If the resolved provider is not a TTS provider.
+        """
+        prov = await self.provider_manager.get_using_provider_async(
+            provider_type=ProviderType.TEXT_TO_SPEECH,
+            umo=umo,
+        )
+        if prov and not isinstance(prov, TTSProvider):
+            raise ValueError("返回的 Provider 不是 TTSProvider 类型")
+        return prov
+
+    @deprecated(reason="Use get_using_stt_provider_async() instead.")
     def get_using_stt_provider(self, umo: str | None = None) -> STTProvider | None:
         """获取当前使用的用于 STT 任务的 Provider。
 
@@ -488,6 +571,29 @@ class Context:
             raise ValueError("返回的 Provider 不是 STTProvider 类型")
         return prov
 
+    async def get_using_stt_provider_async(
+        self,
+        umo: str | None = None,
+    ) -> STTProvider | None:
+        """Asynchronously get the current speech-to-text provider.
+
+        Args:
+            umo: Unified message origin used for session-specific preferences.
+
+        Returns:
+            Current STT provider, or None if no provider is available.
+
+        Raises:
+            ValueError: If the resolved provider is not an STT provider.
+        """
+        prov = await self.provider_manager.get_using_provider_async(
+            provider_type=ProviderType.SPEECH_TO_TEXT,
+            umo=umo,
+        )
+        if prov and not isinstance(prov, STTProvider):
+            raise ValueError("返回的 Provider 不是 STTProvider 类型")
+        return prov
+
     def get_config(self, umo: str | None = None) -> AstrBotConfig:
         """获取 AstrBot 的配置。
 

astrbot/core/star/star_manager.py

@@ -1574,7 +1574,7 @@ class PluginManager:
 
             if plugin_id:
                 try:
-                    await self.context.get_db().clear_preferences("plugin", plugin_id)
+                    await sp.clear_async("plugin", plugin_id)
                     logger.info(
                         f"Cleared KV data for plugin {plugin_label} ({plugin_id})"
                     )

astrbot/core/star/star_tools.py

@@ -5,6 +5,8 @@ from collections.abc import Awaitable, Callable
 from pathlib import Path
 from typing import Any, ClassVar
 
+from deprecated import deprecated
+
 from astrbot.api.platform import AstrBotMessage, MessageMember, MessageType
 from astrbot.core.message.components import BaseMessageComponent
 from astrbot.core.message.message_event_result import MessageChain
@@ -132,6 +134,7 @@ class StarTools:
         adapter.commit_event(event)
 
     @classmethod
+    @deprecated(reason="Use activate_llm_tool_async() instead.")
     def activate_llm_tool(cls, name: str) -> bool:
         """Activates a registered function-calling tool.
 
@@ -149,6 +152,24 @@ class StarTools:
         return cls._context.activate_llm_tool(name)
 
     @classmethod
+    async def activate_llm_tool_async(cls, name: str) -> bool:
+        """Asynchronously activates a registered function-calling tool.
+
+        Args:
+            name: Tool name.
+
+        Returns:
+            Whether the tool was activated successfully.
+
+        Raises:
+            ValueError: If StarTools is not initialized.
+        """
+        if cls._context is None:
+            raise ValueError("StarTools not initialized")
+        return await cls._context.activate_llm_tool_async(name)
+
+    @classmethod
+    @deprecated(reason="Use deactivate_llm_tool_async() instead.")
     def deactivate_llm_tool(cls, name: str) -> bool:
         """Deactivates a registered function-calling tool.
 
@@ -165,6 +186,23 @@ class StarTools:
             raise ValueError("StarTools not initialized")
         return cls._context.deactivate_llm_tool(name)
 
+    @classmethod
+    async def deactivate_llm_tool_async(cls, name: str) -> bool:
+        """Asynchronously deactivates a registered function-calling tool.
+
+        Args:
+            name: Tool name.
+
+        Returns:
+            Whether the tool was deactivated successfully.
+
+        Raises:
+            ValueError: If StarTools is not initialized.
+        """
+        if cls._context is None:
+            raise ValueError("StarTools not initialized")
+        return await cls._context.deactivate_llm_tool_async(name)
+
     @classmethod
     def register_llm_tool(
         cls,

astrbot/core/utils/shared_preferences.py

@@ -2,17 +2,28 @@ import asyncio
 import os
 import threading
 from collections import defaultdict
+from copy import deepcopy
 from typing import Any, TypeVar, overload
 
 from apscheduler.schedulers.background import BackgroundScheduler
 from deprecated import deprecated
 
+from astrbot import logger
 from astrbot.core.db import BaseDatabase
 from astrbot.core.db.po import Preference
 
 from .astrbot_path import get_astrbot_data_path
 
 _VT = TypeVar("_VT")
+_MISSING = object()
+_WriteOperation = tuple[
+    str,
+    str,
+    str,
+    str | None,
+    Any,
+    asyncio.Future[None] | None,
+]
 
 
 class SharedPreferences:
@@ -27,9 +38,22 @@ class SharedPreferences:
         self.temporary_cache: dict[str, dict[str, Any]] = defaultdict(dict)
         """automatically clear per 24 hours. Might be helpful in some cases XD"""
 
-        self._sync_loop = asyncio.new_event_loop()
-        t = threading.Thread(target=self._sync_loop.run_forever, daemon=True)
-        t.start()
+        # In-memory mirror of persistent preferences. It lets synchronous APIs
+        # read and update values without blocking on the async database, provides
+        # immediate read-after-write visibility, and also serves async point reads.
+        # Unlike temporary_cache, it is preloaded at startup, persisted to the
+        # database, and never periodically cleared by the scheduler.
+        # See https://github.com/AstrBotDevs/AstrBot/pull/9649 for the deadlock
+        # scenario and design rationale.
+        self._cache: dict[tuple[str, str, str], Any] = {}
+        self._cache_lock = threading.RLock()
+        self._cache_initialized = False
+        self._initializing = False
+        self._initialize_lock = asyncio.Lock()
+        self._loop: asyncio.AbstractEventLoop | None = None
+        self._write_queue: asyncio.Queue[_WriteOperation] | None = None
+        self._writer_task: asyncio.Task[None] | None = None
+        self._pending_writes: list[_WriteOperation] = []
 
         self._scheduler = BackgroundScheduler()
         self._scheduler.add_job(
@@ -40,6 +64,206 @@ class SharedPreferences:
     def _clear_temporary_cache(self) -> None:
         self.temporary_cache.clear()
 
+    def _apply_cache_operation(self, operation: _WriteOperation) -> None:
+        """Apply one preference mutation to the in-memory cache.
+
+        Args:
+            operation: Queued write operation to reflect in memory.
+        """
+        action, scope, scope_id, key, value, _ = operation
+        with self._cache_lock:
+            if action == "put" and key is not None:
+                self._cache[(scope, scope_id, key)] = deepcopy(value)
+            elif action == "remove" and key is not None:
+                self._cache.pop((scope, scope_id, key), None)
+            elif action == "clear":
+                keys = [
+                    cache_key
+                    for cache_key in self._cache
+                    if cache_key[0] == scope and cache_key[1] == scope_id
+                ]
+                for cache_key in keys:
+                    self._cache.pop(cache_key, None)
+
+    def _schedule_write(self, operation: _WriteOperation) -> None:
+        """Schedule a preference write on the owning event loop.
+
+        Args:
+            operation: Preference mutation to persist.
+        """
+        loop = self._loop
+        queue = self._write_queue
+        if loop is None or queue is None or not loop.is_running() or self._initializing:
+            with self._cache_lock:
+                self._pending_writes.append(operation)
+            return
+
+        def enqueue() -> None:
+            queue.put_nowait(operation)
+            if self._writer_task is None or self._writer_task.done():
+                self._writer_task = loop.create_task(
+                    self._drain_write_queue(),
+                    name="shared_preferences_writer",
+                )
+
+        try:
+            running_loop = asyncio.get_running_loop()
+        except RuntimeError:
+            running_loop = None
+        if running_loop is loop:
+            enqueue()
+        else:
+            try:
+                loop.call_soon_threadsafe(enqueue)
+            except RuntimeError:
+                with self._cache_lock:
+                    self._pending_writes.append(operation)
+
+    def _submit_write(self, operation: _WriteOperation) -> None:
+        """Update the cache and schedule persistence in the same order.
+
+        Args:
+            operation: Preference mutation to apply and persist.
+        """
+        with self._cache_lock:
+            self._apply_cache_operation(operation)
+            self._schedule_write(operation)
+
+    async def _drain_write_queue(self) -> None:
+        """Persist queued preference mutations in FIFO order."""
+        queue = self._write_queue
+        if queue is None:
+            return
+        while True:
+            try:
+                operation = queue.get_nowait()
+            except asyncio.QueueEmpty:
+                return
+
+            action, scope, scope_id, key, value, completion = operation
+            try:
+                if action == "put" and key is not None:
+                    await self.db_helper.insert_preference_or_update(
+                        scope,
+                        scope_id,
+                        key,
+                        {"val": value},
+                    )
+                elif action == "remove" and key is not None:
+                    await self.db_helper.remove_preference(scope, scope_id, key)
+                elif action == "clear":
+                    await self.db_helper.clear_preferences(scope, scope_id)
+                else:
+                    raise ValueError(f"Unknown preference write operation: {action}")
+            except Exception as exc:
+                logger.error(
+                    "Failed to persist shared preference operation %s for %s/%s: %s",
+                    action,
+                    scope,
+                    scope_id,
+                    exc,
+                    exc_info=True,
+                )
+                if completion is not None and not completion.done():
+                    completion.set_exception(exc)
+            else:
+                if completion is not None and not completion.done():
+                    completion.set_result(None)
+            finally:
+                queue.task_done()
+
+    async def initialize(self) -> None:
+        """Load persisted preferences and bind writes to the current event loop.
+
+        Raises:
+            RuntimeError: If another running event loop already owns the store.
+        """
+        loop = asyncio.get_running_loop()
+        async with self._initialize_lock:
+            if self._loop is loop and self._cache_initialized:
+                return
+            with self._cache_lock:
+                self._initializing = True
+            try:
+                if self._loop is not None and self._loop is not loop:
+                    if self._loop.is_running():
+                        raise RuntimeError(
+                            "SharedPreferences is already bound to another running "
+                            "event loop."
+                        )
+                    old_queue = self._write_queue
+                    if old_queue is not None:
+                        with self._cache_lock:
+                            while True:
+                                try:
+                                    operation = old_queue.get_nowait()
+                                    self._pending_writes.append((*operation[:-1], None))
+                                    old_queue.task_done()
+                                except asyncio.QueueEmpty:
+                                    break
+
+                self._loop = loop
+                self._write_queue = asyncio.Queue()
+                self._writer_task = None
+                if not self._cache_initialized:
+                    preferences = await self.db_helper.get_preferences()
+                    loaded_cache = {
+                        (item.scope, item.scope_id, item.key): deepcopy(
+                            item.value["val"]
+                        )
+                        for item in preferences
+                    }
+
+                with self._cache_lock:
+                    pending_writes = list(self._pending_writes)
+                    self._pending_writes.clear()
+                    if not self._cache_initialized:
+                        self._cache = loaded_cache
+                        for operation in pending_writes:
+                            self._apply_cache_operation(operation)
+                        self._cache_initialized = True
+                    self._initializing = False
+            except BaseException:
+                with self._cache_lock:
+                    self._initializing = False
+                raise
+
+            for operation in pending_writes:
+                self._schedule_write(operation)
+            await self.flush()
+
+    async def flush(self) -> None:
+        """Wait until all queued synchronous preference writes are persisted.
+
+        Raises:
+            RuntimeError: If called from a different running event loop.
+        """
+        if self._loop is None or self._write_queue is None:
+            with self._cache_lock:
+                has_pending_writes = bool(self._pending_writes)
+            if has_pending_writes:
+                await self.initialize()
+            return
+        loop = asyncio.get_running_loop()
+        if loop is not self._loop:
+            if self._loop.is_running():
+                raise RuntimeError(
+                    "SharedPreferences writes must be flushed on their owning "
+                    "event loop."
+                )
+            await self.initialize()
+            return
+        await asyncio.sleep(0)
+        await self._write_queue.join()
+        if self._writer_task is not None:
+            await self._writer_task
+
+    async def close(self) -> None:
+        """Flush pending writes and stop the temporary-cache scheduler."""
+        await self.flush()
+        if self._scheduler.running:
+            self._scheduler.shutdown(wait=False)
+
     async def get_async(
         self,
         scope: str,
@@ -48,13 +272,12 @@ class SharedPreferences:
         default: _VT = None,
     ) -> _VT:
         """获取指定范围和键的偏好设置"""
-        if scope_id is not None and key is not None:
-            result = await self.db_helper.get_preference(scope, scope_id, key)
-            if result:
-                ret = result.value["val"]
-            else:
-                ret = default
-            return ret
+        await self.initialize()
+        if scope_id is None or key is None:
+            return default
+        with self._cache_lock:
+            value = self._cache.get((scope, scope_id, key), _MISSING)
+            return default if value is _MISSING else deepcopy(value)
 
     async def range_get_async(
         self,
@@ -65,6 +288,8 @@ class SharedPreferences:
         """获取指定范围的偏好设置
         Note: 返回 Preference 列表，其中的 value 属性是一个 dict，value["val"] 为值。scope_id 和 key 可以为 None，这时返回该范围下所有的偏好设置。
         """
+        await self.initialize()
+        await self.flush()
         ret = await self.db_helper.get_preferences(scope, scope_id, key)
         return ret
 
@@ -135,12 +360,18 @@ class SharedPreferences:
 
     async def put_async(self, scope: str, scope_id: str, key: str, value: Any) -> None:
         """设置指定范围和键的偏好设置"""
-        await self.db_helper.insert_preference_or_update(
+        await self.initialize()
+        completion = asyncio.get_running_loop().create_future()
+        operation: _WriteOperation = (
+            "put",
             scope,
             scope_id,
             key,
-            {"val": value},
+            deepcopy(value),
+            completion,
         )
+        self._submit_write(operation)
+        await completion
 
     async def session_put(self, umo: str, key: str, value: Any) -> None:
         await self.put_async("umo", umo, key, value)
@@ -150,7 +381,18 @@ class SharedPreferences:
 
     async def remove_async(self, scope: str, scope_id: str, key: str) -> None:
         """删除指定范围和键的偏好设置"""
-        await self.db_helper.remove_preference(scope, scope_id, key)
+        await self.initialize()
+        completion = asyncio.get_running_loop().create_future()
+        operation: _WriteOperation = (
+            "remove",
+            scope,
+            scope_id,
+            key,
+            None,
+            completion,
+        )
+        self._submit_write(operation)
+        await completion
 
     async def session_remove(self, umo: str, key: str) -> None:
         await self.remove_async("umo", umo, key)
@@ -161,7 +403,18 @@ class SharedPreferences:
 
     async def clear_async(self, scope: str, scope_id: str) -> None:
         """清空指定范围的所有偏好设置"""
-        await self.db_helper.clear_preferences(scope, scope_id)
+        await self.initialize()
+        completion = asyncio.get_running_loop().create_future()
+        operation: _WriteOperation = (
+            "clear",
+            scope,
+            scope_id,
+            None,
+            None,
+            completion,
+        )
+        self._submit_write(operation)
+        await completion
 
     # ====
     # DEPRECATED METHODS
@@ -186,12 +439,12 @@ class SharedPreferences:
             raise ValueError(
                 "scope_id and key cannot be None when getting a specific preference.",
             )
-        result = asyncio.run_coroutine_threadsafe(
-            self.get_async(scope or "unknown", scope_id or "unknown", key, default),
-            self._sync_loop,
-        ).result()
-
-        return result if result is not None else default
+        with self._cache_lock:
+            value = self._cache.get(
+                (scope or "unknown", scope_id or "unknown", key),
+                _MISSING,
+            )
+            return default if value is _MISSING or value is None else deepcopy(value)
 
     @deprecated(version="4.0.0", reason="Use range_get_async() instead.")
     def range_get(
@@ -201,12 +454,27 @@ class SharedPreferences:
         key: str | None = None,
     ) -> list[Preference]:
         """获取指定范围的偏好设置（已弃用）"""
-        result = asyncio.run_coroutine_threadsafe(
-            self.range_get_async(scope, scope_id, key),
-            self._sync_loop,
-        ).result()
-
-        return result
+        with self._cache_lock:
+            values = [
+                (cache_scope, cache_scope_id, cache_key, deepcopy(value))
+                for (
+                    cache_scope,
+                    cache_scope_id,
+                    cache_key,
+                ), value in self._cache.items()
+                if cache_scope == scope
+                and (scope_id is None or cache_scope_id == scope_id)
+                and (key is None or cache_key == key)
+            ]
+        return [
+            Preference(
+                scope=cache_scope,
+                scope_id=cache_scope_id,
+                key=cache_key,
+                value={"val": value},
+            )
+            for cache_scope, cache_scope_id, cache_key, value in values
+        ]
 
     @deprecated(
         version="4.0.0",
@@ -216,10 +484,15 @@ class SharedPreferences:
         self, key, value, scope: str | None = None, scope_id: str | None = None
     ) -> None:
         """设置偏好设置（已弃用）"""
-        asyncio.run_coroutine_threadsafe(
-            self.put_async(scope or "unknown", scope_id or "unknown", key, value),
-            self._sync_loop,
-        ).result()
+        operation: _WriteOperation = (
+            "put",
+            scope or "unknown",
+            scope_id or "unknown",
+            key,
+            deepcopy(value),
+            None,
+        )
+        self._submit_write(operation)
 
     @deprecated(
         version="4.0.0",
@@ -229,15 +502,25 @@ class SharedPreferences:
         self, key, scope: str | None = None, scope_id: str | None = None
     ) -> None:
         """删除偏好设置（已弃用）"""
-        asyncio.run_coroutine_threadsafe(
-            self.remove_async(scope or "unknown", scope_id or "unknown", key),
-            self._sync_loop,
-        ).result()
+        operation: _WriteOperation = (
+            "remove",
+            scope or "unknown",
+            scope_id or "unknown",
+            key,
+            None,
+            None,
+        )
+        self._submit_write(operation)
 
     @deprecated(version="4.0.0", reason="Use clear_async() instead.")
     def clear(self, scope: str | None = None, scope_id: str | None = None) -> None:
         """清空偏好设置（已弃用）"""
-        asyncio.run_coroutine_threadsafe(
-            self.clear_async(scope or "unknown", scope_id or "unknown"),
-            self._sync_loop,
-        ).result()
+        operation: _WriteOperation = (
+            "clear",
+            scope or "unknown",
+            scope_id or "unknown",
+            None,
+            None,
+            None,
+        )
+        self._submit_write(operation)

astrbot/dashboard/services/tools_service.py

@@ -320,7 +320,18 @@ class ToolsService:
                 f"Failed to update tool permission: {exc!s}"
             ) from exc
 
-    def toggle_tool(self, data: Any) -> str:
+    async def toggle_tool(self, data: Any) -> str:
+        """Toggle a tool and wait for its preference change to persist.
+
+        Args:
+            data: Mapping containing the tool name and activation state.
+
+        Returns:
+            Operation result message.
+
+        Raises:
+            ToolsServiceError: If validation or the tool operation fails.
+        """
         try:
             tool_name = data.get("name")
             action = data.get("activate")
@@ -335,13 +346,16 @@ class ToolsService:
 
             if action:
                 try:
-                    ok = self.tool_mgr.activate_llm_tool(tool_name, star_map=star_map)
+                    ok = await self.tool_mgr.activate_llm_tool_async(
+                        tool_name,
+                        star_map=star_map,
+                    )
                 except ValueError as exc:
                     raise ToolsServiceError(
                         f"Failed to activate tool: {exc!s}"
                     ) from exc
             else:
-                ok = self.tool_mgr.deactivate_llm_tool(tool_name)
+                ok = await self.tool_mgr.deactivate_llm_tool_async(tool_name)
 
             if ok:
                 return "Operation successful."

docs/zh/dev/star/plugin.md

@@ -1152,17 +1152,21 @@ await empty_mention_waiter(event, session_filter=CustomFilter()) # 这里传入
 
 获取提供商有以下几种方式:
 
-- 获取当前使用的大语言模型提供商: `self.context.get_using_provider(umo=event.unified_msg_origin)`。
+- 异步获取当前使用的大语言模型提供商: `await self.context.get_using_provider_async(umo=event.unified_msg_origin)`。
 - 根据 ID 获取大语言模型提供商: `self.context.get_provider_by_id(provider_id="xxxx")`。
 - 获取所有大语言模型提供商: `self.context.get_all_providers()`。
 
+异步事件处理函数中请使用 `get_using_provider_async()`。为兼容现有插件，`get_using_provider()` 同步接口仍然可用，但已标记为弃用。
+
 ```python
 from astrbot.api.event import filter, AstrMessageEvent
 
 @filter.command("test")
 async def test(self, event: AstrMessageEvent):
     # func_tools_mgr = self.context.get_llm_tool_manager()
-    prov = self.context.get_using_provider(umo=event.unified_msg_origin)
+    prov = await self.context.get_using_provider_async(
+        umo=event.unified_msg_origin
+    )
     if prov:
         llm_resp = await prov.text_chat(
             prompt="Hi!",
@@ -1287,12 +1291,14 @@ class LLMResponse:
 
 > 嵌入、重排序 没有 “当前使用”。这两个提供商主要用于知识库。
 
-- 获取当前使用的语音识别提供商(STTProvider): `self.context.get_using_stt_provider(umo=event.unified_msg_origin)`。
-- 获取当前使用的语音合成提供商(TTSProvider): `self.context.get_using_tts_provider(umo=event.unified_msg_origin)`。
+- 异步获取当前使用的语音识别提供商(STTProvider): `await self.context.get_using_stt_provider_async(umo=event.unified_msg_origin)`。
+- 异步获取当前使用的语音合成提供商(TTSProvider): `await self.context.get_using_tts_provider_async(umo=event.unified_msg_origin)`。
 - 获取所有语音识别提供商: `self.context.get_all_stt_providers()`。
 - 获取所有语音合成提供商: `self.context.get_all_tts_providers()`。
 - 获取所有嵌入提供商: `self.context.get_all_embedding_providers()`。
 
+同步接口 `get_using_stt_provider()` 和 `get_using_tts_provider()` 仍然保留，用于兼容已有插件，但已标记为弃用。
+
 ::: details STTProvider / TTSProvider / EmbeddingProvider 类型定义
 
 ```py

tests/conftest.py

@@ -305,6 +305,12 @@ async def mock_context(
 
     provider_manager = MagicMock()
     provider_manager.get_using_provider = MagicMock(return_value=mock_provider)
+    provider_manager.get_using_provider_async = AsyncMock(
+        side_effect=lambda *args, **kwargs: provider_manager.get_using_provider(
+            *args,
+            **kwargs,
+        )
+    )
     provider_manager.get_provider_by_id = MagicMock(return_value=mock_provider)
 
     platform_manager = MagicMock()

tests/test_fastapi_v1_dashboard.py

@@ -165,9 +165,15 @@ class FakeLlmTools:
     def activate_llm_tool(self, _tool_name: str, *, star_map) -> bool:
         return True
 
+    async def activate_llm_tool_async(self, _tool_name: str, *, star_map) -> bool:
+        return True
+
     def deactivate_llm_tool(self, _tool_name: str) -> bool:
         return True
 
+    async def deactivate_llm_tool_async(self, _tool_name: str) -> bool:
+        return True
+
 
 class FakeProviderManager:
     def __init__(self, config: dict) -> None:

tests/test_plugin_manager.py

@@ -1802,11 +1802,10 @@ async def test_cleanup_plugin_optional_artifacts_clears_kv_when_plugin_id_presen
 ):
     cleared = []
 
-    class MockDB:
-        async def clear_preferences(self, scope, scope_id):
-            cleared.append((scope, scope_id))
+    async def clear_preferences(scope, scope_id):
+        cleared.append((scope, scope_id))
 
-    monkeypatch.setattr(plugin_manager_pm.context, "get_db", MockDB, raising=False)
+    monkeypatch.setattr(star_manager_module.sp, "clear_async", clear_preferences)
 
     await plugin_manager_pm._cleanup_plugin_optional_artifacts(
         root_dir_name="test_plugin",
@@ -1825,11 +1824,10 @@ async def test_cleanup_plugin_optional_artifacts_skips_kv_when_plugin_id_none(
 ):
     cleared = []
 
-    class MockDB:
-        async def clear_preferences(self, scope, scope_id):
-            cleared.append((scope, scope_id))
+    async def clear_preferences(scope, scope_id):
+        cleared.append((scope, scope_id))
 
-    monkeypatch.setattr(plugin_manager_pm.context, "get_db", MockDB, raising=False)
+    monkeypatch.setattr(star_manager_module.sp, "clear_async", clear_preferences)
 
     await plugin_manager_pm._cleanup_plugin_optional_artifacts(
         root_dir_name="test_plugin",

tests/test_qqofficial_group_message_create.py

@@ -416,7 +416,9 @@ async def test_result_decorate_segments_qqofficial_ws_plain_result():
         "ctx",
         SimpleNamespace(
             plugin_manager=SimpleNamespace(
-                context=SimpleNamespace(get_using_tts_provider=lambda _umo: None)
+                context=SimpleNamespace(
+                    get_using_tts_provider_async=AsyncMock(return_value=None)
+                )
             ),
             astrbot_config={
                 "provider_tts_settings": {

tests/unit/test_astr_main_agent.py

@@ -16,7 +16,9 @@ from astrbot.core.message.components import File, Image, Plain, Reply, Video
 from astrbot.core.platform.astr_message_event import AstrMessageEvent
 from astrbot.core.platform.platform_metadata import PlatformMetadata
 from astrbot.core.provider import Provider
-from astrbot.core.provider.entities import ProviderRequest
+from astrbot.core.provider import manager as provider_manager_module
+from astrbot.core.provider.entities import ProviderRequest, ProviderType
+from astrbot.core.provider.manager import ProviderManager
 from astrbot.core.skills.skill_manager import SkillInfo
 from astrbot.core.star.star import StarMetadata
 
@@ -38,6 +40,9 @@ def mock_context():
     """Create a mock Context."""
     ctx = MagicMock()
     ctx.get_config.return_value = {}
+    ctx.get_using_provider_async = AsyncMock(
+        side_effect=lambda *args, **kwargs: ctx.get_using_provider(*args, **kwargs)
+    )
     ctx.conversation_manager = MagicMock()
     ctx.persona_manager = MagicMock()
     ctx.persona_manager.personas_v3 = []
@@ -257,7 +262,13 @@ class TestMainAgentBuildConfig:
 class TestSelectProvider:
     """Tests for _select_provider function."""
 
-    def test_select_provider_by_id(self, mock_event, mock_context, mock_provider):
+    @pytest.mark.asyncio
+    async def test_select_provider_by_id(
+        self,
+        mock_event,
+        mock_context,
+        mock_provider,
+    ):
         """Test selecting provider by ID from event extra."""
         module = ama
         mock_event.get_extra.side_effect = lambda k: (
@@ -265,12 +276,13 @@ class TestSelectProvider:
         )
         mock_context.get_provider_by_id.return_value = mock_provider
 
-        result = module._select_provider(mock_event, mock_context)
+        result = await module._select_provider(mock_event, mock_context)
 
         assert result == mock_provider
         mock_context.get_provider_by_id.assert_called_once_with("test-provider")
 
-    def test_select_provider_not_found(self, mock_event, mock_context):
+    @pytest.mark.asyncio
+    async def test_select_provider_not_found(self, mock_event, mock_context):
         """Test selecting provider when ID is not found."""
         module = ama
         mock_event.get_extra.side_effect = lambda k: (
@@ -278,7 +290,7 @@ class TestSelectProvider:
         )
         mock_context.get_provider_by_id.return_value = None
 
-        result = module._select_provider(mock_event, mock_context)
+        result = await module._select_provider(mock_event, mock_context)
 
         assert result is None
         mock_event.set_extra.assert_called_with(
@@ -286,7 +298,8 @@ class TestSelectProvider:
             "LLM 请求失败：未找到指定的提供商 `non-existent`。请检查提供商配置或重新选择可用模型。",
         )
 
-    def test_select_provider_invalid_type(self, mock_event, mock_context):
+    @pytest.mark.asyncio
+    async def test_select_provider_invalid_type(self, mock_event, mock_context):
         """Test selecting provider when result is not a Provider instance."""
         module = ama
         mock_event.get_extra.side_effect = lambda k: (
@@ -294,7 +307,7 @@ class TestSelectProvider:
         )
         mock_context.get_provider_by_id.return_value = "not a provider"
 
-        result = module._select_provider(mock_event, mock_context)
+        result = await module._select_provider(mock_event, mock_context)
 
         assert result is None
         mock_event.set_extra.assert_called_with(
@@ -302,26 +315,33 @@ class TestSelectProvider:
             "LLM 请求失败：选择的提供商类型无效（str），已跳过本次请求。",
         )
 
-    def test_select_provider_fallback(self, mock_event, mock_context, mock_provider):
+    @pytest.mark.asyncio
+    async def test_select_provider_fallback(
+        self,
+        mock_event,
+        mock_context,
+        mock_provider,
+    ):
         """Test provider selection fallback to using provider."""
         module = ama
         mock_event.get_extra.return_value = None
         mock_context.get_using_provider.return_value = mock_provider
 
-        result = module._select_provider(mock_event, mock_context)
+        result = await module._select_provider(mock_event, mock_context)
 
         assert result == mock_provider
         mock_context.get_using_provider.assert_called_once_with(
             umo=mock_event.unified_msg_origin
         )
 
-    def test_select_provider_fallback_error(self, mock_event, mock_context):
+    @pytest.mark.asyncio
+    async def test_select_provider_fallback_error(self, mock_event, mock_context):
         """Test provider selection when fallback raises ValueError."""
         module = ama
         mock_event.get_extra.return_value = None
         mock_context.get_using_provider.side_effect = ValueError("Test error")
 
-        result = module._select_provider(mock_event, mock_context)
+        result = await module._select_provider(mock_event, mock_context)
 
         assert result is None
         mock_event.set_extra.assert_called_with(
@@ -330,6 +350,30 @@ class TestSelectProvider:
         )
 
 
+@pytest.mark.asyncio
+async def test_provider_manager_async_selection_uses_session_preference(monkeypatch):
+    preferred_provider = object()
+    manager = ProviderManager.__new__(ProviderManager)
+    manager.inst_map = {"preferred": preferred_provider}
+    manager.acm = MagicMock()
+
+    get_async = AsyncMock(return_value="preferred")
+    monkeypatch.setattr(provider_manager_module.sp, "get_async", get_async)
+
+    result = await manager.get_using_provider_async(
+        ProviderType.CHAT_COMPLETION,
+        "session-1",
+    )
+
+    assert result is preferred_provider
+    get_async.assert_awaited_once_with(
+        "umo",
+        "session-1",
+        "provider_perf_chat_completion",
+        None,
+    )
+
+
 class TestGetSessionConv:
     """Tests for _get_session_conv function."""
 

tests/unit/test_func_tool_manager.py

@@ -48,6 +48,41 @@ def test_builtin_tool_ignores_inactivated_llm_tools():
         sp.put("inactivated_llm_tools", [], scope="global", scope_id="global")
 
 
+@pytest.mark.asyncio
+async def test_async_tool_toggle_waits_for_preference_persistence(monkeypatch):
+    manager = FunctionToolManager()
+
+    async def handler():
+        return None
+
+    manager.add_func("custom_tool", [], "Custom tool", handler)
+    get_async = AsyncMock(return_value=[])
+    put_async = AsyncMock()
+    monkeypatch.setattr(ftm.sp, "get_async", get_async)
+    monkeypatch.setattr(ftm.sp, "put_async", put_async)
+
+    assert await manager.deactivate_llm_tool_async("custom_tool") is True
+    assert manager.get_func("custom_tool").active is False
+    put_async.assert_awaited_once_with(
+        "global",
+        "global",
+        "inactivated_llm_tools",
+        ["custom_tool"],
+    )
+
+    get_async.return_value = ["custom_tool"]
+    put_async.reset_mock()
+
+    assert await manager.activate_llm_tool_async("custom_tool", {}) is True
+    assert manager.get_func("custom_tool").active is True
+    put_async.assert_awaited_once_with(
+        "global",
+        "global",
+        "inactivated_llm_tools",
+        [],
+    )
+
+
 def test_computer_tools_are_registered_as_builtin_tools():
     manager = FunctionToolManager()
 

tests/unit/test_group_chat_context_wiring.py

@@ -13,6 +13,12 @@ from astrbot.core.platform.message_type import MessageType
 def make_main_with_conversation_manager(conv_mgr):
     main = Main.__new__(Main)
     main.context = MagicMock()
+    main.context.get_using_provider_async = AsyncMock(
+        side_effect=lambda *args, **kwargs: main.context.get_using_provider(
+            *args,
+            **kwargs,
+        )
+    )
     main.context.conversation_manager = conv_mgr
     return main
 

tests/unit/test_shared_preferences.py

@@ -0,0 +1,166 @@
+import asyncio
+import time
+
+import pytest
+import pytest_asyncio
+
+from astrbot.core.db.sqlite import SQLiteDatabase
+from astrbot.core.utils.shared_preferences import SharedPreferences
+
+
+@pytest_asyncio.fixture
+async def preferences(tmp_path):
+    database = SQLiteDatabase(str(tmp_path / "preferences.db"))
+    await database.initialize()
+    store = SharedPreferences(database, tmp_path / "preferences.json")
+    await store.initialize()
+    try:
+        yield store, database
+    finally:
+        await store.close()
+        await database.engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_sync_put_updates_cache_and_persists_without_blocking(preferences):
+    store, database = preferences
+
+    started = time.monotonic()
+    store.put("theme", "dark", scope="global", scope_id="global")
+
+    assert time.monotonic() - started < 0.1
+    assert store.get("theme", scope="global", scope_id="global") == "dark"
+
+    await store.flush()
+    persisted = await database.get_preference("global", "global", "theme")
+    assert persisted is not None
+    assert persisted.value == {"val": "dark"}
+
+
+@pytest.mark.asyncio
+async def test_async_put_waits_for_persistence(preferences):
+    store, database = preferences
+
+    await store.put_async("global", "global", "theme", "dark")
+
+    persisted = await database.get_preference("global", "global", "theme")
+    assert persisted is not None
+    assert persisted.value == {"val": "dark"}
+
+
+@pytest.mark.asyncio
+async def test_sync_get_does_not_wait_for_an_exhausted_connection_pool(preferences):
+    store, database = preferences
+    pool = database.engine.pool
+    capacity = pool.size() + pool._max_overflow
+    connections = [await database.engine.connect() for _ in range(capacity)]
+    released = asyncio.Event()
+
+    async def release_connection():
+        await asyncio.sleep(0.01)
+        await connections.pop().close()
+        released.set()
+
+    release_task = asyncio.create_task(release_connection())
+    try:
+        assert (
+            store.get(
+                "missing",
+                "default",
+                scope="global",
+                scope_id="global",
+            )
+            == "default"
+        )
+        await asyncio.wait_for(released.wait(), timeout=0.5)
+    finally:
+        await release_task
+        for connection in connections:
+            await connection.close()
+
+
+@pytest.mark.asyncio
+async def test_sync_writes_from_worker_threads_keep_fifo_order(preferences):
+    store, database = preferences
+
+    await asyncio.to_thread(
+        store.put,
+        "ordered",
+        "first",
+        "global",
+        "global",
+    )
+    await asyncio.to_thread(
+        store.put,
+        "ordered",
+        "second",
+        "global",
+        "global",
+    )
+    await store.flush()
+
+    assert store.get("ordered", scope="global", scope_id="global") == "second"
+    persisted = await database.get_preference("global", "global", "ordered")
+    assert persisted is not None
+    assert persisted.value == {"val": "second"}
+
+
+@pytest.mark.asyncio
+async def test_concurrent_sync_writes_keep_submission_order(
+    preferences,
+    monkeypatch,
+):
+    store, database = preferences
+    original_schedule_write = store._schedule_write
+
+    def delay_first_write(operation):
+        if operation[4] == "first":
+            time.sleep(0.05)
+        original_schedule_write(operation)
+
+    monkeypatch.setattr(store, "_schedule_write", delay_first_write)
+
+    first = asyncio.create_task(
+        asyncio.to_thread(
+            store.put,
+            "ordered",
+            "first",
+            "global",
+            "global",
+        )
+    )
+    await asyncio.sleep(0.01)
+    second = asyncio.create_task(
+        asyncio.to_thread(
+            store.put,
+            "ordered",
+            "second",
+            "global",
+            "global",
+        )
+    )
+    await asyncio.gather(first, second)
+    await store.flush()
+
+    persisted = await database.get_preference("global", "global", "ordered")
+    assert persisted is not None
+    assert persisted.value == {"val": "second"}
+
+
+@pytest.mark.asyncio
+async def test_initialize_preloads_values_for_sync_reads(tmp_path):
+    database = SQLiteDatabase(str(tmp_path / "preload.db"))
+    await database.initialize()
+    await database.insert_preference_or_update(
+        "umo",
+        "session",
+        "provider",
+        {"val": "provider-1"},
+    )
+    store = SharedPreferences(database, tmp_path / "preferences.json")
+    try:
+        await store.initialize()
+        assert store.get("provider", scope="umo", scope_id="session") == "provider-1"
+    finally:
+        await store.close()
+        await database.engine.dispose()