CmdForge/src/cmdforge/skills.py

"""Provider-attached skills in the Agent Skills open standard.

Each provider can have skills at::

    ~/.cmdforge/providers/<provider>/skills/<skill-name>/SKILL.md

Every ``SKILL.md`` must contain the required Agent Skills frontmatter. The
directory name is the canonical skill identifier and must match ``name``.
"""

import re
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Optional

import yaml

PROVIDERS_DIR = Path.home() / ".cmdforge" / "providers"

_PROVIDER_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]*$")
_SKILL_NAME_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
_MISSING_POLICIES = {"warn", "error", "ignore"}


@dataclass(frozen=True)
class Skill:
    """A single loaded skill."""

    name: str
    directory: str
    description: str
    content: str = ""

    def __str__(self) -> str:
        return self.content


@dataclass
class SkillSet:
    """Collection of skills for one provider, deterministically ordered."""

    provider_name: str
    skills: List[Skill] = field(default_factory=list)

    def get(
        self,
        requested: Optional[List[str]] = None,
        on_missing: str = "warn",
    ) -> List[Skill]:
        """Return enabled skills, reporting missing ones."""
        if on_missing not in _MISSING_POLICIES:
            raise ValueError(
                f"Invalid missing-skill policy '{on_missing}'; expected one of "
                f"{sorted(_MISSING_POLICIES)}"
            )

        available = {skill.name: skill for skill in self.skills}
        if requested is None:
            return list(self.skills)

        enabled: List[Skill] = []
        for name in requested:
            if name in available:
                enabled.append(available[name])
                continue

            message = (
                f"Skill '{name}' requested for provider '{self.provider_name}' "
                "but not found in skills directory"
            )
            if on_missing == "error":
                raise KeyError(message)
            if on_missing == "warn":
                print(f"[skills] {message}", file=sys.stderr)
        return enabled


def _validate_provider_name(name: str) -> None:
    """Validate a provider name before using it as a path component."""
    if not isinstance(name, str) or not _PROVIDER_NAME_RE.fullmatch(name):
        raise ValueError(
            f"Invalid provider name '{name}'. Must match {_PROVIDER_NAME_RE.pattern}"
        )


def _validate_skill_name(name: str, provider: str = "") -> None:
    """Validate the Agent Skills canonical name rules."""
    context = f" under provider '{provider}'" if provider else ""
    if (
        not isinstance(name, str)
        or not 1 <= len(name) <= 64
        or not _SKILL_NAME_RE.fullmatch(name)
    ):
        raise ValueError(
            f"Invalid skill name '{name}'{context}. Skill names must be 1-64 "
            "lowercase letters, numbers, or single hyphens, and may not start "
            "or end with a hyphen"
        )


def _parse_frontmatter(text: str) -> tuple[dict, str]:
    """Parse the mandatory YAML frontmatter from a ``SKILL.md`` file."""
    lines = text.splitlines(keepends=True)
    if not lines or lines[0].rstrip("\r\n") != "---":
        raise ValueError("SKILL.md must begin with YAML frontmatter")

    closing_index = next(
        (index for index, line in enumerate(lines[1:], start=1)
         if line.rstrip("\r\n") == "---"),
        None,
    )
    if closing_index is None:
        raise ValueError("SKILL.md frontmatter is not terminated")

    frontmatter_raw = "".join(lines[1:closing_index]).strip()
    if not frontmatter_raw:
        raise ValueError("SKILL.md frontmatter is empty; metadata is required")
    try:
        metadata = yaml.safe_load(frontmatter_raw)
    except yaml.YAMLError as exc:
        raise ValueError(f"Invalid YAML in SKILL.md frontmatter: {exc}") from exc
    if not isinstance(metadata, dict):
        raise ValueError("SKILL.md frontmatter must be a YAML mapping")

    content = "".join(lines[closing_index + 1:]).lstrip("\r\n").rstrip()
    return metadata, content


def _validate_metadata(metadata: dict, directory_name: str, provider: str) -> tuple[str, str]:
    """Validate required Agent Skills metadata and return normalized values."""
    if "name" not in metadata:
        raise ValueError(f"SKILL.md for '{directory_name}' is missing required 'name'")
    name = metadata["name"]
    _validate_skill_name(name, provider)
    if name != directory_name:
        raise ValueError(
            f"SKILL.md frontmatter name '{name}' does not match skill directory "
            f"name '{directory_name}' under provider '{provider}'"
        )

    if "description" not in metadata:
        raise ValueError(
            f"SKILL.md for '{directory_name}' is missing required 'description'"
        )
    description = metadata["description"]
    if not isinstance(description, str) or not description.strip():
        raise ValueError(
            f"SKILL.md description for '{directory_name}' must be a non-empty string"
        )
    if len(description) > 1024:
        raise ValueError(
            f"SKILL.md description for '{directory_name}' must not exceed 1024 characters"
        )
    return name, description.strip()


def _reject_symlink(path: Path, label: str) -> None:
    if path.is_symlink():
        raise ValueError(f"Refusing symlinked {label}: {path}")


def load_provider_skills(provider_name: str) -> SkillSet:
    """Load and validate all skills configured for a provider."""
    _validate_provider_name(provider_name)

    provider_dir = PROVIDERS_DIR / provider_name
    skills_dir = provider_dir / "skills"
    _reject_symlink(provider_dir, "provider directory")
    _reject_symlink(skills_dir, "skills directory")
    if not skills_dir.is_dir():
        return SkillSet(provider_name=provider_name)

    skills: List[Skill] = []
    for entry in sorted(skills_dir.iterdir(), key=lambda path: path.name):
        _reject_symlink(entry, "skill directory")
        if not entry.is_dir():
            continue
        _validate_skill_name(entry.name, provider_name)

        skill_md = entry / "SKILL.md"
        _reject_symlink(skill_md, "SKILL.md")
        if not skill_md.is_file():
            continue
        try:
            text = skill_md.read_text(encoding="utf-8")
        except OSError as exc:
            raise OSError(f"Cannot read {skill_md}: {exc}") from exc

        metadata, content = _parse_frontmatter(text)
        name, description = _validate_metadata(
            metadata, entry.name, provider_name
        )
        skills.append(
            Skill(
                name=name,
                directory=str(entry),
                description=description,
                content=content,
            )
        )

    return SkillSet(provider_name=provider_name, skills=skills)


def inject_skills(
    prompt: str,
    provider_name: str,
    skill_names: Optional[List[str]] = None,
    on_missing: str = "warn",
) -> str:
    """Prepend enabled provider skills to a prompt."""
    skillset = load_provider_skills(provider_name)
    enabled = skillset.get(skill_names, on_missing=on_missing)
    if not enabled:
        return prompt

    skill_context = "\n\n".join(
        f"## Skill: {skill.name}\n\n{skill.content}" for skill in enabled
    )
    return f"{skill_context}\n\n---\n\n{prompt}"