fix(skill): 修复 install_skill 同步异步混用及数据库连接问题

- 将 install_skill 改为 async def，消除 asyncio.run() 事件循环冲突
- 修复 agent_config_repository 表名及 jsonb 类型转换
- 增加数据库连接池配置，避免高并发耗尽连接
- 修复远程 skill 批量安装列表引用共享 bug
- 修复 slug 冲突 warning 误报（目录名 vs SKILL.md name）
- 新增 UserRepository 支持外部传入 db 会话
- 更新测试适配 async 签名

backend/package/yuxi/agents/toolkits/buildin/install_skill.py

@@ -1,11 +1,9 @@
-"""Agent 会话中安装 Skill 的工具"""
-
-from __future__ import annotations
-
 import shutil
 import tempfile
 from pathlib import Path, PurePosixPath
-from typing import Annotated
+from typing import Annotated, Any
+
+from langchain_core.tools import InjectedToolArg
 
 from langchain.tools import InjectedToolCallId
 from langchain_core.messages import ToolMessage
@@ -23,7 +21,6 @@ from yuxi.utils.logging_config import logger
 
 ADMIN_ROLES = {"admin", "superadmin"}
 
-
 class InstallSkillInput(BaseModel):
     source: str = Field(
         description="Skill 来源，支持两种格式:\n"
@@ -35,16 +32,14 @@ class InstallSkillInput(BaseModel):
         description="Git 安装时指定要安装的 skill slug 列表（至少一个）。Sandbox 路径安装时忽略此参数。"
     )
 
-
-async def _assert_admin(user_id: str) -> None:
+async def _assert_admin(db, user_id: str) -> None:
     """验证用户是管理员，否则抛出 ValueError。"""
-    async with pg_manager.get_async_session_context() as db:
-        repo = UserRepository(db)
-        user = await repo.get_by_user_id(user_id)
-        if user is None:
-            raise ValueError("用户不存在")
-        if user.role not in ADMIN_ROLES:
-            raise ValueError("仅管理员可以安装 skill")
+    repo = UserRepository()
+    user = await repo.get_by_id_with_db(db, int(user_id))
+    if user is None:
+        raise ValueError("用户不存在")
+    if user.role not in ADMIN_ROLES:
+        raise ValueError("仅管理员可以安装 skill")
 
 
 def _download_skill_dir(backend, remote_dir: str, local_dir: Path) -> None:
@@ -62,10 +57,14 @@ def _download_skill_dir(backend, remote_dir: str, local_dir: Path) -> None:
                 (local_dir / PurePosixPath(path).name).write_bytes(resp[0].content)
 
 
-async def _install_skill_from_sandbox(sandbox_path: str, thread_id: str, user_id: str) -> str:
-    """从 Sandbox 路径安装技能。返回最终安装的 slug（可能与传入的不同）。"""
+async def _install_skill_from_sandbox(db, sandbox_path: str, thread_id: str, user_id: str) -> tuple[str, bool]:
+    """从 Sandbox 路径安装 skill。返回 (slug, 是否因冲突被重命名)。"""
     from yuxi.agents.backends.sandbox import ProvisionerSandboxBackend, resolve_virtual_path
-    from yuxi.services.skill_service import import_skill_dir, is_valid_skill_slug
+    from yuxi.services.skill_service import (
+        _parse_skill_markdown,
+        import_skill_dir,
+        is_valid_skill_slug,
+    )
 
     slug = PurePosixPath(sandbox_path.rstrip("/")).name
     if not is_valid_skill_slug(slug):
