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.
"""

import subprocess
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_cli(
        self, skill_name: str, *args: str
    ) -> Optional[str]:
        """
        Invoke a skill via Claude Code CLI.

        Requires claude-code CLI to be installed and in PATH.
        For production, integrate with the Agent's LLM directly.
        """
        # Validate skill_name
        if not skill_name or not skill_name.replace("-", "").replace("_", "").isalnum():
            raise ValueError(
                "Invalid skill name: must contain only alphanumeric, "
                "hyphens, and underscores"
            )

        # Validate arguments don't contain shell metacharacters
        for arg in args:
            if any(char in str(arg) for char in ['&', '|', ';', '$', '`', '\n', '\r']):
                raise ValueError(
                    "Invalid argument: contains shell metacharacters"
                )

        try:
            cmd = ["claude-code", f"/{skill_name}"] + list(args)
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                cwd=self.project_root,
                timeout=60,  # Add timeout to prevent hanging
            )
            return result.stdout if result.returncode == 0 else None
        except FileNotFoundError:
            print("[SkillInvoker] claude-code CLI not found")
            return None
        except subprocess.TimeoutExpired:
            print(f"[SkillInvoker] Skill {skill_name} timed out")
            return None

    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


if __name__ == "__main__":
    invoker = SkillInvoker()

    print("Available skills:")
    for skill in invoker.list_available_skills():
        info = invoker.get_skill_info(skill)
        print(f"  /{skill}")
        if info:
            print(
                f"    Description: {info.get('description', 'N/A')}"
            )
            print(
                f"    User-invocable: "
                f"{info.get('user-invocable', 'N/A')}"
            )