Load and Discover SKILL.md Content

from pathlib import Path from typing import Dict, List, Optional, Set, Tuple, Union

import yaml

from camel.logger import get_logger from camel.toolkits.base import BaseToolkit from camel.toolkits.function_tool import FunctionTool

logger = get_logger(name)

class SkillToolkit(BaseToolkit): r"""Toolkit for loading SKILL.md content.

Skills are discovered from filesystem roots and loaded on-demand.
"""

def __init__(
    self,
    working_directory: Optional[str] = None,
    allowed_skills: Optional[Set[str]] = None,
    timeout: Optional[float] = None,
) -> None:
    super().__init__(timeout=timeout)
    self.working_directory = (
        Path(working_directory).resolve()
        if working_directory
        else Path.cwd().resolve()
    )
    self._skills_cache: Optional[Dict[str, Dict[str, str]]] = None
    self._allowed_skills = (
        None
        if allowed_skills is None
        else {name.strip() for name in allowed_skills if name.strip()}
    )

def _build_description(self) -> str:
    skills = self._get_skills()
    if not skills:
        return (
            "Load a skill to get detailed instructions for a specific "
            "task. No skills are currently available."
        )
    # Build skill summary lines
    summary_lines = [
        "Load a skill to get detailed instructions for a specific task.",
        "Skills provide specialized knowledge and step-by-step guidance.",
        "Use this when a task matches an available skill's description.",
        "Only the skills listed here are available:",
        "<available_skills>",
    ]
    for skill in skills.values():
        extra_files = self._list_skill_entries(Path(skill["path"]).parent)
        files_text = ", ".join(extra_files) if extra_files else "(none)"
        summary_lines.extend(
            [
                "  <skill>",
                f"    <name>{skill['name']}</name>",
                f"    <description>{skill['description']}</description>",
                f"    <path>{skill['path']}</path>",
                f"    <files>{files_text}</files>",
                "  </skill>",
            ]
        )
    summary_lines.append("</available_skills>")
    # Build markdown list for display
    available = ""
    for skill in skills.values():
        extra_files = self._list_skill_entries(Path(skill["path"]).parent)
        files_text = ", ".join(extra_files) if extra_files else "(none)"
        available += (
            f"- {skill['name']}: {skill['description']} "
            f"({skill['path']})\n"
            f"  - extra files: {files_text}\n"
        )
    return (
        "## Skills\n"
        "A skill is a set of local instructions to follow that is stored "
        "in a `SKILL.md` file. Below is the list of skills that can be "
        "used. Each entry includes a name, description, and file path so "
        "you can open the source for full instructions when using a "
        "specific skill. It also includes additional files in each skill "
        "folder so other files can be loaded on demand. This function can "
        "be used to load Skill.md content, for additional files you can "
        "use other tools to load\n"
        "### Available skills\n"
        f"{available}\n"
        "### How to use skills\n"
        "- Discovery: The list above is the skills available in this "
        "session (name + description + file path). Skill bodies live on "
        "disk at the listed paths.\n"
        "- Trigger rules: If the user names a skill or the task clearly "
        "matches a skill's description shown above, you should use that "
        "skill for this turn.\n"
        "- Missing/blocked: If a named skill isn't in the list or the "
        "path can't be read, say so briefly and continue with the best "
        "fallback.\n"
        "- How to use a skill (progressive disclosure):\n"
        "  1) After deciding to use a skill, open its `SKILL.md` from the "
        "filesystem. Read only enough to follow the workflow.\n"
        "  2) If `SKILL.md` points to extra folders such as `references/` "
        ", load only the specific files needed for the request; don't "
        "bulk-load everything.\n"
        "  3) If `scripts/` exist, use available execution tools "
        "(terminal, code runner) to run them instead of rewriting code.\n"
        "  4) If `assets/` or templates exist, reuse them instead of "
        "recreating from scratch.\n"
        "- Coordination and sequencing:\n"
        "  - If multiple skills apply, choose the minimal set that covers "
        "the request and state the order you'll use them.\n"
        "  - Announce which skill(s) you're using and why "
        "(one short line). If you skip an obvious skill, say why.\n"
        "- Context hygiene:\n"
        "  - Keep context small: summarize long sections instead of "
        "pasting them; only load extra files when needed.\n"
        "  - Avoid deep reference-chasing: prefer opening only files "
        "directly linked from `SKILL.md` unless you're blocked.\n"
        "  - When variants exist (frameworks, providers, domains), pick "
        "only the relevant reference file(s) and note that choice.\n"
        "- Safety and fallback: If a skill can't be applied cleanly "
        "(missing files, unclear instructions), state the issue, pick the "
        "next-best approach, and continue.\n"
        "\n" + "\n".join(summary_lines)
    )

