from __future__ import annotations import json import os import re import shlex import shutil import tempfile import uuid import zipfile from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path, PurePosixPath import yaml from astrbot.core.utils.astrbot_path import ( AstrbotPaths, get_astrbot_data_path, get_astrbot_plugin_path, get_astrbot_skills_path, get_astrbot_temp_path, ) SKILLS_CONFIG_FILENAME = "skills.json" SANDBOX_SKILLS_CACHE_FILENAME = "sandbox_skills_cache.json" DEFAULT_SKILLS_CONFIG: dict[str, dict] = {"skills": {}} SANDBOX_SKILLS_ROOT = "skills" SANDBOX_WORKSPACE_ROOT = "/workspace" WORKSPACE_SKILLS_ROOT = "skills" WORKSPACE_SKILL_FRONTMATTER_MAX_CHARS = 64 * 1024 _SANDBOX_SKILLS_CACHE_VERSION = 1 _SKILL_NAME_RE = re.compile(r"^[\w.-]+$") def _normalize_skill_name(name: str | None) -> str: raw = str(name or "") return re.sub(r"\s+", "_", raw.strip()) def _default_sandbox_skill_path(name: str) -> str: return f"{SANDBOX_WORKSPACE_ROOT}/{SANDBOX_SKILLS_ROOT}/{name}/SKILL.md" def _normalize_cached_sandbox_skill_path(name: str, path: str) -> str: normalized = str(path or "").strip().replace("\\", "/") if not normalized: return _default_sandbox_skill_path(name) pure_path = PurePosixPath(normalized) if ".." in pure_path.parts: return _default_sandbox_skill_path(name) if pure_path.name != "SKILL.md": return _default_sandbox_skill_path(name) if pure_path.parent.name != name: return _default_sandbox_skill_path(name) return str(pure_path) def _is_ignored_zip_entry(name: str) -> bool: parts = PurePosixPath(name).parts if not parts: return True return parts[0] == "__MACOSX" def _normalize_skill_markdown_path( skill_dir: Path, *, rename_legacy: bool = True, ) -> Path | None: """Return the canonical `SKILL.md` path for a skill directory. If only legacy `skill.md` exists, it is renamed to `SKILL.md` in-place. """ canonical = skill_dir / "SKILL.md" entries = set() if skill_dir.exists(): entries = {entry.name for entry in skill_dir.iterdir()} if "SKILL.md" in entries: return canonical legacy = skill_dir / "skill.md" if "skill.md" not in entries: return None try: if not rename_legacy: return legacy tmp = skill_dir / f".{uuid.uuid4().hex}.tmp_skill_md" legacy.rename(tmp) tmp.rename(canonical) except OSError: return legacy return canonical @dataclass class SkillInfo: name: str description: str path: str active: bool source_type: str = "local_only" source_label: str = "local" local_exists: bool = True sandbox_exists: bool = False plugin_name: str = "" readonly: bool = False input_schema: dict | None = None output_schema: dict | None = None def _parse_frontmatter(text: str) -> dict: """Extract metadata from YAML frontmatter. Expects the standard SKILL.md format used by OpenAI Codex CLI and Anthropic Claude Skills:: --- name: my-skill description: What this skill does and when to use it. input_schema: ... output_schema: ... --- """ if not text.startswith("---"): return {} lines = text.splitlines() if not lines or lines[0].strip() != "---": return {} end_idx = None for i in range(1, len(lines)): if lines[i].strip() == "---": end_idx = i break if end_idx is None: return {} frontmatter = "\n".join(lines[1:end_idx]) try: payload = yaml.safe_load(frontmatter) or {} except yaml.YAMLError: return {} if not isinstance(payload, dict): return {} return payload def _parse_frontmatter_description(text: str) -> str: """Extract the ``description`` value from YAML frontmatter. Expects the standard SKILL.md format used by OpenAI Codex CLI and Anthropic Claude Skills:: --- name: my-skill description: What this skill does and when to use it. --- """ if not text.startswith("---"): return "" lines = text.splitlines() if not lines or lines[0].strip() != "---": return "" end_idx = None for i in range(1, len(lines)): if lines[i].strip() == "---": end_idx = i break if end_idx is None: return "" frontmatter = "\n".join(lines[1:end_idx]) try: payload = yaml.safe_load(frontmatter) or {} except yaml.YAMLError: return "" if not isinstance(payload, dict): return "" description = payload.get("description", "") if not isinstance(description, str): return "" return description.strip() # Regex for sanitizing paths used in prompt examples — only allow # safe path characters to prevent prompt injection via crafted skill paths. _SAFE_PATH_RE = re.compile(r"[^\w./ ,()'\-]", re.UNICODE) _WINDOWS_DRIVE_PATH_RE = re.compile(r"^[A-Za-z]:(?:/|\\)") _WINDOWS_UNC_PATH_RE = re.compile(r"^(//|\\\\)[^/\\]+[/\\][^/\\]+") _CONTROL_CHARS_RE = re.compile(r"[\x00-\x1F\x7F]") def _is_windows_prompt_path(path: str) -> bool: if os.name != "nt": return False return bool(_WINDOWS_DRIVE_PATH_RE.match(path) or _WINDOWS_UNC_PATH_RE.match(path)) def _sanitize_prompt_path_for_prompt(path: str) -> str: if not path: return "" if _WINDOWS_DRIVE_PATH_RE.match(path) or _WINDOWS_UNC_PATH_RE.match(path): path = path.replace("\\", "/") drive_prefix = "" if _WINDOWS_DRIVE_PATH_RE.match(path): drive_prefix = path[:2] path = path[2:] path = path.replace("`", "") path = _CONTROL_CHARS_RE.sub("", path) sanitized = _SAFE_PATH_RE.sub("", path) return f"{drive_prefix}{sanitized}" def _sanitize_prompt_description(description: str) -> str: description = description.replace("`", "") description = _CONTROL_CHARS_RE.sub(" ", description) description = " ".join(description.split()) return description def _sanitize_skill_display_name(name: str) -> str: if _SKILL_NAME_RE.fullmatch(name): return name return "" def _build_skill_read_command_example(path: str) -> str: if path == "//SKILL.md": return f"cat {path}" if _is_windows_prompt_path(path): command = "type" path_arg = f'"{os.path.normpath(path)}"' else: command = "cat" path_arg = shlex.quote(path) return f"{command} {path_arg}" def build_skills_prompt(skills: list[SkillInfo]) -> str: """Build the skills section of the system prompt. Generates a markdown-formatted skill inventory for the LLM. Only ``name`` and ``description`` are shown upfront; the LLM must read the full ``SKILL.md`` before execution (progressive disclosure). """ skills_lines: list[str] = [] example_path = "" for skill in skills: display_name = _sanitize_skill_display_name(skill.name) description = skill.description or "No description" if skill.source_type in {"sandbox_only", "workspace"}: description = _sanitize_prompt_description(description) if not description: description = "Read SKILL.md for details." if skill.source_type == "sandbox_only": # Prefer the actual path from sandbox cache if available rendered_path = _sanitize_prompt_path_for_prompt(skill.path) if not rendered_path: rendered_path = _default_sandbox_skill_path(skill.name) else: rendered_path = _sanitize_prompt_path_for_prompt(skill.path) if not rendered_path: rendered_path = "//SKILL.md" entry = f"- **{display_name}**: {description}\n File: `{rendered_path}`" if skill.input_schema: entry += f"\n Input Schema: {json.dumps(skill.input_schema, ensure_ascii=False)}" if skill.output_schema: entry += f"\n Output Schema: {json.dumps(skill.output_schema, ensure_ascii=False)}" skills_lines.append(entry) if not example_path: example_path = rendered_path skills_block = "\n".join(skills_lines) # Sanitize example_path — it may originate from sandbox cache (untrusted) if example_path == "//SKILL.md": example_path = "//SKILL.md" else: example_path = _sanitize_prompt_path_for_prompt(example_path) example_path = example_path or "//SKILL.md" example_command = _build_skill_read_command_example(example_path) return ( "## Skills\n\n" "You have specialized skills — reusable instruction bundles stored " "in `SKILL.md` files. Each skill has a **name** and a **description** " "that tells you what it does and when to use it.\n\n" "### Available skills\n\n" f"{skills_block}\n\n" "### Skill rules\n\n" "1. **Discovery** — The list above is the complete skill inventory " "for this session. Full instructions are in the referenced " "`SKILL.md` file.\n" "2. **When to trigger** — Use a skill if the user names it " "explicitly, or if the task clearly matches the skill's description. " "*Never silently skip a matching skill* — either use it or briefly " "explain why you chose not to.\n" "3. **Mandatory grounding** — Before executing any skill you MUST " "first read its `SKILL.md` by running a shell command compatible " "with the current runtime shell and using the **absolute path** " f"shown above (e.g. `{example_command}`). " "Never rely on memory or assumptions about a skill's content.\n" "4. **Progressive disclosure** — Load only what is directly " "referenced from `SKILL.md`:\n" " - If `scripts/` exist, prefer running or patching them over " "rewriting code from scratch.\n" " - If `assets/` or templates exist, reuse them.\n" " - Do NOT bulk-load every file in the skill directory.\n" "5. **Coordination** — When multiple skills apply, pick the minimal " "set needed. Announce which skill(s) you are using and why " "(one short line). Prefer `astrbot_*` tools when running skill " "scripts.\n" "6. **Context hygiene** — Avoid deep reference chasing; open only " "files that are directly linked from `SKILL.md`.\n" "7. **Failure handling** — If a skill cannot be applied, state the " "issue clearly and continue with the best alternative.\n" ) class SkillManager: def __init__( self, skills_root: str | None = None, plugins_root: str | None = None, astrbot_paths: AstrbotPaths | None = None, ) -> None: if astrbot_paths is not None: self.skills_root = skills_root or str(astrbot_paths.skills) self.plugins_root = plugins_root or get_astrbot_plugin_path() self.config_path = str(astrbot_paths.config / SKILLS_CONFIG_FILENAME) self.sandbox_skills_cache_path = str( astrbot_paths.data / SANDBOX_SKILLS_CACHE_FILENAME, ) else: self.skills_root = skills_root or get_astrbot_skills_path() self.plugins_root = plugins_root or get_astrbot_plugin_path() data_path = Path(get_astrbot_data_path()) self.config_path = str(data_path / SKILLS_CONFIG_FILENAME) self.sandbox_skills_cache_path = str( data_path / SANDBOX_SKILLS_CACHE_FILENAME, ) os.makedirs(self.skills_root, exist_ok=True) def _iter_plugin_skill_dirs(self) -> list[tuple[str, str, Path]]: """Return plugin-provided skill directories as (skill, plugin, dir).""" plugins_root = Path(self.plugins_root) if not plugins_root.is_dir(): return [] result: list[tuple[str, str, Path]] = [] for plugin_dir in sorted(plugins_root.iterdir(), key=lambda item: item.name): if not plugin_dir.is_dir(): continue plugin_name = plugin_dir.name skills_dir = plugin_dir / "skills" if not skills_dir.is_dir(): continue direct_skill_md = _normalize_skill_markdown_path( skills_dir, rename_legacy=False, ) if direct_skill_md is not None and _SKILL_NAME_RE.match(plugin_name): result.append((plugin_name, plugin_name, skills_dir)) for skill_dir in sorted(skills_dir.iterdir(), key=lambda item: item.name): if not skill_dir.is_dir(): continue skill_name = skill_dir.name if not _SKILL_NAME_RE.match(skill_name): continue if ( _normalize_skill_markdown_path(skill_dir, rename_legacy=False) is None ): continue result.append((skill_name, plugin_name, skill_dir)) return result def _get_plugin_skill_dir(self, name: str) -> Path | None: for skill_name, _plugin_name, skill_dir in self._iter_plugin_skill_dirs(): if skill_name == name: return skill_dir return None def list_workspace_skills( self, workspace_root: str | Path | None ) -> list[SkillInfo]: """List request-scoped skills from a session workspace. Args: workspace_root: The current session workspace directory. Returns: Skills discovered under ``/skills``. """ if not workspace_root: return [] raw_workspace_root = Path(workspace_root) skills_root = raw_workspace_root / WORKSPACE_SKILLS_ROOT if not skills_root.is_dir(): return [] try: resolved_workspace_root = raw_workspace_root.resolve(strict=True) resolved_skills_root = skills_root.resolve(strict=True) if not resolved_skills_root.is_relative_to(resolved_workspace_root): return [] skill_dirs = sorted( resolved_skills_root.iterdir(), key=lambda item: item.name ) except OSError: return [] skills: list[SkillInfo] = [] for skill_dir in skill_dirs: if not skill_dir.is_dir(): continue skill_name = skill_dir.name if not _SKILL_NAME_RE.match(skill_name): continue try: entry_names = {entry.name for entry in skill_dir.iterdir()} except OSError: continue if "SKILL.md" not in entry_names: continue skill_md = skill_dir / "SKILL.md" if not skill_md.is_file(): continue try: resolved_skill_md = skill_md.resolve(strict=True) except OSError: continue if not resolved_skill_md.is_relative_to(resolved_skills_root): continue description = "" try: with resolved_skill_md.open(encoding="utf-8") as f: content = f.read(WORKSPACE_SKILL_FRONTMATTER_MAX_CHARS) description = _parse_frontmatter_description(content) except (OSError, UnicodeError): description = "" skills.append( SkillInfo( name=skill_name, description=description, path=resolved_skill_md.as_posix(), active=True, source_type="workspace", source_label="workspace", local_exists=True, readonly=True, ) ) return skills def _load_config(self) -> dict: if not os.path.exists(self.config_path): self._save_config(DEFAULT_SKILLS_CONFIG.copy()) return DEFAULT_SKILLS_CONFIG.copy() with open(self.config_path, encoding="utf-8") as f: data = json.load(f) if not isinstance(data, dict) or "skills" not in data: return DEFAULT_SKILLS_CONFIG.copy() return data def _save_config(self, config: dict) -> None: with open(self.config_path, "w", encoding="utf-8") as f: json.dump(config, f, ensure_ascii=False, indent=4) def _load_sandbox_skills_cache(self) -> dict: if not os.path.exists(self.sandbox_skills_cache_path): return {"version": _SANDBOX_SKILLS_CACHE_VERSION, "skills": []} try: with open(self.sandbox_skills_cache_path, encoding="utf-8") as f: data = json.load(f) if not isinstance(data, dict): return {"version": _SANDBOX_SKILLS_CACHE_VERSION, "skills": []} skills = data.get("skills", []) if not isinstance(skills, list): skills = [] return { "version": int(data.get("version", _SANDBOX_SKILLS_CACHE_VERSION)), "skills": skills, "updated_at": data.get("updated_at"), } except Exception: return {"version": _SANDBOX_SKILLS_CACHE_VERSION, "skills": []} def _save_sandbox_skills_cache(self, cache: dict) -> None: cache["version"] = _SANDBOX_SKILLS_CACHE_VERSION cache["updated_at"] = datetime.now(timezone.utc).isoformat() with open(self.sandbox_skills_cache_path, "w", encoding="utf-8") as f: json.dump(cache, f, ensure_ascii=False, indent=2) def set_sandbox_skills_cache(self, skills: list[dict]) -> None: """Persist sandbox skill metadata discovered from runtime side.""" deduped: dict[str, dict[str, str]] = {} for item in skills: if not isinstance(item, dict): continue name = str(item.get("name", "")).strip() if not name or not _SKILL_NAME_RE.match(name): continue description = str(item.get("description", "") or "") path = _normalize_cached_sandbox_skill_path( name, str(item.get("path", "") or ""), ) deduped[name] = { "name": name, "description": description, "path": path, } cache = { "version": _SANDBOX_SKILLS_CACHE_VERSION, "skills": [deduped[name] for name in sorted(deduped)], } self._save_sandbox_skills_cache(cache) def get_sandbox_skills_cache_status(self) -> dict[str, object]: cache = self._load_sandbox_skills_cache() skills = cache.get("skills", []) count = len(skills) if isinstance(skills, list) else 0 return { "exists": os.path.exists(self.sandbox_skills_cache_path), "ready": count > 0, "count": count, "updated_at": cache.get("updated_at"), } def list_skills( self, *, active_only: bool = False, runtime: str = "local", show_sandbox_path: bool = True, ) -> list[SkillInfo]: """List all skills. show_sandbox_path: If True and runtime is "sandbox", return the path as it would appear in the sandbox environment, otherwise return the local filesystem path. """ config = self._load_config() skill_configs = config.get("skills", {}) modified = False skills_by_name: dict[str, SkillInfo] = {} sandbox_cached_paths: dict[str, str] = {} sandbox_cached_descriptions: dict[str, str] = {} cache_for_paths = self._load_sandbox_skills_cache() for item in cache_for_paths.get("skills", []): if not isinstance(item, dict): continue name = str(item.get("name", "") or "").strip() path = _normalize_cached_sandbox_skill_path( name, str(item.get("path", "") or ""), ) if not name or not _SKILL_NAME_RE.match(name): continue sandbox_cached_descriptions[name] = str(item.get("description", "") or "") sandbox_cached_paths[name] = path for entry in sorted(Path(self.skills_root).iterdir()): if not entry.is_dir(): continue skill_name = entry.name skill_md = _normalize_skill_markdown_path(entry) if skill_md is None: continue active = skill_configs.get(skill_name, {}).get("active", True) if skill_name not in skill_configs: skill_configs[skill_name] = {"active": active} modified = True if active_only and not active: continue description = "" input_schema = None output_schema = None try: content = skill_md.read_text(encoding="utf-8") meta = _parse_frontmatter(content) description = meta.get("description", "") if not isinstance(description, str): description = "" description = description.strip() input_schema = meta.get("input_schema") output_schema = meta.get("output_schema") except Exception: description = "" input_schema = None output_schema = None sandbox_exists = ( runtime == "sandbox" and skill_name in sandbox_cached_descriptions ) source_type = "both" if sandbox_exists else "local_only" source_label = "synced" if sandbox_exists else "local" if runtime == "sandbox" and show_sandbox_path: path_str = sandbox_cached_paths.get( skill_name, ) or _default_sandbox_skill_path(skill_name) else: path_str = str(skill_md) path_str = path_str.replace("\\", "/") skills_by_name[skill_name] = SkillInfo( name=skill_name, description=description, path=path_str, active=active, source_type=source_type, source_label=source_label, local_exists=True, sandbox_exists=sandbox_exists, input_schema=input_schema, output_schema=output_schema, ) for skill_name, plugin_name, skill_dir in self._iter_plugin_skill_dirs(): if skill_name in skills_by_name: continue skill_md = _normalize_skill_markdown_path(skill_dir, rename_legacy=False) if skill_md is None: continue active = skill_configs.get(skill_name, {}).get("active", True) if skill_name not in skill_configs: skill_configs[skill_name] = {"active": active} modified = True if active_only and not active: continue description = "" try: content = skill_md.read_text(encoding="utf-8") description = _parse_frontmatter_description(content) except Exception: description = "" sandbox_exists = ( runtime == "sandbox" and skill_name in sandbox_cached_descriptions ) if runtime == "sandbox" and show_sandbox_path: path_str = sandbox_cached_paths.get( skill_name, ) or _default_sandbox_skill_path(skill_name) else: path_str = str(skill_md) skills_by_name[skill_name] = SkillInfo( name=skill_name, description=description, path=path_str.replace("\\", "/"), active=active, source_type="plugin", source_label=plugin_name, local_exists=True, sandbox_exists=sandbox_exists, plugin_name=plugin_name, readonly=True, ) if runtime == "sandbox": cache = self._load_sandbox_skills_cache() for item in cache.get("skills", []): if not isinstance(item, dict): continue skill_name = str(item.get("name", "")).strip() if ( not skill_name or skill_name in skills_by_name or not _SKILL_NAME_RE.match(skill_name) ): continue active = skill_configs.get(skill_name, {}).get("active", True) if skill_name not in skill_configs: skill_configs[skill_name] = {"active": active} modified = True if active_only and not active: continue description = sandbox_cached_descriptions.get(skill_name, "") # For sandbox_only skills, show_sandbox_path is implicitly True # since there is no local path to show. Always prefer the # actual path from sandbox cache. path_str = sandbox_cached_paths.get( skill_name, ) or _default_sandbox_skill_path(skill_name) skills_by_name[skill_name] = SkillInfo( name=skill_name, description=description, path=path_str.replace("\\", "/"), active=active, source_type="sandbox_only", source_label="sandbox_preset", local_exists=False, sandbox_exists=True, ) if modified: config["skills"] = skill_configs self._save_config(config) return [skills_by_name[name] for name in sorted(skills_by_name)] def is_sandbox_only_skill(self, name: str) -> bool: skill_dir = Path(self.skills_root) / name skill_md_exists = _normalize_skill_markdown_path(skill_dir) is not None if skill_md_exists: return False cache = self._load_sandbox_skills_cache() skills = cache.get("skills", []) if not isinstance(skills, list): return False for item in skills: if not isinstance(item, dict): continue if str(item.get("name", "")).strip() == name: return True return False def is_plugin_skill(self, name: str) -> bool: return self._get_plugin_skill_dir(name) is not None def set_skill_active(self, name: str, active: bool) -> None: if self.is_sandbox_only_skill(name): raise PermissionError( "Sandbox preset skill cannot be enabled/disabled from local skill management.", ) config = self._load_config() config.setdefault("skills", {}) config["skills"][name] = {"active": bool(active)} self._save_config(config) def _remove_skill_from_sandbox_cache(self, name: str) -> None: cache = self._load_sandbox_skills_cache() skills = cache.get("skills", []) if not isinstance(skills, list): return filtered = [ item for item in skills if not ( isinstance(item, dict) and str(item.get("name", "")).strip() == name ) ] if len(filtered) != len(skills): cache["skills"] = filtered self._save_sandbox_skills_cache(cache) def delete_skill(self, name: str) -> None: if self.is_sandbox_only_skill(name): raise PermissionError( "Sandbox preset skill cannot be deleted from local skill management.", ) if self.is_plugin_skill(name): raise PermissionError( "Plugin-provided skill cannot be deleted from local skill management.", ) skill_dir = Path(self.skills_root) / name if skill_dir.exists(): shutil.rmtree(skill_dir) # Ensure UI consistency even when there is no active sandbox session # to refresh cache from runtime side. self._remove_skill_from_sandbox_cache(name) config = self._load_config() if name in config.get("skills", {}): config["skills"].pop(name, None) self._save_config(config) def install_skill_from_zip( self, zip_path: str, *, overwrite: bool = True, skill_name_hint: str | None = None, ) -> str: zip_path_obj = Path(zip_path) if not zip_path_obj.exists(): raise FileNotFoundError(f"Zip file not found: {zip_path}") if not zipfile.is_zipfile(zip_path): raise ValueError("Uploaded file is not a valid zip archive.") installed_skills = [] with zipfile.ZipFile(zip_path) as zf: names = [ name for name in (entry.replace("\\", "/") for entry in zf.namelist()) if name and not _is_ignored_zip_entry(name) ] file_names = [name for name in names if name and not name.endswith("/")] if not file_names: raise ValueError("Zip archive is empty.") has_root_skill_md = any( len(parts := PurePosixPath(name).parts) == 1 and parts[0] in {"SKILL.md", "skill.md"} for name in file_names ) root_mode = has_root_skill_md archive_skill_name = None if skill_name_hint is not None: archive_skill_name = _normalize_skill_name(skill_name_hint) if archive_skill_name and not _SKILL_NAME_RE.fullmatch( archive_skill_name, ): raise ValueError("Invalid skill name.") for name in names: if not name: continue if name.startswith("/") or re.match(r"^[A-Za-z]:", name): raise ValueError("Zip archive contains absolute paths.") parts = PurePosixPath(name).parts if ".." in parts: raise ValueError("Zip archive contains invalid relative paths.") if not root_mode and not overwrite: top_dirs = {PurePosixPath(n).parts[0] for n in file_names if n.strip()} conflict_dirs: list[str] = [] for src_dir_name in top_dirs: if ( f"{src_dir_name}/SKILL.md" not in file_names and f"{src_dir_name}/skill.md" not in file_names ): continue candidate_name = _normalize_skill_name(src_dir_name) if not candidate_name or not _SKILL_NAME_RE.fullmatch( candidate_name, ): continue if archive_skill_name and len(top_dirs) == 1: target_name = archive_skill_name else: target_name = candidate_name dest_dir = Path(self.skills_root) / target_name if dest_dir.exists(): conflict_dirs.append(str(dest_dir)) if conflict_dirs: raise FileExistsError( "One or more skills from the archive already exist and " "overwrite=False. No skills were installed. Conflicting " f"paths: {', '.join(conflict_dirs)}", ) with tempfile.TemporaryDirectory(dir=get_astrbot_temp_path()) as tmp_dir: for member in zf.infolist(): member_name = member.filename.replace("\\", "/") if not member_name or _is_ignored_zip_entry(member_name): continue zf.extract(member, tmp_dir) if root_mode: archive_hint = _normalize_skill_name( archive_skill_name or zip_path_obj.stem, ) if not archive_hint or not _SKILL_NAME_RE.fullmatch(archive_hint): raise ValueError("Invalid skill name.") skill_name = archive_hint src_dir = Path(tmp_dir) normalized_path = _normalize_skill_markdown_path(src_dir) if normalized_path is None: raise ValueError( "SKILL.md not found in the root of the zip archive.", ) dest_dir = Path(self.skills_root) / skill_name if dest_dir.exists() and overwrite: shutil.rmtree(dest_dir) elif dest_dir.exists() and not overwrite: raise FileExistsError(f"Skill {skill_name} already exists.") shutil.move(str(src_dir), str(dest_dir)) self.set_skill_active(skill_name, True) installed_skills.append(skill_name) else: top_dirs = { PurePosixPath(n).parts[0] for n in file_names if n.strip() } for archive_root_name in top_dirs: archive_root_name_normalized = _normalize_skill_name( archive_root_name, ) if ( f"{archive_root_name}/SKILL.md" not in file_names and f"{archive_root_name}/skill.md" not in file_names ): continue if archive_root_name in {".", "..", ""} or not ( _SKILL_NAME_RE.fullmatch(archive_root_name_normalized) ): continue if archive_skill_name and len(top_dirs) == 1: skill_name = archive_skill_name else: skill_name = archive_root_name_normalized src_dir = Path(tmp_dir) / archive_root_name normalized_path = _normalize_skill_markdown_path(src_dir) if normalized_path is None: continue dest_dir = Path(self.skills_root) / skill_name if dest_dir.exists(): if not overwrite: raise FileExistsError( f"Skill {skill_name} already exists.", ) shutil.rmtree(dest_dir) shutil.move(str(src_dir), str(dest_dir)) self.set_skill_active(skill_name, True) installed_skills.append(skill_name) if not installed_skills: raise ValueError( "No valid SKILL.md found in any folder of the zip archive.", ) return ", ".join(installed_skills)