ajarbot/adapters/skill_integration.py

"""
Integration layer for using Claude Code skills from within ajarbot adapters.

Allows the Agent to invoke local skills programmatically,
enabling advanced automation and dynamic behavior.
"""

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


class SkillInvoker:
    """
    Invokes Claude Code skills programmatically.

    Skills are local-only (no registry) and live in:
    - .claude/skills/<skill-name>/SKILL.md (project)
    - ~/.claude/skills/<skill-name>/SKILL.md (personal)
    """

    def __init__(self, project_root: Optional[str] = None) -> None:
        self.project_root = Path(project_root or Path.cwd())
        self.skills_dir = self.project_root / ".claude" / "skills"

    def list_available_skills(self) -> List[str]:
        """List all available local skills."""
        if not self.skills_dir.exists():
            return []

        return [
            skill_dir.name
            for skill_dir in self.skills_dir.iterdir()
            if skill_dir.is_dir()
            and (skill_dir / "SKILL.md").exists()
        ]

    def get_skill_info(
        self, skill_name: str
    ) -> Optional[Dict[str, Any]]:
        """Get information about a skill."""
        # Validate skill_name to prevent path traversal
        if not skill_name or not skill_name.replace("-", "").replace("_", "").isalnum():
            raise ValueError(
                "Invalid skill name: must contain only alphanumeric, "
                "hyphens, and underscores"
            )

        skill_path = self.skills_dir / skill_name / "SKILL.md"

        # Verify the resolved path is within skills_dir
        try:
            resolved = skill_path.resolve()
            if not resolved.is_relative_to(self.skills_dir.resolve()):
                raise ValueError("Path traversal detected in skill name")
        except (ValueError, OSError) as e:
            raise ValueError(f"Invalid skill path: {e}")

        if not skill_path.exists():
            return None

        with open(skill_path) as f:
            content = f.read()

        if not content.startswith("---"):
            return None

        parts = content.split("---", 2)
        if len(parts) < 3:
            return None

        frontmatter = parts[1].strip()
        body = parts[2].strip()

        # Simple YAML parsing (key: value pairs)
        info: Dict[str, Any] = {}
        for line in frontmatter.split("\n"):
            if ":" in line:
                key, value = line.split(":", 1)
                info[key.strip()] = value.strip()

        info["body"] = body
        info["path"] = str(skill_path)
        return info

    def invoke_skill_via_agent(
        self, skill_name: str, agent: Any, *args: str
    ) -> str:
        """
        Invoke a skill by injecting it into the Agent's context.

        This is the recommended approach - it uses the Agent's existing
        LLM connection without requiring the CLI.
        """
        skill_info = self.get_skill_info(skill_name)

        if not skill_info:
            return f"Skill '{skill_name}' not found"

        # Validate and sanitize arguments to prevent prompt injection
        sanitized_args = []
        for arg in args:
            # Limit argument length
            arg_str = str(arg)[:1000]
            # Wrap in XML-like tags to clearly delimit user input
            sanitized_args.append(arg_str)

        arguments = " ".join(sanitized_args)
        skill_instructions = skill_info.get("body", "")

        # Replace argument placeholders with delimited user input
        # Wrap arguments in XML tags to prevent prompt injection
        safe_arguments = f"<user_input>{arguments}</user_input>"
        skill_instructions = skill_instructions.replace(
            "$ARGUMENTS", safe_arguments
        )
        for i, arg in enumerate(sanitized_args):
            safe_arg = f"<user_input>{arg}</user_input>"
            skill_instructions = skill_instructions.replace(
                f"${i}", safe_arg
            )

        # Use actual username instead of privileged "skill-invoker"
        return agent.chat(
            user_message=skill_instructions,
            username="default",  # Changed from "skill-invoker"
        )


def skill_based_preprocessor(
    skill_invoker: SkillInvoker, agent: Any
) -> Callable:
    """
    Create a preprocessor that invokes skills based on message patterns.

    Messages starting with /skill-name will trigger the corresponding skill.
    """

    def preprocessor(message):
        if not message.text.startswith("/"):
            return message

        parts = message.text.split(maxsplit=1)
        skill_name = parts[0][1:]  # Remove leading /
        args = parts[1] if len(parts) > 1 else ""

        if skill_name in skill_invoker.list_available_skills():
            result = skill_invoker.invoke_skill_via_agent(
                skill_name, agent, args
            )
            message.text = result

        return message

    return preprocessor