diff --git a/backend/package/yuxi/agents/toolkits/buildin/install_skill.py b/backend/package/yuxi/agents/toolkits/buildin/install_skill.py index 50b85087..18306409 100644 --- a/backend/package/yuxi/agents/toolkits/buildin/install_skill.py +++ b/backend/package/yuxi/agents/toolkits/buildin/install_skill.py @@ -1,9 +1,7 @@ import shutil import tempfile from pathlib import Path, PurePosixPath -from typing import Annotated, Any - -from langchain_core.tools import InjectedToolArg +from typing import Annotated from langchain.tools import InjectedToolCallId from langchain_core.messages import ToolMessage @@ -20,18 +18,23 @@ from yuxi.utils.logging_config import logger ADMIN_ROLES = {"admin", "superadmin"} +SANDBOX_PATH_HINT = ( + "请使用 /home/gem/user-data/workspace/...、/home/gem/user-data/uploads/... 或 /home/gem/user-data/outputs/..." +) + 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" + "1. Sandbox 路径: /home/gem/user-data/workspace/...、" + "/home/gem/user-data/uploads/... 或 /home/gem/user-data/outputs/...（/ 开头）\n" + "2. Git 仓库: owner/repo 或完整 GitHub URL" ) skill_names: list[str] | None = Field( - default=None, - description="Git 安装时指定要安装的 skill slug 列表（至少一个）。Sandbox 路径安装时忽略此参数。" + default=None, description="Git 安装时指定要安装的 skill slug 列表（至少一个）。Sandbox 路径安装时忽略此参数。" ) + async def _assert_admin(db, user_id: str) -> None: """验证用户是管理员，否则抛出 ValueError。""" repo = UserRepository() @@ -42,27 +45,53 @@ async def _assert_admin(db, user_id: str) -> None: raise ValueError("仅管理员可以安装 skill") -def _download_skill_dir(backend, remote_dir: str, local_dir: Path) -> None: - """递归通过沙盒 API 下载 skill 目录到本地。""" +def _collect_sandbox_file_paths(backend, remote_dir: str) -> list[str]: entries = backend.ls_info(remote_dir) + file_paths: list[str] = [] 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) + file_paths.extend(_collect_sandbox_file_paths(backend, path)) else: - resp = backend.download_files([path]) - if resp and not resp[0].error: - (local_dir / PurePosixPath(path).name).write_bytes(resp[0].content) + file_paths.append(path) + return file_paths -async def _install_skill_from_sandbox(db, sandbox_path: str, thread_id: str, user_id: str) -> tuple[str, bool]: - """从 Sandbox 路径安装 skill。返回 (slug, 是否因冲突被重命名)。""" +def _download_skill_dir(backend, remote_dir: str, local_dir: Path) -> None: + """递归通过沙盒 API 下载 skill 目录到本地。""" + remote_root = PurePosixPath(remote_dir.rstrip("/")) + file_paths = _collect_sandbox_file_paths(backend, remote_dir) + if not file_paths: + raise ValueError(f"沙盒路径 {remote_dir} 中未发现可下载文件") + + responses = backend.download_files(file_paths) + if len(responses) != len(file_paths): + raise ValueError("沙盒文件下载结果数量不匹配") + + for expected_path, response in zip(file_paths, responses): + error = getattr(response, "error", None) + content = getattr(response, "content", None) + if error or content is None: + raise ValueError(f"下载沙盒文件失败: {expected_path} ({error or 'empty_content'})") + + pure_path = PurePosixPath(expected_path) + try: + relative_path = pure_path.relative_to(remote_root) + except ValueError: + relative_path = PurePosixPath(pure_path.name) + + target_path = local_dir / Path(relative_path.as_posix()) + target_path.parent.mkdir(parents=True, exist_ok=True) + target_path.write_bytes(content) + + +def _prepare_skill_from_sandbox( + sandbox_path: str, thread_id: str, user_id: str, staging_root: Path +) -> tuple[Path, str]: + """从 Sandbox 路径准备 skill 目录。返回 (本地目录, 原始 skill name)。""" from yuxi.agents.backends.sandbox import ProvisionerSandboxBackend, resolve_virtual_path from yuxi.services.skill_service import ( _parse_skill_markdown, - import_skill_dir, is_valid_skill_slug, ) @@ -71,50 +100,42 @@ async def _install_skill_from_sandbox(db, sandbox_path: str, thread_id: str, use 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/..." - ) + raise ValueError(f"不支持的沙盒路径: {sandbox_path}。{SANDBOX_PATH_HINT}") - with tempfile.TemporaryDirectory(prefix=".skill-install-") as tmp: - staging = Path(tmp) / slug + staging = staging_root / 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") + # 优先尝试共享卷路径（性能更好，无需走沙盒 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") - 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 + content = (staging / "SKILL.md").read_text(encoding="utf-8") + parsed_name, _, _ = _parse_skill_markdown(content) + return staging, parsed_name -async def _enable_skill_in_current_config(db, user_id: str, thread_id: str, skill_slug: str) -> bool: +async def _enable_skills_in_current_config(db, thread_id: str, skill_slugs: list[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] - ) + result = await config_repo.add_skills_to_config_json(agent_config_id=agent_config_id, new_slugs=skill_slugs) return result @@ -126,88 +147,111 @@ async def _run_install_task( ) -> 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 + from yuxi.services.remote_skill_install_service import prepare_remote_skills_batch + from yuxi.services.skill_service import import_skill_dir, sync_thread_visible_skills 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)] - }) + 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) - installed_slugs: list[str] = [] - failed_items: list[dict] = [] - slug_warnings: list[str] = [] + installed_slugs: list[str] = [] + failed_items: list[dict] = [] + slug_warnings: list[str] = [] + config_success = True - 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")] + if source.startswith("/"): + with tempfile.TemporaryDirectory(prefix=".skill-install-") as tmp: + source_dir, parsed_name = _prepare_skill_from_sandbox(source, thread_id, user_id, Path(tmp)) + async with pg_manager.get_async_session_context() as db: + await _assert_admin(db, user_id) + item = await import_skill_dir(db, source_dir=source_dir, created_by=user_id) + installed_slugs = [item.slug] + if item.slug != parsed_name: + slug_warnings.append(f"⚠️ 技能 slug '{item.slug}' 已存在，已自动重命名安装") + config_success = await _enable_skills_in_current_config(db, thread_id, installed_slugs) + else: + _skill_names = skill_names or [] + if not _skill_names: + return Command( + update={ + "messages": [ + ToolMessage( + content="❌ 错误: 从 Git 安装时必须通过 skill_names 指定技能名称", + tool_call_id=tool_call_id, + ) + ] + } + ) - # 持久化 - 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 + preparation = await prepare_remote_skills_batch(source=source, skills=_skill_names) + try: + async with pg_manager.get_async_session_context() as db: + await _assert_admin(db, user_id) + for result in preparation.results: + if not result.get("success"): + failed_items.append(result) + continue - # 文件同步 - current_skills = normalize_selected_skills( - getattr(runtime.context, "skills", None) - ) - sync_thread_visible_skills(thread_id, current_skills + installed_slugs) + try: + item = await import_skill_dir(db, source_dir=result["source_dir"], created_by=user_id) + installed_slugs.append(item.slug) + except Exception as e: + await db.rollback() + failed_items.append({"slug": result["slug"], "success": False, "error": str(e)}) - # 响应 - 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("ℹ️ 未发现需要安装的技能") + config_success = True + if installed_slugs: + config_success = await _enable_skills_in_current_config(db, thread_id, installed_slugs) + finally: + preparation.cleanup() - return Command(update={ + # 文件同步 + current_skills = normalize_selected_skills(getattr(runtime.context, "skills", None)) + sync_thread_visible_skills(thread_id, normalize_selected_skills(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, - )] - }) + return Command( + update={ + "messages": [ + ToolMessage( + content=f"❌ 安装异常: {str(e)}", + tool_call_id=tool_call_id, + ) + ] + } + ) @tool( @@ -226,12 +270,13 @@ async def install_skill( 参数说明: - source: 必填。支持两种格式: - 1. Sandbox 路径: 例如 "/tmp/my-skill" + 1. Sandbox 路径: /home/gem/user-data/workspace/...、/home/gem/user-data/uploads/... + 或 /home/gem/user-data/outputs/... 2. Git 仓库: 例如 "owner/repo" 或 "https://github.com/owner/repo" - skill_names: 从 Git 仓库安装时必填，指定要安装的技能列表。 注意: - - 仅超级管理员 (superadmin) 有权执行此操作。 + - 仅管理员 (admin/superadmin) 有权执行此操作。 - 安装成功后，该技能会自动在当前会话 (thread) 中激活。 """ return await _run_install_task(source, runtime, tool_call_id, skill_names) diff --git a/backend/package/yuxi/repositories/agent_config_repository.py b/backend/package/yuxi/repositories/agent_config_repository.py index f5dd10fc..6cc4135a 100644 --- a/backend/package/yuxi/repositories/agent_config_repository.py +++ b/backend/package/yuxi/repositories/agent_config_repository.py @@ -1,6 +1,6 @@ from __future__ import annotations -from sqlalchemy import select, update, text +from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession from yuxi.storage.postgres.models_business import AgentConfig @@ -10,6 +10,20 @@ from yuxi.utils.datetime_utils import utc_now_naive DEFAULT_CONFIG_NAME = "初始配置" +def _merge_skill_slugs(current_slugs: object, new_slugs: list[str]) -> list[str]: + merged: list[str] = [] + seen: set[str] = set() + for value in [*(current_slugs if isinstance(current_slugs, list) else []), *new_slugs]: + if not isinstance(value, str): + continue + slug = value.strip() + if not slug or slug in seen: + continue + seen.add(slug) + merged.append(slug) + return merged + + class AgentConfigRepository: def __init__(self, db_session: AsyncSession): self.db = db_session @@ -26,6 +40,10 @@ class AgentConfigRepository: result = await self.db.execute(select(AgentConfig).where(AgentConfig.id == config_id)) return result.scalar_one_or_none() + async def _get_by_id_for_update(self, config_id: int) -> AgentConfig | None: + result = await self.db.execute(select(AgentConfig).where(AgentConfig.id == config_id).with_for_update()) + return result.scalar_one_or_none() + async def set_default(self, *, config: AgentConfig, updated_by: str | None = None) -> AgentConfig: if config.is_default: return config @@ -229,32 +247,26 @@ class AgentConfigRepository: 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。 + """在 config_json.context.skills 中追加 skills，并保持顺序去重。 Args: agent_config_id: AgentConfig 的 ID new_slugs: 要追加的技能 slug 列表，会自动去重 Returns: - 是否成功更新（至少有一条记录被修改） + 是否找到并更新了配置 """ - import json - sql = text(""" - UPDATE agent_configs - SET config_json = jsonb_set( - COALESCE(CAST(config_json AS jsonb), '{}'::jsonb), - '{context}', - jsonb_set( - COALESCE(CAST(config_json->'context' AS jsonb), '{}'::jsonb), - '{skills}', - COALESCE(CAST(config_json->'context'->'skills' AS jsonb), CAST('[]' AS jsonb)) || CAST(:new_slugs_json AS jsonb), - true - ), - true - ) - WHERE id = :id - """) - new_slugs_json = json.dumps(new_slugs) - result = await self.db.execute(sql, {"id": agent_config_id, "new_slugs_json": new_slugs_json}) + config = await self._get_by_id_for_update(agent_config_id) + if not config: + return False + + config_json = dict(config.config_json or {}) + context = dict(config_json.get("context") or {}) + context["skills"] = _merge_skill_slugs(context.get("skills"), new_slugs) + config_json["context"] = context + + config.config_json = config_json + config.updated_at = utc_now_naive() await self.db.commit() - return result.rowcount > 0 + await self.db.refresh(config) + return True diff --git a/backend/package/yuxi/services/remote_skill_install_service.py b/backend/package/yuxi/services/remote_skill_install_service.py index 8563624a..7a65383d 100644 --- a/backend/package/yuxi/services/remote_skill_install_service.py +++ b/backend/package/yuxi/services/remote_skill_install_service.py @@ -5,6 +5,7 @@ import os import re import shutil import tempfile +from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING @@ -19,6 +20,16 @@ CONTROL_SEQUENCE_RE = re.compile(r"\x1B\][^\x07]*(?:\x07|\x1B\\)|\x1B[\(\)][A-Za CLI_TIMEOUT_SECONDS = 300 +@dataclass(slots=True) +class RemoteSkillsBatchPreparation: + temp_home: str | None + results: list[dict] + + def cleanup(self) -> None: + if self.temp_home: + shutil.rmtree(self.temp_home, ignore_errors=True) + + def _normalize_source(source: str) -> str: value = str(source or "").strip() if not value: @@ -229,6 +240,37 @@ async def install_remote_skills_batch( Returns: 每个 skill 的安装结果列表，顺序与请求一致: ``[{slug, success, error?}, ...]`` """ + preparation = await prepare_remote_skills_batch(source=source, skills=skills) + try: + results = preparation.results + for index, result in enumerate(results): + if not result.get("success"): + continue + + source_dir = result.get("source_dir") + try: + item = await import_skill_dir( + db, + source_dir=source_dir, + created_by=created_by, + ) + results[index] = {"slug": item.slug, "success": True} + except Exception as e: + if hasattr(db, "rollback"): + await db.rollback() + results[index] = {"slug": result["slug"], "success": False, "error": str(e)} + + return results + finally: + preparation.cleanup() + + +async def prepare_remote_skills_batch( + *, + source: str, + skills: list[str], +) -> RemoteSkillsBatchPreparation: + """批量从远程仓库拉取 skill 目录，但不写数据库。""" normalized_source = _normalize_source(source) if not skills: raise ValueError("skills 列表不能为空") @@ -245,11 +287,10 @@ async def install_remote_skills_batch( results[i] = {"slug": skill, "success": False, "error": str(e)} if not normalized_skills: - return results + return RemoteSkillsBatchPreparation(temp_home=None, results=results) temp_home, env, workdir = _create_isolated_workdir() try: - # Step 1: 一次 npx 调用安装所有 skill（克隆一次） skill_args: list[str] = [] for name in normalized_skills: skill_args.extend(["--skill", name]) @@ -275,31 +316,19 @@ async def install_remote_skills_batch( # CLI 对不匹配的 skill 会退出码非零，但已安装的目录仍在 cli_failed = True - # Step 2: 从临时目录中找到各 skill 的安装目录并逐个导入 - base_dir = Path(temp_home).resolve() - skills_dir = base_dir / ".agents" / "skills" - + skills_dir = Path(temp_home).resolve() / ".agents" / "skills" for original_index, name in zip(valid_indices, normalized_skills): installed_dir = _find_skill_dir(skills_dir, name) if installed_dir is None: error_msg = "CLI 安装失败" if cli_failed else "skills CLI 未生成预期的技能目录" results[original_index] = {"slug": name, "success": False, "error": error_msg} - continue + else: + results[original_index] = {"slug": name, "success": True, "source_dir": installed_dir} - try: - item = await import_skill_dir( - db, - source_dir=installed_dir, - created_by=created_by, - ) - results[original_index] = {"slug": item.slug, "success": True} - except Exception as e: - await db.rollback() - results[original_index] = {"slug": name, "success": False, "error": str(e)} - - return results - finally: + return RemoteSkillsBatchPreparation(temp_home=temp_home, results=results) + except Exception: shutil.rmtree(temp_home, ignore_errors=True) + raise def _find_skill_dir(skills_dir: Path, name: str) -> Path | None: diff --git a/backend/test/unit/agents/toolkits/buildin/test_install_skill.py b/backend/test/unit/agents/toolkits/buildin/test_install_skill.py index 54acb908..b2b37308 100644 --- a/backend/test/unit/agents/toolkits/buildin/test_install_skill.py +++ b/backend/test/unit/agents/toolkits/buildin/test_install_skill.py @@ -2,6 +2,7 @@ from __future__ import annotations +from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -12,6 +13,7 @@ from yuxi.agents.toolkits.buildin.install_skill import ( ADMIN_ROLES, InstallSkillInput, _assert_admin, + _download_skill_dir, install_skill, ) @@ -19,10 +21,26 @@ from yuxi.agents.toolkits.buildin.install_skill import ( _install_skill_func = install_skill.coroutine +class AsyncSessionContext: + def __init__(self, session=None): + self.session = session or MagicMock() + + async def __aenter__(self): + return self.session + + async def __aexit__(self, exc_type, exc, tb): + return False + + +def _mock_pg_session(mock_pg, session=None): + mock_pg.get_async_session_context.return_value = AsyncSessionContext(session) + + # ============================================================================= # Tests for ADMIN_ROLES # ============================================================================= + def test_admin_roles_contains_admin(): """ADMIN_ROLES should contain 'admin'.""" assert "admin" in ADMIN_ROLES @@ -42,16 +60,17 @@ def test_admin_roles_is_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 @@ -61,13 +80,13 @@ def test_install_skill_input_source_field(): 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 @@ -88,12 +107,59 @@ def test_install_skill_input_with_skill_names(): assert input_data.skill_names == ["skill1", "skill2"] +def test_download_skill_dir_preserves_nested_files(tmp_path): + """_download_skill_dir should batch download files and preserve relative paths.""" + calls = [] + + class Backend: + def ls_info(self, path): + if path.endswith("/demo"): + return [ + {"path": f"{path}/SKILL.md", "is_dir": False}, + {"path": f"{path}/scripts", "is_dir": True}, + ] + return [{"path": f"{path}/run.py", "is_dir": False}] + + def download_files(self, paths): + calls.append(paths) + return [ + SimpleNamespace(path=path, content=f"content:{Path(path).name}".encode(), error=None) for path in paths + ] + + _download_skill_dir(Backend(), "/home/gem/user-data/workspace/demo", tmp_path) + + assert calls == [ + [ + "/home/gem/user-data/workspace/demo/SKILL.md", + "/home/gem/user-data/workspace/demo/scripts/run.py", + ] + ] + assert (tmp_path / "SKILL.md").read_bytes() == b"content:SKILL.md" + assert (tmp_path / "scripts" / "run.py").read_bytes() == b"content:run.py" + + +def test_download_skill_dir_raises_on_download_error(tmp_path): + """_download_skill_dir should fail instead of importing a partial skill.""" + + class Backend: + def ls_info(self, path): + return [{"path": f"{path}/SKILL.md", "is_dir": False}] + + def download_files(self, paths): + return [SimpleNamespace(path=paths[0], content=None, error="file_not_found")] + + with pytest.raises(ValueError, match="下载沙盒文件失败"): + _download_skill_dir(Backend(), "/home/gem/user-data/workspace/demo", tmp_path) + + # ============================================================================= # 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 @@ -104,15 +170,15 @@ 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_id_with_db = AsyncMock(return_value=mock_user) mock_repo_cls.return_value = mock_repo - + with pytest.raises(ValueError, match="用户不存在"): await _assert_admin(mock_session, "1") @@ -122,15 +188,15 @@ 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_id_with_db = AsyncMock(return_value=mock_user) mock_repo_cls.return_value = mock_repo - + with pytest.raises(ValueError, match="仅管理员可以安装 skill"): await _assert_admin(mock_session, "1") @@ -140,15 +206,15 @@ 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_id_with_db = AsyncMock(return_value=mock_user) mock_repo_cls.return_value = mock_repo - + # Should not raise await _assert_admin(mock_session, "1") @@ -158,15 +224,15 @@ 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_id_with_db = AsyncMock(return_value=mock_user) mock_repo_cls.return_value = mock_repo - + # Should not raise await _assert_admin(mock_session, "1") @@ -175,19 +241,20 @@ async def test_assert_admin_superadmin_passes(): # Tests for install_skill function (使用 .func 访问底层函数) # ============================================================================= + @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 = await _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 @@ -198,13 +265,13 @@ async def test_install_skill_no_user_id_returns_error(): runtime = MagicMock() runtime.context.thread_id = "test-thread-id" runtime.context.user_id = None - + result = await _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 @@ -215,13 +282,13 @@ async def test_install_skill_no_context_returns_error(): runtime = MagicMock() runtime.context = SimpleNamespace() # No thread_id or user_id attributes - + result = await _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 @@ -230,21 +297,22 @@ async def test_install_skill_no_context_returns_error(): @patch("yuxi.agents.toolkits.buildin.install_skill.pg_manager") async def test_install_skill_git_no_skill_names_returns_error(mock_pg): """install_skill should return error Command when using Git source without skill_names.""" + _mock_pg_session(mock_pg) 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 = await _install_skill_func( source="owner/repo", # Git format, not starting with "/" - skill_names=None, # Missing skill_names + 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 @@ -256,31 +324,40 @@ async def test_install_skill_git_no_skill_names_returns_error(mock_pg): @patch("yuxi.agents.toolkits.buildin.install_skill.pg_manager") async def test_install_skill_git_with_skill_names_passes_admin_check(mock_pg): """install_skill with Git source and skill_names should pass admin check (but may fail other steps).""" + _mock_pg_session(mock_pg) 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.services.remote_skill_install_service.install_remote_skills_batch") 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 = await _install_skill_func( - source="owner/repo", - skill_names=["test-skill"], - runtime=runtime, - tool_call_id="test-call-id", - ) - + + preparation = SimpleNamespace( + results=[{"slug": "test-skill", "success": True, "source_dir": Path("/tmp/test-skill")}], + cleanup=MagicMock(), + ) + with patch("yuxi.services.remote_skill_install_service.prepare_remote_skills_batch") as mock_prepare: + mock_prepare.return_value = preparation + + with patch("yuxi.services.skill_service.import_skill_dir") as mock_import: + mock_import.return_value = SimpleNamespace(slug="test-skill") + + with patch( + "yuxi.agents.toolkits.buildin.install_skill._enable_skills_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 = await _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 @@ -290,27 +367,33 @@ async def test_install_skill_git_with_skill_names_passes_admin_check(mock_pg): @patch("yuxi.agents.toolkits.buildin.install_skill.pg_manager") async def test_install_skill_sandbox_success(mock_pg): """install_skill with valid sandbox path should work (full mock).""" + _mock_pg_session(mock_pg) 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", False) - - 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 = await _install_skill_func( - source="/home/gem/user-data/workspace/my-skill", - runtime=runtime, - tool_call_id="test-call-id", - ) - + + with patch("yuxi.agents.toolkits.buildin.install_skill._prepare_skill_from_sandbox") as mock_prepare: + mock_prepare.return_value = (Path("/tmp/my-skill"), "my-skill") + + with patch("yuxi.services.skill_service.import_skill_dir") as mock_import: + mock_import.return_value = SimpleNamespace(slug="my-skill") + + with patch( + "yuxi.agents.toolkits.buildin.install_skill._enable_skills_in_current_config" + ) as mock_enable: + mock_enable.return_value = True + + with patch("yuxi.services.skill_service.sync_thread_visible_skills"): + result = await _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"] @@ -320,20 +403,21 @@ async def test_install_skill_sandbox_success(mock_pg): @patch("yuxi.agents.toolkits.buildin.install_skill.pg_manager") async def test_install_skill_value_error_handling(mock_pg): """install_skill should handle ValueError from admin check gracefully.""" + _mock_pg_session(mock_pg) 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 = await _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 @@ -345,20 +429,21 @@ async def test_install_skill_value_error_handling(mock_pg): @patch("yuxi.agents.toolkits.buildin.install_skill.pg_manager") async def test_install_skill_exception_handling(mock_pg): """install_skill should handle unexpected exceptions gracefully.""" + _mock_pg_session(mock_pg) 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 = await _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 @@ -370,28 +455,34 @@ async def test_install_skill_exception_handling(mock_pg): @patch("yuxi.agents.toolkits.buildin.install_skill.pg_manager") async def test_install_skill_partial_config_failure(mock_pg): """install_skill should handle partial config persistence failure.""" + _mock_pg_session(mock_pg) 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", False) - - 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 = await _install_skill_func( - source="/home/gem/user-data/workspace/my-skill", - runtime=runtime, - tool_call_id="test-call-id", - ) - + + with patch("yuxi.agents.toolkits.buildin.install_skill._prepare_skill_from_sandbox") as mock_prepare: + mock_prepare.return_value = (Path("/tmp/my-skill"), "my-skill") + + with patch("yuxi.services.skill_service.import_skill_dir") as mock_import: + mock_import.return_value = SimpleNamespace(slug="my-skill") + + with patch( + "yuxi.agents.toolkits.buildin.install_skill._enable_skills_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 = await _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 @@ -402,28 +493,34 @@ async def test_install_skill_partial_config_failure(mock_pg): @patch("yuxi.agents.toolkits.buildin.install_skill.pg_manager") async def test_install_skill_slug_warning_for_renamed(mock_pg): """install_skill should include warning when skill is renamed.""" + _mock_pg_session(mock_pg) 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: + + with patch("yuxi.agents.toolkits.buildin.install_skill._prepare_skill_from_sandbox") as mock_prepare: # Simulate skill being renamed during installation - mock_install.return_value = ("my-skill-v2", True) - - 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 = await _install_skill_func( - source="/home/gem/user-data/workspace/my-skill", - runtime=runtime, - tool_call_id="test-call-id", - ) - + mock_prepare.return_value = (Path("/tmp/my-skill"), "my-skill") + + with patch("yuxi.services.skill_service.import_skill_dir") as mock_import: + mock_import.return_value = SimpleNamespace(slug="my-skill-v2") + + with patch( + "yuxi.agents.toolkits.buildin.install_skill._enable_skills_in_current_config" + ) as mock_enable: + mock_enable.return_value = True + + with patch("yuxi.services.skill_service.sync_thread_visible_skills"): + result = await _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 diff --git a/backend/test/unit/repositories/test_agent_config_repository.py b/backend/test/unit/repositories/test_agent_config_repository.py new file mode 100644 index 00000000..8de888ad --- /dev/null +++ b/backend/test/unit/repositories/test_agent_config_repository.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from sqlalchemy.dialects import postgresql + +from yuxi.repositories.agent_config_repository import AgentConfigRepository, _merge_skill_slugs + + +class FakeScalarResult: + def __init__(self, value): + self.value = value + + def scalar_one_or_none(self): + return self.value + + +class FakeDb: + def __init__(self, config): + self.config = config + self.statements = [] + self.commit = AsyncMock() + self.refresh = AsyncMock() + + async def execute(self, statement): + self.statements.append(statement) + return FakeScalarResult(self.config) + + +def test_merge_skill_slugs_preserves_order_and_dedupes(): + assert _merge_skill_slugs(["old", "new", "old", "", None], ["new", "third"]) == [ + "old", + "new", + "third", + ] + + +@pytest.mark.asyncio +async def test_get_by_id_for_update_uses_row_lock(): + config = SimpleNamespace(config_json={}, updated_at=None) + db = FakeDb(config) + repo = AgentConfigRepository(db) + + result = await repo._get_by_id_for_update(1) + + assert result is config + compiled_sql = str(db.statements[0].compile(dialect=postgresql.dialect())) + assert "FOR UPDATE" in compiled_sql + + +@pytest.mark.asyncio +async def test_add_skills_to_config_json_writes_context_skills_without_duplicates(): + config = SimpleNamespace(config_json={"context": {"skills": ["existing", "new"]}}, updated_at=None) + db = FakeDb(config) + repo = AgentConfigRepository(db) + + result = await repo.add_skills_to_config_json(agent_config_id=1, new_slugs=["new", "extra"]) + + assert result is True + assert config.config_json == {"context": {"skills": ["existing", "new", "extra"]}} + db.commit.assert_awaited_once() + db.refresh.assert_awaited_once_with(config) + + +@pytest.mark.asyncio +async def test_add_skills_to_config_json_creates_missing_context(): + config = SimpleNamespace(config_json={}, updated_at=None) + db = FakeDb(config) + repo = AgentConfigRepository(db) + + result = await repo.add_skills_to_config_json(agent_config_id=1, new_slugs=["demo-skill"]) + + assert result is True + assert config.config_json == {"context": {"skills": ["demo-skill"]}} + + +@pytest.mark.asyncio +async def test_add_skills_to_config_json_returns_false_when_missing_config(): + db = FakeDb(None) + repo = AgentConfigRepository(db) + + result = await repo.add_skills_to_config_json(agent_config_id=404, new_slugs=["demo-skill"]) + + assert result is False + db.commit.assert_not_awaited() + db.refresh.assert_not_awaited()