def _get_skills(self) -> Dict[str, Dict[str, str]]:
    if self._skills_cache is None:
        self._skills_cache = self._scan_skills()
    if self._allowed_skills is None:
        return self._skills_cache
    return {
        name: skill
        for name, skill in self._skills_cache.items()
        if name in self._allowed_skills
    }

def clear_cache(self) -> None:
    r"""Clear the cached skills to force rescanning on next access."""
    self._skills_cache = None

def _scan_skills(self) -> Dict[str, Dict[str, str]]:
    r"""Scan all skill roots and collect skill metadata.

    Skills are discovered from roots in priority order:
    repo > user > system. If multiple skills have the same name,
    the first one found takes priority.

    Returns:
        Dict[str, Dict[str, str]]: Mapping of skill name to metadata.
    """
    skills: Dict[str, Dict[str, str]] = {}
    for scope, root in self._skill_roots():
        if not root.is_dir():
            continue
        for path in root.rglob("SKILL.md"):
            if self._is_hidden_path(path, root):
                continue
            skill = self._parse_skill(path)
            if not skill:
                continue
            name = skill["name"]
            # First skill with this name wins (earlier scope has priority)
            if name in skills:
                continue
            skills[name] = {
                "name": name,
                "description": skill["description"],
                "path": str(path),
                "scope": scope,
            }
    return skills

def _skill_roots(self) -> List[Tuple[str, Path]]:
    r"""Return skill root directories with their scope.

    Scopes are checked in order: repo -> user -> system.
    Earlier scopes take priority when skills have the same name.

    Returns:
        List[Tuple[str, Path]]: List of (scope, path) tuples.
    """
    # Skills are discovered in priority order:
    # 1. repo scope: project-specific skills in working directory
    # 2. user scope: user-level skills in home directory
    # 3. system scope: system-wide skills in /etc
    roots: List[Tuple[str, Path]] = [
        # Repo scope - project-specific skills
        ("repo", self.working_directory / ".camel" / "skills"),
        ("repo", self.working_directory / ".agents" / "skills"),
        # User scope - user-level skills
        ("user", Path.home() / ".camel" / "skills"),
        ("user", Path.home() / ".config" / "camel" / "skills"),
        # System scope - system-wide skills
        ("system", Path("/etc/camel/skills")),
    ]
    return roots

def _is_hidden_path(self, path: Path, root: Path) -> bool:
    r"""Check if a path contains hidden directories (starting with dot).

    Args:
        path (Path): The path to check.
        root (Path): The root directory to compute relative path from.

    Returns:
        bool: True if the path contains hidden directories.
    """
    try:
        rel = path.relative_to(root)
    except ValueError:
        rel = path
    return any(part.startswith(".") for part in rel.parts)

def _parse_skill(self, path: Path) -> Optional[Dict[str, str]]:
    r"""Parse a SKILL.md file and extract metadata.

    Args:
        path (Path): Path to the SKILL.md file.

    Returns:
        Optional[Dict[str, str]]: Parsed skill data with name,
            description, and body. Returns None if parsing fails.
    """
    try:
        contents = path.read_text(encoding="utf-8")
    except OSError as exc:
        logger.warning("Failed to read skill %s: %s", path, exc)
        return None

    frontmatter_text, body = self._split_frontmatter(contents)
    if frontmatter_text is None:
        logger.warning("Missing YAML frontmatter in %s", path)
        return None

    try:
        data = yaml.safe_load(frontmatter_text) or {}
    except yaml.YAMLError as exc:
        logger.warning("Invalid YAML frontmatter in %s: %s", path, exc)
        return None

    if not isinstance(data, dict):
        logger.warning("Frontmatter must be a mapping in %s", path)
        return None

    name = data.get("name")
    description = data.get("description")
    if not isinstance(name, str) or not isinstance(description, str):
        logger.warning("Skill missing name/description in %s", path)
        return None

    return {
        "name": name.strip(),
        "description": description.strip(),
        "body": body,
    }

