shira-hermes/api/services/skills.py

"""Skills engine — list, view, manage, and register skill tools."""

from __future__ import annotations

import logging
import os
import re
from pathlib import Path

import yaml

logger = logging.getLogger("shira.skills")

SKILLS_DIR = os.environ.get("SHIRA_DATA_DIR", "/opt/data") + "/skills"

_SAFE_NAME_RE = re.compile(r"^[a-zA-Z0-9_-]+$")


def _safe_skill_name(name: str) -> str:
    """Validate and normalize a skill name for use in file paths."""
    name = name.lower().strip()
    if not name or not _SAFE_NAME_RE.match(name):
        raise ValueError(f"Invalid skill name: {name!r}")
    return name


def _parse_frontmatter(text: str) -> tuple[dict, str]:
    """Parse YAML frontmatter from a SKILL.md file. Returns (meta, body)."""
    if not text.startswith("---"):
        return {}, text
    end = text.find("---", 3)
    if end == -1:
        return {}, text
    try:
        meta = yaml.safe_load(text[3:end]) or {}
    except yaml.YAMLError:
        meta = {}
    body = text[end + 3:].lstrip("\n")
    return meta, body


def list_skills() -> list[dict]:
    """Scan the skills directory and return metadata for each skill."""
    skills_path = Path(SKILLS_DIR)
    if not skills_path.exists():
        return []

    result = []
    for entry in sorted(skills_path.iterdir()):
        skill_file = entry / "SKILL.md" if entry.is_dir() else None
        if not skill_file or not skill_file.exists():
            continue
        try:
            meta, _ = _parse_frontmatter(skill_file.read_text(encoding="utf-8"))
            result.append({
                "name": meta.get("name", entry.name),
                "description": meta.get("description", ""),
                "version": meta.get("version", "1.0.0"),
            })
        except Exception as e:
            logger.warning("Failed to parse skill %s: %s", entry.name, e)
    return result


def view_skill(skill_name: str) -> str | None:
    """Load the full SKILL.md content for a given skill name."""
    skill_file = Path(SKILLS_DIR) / _safe_skill_name(skill_name) / "SKILL.md"
    if not skill_file.exists():
        return None
    return skill_file.read_text(encoding="utf-8")


def save_skill(skill_name: str, content: str) -> None:
    """Create or overwrite a skill file."""
    safe_name = _safe_skill_name(skill_name)
    skill_dir = Path(SKILLS_DIR) / safe_name
    skill_dir.mkdir(parents=True, exist_ok=True)
    (skill_dir / "SKILL.md").write_text(content, encoding="utf-8")
    logger.info("Saved skill: %s", safe_name)


def register_skill_tools(tools: dict) -> None:
    """Register skill-related tools into the tools dict."""

    async def skills_list() -> str:
        import json
        skills = list_skills()
        if not skills:
            return "אין skills זמינים כרגע."
        lines = ["Skills זמינים:"]
        for s in skills:
            lines.append(f'- **{s["name"]}**: {s["description"]}')
        return "\n".join(lines)

    async def skill_view(skill_name: str) -> str:
        content = view_skill(skill_name)
        if content is None:
            return f'❌ Skill "{skill_name}" לא נמצא.'
        return content

    async def skill_manage(action: str, skill_name: str, content: str = "") -> str:
        if action == "create":
            if not content:
                return "❌ צריך לספק תוכן ל-skill חדש."
            existing = view_skill(skill_name)
            if existing:
                return f'❌ Skill "{skill_name}" כבר קיים. השתמש ב-action "update" כדי לעדכן.'
            save_skill(skill_name, content)
            return f'✅ Skill "{skill_name}" נוצר בהצלחה.'
        elif action == "update":
            existing = view_skill(skill_name)
            if not existing:
                return f'❌ Skill "{skill_name}" לא נמצא. השתמש ב-action "create" כדי ליצור.'
            if not content:
                return "❌ צריך לספק תוכן מעודכן."
            save_skill(skill_name, content)
            return f'✅ Skill "{skill_name}" עודכן בהצלחה.'
        else:
            return f'❌ פעולה לא מוכרת: {action}. השתמש ב-"create" או "update".'

    tools["skills_list"] = {
        "description": "List all available skills (learned workflows and templates). Returns skill names and descriptions.",
        "parameters": {"type": "object", "properties": {}, "required": []},
        "handler": skills_list,
    }

    tools["skill_view"] = {
        "description": "View the full content of a specific skill. Use after skills_list to load a relevant workflow before a complex task.",
        "parameters": {
            "type": "object",
            "properties": {
                "skill_name": {"type": "string", "description": "The skill name (from skills_list)"},
            },
            "required": ["skill_name"],
        },
        "handler": skill_view,
    }

    tools["skill_manage"] = {
        "description": "Create or update a skill. Use when you discover a reusable workflow pattern during a conversation.",
        "parameters": {
            "type": "object",
            "properties": {
                "action": {"type": "string", "enum": ["create", "update"], "description": "Create new or update existing"},
                "skill_name": {"type": "string", "description": "Skill name (lowercase-dashes)"},
                "content": {"type": "string", "description": "Full skill content in markdown with YAML frontmatter"},
            },
            "required": ["action", "skill_name", "content"],
        },
        "handler": skill_manage,
    }