client/src/toolstore/skill_manager.py

"""
Agent Skills support for ToolStore.

Implements the agentskills.io open standard:
- Parses SKILL.md files (YAML frontmatter + Markdown body)
- Validates against the spec
- Manages skill directories
- Registers skills as first-class tools (type: "skill")
- Supports progressive disclosure (metadata → instructions → resources)
"""

from __future__ import annotations

import re
from pathlib import Path
from typing import Dict, List, Any, Optional


# ---------------------------------------------------------------------------
# YAML frontmatter parser (lightweight, no PyYAML dependency needed for
# simple key-value YAML used in SKILL.md)
# ---------------------------------------------------------------------------

_FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)

# Allowed fields per the agentskills.io spec
_ALLOWED_FRONTMATTER = {
    "name", "description", "license", "compatibility",
    "metadata", "allowed-tools",
}


def _parse_yaml_simple(text: str) -> Dict[str, Any]:
    """Parse a flat YAML mapping (keys: values only, no nesting beyond
    the metadata dict)."""
    result: Dict[str, Any] = {}
    current_key: Optional[str] = None
    current_nested: Dict[str, str] = {}

    for line in text.split("\n"):
        # Skip empty lines and comments
        stripped = line.strip()
        if not stripped or stripped.startswith("#"):
            continue

        # Nested under 'metadata:'
        if line.startswith("  ") and current_key == "metadata":
            m = re.match(r"  (\w[\w-]*):\s*(.*)", line)
            if m:
                current_nested[m.group(1)] = m.group(2).strip()
            continue

        # Flush nested dict
        if current_key == "metadata" and current_nested:
            result["metadata"] = dict(current_nested)
            current_nested = {}
            current_key = None

        # Top-level key: value
        m = re.match(r"^(\w[\w-]*):\s*(.*)", stripped)
        if m:
            key = m.group(1)
            value = m.group(2).strip().strip('"').strip("'")
            if key == "metadata":
                current_key = key
                current_nested = {}
            else:
                result[key] = value

    # Flush any trailing nested
    if current_key == "metadata" and current_nested:
        result["metadata"] = dict(current_nested)

    return result


# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------

_NAME_RE = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*$")


def validate_skill_name(name: str) -> List[str]:
    """Validate a skill name against the agentskills.io spec. Returns a
    list of error messages (empty = valid)."""
    errors = []
    if not name:
        errors.append("name is required")
        return errors
    if len(name) > 64:
        errors.append(f"name too long ({len(name)} > 64)")
    if not _NAME_RE.match(name):
        errors.append(f"name {name!r} must be lowercase letters, numbers, "
                       "single hyphens only")
    return errors


def validate_skill_description(desc: str) -> List[str]:
    errors = []
    if not desc or not desc.strip():
        errors.append("description is required and must be non-empty")
    elif len(desc) > 1024:
        errors.append(f"description too long ({len(desc)} > 1024)")
    return errors


# ---------------------------------------------------------------------------
# SKILL.md parser
# ---------------------------------------------------------------------------