def _split_frontmatter(self, contents: str) -> Tuple[Optional[str], str]:
    r"""Split YAML frontmatter from the body of a SKILL.md file.

    Args:
        contents (str): The full contents of the file.

    Returns:
        Tuple[Optional[str], str]: A tuple of (frontmatter, body).
            frontmatter is None if no valid frontmatter delimiter is found.
    """
    lines = contents.splitlines()
    if not lines or lines[0].strip() != "---":
        return None, contents
    for i in range(1, len(lines)):
        if lines[i].strip() == "---":
            frontmatter = "\n".join(lines[1:i])
            body = "\n".join(lines[i + 1 :])
            return frontmatter, body
    return None, contents

def list_skills(self) -> List[Dict[str, str]]:
    r"""List available skills without loading their full content.

    Returns:
        List[Dict[str, str]]: Skill metadata entries including name,
            description, path, and scope.
    """
    skills = self._get_skills()
    return [
        {
            "name": skill["name"],
            "description": skill["description"],
            "path": skill["path"],
            "scope": skill["scope"],
        }
        for skill in skills.values()
    ]

def list_skill_files(self, name: str) -> str:
    r"""List files and directories in a skill folder.

    Args:
        name (str): The skill identifier.

    Returns:
        str: Formatted list of files/directories, or error message.
    """
    skills = self._get_skills()
    skill = skills.get(name)
    if not skill:
        available = ", ".join(sorted(skills.keys())) or "none"
        return f"Error: Skill \"{name}\" not found. Available: {available}"

    skill_dir = Path(skill["path"]).parent
    lines = [f"Files in {name}:"]
    for entry in self._list_skill_entries(
        skill_dir, include_skill_md=True
    ):
        lines.append(f"  - {entry}")
    return "\n".join(lines)

def _list_skill_entries(
    self, skill_dir: Path, include_skill_md: bool = False
) -> List[str]:
    r"""List direct child entries in a skill directory.

    Args:
        skill_dir (Path): The skill directory to inspect.
        include_skill_md (bool): Whether to include ``SKILL.md``.

    Returns:
        List[str]: Sorted entry names. Directories end with ``/``.
    """
    entries: List[str] = []
    for item in sorted(skill_dir.iterdir()):
        if not include_skill_md and item.name == "SKILL.md":
            continue
        entries.append(f"{item.name}/" if item.is_dir() else item.name)
    return entries

def _load_single_skill(self, name: str) -> str:
    r"""Load a single skill by name.

    Args:
        name (str): The skill identifier.

    Returns:
        str: The skill content, or error message if not found.
    """
    skills = self._get_skills()
    skill = skills.get(name)
    if not skill:
        available = ", ".join(sorted(skills.keys())) or "none"
        return f"Error: Skill \"{name}\" not found. Available: {available}"

    path = Path(skill["path"])
    parsed = self._parse_skill(path)
    if not parsed:
        return f"Error: Failed to load skill from {path}"

    base_dir = path.parent
    body = parsed.get("body", "").strip()

    # List files in skill directory
    files_info = [
        f"  - {entry}" for entry in self._list_skill_entries(base_dir)
    ]

    output_lines = [
        f"## Skill: {parsed['name']}",
        "",
        f"**Base directory**: {base_dir}",
    ]

    if files_info:
        output_lines.append("")
        output_lines.append("**Available files**:")
        output_lines.extend(files_info)

    output_lines.append("")
    output_lines.append(body)

    return "\n".join(output_lines).strip()

def load_skill(self, name: Union[str, List[str]]) -> str:
    r"""Load one or more skills by name.

    Args:
        name (Union[str, List[str]]): A single skill name or list of names.

    Returns:
        str: The skill content(s), or error message if not found.
    """
    if isinstance(name, str):
        return self._load_single_skill(name)

    # Load multiple skills
    results = []
    for skill_name in name:
        results.append(self._load_single_skill(skill_name))

    return "\n\n---\n\n".join(results)

def get_tools(self) -> List[FunctionTool]:
    r"""Return the skill tools with injected available skills."""
    list_tool = FunctionTool(self.list_skills)
    load_tool = FunctionTool(self.load_skill)
    schema = load_tool.get_openai_tool_schema()
    schema["function"]["description"] = self._build_description()
    load_tool.set_openai_tool_schema(schema)
    return [list_tool, load_tool]