feat(skill): 新增 Agent 会话中安装 Skill 功能

- 新增 install_skill 工具，支持从 Sandbox 路径或 Git 仓库安装 skill
- 支持管理员权限校验（admin/superadmin）
- 安装后自动持久化到 agent config，新对话自动生效
- 动态激活当前会话，无需重启
- 新增 AgentConfigRepository.add_skills_to_config_json() 原子 JSONB 操作
- 新增 21 个单元测试，覆盖权限、Schema、安装流程等场景

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

@@ -1,9 +1,11 @@
 # buildin 工具包
+from .install_skill import install_skill
 from .tools import ask_user_question, calculator, present_artifacts, query_knowledge_graph, text_to_img_qwen_image
 
 __all__ = [
     "ask_user_question",
     "calculator",
+    "install_skill",
     "present_artifacts",
     "query_knowledge_graph",
     "text_to_img_qwen_image",

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

@@ -0,0 +1,246 @@
+"""Agent 会话中安装 Skill 的工具"""
+
+from __future__ import annotations
+
+import shutil
+import tempfile
+from pathlib import Path, PurePosixPath
+from typing import Annotated
+
+from langchain.tools import InjectedToolCallId
+from langchain_core.messages import ToolMessage
+from langgraph.prebuilt.tool_node import ToolRuntime
+from langgraph.types import Command
+from pydantic import BaseModel, Field
+
+from yuxi.agents.toolkits.registry import tool
+from yuxi.repositories.agent_config_repository import AgentConfigRepository
+from yuxi.repositories.conversation_repository import ConversationRepository
+from yuxi.repositories.user_repository import UserRepository
+from yuxi.storage.postgres.manager import pg_manager
+from yuxi.utils.logging_config import logger
+
+
+ADMIN_ROLES = {"admin", "superadmin"}
+
+
+class InstallSkillInput(BaseModel):
+    source: str = Field(
+        description="Skill 来源，支持两种格式:\n"
+                    "1. Sandbox 路径: /home/gem/user-data/workspace/my-skill （/ 开头）\n"
+                    "2. Git 仓库: owner/repo 或完整 GitHub URL"
+    )
+    skill_names: list[str] | None = Field(
+        default=None,
+        description="Git 安装时指定要安装的 skill slug 列表（至少一个）。Sandbox 路径安装时忽略此参数。"
+    )
+
+
+async def _assert_admin(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")
+
+
+def _download_skill_dir(backend, remote_dir: str, local_dir: Path) -> None:
+    """递归通过沙盒 API 下载 skill 目录到本地。"""
+    entries = backend.ls_info(remote_dir)
+    for entry in entries:
+        path = entry["path"]
+        if entry.get("is_dir"):
+            sub = local_dir / PurePosixPath(path).name
+            sub.mkdir(parents=True, exist_ok=True)
+            _download_skill_dir(backend, path, sub)
+        else:
+            resp = backend.download_files([path])
+            if resp and not resp[0].error:
+                (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（可能与传入的不同）。"""
+    from yuxi.agents.backends.sandbox import ProvisionerSandboxBackend, resolve_virtual_path
+    from yuxi.services.skill_service import import_skill_dir, is_valid_skill_slug
+
+    slug = PurePosixPath(sandbox_path.rstrip("/")).name
+    if not is_valid_skill_slug(slug):
+        raise ValueError(f"slug '{slug}' 不合法（仅允许小写字母、数字和连字符）")
+
+    if not sandbox_path.startswith("/home/gem/user-data/"):
+        raise ValueError(
+            f"不支持的沙盒路径: {sandbox_path}。"
+            "请使用 /home/gem/user-data/workspace/...、/home/gem/user-data/uploads/... "
+            "或 /home/gem/user-data/outputs/..."
+        )
+
+    with tempfile.TemporaryDirectory(prefix=".skill-install-") as tmp:
+        staging = Path(tmp) / slug
+
+        # 优先尝试共享卷路径（性能更好，无需走沙盒 API）
+        try:
+            local_path = resolve_virtual_path(thread_id, sandbox_path, user_id=user_id)
+            if (local_path / "SKILL.md").exists():
+                shutil.copytree(local_path, staging)
+            else:
+                raise FileNotFoundError(f"{local_path} 中未找到 SKILL.md")
+        except (ValueError, FileNotFoundError):
+            staging.mkdir(parents=True, exist_ok=True)
+            backend = ProvisionerSandboxBackend(thread_id=thread_id, user_id=user_id)
+            _download_skill_dir(backend, sandbox_path, staging)
+            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
+
+
+async def _install_git_skills(source: str, skill_names: list[str], created_by: str) -> list[dict]:
+    """从 Git 仓库安装多个 skill。返回结果列表。"""
+    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
+        )
+
+
+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
+
+        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
+
+        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
+
+
+@tool(
+    category="buildin",
+    tags=["skill", "安装"],
+    display_name="安装技能",
+    args_schema=InstallSkillInput,
+)
+def install_skill(
+    source: str,
+    runtime: ToolRuntime,
+    tool_call_id: Annotated[str, InjectedToolCallId],
+    skill_names: list[str] | None = None,
+) -> Command:
+    """从 Sandbox 路径或 Git 仓库安装 skill 到平台。
+
+    管理员安装后自动启用，新对话中立即生效。
+
+    Sandbox 路径（以 / 开头）：
+      指定沙盒中包含 SKILL.md 的目录，支持所有 /home/gem/... 路径。
+      例如 /home/gem/user-data/workspace/my-skill
+
+    Git 仓库（不以 / 开头）：
+      支持 owner/repo 格式或完整 GitHub URL。
+      必须指定 skill_names 选择要安装的 skill，至少一个。
+    """
+    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)]
+        })

backend/package/yuxi/repositories/agent_config_repository.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from sqlalchemy import select, update
+from sqlalchemy import select, update, text
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from yuxi.storage.postgres.models_business import AgentConfig
@@ -227,3 +227,27 @@ class AgentConfigRepository:
 
         if was_default:
             await self.set_default(config=remaining[0], updated_by=updated_by)
+
+    async def add_skills_to_config_json(self, *, agent_config_id: int, new_slugs: list[str]) -> bool:
+        """使用 PostgreSQL JSONB 原子操作追加 skills。
+
+        Args:
+            agent_config_id: AgentConfig 的 ID
+            new_slugs: 要追加的技能 slug 列表，会自动去重
+
+        Returns:
+            是否成功更新（至少有一条记录被修改）
+        """
+        sql = text("""
+            UPDATE agent_config
+            SET config_json = jsonb_set(
+                config_json,
+                '{skills}',
+                COALESCE(config_json->'skills', '[]'::jsonb) || to_jsonb(:new_slugs::text[])::jsonb,
+                true
+            )
+            WHERE id = :id
+        """)
+        result = await self.db.execute(sql, {"id": agent_config_id, "new_slugs": new_slugs})
+        await self.db.commit()
+        return result.rowcount > 0

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

@@ -0,0 +1,413 @@
+"""Tests for install_skill tool."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from langgraph.types import Command
+
+from yuxi.agents.toolkits.buildin.install_skill import (
+    ADMIN_ROLES,
+    InstallSkillInput,
+    _assert_admin,
+    install_skill,
+)
+
+# 获取底层函数（@tool 装饰器包装为 StructuredTool 后不可直接调用）
+_install_skill_func = install_skill.func
+
+
+# =============================================================================
+# Tests for ADMIN_ROLES
+# =============================================================================
+
+def test_admin_roles_contains_admin():
+    """ADMIN_ROLES should contain 'admin'."""
+    assert "admin" in ADMIN_ROLES
+
+
+def test_admin_roles_contains_superadmin():
+    """ADMIN_ROLES should contain 'superadmin'."""
+    assert "superadmin" in ADMIN_ROLES
+
+
+def test_admin_roles_is_set():
+    """ADMIN_ROLES should be a set."""
+    assert isinstance(ADMIN_ROLES, set)
+
+
+# =============================================================================
+# Tests for InstallSkillInput schema
+# =============================================================================
+
+def test_install_skill_input_source_field():
+    """InstallSkillInput should have required 'source' field with proper description."""
+    schema = InstallSkillInput.model_json_schema()
+    
+    # source field should exist
+    assert "source" in schema["properties"]
+    
+    # source field should be required
+    assert "source" in schema["required"]
+    
+    # Check description contains sandbox and git format descriptions
+    description = schema["properties"]["source"].get("description", "")
+    assert "Sandbox" in description or "沙盒" in description
+    assert "Git" in description or "owner/repo" in description
+
+
+def test_install_skill_input_skill_names_field():
+    """InstallSkillInput should have skill_names field that is nullable with default None."""
+    schema = InstallSkillInput.model_json_schema()
+    
+    # skill_names field should exist
+    assert "skill_names" in schema["properties"]
+    
+    # skill_names should NOT be required
+    assert "skill_names" not in schema["required"]
+    
+    # skill_names should be an array type
+    skill_names_schema = schema["properties"]["skill_names"]
+    assert skill_names_schema.get("type") == "array" or "anyOf" in skill_names_schema
+
+
+def test_install_skill_input_default_values():
+    """InstallSkillInput should have correct default values."""
+    # Test with only required field
+    input_data = InstallSkillInput(source="/path/to/skill")
+    assert input_data.source == "/path/to/skill"
+    assert input_data.skill_names is None
+
+
+def test_install_skill_input_with_skill_names():
+    """InstallSkillInput should accept skill_names list."""
+    input_data = InstallSkillInput(source="owner/repo", skill_names=["skill1", "skill2"])
+    assert input_data.source == "owner/repo"
+    assert input_data.skill_names == ["skill1", "skill2"]
+
+
+# =============================================================================
+# Tests for _assert_admin
+# =============================================================================
+
+class MockUser:
+    """Mock user object for testing."""
+    def __init__(self, role="user", user_id="test-user"):
+        self.role = role
+        self.user_id = user_id
+
+
+@pytest.mark.asyncio
+async def test_assert_admin_missing_user_raises_error():
+    """_assert_admin should raise ValueError when user does not exist."""
+    mock_session = MagicMock()
+    mock_user = None  # User not found
+    
+    with patch("yuxi.agents.toolkits.buildin.install_skill.pg_manager") as mock_pg:
+        mock_pg.get_async_session_context.return_value.__aenter__.return_value = mock_session
+        
+        with patch("yuxi.agents.toolkits.buildin.install_skill.UserRepository") as mock_repo_cls:
+            mock_repo = MagicMock()
+            mock_repo.get_by_user_id = AsyncMock(return_value=mock_user)
+            mock_repo_cls.return_value = mock_repo
+            
+            with pytest.raises(ValueError, match="用户不存在"):
+                await _assert_admin("non-existent-user")
+
+
+@pytest.mark.asyncio
+async def test_assert_admin_non_admin_raises_error():
+    """_assert_admin should raise ValueError when user is not an admin."""
+    mock_session = MagicMock()
+    mock_user = MockUser(role="user", user_id="test-user")
+    
+    with patch("yuxi.agents.toolkits.buildin.install_skill.pg_manager") as mock_pg:
+        mock_pg.get_async_session_context.return_value.__aenter__.return_value = mock_session
+        
+        with patch("yuxi.agents.toolkits.buildin.install_skill.UserRepository") as mock_repo_cls:
+            mock_repo = MagicMock()
+            mock_repo.get_by_user_id = AsyncMock(return_value=mock_user)
+            mock_repo_cls.return_value = mock_repo
+            
+            with pytest.raises(ValueError, match="仅管理员可以安装 skill"):
+                await _assert_admin("test-user")
+
+
+@pytest.mark.asyncio
+async def test_assert_admin_admin_passes():
+    """_assert_admin should NOT raise for admin users."""
+    mock_session = MagicMock()
+    mock_user = MockUser(role="admin", user_id="test-admin-user")
+    
+    with patch("yuxi.agents.toolkits.buildin.install_skill.pg_manager") as mock_pg:
+        mock_pg.get_async_session_context.return_value.__aenter__.return_value = mock_session
+        
+        with patch("yuxi.agents.toolkits.buildin.install_skill.UserRepository") as mock_repo_cls:
+            mock_repo = MagicMock()
+            mock_repo.get_by_user_id = AsyncMock(return_value=mock_user)
+            mock_repo_cls.return_value = mock_repo
+            
+            # Should not raise
+            await _assert_admin("test-admin-user")
+
+
+@pytest.mark.asyncio
+async def test_assert_admin_superadmin_passes():
+    """_assert_admin should NOT raise for superadmin users."""
+    mock_session = MagicMock()
+    mock_user = MockUser(role="superadmin", user_id="test-superadmin-user")
+    
+    with patch("yuxi.agents.toolkits.buildin.install_skill.pg_manager") as mock_pg:
+        mock_pg.get_async_session_context.return_value.__aenter__.return_value = mock_session
+        
+        with patch("yuxi.agents.toolkits.buildin.install_skill.UserRepository") as mock_repo_cls:
+            mock_repo = MagicMock()
+            mock_repo.get_by_user_id = AsyncMock(return_value=mock_user)
+            mock_repo_cls.return_value = mock_repo
+            
+            # Should not raise
+            await _assert_admin("test-superadmin-user")
+
+
+# =============================================================================
+# Tests for install_skill function (使用 .func 访问底层函数)
+# =============================================================================
+
+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(
+        source="/home/gem/user-data/workspace/test",
+        runtime=runtime,
+        tool_call_id="test-call-id",
+    )
+    
+    assert isinstance(result, Command)
+    assert "messages" in result.update
+
+
+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(
+        source="/home/gem/user-data/workspace/test",
+        runtime=runtime,
+        tool_call_id="test-call-id",
+    )
+    
+    assert isinstance(result, Command)
+    assert "messages" in result.update
+
+
+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(
+        source="/home/gem/user-data/workspace/test",
+        runtime=runtime,
+        tool_call_id="test-call-id",
+    )
+    
+    assert isinstance(result, Command)
+    assert "messages" in result.update
+
+
+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"
+    runtime.context.user_id = "test-user"
+    
+    with patch("yuxi.agents.toolkits.buildin.install_skill._assert_admin") as mock_assert:
+        # Mock _assert_admin to not raise (simulate admin user)
+        mock_assert.return_value = None
+        
+        result = _install_skill_func(
+            source="owner/repo",  # Git format, not starting with "/"
+            skill_names=None,    # Missing skill_names
+            runtime=runtime,
+            tool_call_id="test-call-id",
+        )
+    
+    assert isinstance(result, Command)
+    assert "messages" in result.update
+    # Check that error mentions skill_names
+    error_content = str(result.update)
+    assert "skill_names" in error_content or "Git" in error_content
+
+
+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"
+    runtime.context.user_id = "test-user"
+    runtime.context.skills = None
+    
+    with patch("yuxi.agents.toolkits.buildin.install_skill._assert_admin") as mock_assert:
+        # Mock _assert_admin to not raise
+        mock_assert.return_value = None
+        
+        with patch("yuxi.agents.toolkits.buildin.install_skill._install_git_skills") as mock_install:
+            # Mock the git installation to return success
+            mock_install.return_value = [{"slug": "test-skill", "success": True}]
+            
+            with patch("yuxi.agents.toolkits.buildin.install_skill._enable_skill_in_current_config") as mock_enable:
+                # Mock enabling skill in config
+                mock_enable.return_value = True
+                
+                with patch("yuxi.services.skill_service.sync_thread_visible_skills"):
+                    result = _install_skill_func(
+                        source="owner/repo",
+                        skill_names=["test-skill"],
+                        runtime=runtime,
+                        tool_call_id="test-call-id",
+                    )
+    
+    assert isinstance(result, Command)
+    result_data = result.update
+    assert "messages" in result_data or "activated_skills" in result_data
+
+
+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"
+    runtime.context.user_id = "test-user"
+    runtime.context.skills = None
+    
+    with patch("yuxi.agents.toolkits.buildin.install_skill._assert_admin") as mock_assert:
+        mock_assert.return_value = None
+        
+        with patch("yuxi.agents.toolkits.buildin.install_skill._install_skill_from_sandbox") as mock_install:
+            mock_install.return_value = "my-skill"
+            
+            with patch("yuxi.agents.toolkits.buildin.install_skill._enable_skill_in_current_config") as mock_enable:
+                mock_enable.return_value = True
+                
+                with patch("yuxi.services.skill_service.sync_thread_visible_skills"):
+                    result = _install_skill_func(
+                        source="/home/gem/user-data/workspace/my-skill",
+                        runtime=runtime,
+                        tool_call_id="test-call-id",
+                    )
+    
+    assert isinstance(result, Command)
+    assert "activated_skills" in result.update
+    assert "my-skill" in result.update["activated_skills"]
+
+
+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"
+    runtime.context.user_id = "test-user"
+    
+    with patch("yuxi.agents.toolkits.buildin.install_skill._assert_admin") as mock_assert:
+        # Simulate admin check failure
+        mock_assert.side_effect = ValueError("仅管理员可以安装 skill")
+        
+        result = _install_skill_func(
+            source="/home/gem/user-data/workspace/test",
+            runtime=runtime,
+            tool_call_id="test-call-id",
+        )
+    
+    assert isinstance(result, Command)
+    assert "messages" in result.update
+    # Error message should contain the ValueError message
+    result_str = str(result.update)
+    assert "仅管理员可以安装 skill" in result_str
+
+
+def test_install_skill_exception_handling():
+    """install_skill should handle unexpected exceptions gracefully."""
+    runtime = MagicMock()
+    runtime.context.thread_id = "test-thread-id"
+    runtime.context.user_id = "test-user"
+    
+    with patch("yuxi.agents.toolkits.buildin.install_skill._assert_admin") as mock_assert:
+        # Simulate unexpected exception
+        mock_assert.side_effect = RuntimeError("Unexpected error")
+        
+        result = _install_skill_func(
+            source="/home/gem/user-data/workspace/test",
+            runtime=runtime,
+            tool_call_id="test-call-id",
+        )
+    
+    assert isinstance(result, Command)
+    assert "messages" in result.update
+    # Error message should indicate an exception occurred
+    result_str = str(result.update)
+    assert "安装异常" in result_str
+
+
+def test_install_skill_partial_config_failure():
+    """install_skill should handle partial config persistence failure."""
+    runtime = MagicMock()
+    runtime.context.thread_id = "test-thread-id"
+    runtime.context.user_id = "test-user"
+    runtime.context.skills = None
+    
+    with patch("yuxi.agents.toolkits.buildin.install_skill._assert_admin") as mock_assert:
+        mock_assert.return_value = None
+        
+        with patch("yuxi.agents.toolkits.buildin.install_skill._install_skill_from_sandbox") as mock_install:
+            mock_install.return_value = "my-skill"
+            
+            with patch("yuxi.agents.toolkits.buildin.install_skill._enable_skill_in_current_config") as mock_enable:
+                # Simulate config persistence failure
+                mock_enable.return_value = False
+                
+                with patch("yuxi.services.skill_service.sync_thread_visible_skills"):
+                    result = _install_skill_func(
+                        source="/home/gem/user-data/workspace/my-skill",
+                        runtime=runtime,
+                        tool_call_id="test-call-id",
+                    )
+    
+    assert isinstance(result, Command)
+    result_str = str(result.update)
+    # Should indicate config persistence issue
+    assert "持久化" in result_str or "Skill 已安装" in result_str
+
+
+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"
+    runtime.context.user_id = "test-user"
+    runtime.context.skills = None
+    
+    with patch("yuxi.agents.toolkits.buildin.install_skill._assert_admin") as mock_assert:
+        mock_assert.return_value = None
+        
+        with patch("yuxi.agents.toolkits.buildin.install_skill._install_skill_from_sandbox") as mock_install:
+            # Simulate skill being renamed during installation
+            mock_install.return_value = "my-skill-v2"
+            
+            with patch("yuxi.agents.toolkits.buildin.install_skill._enable_skill_in_current_config") as mock_enable:
+                mock_enable.return_value = True
+                
+                with patch("yuxi.services.skill_service.sync_thread_visible_skills"):
+                    result = _install_skill_func(
+                        source="/home/gem/user-data/workspace/my-skill",
+                        runtime=runtime,
+                        tool_call_id="test-call-id",
+                    )
+    
+    assert isinstance(result, Command)
+    result_str = str(result.update)
+    # Should warn about slug rename
+    assert "已安装为" in result_str or "⚠️" in result_str