class SkillDefinition:
    """Parsed representation of a Skill."""

    def __init__(self, skill_dir: Path):
        self.skill_dir = Path(skill_dir).resolve()
        self.skill_file = self.skill_dir / "SKILL.md"
        self.name: str = ""
        self.description: str = ""
        self.license: str = ""
        self.compatibility: str = ""
        self.metadata: Dict[str, str] = {}
        self.allowed_tools: str = ""
        self.body: str = ""
        self.errors: List[str] = []
        self._loaded = False

    def load(self) -> bool:
        """Parse and validate SKILL.md. Returns True if valid."""
        self._loaded = True
        if not self.skill_file.exists():
            self.errors.append(f"SKILL.md not found at {self.skill_file}")
            return False

        raw = self.skill_file.read_text(encoding="utf-8")

        # Extract frontmatter
        fm_match = _FRONTMATTER_RE.match(raw)
        if not fm_match:
            self.errors.append("Missing or malformed YAML frontmatter "
                               "(must start with --- ... ---)")
            return False

        fm_text = fm_match.group(1)
        self.body = raw[fm_match.end():].strip()

        # Parse frontmatter
        fm = _parse_yaml_simple(fm_text)

        # Unknown fields
        for key in fm:
            if key not in _ALLOWED_FRONTMATTER:
                self.errors.append(f"Unknown frontmatter field: {key!r}")

        self.name = fm.get("name", "")
        self.description = fm.get("description", "")
        self.license = fm.get("license", "")
        self.compatibility = fm.get("compatibility", "")
        self.metadata = fm.get("metadata", {})
        self.allowed_tools = fm.get("allowed-tools", "")

        # Validate required fields
        self.errors.extend(validate_skill_name(self.name))
        self.errors.extend(validate_skill_description(self.description))

        # Check directory name matches
        if self.name and self.skill_dir.name != self.name:
            self.errors.append(
                f"Directory name {self.skill_dir.name!r} does not match "
                f"skill name {self.name!r}")

        return len(self.errors) == 0

    def list_files(self) -> List[str]:
        """List all additional files bundled with the skill (excluding SKILL.md)."""
        files = []
        if not self.skill_dir.is_dir():
            return files
        for f in self.skill_dir.rglob("*"):
            if f.is_file() and f.name != "SKILL.md":
                files.append(str(f.relative_to(self.skill_dir)))
        return sorted(files)

    # ------------------------------------------------------------------
    # Convenience accessors for standard subdirectories
    # ------------------------------------------------------------------

    def _dir(self, subdir: str) -> Path:
        """Return the path to a standard subdirectory (may not exist)."""
        return self.skill_dir / subdir

    def _list_dir(self, subdir: str) -> list[str]:
        """List files inside a standard subdirectory, relative to the skill root."""
        d = self._dir(subdir)
        if not d.is_dir():
            return []
        result = []
        for f in sorted(d.rglob("*")):
            if f.is_file():
                result.append(str(f.relative_to(self.skill_dir)))
        return result

    @property
    def references(self) -> list[str]:
        """List files inside ``references/``, relative to skill root."""
        return self._list_dir("references")

    @property
    def scripts(self) -> list[str]:
        """List files inside ``scripts/``, relative to skill root."""
        return self._list_dir("scripts")

    @property
    def assets(self) -> list[str]:
        """List files inside ``assets/``, relative to skill root."""
        return self._list_dir("assets")

    def to_tool_definition(self) -> Dict[str, Any]:
        """Convert to a ToolStore-compatible tool definition (type: skill).

        Includes the full SKILL.md body for progressive disclosure and
        the list of bundled files for discovery — both available in the
        in-memory index without needing to re-read from disk.
        """
        definition = {
            "name": self.name,
            "type": "skill",
            "description": self.description,
            "skill_dir": str(self.skill_dir),
            "license": self.license,
            "compatibility": self.compatibility,
            "metadata": self.metadata,
            "allowed_tools": self.allowed_tools,
            "source": "skill",
            "schema": {
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "script": {
                            "type": "string",
                            "description": "Script to run from scripts/ dir (for action='run')",
                        },
            "action": {
                            "type": "string",
                            "enum": ["load", "files", "file", "run", "list-references", "read-reference"],
                            "description": "load = read full SKILL.md body (progressive disclosure), "
                                           "files = list all bundled files (references/ + scripts/ + assets/), "
                                           "file = read a specific file by path, "
                                           "run = execute a script from scripts/, "
                                           "list-references = list files in references/, "
                                           "read-reference = read a reference file",
                        },
                        "file_path": {
                            "type": "string",
                            "description": "Relative path of file to read "
                                           "(required for action='file')",
                        },
                    },
                    "required": ["action"],
                },
            },
        }

        # Include body for progressive disclosure at load time
        if self.body:
            definition["body"] = self.body

        # Include bundled file list for discovery
        files = self.list_files()
        if files:
            definition["skill_files_list"] = files

        return definition

    def to_upload_dict(self) -> Dict[str, Any]:
        """Serialize the skill for upload to the ToolStore server.

        Includes the full SKILL.md body and all bundled files as a
        {filename: content} dict suitable for the `POST /skills/publish`
        endpoint.
        """
        files: Dict[str, str] = {}
        for rel_path in self.list_files():
            abs_path = (self.skill_dir / rel_path).resolve()
            # Safety: ensure file is within skill dir
            if not str(abs_path).startswith(str(self.skill_dir)):
                continue
            try:
                content = abs_path.read_text(encoding="utf-8")
                files[str(rel_path)] = content
            except (UnicodeDecodeError, OSError):
                # Binary files: read as base64
                import base64
                raw = abs_path.read_bytes()
                files[str(rel_path)] = (
                    f"[base64]{base64.b64encode(raw).decode('ascii')}"
                )

        return {
            "name": self.name,
            "description": self.description,
            "version": "1.0.0",
            "body": self.body,
            "license": self.license,
            "compatibility": self.compatibility,
            "metadata": self.metadata,
            "allowed_tools": self.allowed_tools,
            "skill_files": files if files else None,
        }