@@ -95,50 +94,120 @@ async def _install_skill_from_sandbox(sandbox_path: str, thread_id: str, user_id
             if not (staging / "SKILL.md").exists():
                 raise ValueError(f"沙盒路径 {sandbox_path} 中未找到 SKILL.md")
 
-        async with pg_manager.get_async_session_context() as db:
-            result = await import_skill_dir(db, source_dir=staging, created_by=user_id)
-
-    if isinstance(result, Path):
-        return result.name
-    if isinstance(result, str):
-        return result
-    return slug
+        content = (staging / "SKILL.md").read_text(encoding="utf-8")
+        parsed_name, _, _ = _parse_skill_markdown(content)
+        result = await import_skill_dir(db, source_dir=staging, created_by=user_id)
+    return result.slug, result.slug != parsed_name
 
 
-async def _install_git_skills(source: str, skill_names: list[str], created_by: str) -> list[dict]:
-    """从 Git 仓库安装多个 skill。返回结果列表。"""
+async def _enable_skill_in_current_config(db, user_id: str, thread_id: str, skill_slug: str) -> bool:
+    """在当前会话的配置中启用新安装的 skill"""
+    conv_repo = ConversationRepository(db)
+    conv = await conv_repo.get_conversation_by_thread_id(thread_id)
+    if not conv:
+        return False
+        
+    agent_config_id = (conv.extra_metadata or {}).get("agent_config_id")
+    if not agent_config_id:
+        return False
+
+    config_repo = AgentConfigRepository(db)
+    result = await config_repo.add_skills_to_config_json(
+        agent_config_id=agent_config_id, new_slugs=[skill_slug]
+    )
+    return result
+
+
+async def _run_install_task(
+    source: str,
+    runtime: ToolRuntime,
+    tool_call_id: str,
+    skill_names: list[str] | None = None,
+) -> Command:
+    """执行异步安装任务的核心逻辑"""
+    from yuxi.agents.middlewares.skills_middleware import normalize_selected_skills
+    from yuxi.services.skill_service import sync_thread_visible_skills
     from yuxi.services.remote_skill_install_service import install_remote_skills_batch
 
-    async with pg_manager.get_async_session_context() as db:
-        return await install_remote_skills_batch(
-            db, source=source, skills=skill_names, created_by=created_by
-        )
+    user_id = getattr(runtime.context, "user_id", None)
+    thread_id = getattr(runtime.context, "thread_id", None)
+    
+    logger.info(f"DEBUG: install_skill called with user_id={user_id}, thread_id={thread_id}, source={source}")
+    
+    if not user_id or not thread_id:
+        return Command(update={
+            "messages": [ToolMessage(content="错误：无法获取当前会话信息", tool_call_id=tool_call_id)]
+        })
 
+    try:
+        async with pg_manager.get_async_session_context() as db:
+            await _assert_admin(db, user_id)
 
-async def _enable_skill_in_current_config(user_id: str, thread_id: str, slug: str) -> bool:
-    """将 skill slug 原子追加到当前对话使用的 agent config 中。"""
-    async with pg_manager.get_async_session_context() as db:
-        conv_repo = ConversationRepository(db)
-        conv = await conv_repo.get_conversation_by_thread_id(thread_id)
-        if not conv:
-            logger.warning(f"Conversation {thread_id} not found")
-            return False
+            installed_slugs: list[str] = []
+            failed_items: list[dict] = []
+            slug_warnings: list[str] = []
 
-        agent_config_id = (conv.extra_metadata or {}).get("agent_config_id")
-        if not agent_config_id:
-            logger.warning(f"No agent_config_id found for thread {thread_id}")
-            return False
+            if source.startswith("/"):
+                # Sandbox 路径安装
+                actual_slug, was_renamed = await _install_skill_from_sandbox(db, source, thread_id, user_id)
+                installed_slugs = [actual_slug]
+                if was_renamed:
+                    slug_warnings.append(f"⚠️ 技能 slug '{actual_slug}' 已存在，已自动重命名安装")
+            else:
+                # Git 安装
+                _skill_names = skill_names or []
+                if not _skill_names:
+                    return Command(update={
+                        "messages": [ToolMessage(
+                            content="❌ 错误: 从 Git 安装时必须通过 skill_names 指定技能名称",
+                            tool_call_id=tool_call_id,
+                        )]
+                    })
+                results = await install_remote_skills_batch(db, source=source, skills=_skill_names, created_by=user_id)
+                installed_slugs = [r["slug"] for r in results if r.get("success")]
+                failed_items = [r for r in results if not r.get("success")]
 
-        config_repo = AgentConfigRepository(db)
-        result = await config_repo.add_skills_to_config_json(
-            agent_config_id=agent_config_id,
-            new_slugs=[slug],
-        )
-        if result:
-            logger.info(f"Skill '{slug}' added to agent config {agent_config_id}")
-        else:
-            logger.warning(f"Failed to add skill '{slug}' to config {agent_config_id}")
-        return result
+            # 持久化
+            config_success = True
+            if installed_slugs:
+                for slug in installed_slugs:
+                    ok = await _enable_skill_in_current_config(db, user_id, thread_id, slug)
+                    if not ok:
+                        config_success = False
+
+            # 文件同步
+            current_skills = normalize_selected_skills(
+                getattr(runtime.context, "skills", None)
+            )
+            sync_thread_visible_skills(thread_id, current_skills + installed_slugs)
+
+            # 响应
+            lines = []
+            if installed_slugs:
+                lines.append(f"✅ 成功安装并激活技能: {', '.join(installed_slugs)}")
+            for w in slug_warnings:
+                lines.append(w)
+            if failed_items:
+                for item in failed_items:
+                    lines.append(f"❌ 安装失败 ({item['slug']}): {item.get('error', '未知错误')}")
+            if not config_success:
+                lines.append("⚠️ 技能已安装到系统，但在当前会话配置中激活失败")
+            if not installed_slugs and not failed_items:
+                lines.append("ℹ️ 未发现需要安装的技能")
+
+            return Command(update={
+                "activated_skills": installed_slugs,
+                "messages": [ToolMessage(content="\n".join(lines), tool_call_id=tool_call_id)],
+            })
+
+    except Exception as e:
+        logger.exception("install_skill 异常")
+        return Command(update={
+            "messages": [ToolMessage(
+                content=f"❌ 安装异常: {str(e)}",
+                tool_call_id=tool_call_id,
+            )]
+        })
 
 
 @tool(
@@ -147,100 +216,22 @@ async def _enable_skill_in_current_config(user_id: str, thread_id: str, slug: st
     display_name="安装技能",
     args_schema=InstallSkillInput,
 )
-def install_skill(
+async def install_skill(
     source: str,
-    runtime: ToolRuntime,
-    tool_call_id: Annotated[str, InjectedToolCallId],
     skill_names: list[str] | None = None,
+    runtime: ToolRuntime = None,
+    tool_call_id: Annotated[str, InjectedToolCallId] = "",
 ) -> Command:
-    """从 Sandbox 路径或 Git 仓库安装 skill 到平台。
+    """安装新的技能 (Skill) 到系统中。
 
-    管理员安装后自动启用，新对话中立即生效。
+    参数说明:
+    - source: 必填。支持两种格式:
+      1. Sandbox 路径: 例如 "/tmp/my-skill"
+      2. Git 仓库: 例如 "owner/repo" 或 "https://github.com/owner/repo"
+    - skill_names: 从 Git 仓库安装时必填，指定要安装的技能列表。
 
-    Sandbox 路径（以 / 开头）：
-      指定沙盒中包含 SKILL.md 的目录，支持所有 /home/gem/... 路径。
-      例如 /home/gem/user-data/workspace/my-skill
-
-    Git 仓库（不以 / 开头）：
-      支持 owner/repo 格式或完整 GitHub URL。
-      必须指定 skill_names 选择要安装的 skill，至少一个。
+    注意:
+    - 仅超级管理员 (superadmin) 有权执行此操作。
+    - 安装成功后，该技能会自动在当前会话 (thread) 中激活。
     """
-    import asyncio
-    from yuxi.agents.middlewares.skills_middleware import normalize_selected_skills
-    from yuxi.services.skill_service import sync_thread_visible_skills
-
-    thread_id = getattr(runtime.context, "thread_id", None)
-    user_id = getattr(runtime.context, "user_id", None)
-    if not thread_id or not user_id:
-        return Command(update={
-            "messages": [ToolMessage(content="错误：无法获取当前会话信息", tool_call_id=tool_call_id)]
-        })
-
-    try:
-        asyncio.run(_assert_admin(user_id))
-
-        installed_slugs: list[str] = []
-        failed_items: list[dict] = []
-        slug_warnings: list[str] = []
-
-        if source.startswith("/"):
-            # Sandbox 路径安装
-            original_slug = PurePosixPath(source.strip().rstrip("/")).name
-            actual_slug = asyncio.run(_install_skill_from_sandbox(source, thread_id, user_id))
-            installed_slugs = [actual_slug]
-            if actual_slug != original_slug:
-                slug_warnings.append(f"⚠️ 技能 '{original_slug}' 已存在，已安装为 '{actual_slug}'")
-        else:
-            # Git 仓库安装
-            _skill_names = skill_names or []
-            if not _skill_names:
-                return Command(update={
-                    "messages": [ToolMessage(
-                        content=f"❌ Git 安装需要指定 skill_names 参数。\n"
-                                f"在沙盒中执行 'npx skills list --source {source}' "
-                                f"查看可用的 skill 列表。",
-                        tool_call_id=tool_call_id,
-                    )]
-                })
-            results = asyncio.run(_install_git_skills(source, _skill_names, user_id))
-            installed_slugs = [r["slug"] for r in results if r.get("success")]
-            failed_items = [r for r in results if not r.get("success")]
-
-        # 持久化：每个成功安装的 skill 单独写入 agent config
-        config_success = True
-        if installed_slugs:
-            for slug in installed_slugs:
-                ok = asyncio.run(_enable_skill_in_current_config(user_id, thread_id, slug))
-                if not ok:
-                    config_success = False
-
-            # 文件同步（传递 current_skills + 新 skills，否则会删除已有的）
-            current_skills = normalize_selected_skills(
-                getattr(runtime.context, "skills", None)
-            )
-            sync_thread_visible_skills(thread_id, current_skills + installed_slugs)
-
-        # 构建返回消息
-        lines: list[str] = []
-        if installed_slugs:
-            lines.append(f"✅ Skill '{', '.join(installed_slugs)}' 安装成功！已启用，所有使用当前配置的对话中自动生效。")
-        for w in slug_warnings:
-            lines.append(w)
-        for r in failed_items:
-            lines.append(f"❌ {r['slug']}: {r.get('error', '未知错误')}")
-        if not config_success:
-            lines.append("⚠️ Skill 已安装但持久化到配置失败，请联系管理员。")
-
-        return Command(update={
-            "activated_skills": installed_slugs,
-            "messages": [ToolMessage(content="\n".join(lines), tool_call_id=tool_call_id)],
-        })
-    except ValueError as e:
-        return Command(update={
-            "messages": [ToolMessage(content=f"❌ 安装失败: {e}", tool_call_id=tool_call_id)]
-        })
-    except Exception as e:
-        logger.exception("install_skill 异常")
-        return Command(update={
-            "messages": [ToolMessage(content=f"❌ 安装异常: {e}", tool_call_id=tool_call_id)]
-        })
+    return await _run_install_task(source, runtime, tool_call_id, skill_names)

backend/package/yuxi/repositories/agent_config_repository.py

@@ -238,16 +238,18 @@ class AgentConfigRepository:
         Returns:
             是否成功更新（至少有一条记录被修改）
         """
+        import json
         sql = text("""
-            UPDATE agent_config
+            UPDATE agent_configs
             SET config_json = jsonb_set(
-                config_json,
+                CAST(config_json AS jsonb),
                 '{skills}',
-                COALESCE(config_json->'skills', '[]'::jsonb) || to_jsonb(:new_slugs::text[])::jsonb,
+                COALESCE(CAST(config_json->'skills' AS jsonb), CAST('[]' AS jsonb)) || CAST(:new_slugs_json AS jsonb),
                 true
             )
             WHERE id = :id
         """)
-        result = await self.db.execute(sql, {"id": agent_config_id, "new_slugs": new_slugs})
+        new_slugs_json = json.dumps(new_slugs)
+        result = await self.db.execute(sql, {"id": agent_config_id, "new_slugs_json": new_slugs_json})
         await self.db.commit()
         return result.rowcount > 0

backend/package/yuxi/repositories/user_repository.py

@@ -5,6 +5,7 @@ from datetime import datetime as dt
 from typing import Annotated, Any
 
 from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from yuxi.storage.postgres.manager import pg_manager
 from yuxi.storage.postgres.models_business import User
@@ -19,14 +20,22 @@ class UserRepository:
     async def get_by_id(self, id: int) -> User | None:
         """根据 ID 获取用户"""
         async with pg_manager.get_async_session_context() as session:
-            result = await session.execute(select(User).where(User.id == id))
-            return result.scalar_one_or_none()
+            return await self.get_by_id_with_db(session, id)
+
+    async def get_by_id_with_db(self, db: AsyncSession, id: int) -> User | None:
+        """使用指定的 db 根据 ID 获取用户"""
+        result = await db.execute(select(User).where(User.id == id))
+        return result.scalar_one_or_none()
 
     async def get_by_user_id(self, user_id: str) -> User | None:
         """根据 user_id 获取用户"""
         async with pg_manager.get_async_session_context() as session:
-            result = await session.execute(select(User).where(User.user_id == user_id))
-            return result.scalar_one_or_none()
+            return await self.get_by_user_id_with_db(session, user_id)
+
+    async def get_by_user_id_with_db(self, db: AsyncSession, user_id: str) -> User | None:
+        """使用指定的 db 获取用户"""
+        result = await db.execute(select(User).where(User.user_id == user_id))
+        return result.scalar_one_or_none()
 
     async def get_by_phone(self, phone: str) -> User | None:
         """根据手机号获取用户"""

backend/package/yuxi/services/remote_skill_install_service.py

@@ -234,7 +234,7 @@ async def install_remote_skills_batch(
         raise ValueError("skills 列表不能为空")
 
     # 预分配结果数组（按请求顺序），校验非法名并记录失败
-    results: list[dict] = [{"slug": "", "success": False, "error": "unset"}] * len(skills)
+    results: list[dict] = [{"slug": "", "success": False, "error": "unset"} for _ in range(len(skills))]
     normalized_skills: list[str] = []
     valid_indices: list[int] = []
     for i, skill in enumerate(skills):

backend/package/yuxi/storage/postgres/manager.py

@@ -58,6 +58,8 @@ class PostgresManager(metaclass=SingletonMeta):
                 json_deserializer=json.loads,
                 pool_pre_ping=True,
                 pool_recycle=1800,
+                pool_size=10,
+                max_overflow=20,
             )
 
             # 创建异步会话工厂
@@ -258,13 +260,13 @@ class PostgresManager(metaclass=SingletonMeta):
 
     async def get_async_session(self) -> AsyncSession:
         """获取异步数据库会话"""
-        self._check_initialized()
+        self.initialize()  # 确保已初始化
         return self.AsyncSession()
 
     @asynccontextmanager
     async def get_async_session_context(self):
         """获取异步数据库会话的上下文管理器"""
-        self._check_initialized()
+        self.initialize()  # 确保已初始化
         session = self.AsyncSession()
         try:
             yield session

backend/test/unit/agents/toolkits/buildin/test_install_skill.py

@@ -16,7 +16,7 @@ from yuxi.agents.toolkits.buildin.install_skill import (
 )
 
 # 获取底层函数（@tool 装饰器包装为 StructuredTool 后不可直接调用）
-_install_skill_func = install_skill.func
+_install_skill_func = install_skill.coroutine
 
 
 # =============================================================================
@@ -175,13 +175,14 @@ async def test_assert_admin_superadmin_passes():
 # Tests for install_skill function (使用 .func 访问底层函数)
 # =============================================================================
 
-def test_install_skill_no_thread_id_returns_error():
+@pytest.mark.asyncio
+async def test_install_skill_no_thread_id_returns_error():
     """install_skill should return error Command when thread_id is missing."""
     runtime = MagicMock()
     runtime.context.thread_id = None
     runtime.context.user_id = "test-user"
     
-    result = _install_skill_func(
+    result = await _install_skill_func(
         source="/home/gem/user-data/workspace/test",
         runtime=runtime,
         tool_call_id="test-call-id",
@@ -191,13 +192,14 @@ def test_install_skill_no_thread_id_returns_error():
     assert "messages" in result.update
 
 
-def test_install_skill_no_user_id_returns_error():
+@pytest.mark.asyncio
+async def test_install_skill_no_user_id_returns_error():
     """install_skill should return error Command when user_id is missing."""
     runtime = MagicMock()
     runtime.context.thread_id = "test-thread-id"
     runtime.context.user_id = None
     
-    result = _install_skill_func(
+    result = await _install_skill_func(
         source="/home/gem/user-data/workspace/test",
         runtime=runtime,
         tool_call_id="test-call-id",
@@ -207,13 +209,14 @@ def test_install_skill_no_user_id_returns_error():
     assert "messages" in result.update
 
 
-def test_install_skill_no_context_returns_error():
+@pytest.mark.asyncio
+async def test_install_skill_no_context_returns_error():
     """install_skill should return error Command when runtime.context is missing attributes."""
     runtime = MagicMock()
     runtime.context = SimpleNamespace()
     # No thread_id or user_id attributes
     
-    result = _install_skill_func(
+    result = await _install_skill_func(
         source="/home/gem/user-data/workspace/test",
         runtime=runtime,
         tool_call_id="test-call-id",
@@ -223,7 +226,8 @@ def test_install_skill_no_context_returns_error():
     assert "messages" in result.update
 
 
-def test_install_skill_git_no_skill_names_returns_error():
+@pytest.mark.asyncio
+async def test_install_skill_git_no_skill_names_returns_error():
     """install_skill should return error Command when using Git source without skill_names."""
     runtime = MagicMock()
     runtime.context.thread_id = "test-thread-id"
@@ -233,7 +237,7 @@ def test_install_skill_git_no_skill_names_returns_error():
         # Mock _assert_admin to not raise (simulate admin user)
         mock_assert.return_value = None
         
-        result = _install_skill_func(
+        result = await _install_skill_func(
             source="owner/repo",  # Git format, not starting with "/"
             skill_names=None,    # Missing skill_names
             runtime=runtime,
@@ -247,7 +251,8 @@ def test_install_skill_git_no_skill_names_returns_error():
     assert "skill_names" in error_content or "Git" in error_content
 
 
-def test_install_skill_git_with_skill_names_passes_admin_check():
+@pytest.mark.asyncio
+async def test_install_skill_git_with_skill_names_passes_admin_check():
     """install_skill with Git source and skill_names should pass admin check (but may fail other steps)."""
     runtime = MagicMock()
     runtime.context.thread_id = "test-thread-id"
@@ -267,7 +272,7 @@ def test_install_skill_git_with_skill_names_passes_admin_check():
                 mock_enable.return_value = True
                 
                 with patch("yuxi.services.skill_service.sync_thread_visible_skills"):
-                    result = _install_skill_func(
+                    result = await _install_skill_func(
                         source="owner/repo",
                         skill_names=["test-skill"],
                         runtime=runtime,
@@ -279,7 +284,8 @@ def test_install_skill_git_with_skill_names_passes_admin_check():
     assert "messages" in result_data or "activated_skills" in result_data
 
 
-def test_install_skill_sandbox_success():
+@pytest.mark.asyncio
+async def test_install_skill_sandbox_success():
     """install_skill with valid sandbox path should work (full mock)."""
     runtime = MagicMock()
     runtime.context.thread_id = "test-thread-id"
@@ -296,7 +302,7 @@ def test_install_skill_sandbox_success():
                 mock_enable.return_value = True
                 
                 with patch("yuxi.services.skill_service.sync_thread_visible_skills"):
-                    result = _install_skill_func(
+                    result = await _install_skill_func(
                         source="/home/gem/user-data/workspace/my-skill",
                         runtime=runtime,
                         tool_call_id="test-call-id",
@@ -307,7 +313,8 @@ def test_install_skill_sandbox_success():
     assert "my-skill" in result.update["activated_skills"]
 
 
-def test_install_skill_value_error_handling():
+@pytest.mark.asyncio
+async def test_install_skill_value_error_handling():
     """install_skill should handle ValueError from admin check gracefully."""
     runtime = MagicMock()
     runtime.context.thread_id = "test-thread-id"
@@ -317,7 +324,7 @@ def test_install_skill_value_error_handling():
         # Simulate admin check failure
         mock_assert.side_effect = ValueError("仅管理员可以安装 skill")
         
-        result = _install_skill_func(
+        result = await _install_skill_func(
             source="/home/gem/user-data/workspace/test",
             runtime=runtime,
             tool_call_id="test-call-id",
@@ -330,7 +337,8 @@ def test_install_skill_value_error_handling():
     assert "仅管理员可以安装 skill" in result_str
 
 
-def test_install_skill_exception_handling():
+@pytest.mark.asyncio
+async def test_install_skill_exception_handling():
     """install_skill should handle unexpected exceptions gracefully."""
     runtime = MagicMock()
     runtime.context.thread_id = "test-thread-id"
@@ -340,7 +348,7 @@ def test_install_skill_exception_handling():
         # Simulate unexpected exception
         mock_assert.side_effect = RuntimeError("Unexpected error")
         
-        result = _install_skill_func(
+        result = await _install_skill_func(
             source="/home/gem/user-data/workspace/test",
             runtime=runtime,
             tool_call_id="test-call-id",
@@ -353,7 +361,8 @@ def test_install_skill_exception_handling():
     assert "安装异常" in result_str
 
 
-def test_install_skill_partial_config_failure():
+@pytest.mark.asyncio
+async def test_install_skill_partial_config_failure():
     """install_skill should handle partial config persistence failure."""
     runtime = MagicMock()
     runtime.context.thread_id = "test-thread-id"
@@ -371,7 +380,7 @@ def test_install_skill_partial_config_failure():
                 mock_enable.return_value = False
                 
                 with patch("yuxi.services.skill_service.sync_thread_visible_skills"):
-                    result = _install_skill_func(
+                    result = await _install_skill_func(
                         source="/home/gem/user-data/workspace/my-skill",
                         runtime=runtime,
                         tool_call_id="test-call-id",
@@ -383,7 +392,8 @@ def test_install_skill_partial_config_failure():
     assert "持久化" in result_str or "Skill 已安装" in result_str
 
 
-def test_install_skill_slug_warning_for_renamed():
+@pytest.mark.asyncio
+async def test_install_skill_slug_warning_for_renamed():
     """install_skill should include warning when skill is renamed."""
     runtime = MagicMock()
     runtime.context.thread_id = "test-thread-id"
@@ -401,7 +411,7 @@ def test_install_skill_slug_warning_for_renamed():
                 mock_enable.return_value = True
                 
                 with patch("yuxi.services.skill_service.sync_thread_visible_skills"):
-                    result = _install_skill_func(
+                    result = await _install_skill_func(
                         source="/home/gem/user-data/workspace/my-skill",
                         runtime=runtime,
                         tool_call_id="test-call-id",