Newer
Older
cortex-hub / ai-hub / app / core / skills / fs_loader.py
import os
import yaml
import json
import logging
import collections
from typing import List, Dict, Any, Optional
from app.config import settings

logger = logging.getLogger(__name__)

class FileSystemSkillLoader:
    def __init__(self, base_dirs: List[str]):
        self.base_dirs = base_dirs

    def _ensure_metadata(self, folder_path: str, skill_name: str, is_system: bool = False):
        """
        Auto-generates a .metadata.json file if it's missing.
        Assigns 'admin' as default owner if it's auto-generated during boot.
        """
        meta_path = os.path.join(folder_path, ".metadata.json")
        if not os.path.exists(meta_path):
            default_metadata = {
                "owner_id": "admin",
                "is_system": is_system,
                "extra_metadata": {"emoji": "🛠️"}
            }
            try:
                with open(meta_path, "w") as f:
                    json.dump(default_metadata, f, indent=4)
                logger.info(f"Generated default .metadata.json for skill '{skill_name}'")
            except Exception as e:
                logger.error(f"Failed to generate metadata for {skill_name}: {e}")
            return default_metadata
            
        try:
            with open(meta_path, "r") as f:
                return json.load(f)
        except Exception as e:
            logger.error(f"Error reading {meta_path}: {e}")
            return {}

    def get_all_skills(self) -> List[Dict[str, Any]]:
        """
        Recursively walks the base directories and parses all SKILL.md rules 
        into the common dictionary schema used by the orchestrator.
        Supports both nested hierarchy (feature/skill/SKILL.md) and flat (skill/SKILL.md).
        """
        skills = []
        for base_dir in self.base_dirs:
            if not os.path.exists(base_dir):
                continue

            # Root entries can be features (nested) OR skills (flat)
            for entry in os.listdir(base_dir):
                entry_path = os.path.join(base_dir, entry)
                if not os.path.isdir(entry_path):
                    continue
                
                # Check if this is a Skill directly (Flat Structure)
                skill_md_path = os.path.join(entry_path, "SKILL.md")
                if os.path.exists(skill_md_path):
                    # It's a skill! Try to load it. 
                    # We assume it belongs to the "chat" feature if not nested, 
                    # but frontmatter in SKILL.md will override this later.
                    loaded = self._load_skill(entry_path, entry, default_features=[entry, "chat"])
                    if loaded:
                        skills.append(loaded)
                    continue

                # Check if it's a Feature folder (Nested Structure)
                for skill_id in os.listdir(entry_path):
                    skill_path = os.path.join(entry_path, skill_id)
                    if not os.path.isdir(skill_path):
                        continue
                        
                    skill_md_path = os.path.join(skill_path, "SKILL.md")
                    if os.path.exists(skill_md_path):
                        loaded = self._load_skill(skill_path, skill_id, default_features=[entry])
                        if loaded:
                            skills.append(loaded)

        return skills

    def _load_skill(self, skill_path: str, skill_id: str, default_features: List[str]) -> Optional[Dict[str, Any]]:
        """Parses a skill folder into the standardized dictionary schema."""
        skill_md_path = os.path.join(skill_path, "SKILL.md")
        try:
            with open(skill_md_path, "r", encoding='utf-8') as f:
                skill_content = f.read()

            # Create virtual file schema
            vfs_files = []
            for root, _, files in os.walk(skill_path):
                for file in files:
                    # Hide .metadata.json and macOS .DS_Store from the VFS visible to LLMs
                    if file == ".metadata.json" or file.startswith('.'):
                        continue
                    file_abs = os.path.join(root, file)
                    file_rel = os.path.relpath(file_abs, skill_path).replace('\\', '/')
                    
                    class LazyFileContent(collections.UserString):
                        def __init__(self, fpath):
                            self._fpath = fpath
                            self._content = None
                            
                        @property
                        def data(self):
                            if self._content is None:
                                try:
                                    with open(self._fpath, "r", encoding='utf-8') as ff:
                                        self._content = ff.read()
                                except Exception:
                                    self._content = ""
                            return self._content

                    vfs_files.append({"file_path": file_rel, "content": LazyFileContent(file_abs), "absolute_path": file_abs})
                    
            # Extract or Generate Metadata
            metadata = self._ensure_metadata(skill_path, skill_id, is_system=False)

            # Extract Description directly from SKILL.md (Fallback or yaml frontmatter)
            skill_name = skill_id
            skill_desc = ""
            skill_features = default_features
            
            if skill_content.startswith("---"):
                parts = skill_content.split("---", 2)
                if len(parts) >= 3:
                    frontmatter = yaml.safe_load(parts[1]) or {}
                    skill_name = frontmatter.get("name", skill_name)
                    skill_desc = frontmatter.get("description", "")
                    # M6: Prioritize features defined in SKILL.md frontmatter!
                    if frontmatter.get("features"):
                        skill_features = frontmatter.get("features")
            
            # Generate the legacy internal dict schema
            return {
                "id": f"fs-{skill_id}", 
                "name": skill_name,
                "description": skill_desc or f"Skill loaded dynamically from {skill_id}",
                "skill_type": "local",
                "is_enabled": True,
                "features": skill_features,
                "is_system": metadata.get("is_system", False) or "/app/skills" in skill_path,
                "owner_id": metadata.get("owner_id", "admin"),
                "extra_metadata": metadata.get("extra_metadata", {"emoji": "🛠️"}),
                "files": vfs_files
            }
        except Exception as e:
            logger.error(f"Error parsing FS Skill at {skill_path}: {e}")
            return None

# Create a global instance initialized with both system and data directory configs
fs_loader = FileSystemSkillLoader(base_dirs=[
    os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../skills")),
    os.path.join(settings.DATA_DIR, "skills")
])