# ---------------------------------------------------------------------------
# Skill manager
# ---------------------------------------------------------------------------

class SkillManager:
    """Manage skill directories: scan, load, validate, register."""

    def __init__(self, skill_dirs: List[str] = None):
        self._skill_dirs: List[Path] = [Path(d) for d in (skill_dirs or [])]
        self._skills: Dict[str, SkillDefinition] = {}

    def add_skill_dir(self, path: str) -> None:
        p = Path(path).expanduser().resolve()
        if p not in self._skill_dirs:
            self._skill_dirs.append(p)

    def scan(self) -> List[SkillDefinition]:
        """Scan all configured directories for skills (directories
        containing a SKILL.md file). Parses and validates each."""
        found: List[SkillDefinition] = []
        seen = set()

        for base in self._skill_dirs:
            if not base.is_dir():
                continue
            for skill_file in base.rglob("SKILL.md"):
                skill_dir = skill_file.parent
                key = str(skill_dir.resolve())
                if key in seen:
                    continue
                seen.add(key)
                sd = SkillDefinition(skill_dir)
                if sd.load():
                    self._skills[sd.name] = sd
                    found.append(sd)
        return found

    def get_skill(self, name: str) -> Optional[SkillDefinition]:
        return self._skills.get(name)

    def get_skill_body(self, name: str) -> Optional[str]:
        """Return the full SKILL.md body text for a loaded skill."""
        sd = self._skills.get(name)
        if not sd:
            return None
        # Re-read from disk to get fresh content
        if sd.skill_file.exists():
            raw = sd.skill_file.read_text(encoding="utf-8")
            fm_match = _FRONTMATTER_RE.match(raw)
            if fm_match:
                return raw[fm_match.end():].strip()
        return sd.body

    def get_skill_file(self, skill_name: str, file_path: str) -> Optional[str]:
        """Read a bundled file from a skill directory."""
        sd = self._skills.get(skill_name)
        if not sd:
            return None
        target = (sd.skill_dir / file_path).resolve()
        # Ensure it's within the skill dir (basic traversal guard)
        if not str(target).startswith(str(sd.skill_dir)):
            return None
        if target.is_file():
            return target.read_text(encoding="utf-8")
        return None

    def run_skill_script(self, skill_name: str, script: str) -> str:
        """Execute a script from the skill's scripts/ directory.

        Returns stdout of the script as a string, or an error message.
        """
        import subprocess
        sd = self._skills.get(skill_name)
        if not sd:
            return f"Error: Skill '{skill_name}' not found"
        target = (sd.skill_dir / "scripts" / script).resolve()
        if not str(target).startswith(str(sd.skill_dir)):
            return "Error: Script path escapes skill directory"
        if not target.is_file():
            return f"Error: Script '{script}' not found in skill '{skill_name}'"
        try:
            result = subprocess.run(
                [str(target)], capture_output=True, text=True, timeout=30,
                cwd=str(sd.skill_dir)
            )
            output = result.stdout
            if result.stderr:
                output += "\n[stderr]\n" + result.stderr
            if result.returncode != 0:
                output += f"\n[exit code: {result.returncode}]"
            return output or "(script produced no output)"
        except subprocess.TimeoutExpired:
            return "Error: Script timed out (30s)"
        except Exception as exc:
            return f"Error running script: {str(exc)}"

    def to_tool_definitions(self) -> List[Dict[str, Any]]:
        """Convert all loaded skills to ToolStore tool definitions."""
        return [sd.to_tool_definition() for sd in self._skills.values()]

    def list_skill_names(self) -> List[str]:
        return sorted(self._skills.keys())

    # ------------------------------------------------------------------
    # install — import a skill from an external path into a local skill dir
    # ------------------------------------------------------------------

    @property
    def skill_dirs(self) -> List[Path]:
        return list(self._skill_dirs)

    def install_skill(
        self,
        source_path: str | Path,
        target_base_dir: str | Path | None = None,
    ) -> SkillDefinition | None:
        """Install a skill from *source_path* into a local skill directory.

        1. Validates that *source_path* contains a valid ``SKILL.md``.
        2. Copies the entire directory into *target_base_dir* (defaults to
           the first configured skill dir, or ``~/.toolstore/installed-skills``
           when none are configured).
        3. Adds the target base dir to ``skill_dirs`` if it is not already
           tracked, then rescans so the new skill is immediately available.

        Returns the :class:`SkillDefinition` on success, ``None`` on failure.
        """
        import shutil

        source = Path(source_path).resolve()
        if not source.is_dir():
            return None

        # Validate the skill before copying anything
        sd = SkillDefinition(source)
        if not sd.load():
            return None

        # Determine where to place the skill.
        # Docker persistent volume always wins (survives container restarts),
        # then explicit target, then configured skill dirs, then home fallback.
        from toolstore.config_manager import _DOCKER_PERSISTENT_DIR
        if _DOCKER_PERSISTENT_DIR.parent.exists():
            target_base = _DOCKER_PERSISTENT_DIR / "skills"
        elif target_base_dir:
            target_base = Path(target_base_dir).resolve()
        elif self._skill_dirs:
            target_base = self._skill_dirs[0]
        else:
            target_base = Path.home() / ".toolstore" / "installed-skills"

        target_base.mkdir(parents=True, exist_ok=True)
        target = target_base / sd.name

        # Copy
        if target.exists():
            shutil.rmtree(target)
        shutil.copytree(source, target)

        # Register the target base dir so future scans pick it up
        if target_base not in self._skill_dirs:
            self._skill_dirs.append(target_base)

        # Rescan — the new skill lands in self._skills
        self.scan()

        return self._skills.get(sd.name)

    # ------------------------------------------------------------------
    # ad-hoc discovery (does NOT mutate internal state)
    # ------------------------------------------------------------------

    @staticmethod
    def discover_from_path(path: str | Path, *,
                           recursive: bool = True) -> "DiscoveryResult":
        """Scan an arbitrary path for skills without registering them.

        This is a convenience wrapper around
        :func:`toolstore.skill_discovery.discover_skills`.  It works for
        both single-skill directories and folder trees containing
        multiple skills.

        Returns a :class:`DiscoveryResult` with full validation details.
        """
        # Local import avoids issues when importing skill_manager alone.
        from toolstore.skill_discovery import discover_skills
        return discover_skills(Path(path), recursive=recursive)


# ---------------------------------------------------------------------------
# Singleton
# ---------------------------------------------------------------------------

_skill_manager: Optional[SkillManager] = None


def get_skill_manager(skill_dirs: List[str] = None) -> SkillManager:
    global _skill_manager
    if _skill_manager is None:
        _skill_manager = SkillManager(skill_dirs or [])
    elif skill_dirs:
        for d in skill_dirs:
            _skill_manager.add_skill_dir(d)
    return _skill_manager