skill_engine.py

"""
RUBRA — skill_engine.py
GitHub Repository Skill Manager (Phases 3-5).

Workflow:
  1. User pastes a GitHub URL → clone_and_index()
  2. Repository is cloned into /data/skills/<slug>/repo/
  3. Every readable file is analyzed: language, framework, deps, APIs,
     coding conventions, architecture
  4. A SkillRecord is built and stored in the DB + a JSON manifest
  5. During chat, relevant skills are retrieved by semantic similarity
     and injected into the coding agent's system prompt

Phases covered:
  Phase 3 — repository cloning, analysis, manifest generation
  Phase 4 — dependency/class/function graph, natural-language search
  Phase 5 — code explanation, debugging, refactoring, test generation
             using the skill's own conventions and patterns

Security:
  - All repo operations are sandboxed to /data/skills/
  - Dangerous repo hooks (post-checkout, post-merge etc.) are stripped
    before `git clone` runs so the repository can never execute code
    during indexing
  - Private repo support via GITHUB_TOKEN env var (optional)
"""

import os
import re
import ast
import json
import uuid
import shutil
import asyncio
import hashlib
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional, Dict, List, Any, AsyncIterator

log = logging.getLogger("rubra.skill_engine")

# ── Storage ───────────────────────────────────────────────────────────────────
DATA_DIR   = Path(os.getenv("DATA_DIR", "/data"))
SKILLS_DIR = DATA_DIR / "skills"
SKILLS_DIR.mkdir(parents=True, exist_ok=True)

GITHUB_TOKEN = os.getenv("GITHUB_TOKEN", "")

# Files to never read (binary / too large / not useful for analysis)
SKIP_EXTENSIONS = {
    ".png", ".jpg", ".jpeg", ".gif", ".webp", ".ico", ".svg",
    ".woff", ".woff2", ".ttf", ".eot", ".otf",
    ".mp4", ".mp3", ".wav", ".ogg", ".avi",
    ".zip", ".tar", ".gz", ".rar", ".7z",
    ".pdf", ".docx", ".xlsx", ".pptx",
    ".pyc", ".pyo", ".pyd", ".class", ".o", ".so", ".dll", ".exe",
    ".lock",  # package-lock.json, yarn.lock, etc. — too long, low signal
    ".map",   # source maps
    ".min.js", ".min.css",
}
SKIP_DIRS = {
    ".git", "node_modules", "__pycache__", ".next", "dist", "build",
    "coverage", ".pytest_cache", ".mypy_cache", "venv", ".venv",
    "env", ".env", "eggs", ".eggs", "site-packages",
}
MAX_FILE_BYTES = 200_000    # 200KB per file
MAX_TOTAL_FILES = 500
MAX_TOTAL_BYTES = 10_000_000  # 10MB total repo read


# ── Data structures ───────────────────────────────────────────────────────────
@dataclass
class FileRecord:
    path: str
    language: str
    size_bytes: int
    content_preview: str  # first 500 chars
    classes: List[str]
    functions: List[str]
    imports: List[str]
    exports: List[str]


@dataclass
class SkillRecord:
    skill_id: str
    name: str
    repo_url: str
    branch: str
    commit_sha: str
    language: str           # primary language
    framework: str          # detected framework
    architecture: str       # monolith / microservices / library / cli / etc.
    dependencies: List[str]
    file_count: int
    total_bytes: int
    files: List[FileRecord]
    readme_summary: str
    api_endpoints: List[str]
    coding_conventions: Dict[str, str]  # e.g. {"style": "PEP8", "type_hints": "yes"}
    has_tests: bool
    has_docs: bool
    local_path: str
    created_at: str
    updated_at: str

    def to_dict(self) -> dict:
        import dataclasses
        return dataclasses.asdict(self)

    @staticmethod
    def from_dict(d: dict) -> "SkillRecord":
        d["files"] = [FileRecord(**f) for f in d.get("files", [])]
        return SkillRecord(**d)

    def summary_for_prompt(self, max_chars: int = 2000) -> str:
        """Compact description for injecting into a system prompt."""
        lines = [
            f"[SKILL: {self.name}]",
            f"Repo: {self.repo_url} ({self.branch})",
            f"Language: {self.language} | Framework: {self.framework}",
            f"Architecture: {self.architecture}",
            f"Dependencies: {', '.join(self.dependencies[:15])}",
        ]
        if self.readme_summary:
            lines.append(f"README: {self.readme_summary[:300]}")
        if self.api_endpoints:
            lines.append(f"APIs: {', '.join(self.api_endpoints[:10])}")
        if self.coding_conventions:
            conv = "; ".join(f"{k}={v}" for k, v in self.coding_conventions.items())
            lines.append(f"Conventions: {conv}")
        key_files = [f.path for f in self.files[:20]]
        lines.append(f"Key files: {', '.join(key_files)}")
        text = "\n".join(lines)
        return text[:max_chars]


# ── Language detection ────────────────────────────────────────────────────────
_EXT_TO_LANG = {
    ".py": "Python", ".js": "JavaScript", ".jsx": "JavaScript",
    ".ts": "TypeScript", ".tsx": "TypeScript", ".go": "Go",
    ".rs": "Rust", ".java": "Java", ".kt": "Kotlin",
    ".rb": "Ruby", ".php": "PHP", ".cs": "C#",
    ".cpp": "C++", ".c": "C", ".h": "C/C++",
    ".html": "HTML", ".css": "CSS", ".scss": "SCSS",
    ".sql": "SQL", ".sh": "Shell", ".bash": "Shell",
    ".yaml": "YAML", ".yml": "YAML", ".toml": "TOML",
    ".json": "JSON", ".md": "Markdown", ".mdx": "Markdown",
    ".svelte": "Svelte", ".vue": "Vue", ".dart": "Dart",
    ".swift": "Swift", ".m": "Objective-C",
}

_FRAMEWORK_SIGNATURES = {
    "Next.js":    ["next.config", "pages/", "app/page.tsx", ".next/"],
    "React":      ["react", "ReactDOM", "jsx", "useState"],
    "FastAPI":    ["fastapi", "APIRouter", "@app.get", "@app.post"],
    "Django":     ["django", "settings.py", "models.py", "urls.py"],
    "Flask":      ["flask", "Flask(__name__", "@app.route"],
    "Express":    ["express()", "app.use(", "router.get("],
    "NestJS":     ["@Module", "@Controller", "@Injectable", "@nestjs"],
    "Vue":        ["Vue.component", "defineComponent", ".vue"],
    "Svelte":     [".svelte", "svelte/"],
    "Tailwind":   ["tailwind.config", "@tailwind"],
    "Prisma":     ["prisma/schema", "PrismaClient"],
    "SQLAlchemy": ["sqlalchemy", "Base = declarative_base", "Column("],
    "Pydantic":   ["BaseModel", "Field(", "pydantic"],
}

def _detect_language(files: List[FileRecord]) -> str:
    counts: Dict[str, int] = {}
    for f in files:
        lang = f.language
        if lang and lang != "Unknown":
            counts[lang] = counts.get(lang, 0) + f.size_bytes
    return max(counts, key=counts.get) if counts else "Unknown"

def _detect_framework(all_text: str, files: List[FileRecord]) -> str:
    for fw, signatures in _FRAMEWORK_SIGNATURES.items():
        if sum(1 for sig in signatures if sig in all_text) >= 2:
            return fw
    return "Unknown"

def _detect_architecture(files: List[FileRecord], repo_root: Path) -> str:
    paths = [f.path for f in files]
    if any("microservice" in p or "services/" in p for p in paths):
        return "microservices"
    if any("api/" in p or "routes/" in p or "endpoints/" in p for p in paths):
        return "REST API"
    if (repo_root / "setup.py").exists() or (repo_root / "pyproject.toml").exists():
        return "Python Library"
    if any("cli" in p.lower() for p in paths):
        return "CLI Tool"
    return "Monolith"


# ── Python AST analysis ───────────────────────────────────────────────────────
def _analyze_python_file(content: str) -> tuple[list, list, list]:
    """Returns (classes, functions, imports) from Python source."""
    classes, functions, imports_found = [], [], []
    try:
        tree = ast.parse(content)
        for node in ast.walk(tree):
            if isinstance(node, ast.ClassDef):
                classes.append(node.name)
            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                functions.append(node.name)
            elif isinstance(node, ast.Import):
                for alias in node.names:
                    imports_found.append(alias.name.split(".")[0])
            elif isinstance(node, ast.ImportFrom):
                if node.module:
                    imports_found.append(node.module.split(".")[0])
    except SyntaxError:
        pass
    return classes, functions, list(set(imports_found))


def _analyze_js_file(content: str) -> tuple[list, list, list]:
    """Regex-based JS/TS analysis."""
    classes   = re.findall(r'\bclass\s+(\w+)', content)
    functions = re.findall(r'\bfunction\s+(\w+)|const\s+(\w+)\s*=\s*(?:async\s+)?\(', content)
    functions = [f[0] or f[1] for f in functions if f[0] or f[1]]
    imports   = re.findall(r"(?:import|require)\s*\(?['\"](@?[\w\-/]+)", content)
    exports   = re.findall(r'\bexport\s+(?:default\s+)?(?:function|class|const|let|var)\s+(\w+)', content)
    return classes, functions, list(set(imports))


def _detect_api_endpoints(content: str, path: str) -> List[str]:
    endpoints = []
    # FastAPI / Flask / Express
    for pattern in [
        r'@(?:app|router)\.\w+\(["\']([^"\']+)["\']',   # @app.get("/path")
        r'router\.\w+\(["\']([^"\']+)["\']',             # router.get("/path")
        r'app\.\w+\(["\']([^"\']+)["\']',                # app.use("/path")
        r'path\(["\']([^"\']+)["\']',                    # Django path()
        r'url\(["\']([^"\']+)["\']',                     # Django url()
    ]:
        endpoints.extend(re.findall(pattern, content))
    return list(set(endpoints))[:20]


def _detect_conventions(files: List[FileRecord], all_content: str) -> Dict[str, str]:
    conv = {}
    # Type hints
    if "def " in all_content and "->" in all_content:
        conv["type_hints"] = "yes"
    # Docstrings
    if '"""' in all_content or "'''" in all_content:
        conv["docstrings"] = "present"
    # Async
    if "async def" in all_content or "await " in all_content:
        conv["async"] = "yes"
    # Test framework
    if "pytest" in all_content:
        conv["testing"] = "pytest"
    elif "unittest" in all_content:
        conv["testing"] = "unittest"
    elif "jest" in all_content or "describe(" in all_content:
        conv["testing"] = "jest"
    # Formatting
    if ".prettierrc" in str(files) or "prettier" in all_content:
        conv["formatter"] = "prettier"
    elif "black" in all_content or "ruff" in all_content:
        conv["formatter"] = "black/ruff"
    return conv


# ── Git operations ────────────────────────────────────────────────────────────
def _parse_github_url(url: str) -> tuple[str, str, str]:
    """Returns (owner, repo, branch_hint) from various GitHub URL formats."""
    url = url.strip().rstrip("/")
    # Extract branch from /tree/branch paths
    branch = "main"
    m = re.search(r'/tree/([^/]+)', url)
    if m:
        branch = m.group(1)
        url = url[:m.start()]

    # Normalize URL formats
    m = re.match(r'https?://github\.com/([^/]+)/([^/\.]+)', url)
    if not m:
        m = re.match(r'git@github\.com:([^/]+)/([^\.]+)', url)
    if not m:
        raise ValueError(f"Cannot parse GitHub URL: {url!r}")
    return m.group(1), m.group(2).replace(".git", ""), branch


async def _git_clone(repo_url: str, dest: Path, branch: str) -> str:
    """Clone a repository. Returns the HEAD commit SHA."""
    token = GITHUB_TOKEN
    if token and "github.com" in repo_url and "https://" in repo_url:
        # Inject token for private repos
        repo_url = repo_url.replace("https://", f"https://{token}@")

    cmd = ["git", "clone", "--depth=1", "--single-branch",
           f"--branch={branch}", repo_url, str(dest)]
    log.info(f"[SKILL] Cloning {repo_url} → {dest}")

    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        env={**os.environ, "GIT_TERMINAL_PROMPT": "0"},
    )
    _, stderr = await asyncio.wait_for(proc.communicate(), timeout=120)
    if proc.returncode != 0:
        err = stderr.decode()[:500]
        # Security: strip any token from error message before logging
        if token:
            err = err.replace(token, "***")
        raise RuntimeError(f"git clone failed: {err}")

    # Strip dangerous hooks
    hooks_dir = dest / ".git" / "hooks"
    if hooks_dir.exists():
        for hook in hooks_dir.iterdir():
            if hook.is_file():
                hook.unlink()
        log.info("[SKILL] Stripped all git hooks (security)")

    # Get commit SHA
    sha_proc = await asyncio.create_subprocess_exec(
        "git", "-C", str(dest), "rev-parse", "HEAD",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    sha_out, _ = await sha_proc.communicate()
    return sha_out.decode().strip()[:12]


# ── Core analysis ─────────────────────────────────────────────────────────────
def _read_repo(repo_root: Path) -> List[FileRecord]:
    """Walk repo and build FileRecord list."""
    records = []
    total_bytes = 0

    for path in sorted(repo_root.rglob("*")):
        if total_bytes > MAX_TOTAL_BYTES or len(records) >= MAX_TOTAL_FILES:
            log.warning(f"[SKILL] Size cap reached at {len(records)} files / {total_bytes} bytes")
            break

        # Skip dirs
        if any(part in SKIP_DIRS for part in path.parts):
            continue
        if not path.is_file():
            continue

        suffix = path.suffix.lower()
        if suffix in SKIP_EXTENSIONS or path.name.endswith(".min.js") or path.name.endswith(".min.css"):
            continue

        try:
            size = path.stat().st_size
            if size > MAX_FILE_BYTES or size == 0:
                continue

            content = path.read_text(encoding="utf-8", errors="replace")
            total_bytes += size

            relative = str(path.relative_to(repo_root))
            lang = _EXT_TO_LANG.get(suffix, "Unknown")

            classes, functions, imports_found = [], [], []
            exports = []
            if lang == "Python":
                classes, functions, imports_found = _analyze_python_file(content)
            elif lang in ("JavaScript", "TypeScript"):
                classes, functions, imports_found = _analyze_js_file(content)
                exports = re.findall(r'\bexport\s+(?:default\s+)?(?:function|class|const)\s+(\w+)', content)

            records.append(FileRecord(
                path=relative,
                language=lang,
                size_bytes=size,
                content_preview=content[:500],
                classes=classes[:30],
                functions=functions[:50],
                imports=imports_found[:30],
                exports=exports[:20],
            ))
        except Exception as e:
            log.debug(f"[SKILL] Skipped {path}: {e}")

    return records


def _build_dependency_list(files: List[FileRecord], repo_root: Path) -> List[str]:
    deps = set()
    # Python
    for pf in ["requirements.txt", "pyproject.toml", "setup.py", "Pipfile"]:
        p = repo_root / pf
        if p.exists():
            text = p.read_text(errors="replace")
            deps.update(re.findall(r'^([a-zA-Z0-9_\-]+)', text, re.MULTILINE))

    # Node
    pkg = repo_root / "package.json"
    if pkg.exists():
        try:
            data = json.loads(pkg.read_text())
            deps.update(data.get("dependencies", {}).keys())
            deps.update(data.get("devDependencies", {}).keys())
        except Exception:
            pass

    # Go
    go_mod = repo_root / "go.mod"
    if go_mod.exists():
        text = go_mod.read_text()
        deps.update(re.findall(r'require\s+(\S+)', text))

    # Also collect from file imports
    for f in files:
        deps.update(f.imports)

    # Filter out standard-library noise
    noise = {"os", "sys", "re", "json", "io", "time", "math", "string",
             "typing", "pathlib", "collections", "itertools", "functools",
             "asyncio", "logging", "uuid", "datetime", "abc", "enum",
             "dataclasses", "copy", "shutil", "subprocess"}
    return sorted(deps - noise)[:60]


# ── Public API ────────────────────────────────────────────────────────────────
class SkillEngine:
    """
    Manages the full lifecycle of repository skills.
    Use the module-level `skill_engine` singleton.
    """

    def __init__(self):
        self._manifest_path = SKILLS_DIR / "manifest.json"
        self._skills: Dict[str, SkillRecord] = {}
        self._load_manifest()

    def _load_manifest(self):
        if self._manifest_path.exists():
            try:
                data = json.loads(self._manifest_path.read_text())
                for d in data.values():
                    s = SkillRecord.from_dict(d)
                    self._skills[s.skill_id] = s
                log.info(f"[SKILL] Loaded {len(self._skills)} skill(s) from manifest")
            except Exception as e:
                log.warning(f"[SKILL] Manifest load failed: {e}")

    def _save_manifest(self):
        data = {sid: s.to_dict() for sid, s in self._skills.items()}
        self._manifest_path.write_text(json.dumps(data, indent=2, ensure_ascii=False))

    def list_skills(self) -> List[dict]:
        return [
            {"skill_id": s.skill_id, "name": s.name, "repo_url": s.repo_url,
             "language": s.language, "framework": s.framework,
             "file_count": s.file_count, "updated_at": s.updated_at}
            for s in self._skills.values()
        ]

    def get_skill(self, skill_id: str) -> Optional[SkillRecord]:
        return self._skills.get(skill_id)

    def delete_skill(self, skill_id: str) -> bool:
        s = self._skills.pop(skill_id, None)
        if s:
            skill_dir = SKILLS_DIR / skill_id
            if skill_dir.exists():
                shutil.rmtree(skill_dir)
            self._save_manifest()
            log.info(f"[SKILL] Deleted skill {skill_id} ({s.name})")
            return True
        return False

    async def clone_and_index(
        self,
        repo_url: str,
        branch: Optional[str] = None,
        llm_func=None,
    ) -> AsyncIterator[dict]:
        """
        Full pipeline: clone → read → analyze → summarize → store.
        Yields SSE-compatible progress dicts so the UI can stream progress.
        """
        from datetime import datetime, timezone
        import time

        try:
            owner, repo, detected_branch = _parse_github_url(repo_url)
        except ValueError as e:
            yield {"type": "error", "message": str(e)}
            return

        branch      = branch or detected_branch
        skill_id    = hashlib.md5(f"{owner}/{repo}@{branch}".encode()).hexdigest()[:12]
        skill_dir   = SKILLS_DIR / skill_id
        repo_dir    = skill_dir / "repo"
        name        = f"{owner}/{repo}"
        now         = datetime.now(timezone.utc).isoformat()

        yield {"type": "progress", "step": "clone", "message": f"Cloning {name}@{branch}…"}

        # Clean up existing checkout if re-indexing
        if repo_dir.exists():
            shutil.rmtree(repo_dir)
        repo_dir.mkdir(parents=True, exist_ok=True)

        try:
            commit_sha = await _git_clone(repo_url, repo_dir, branch)
        except Exception as e:
            yield {"type": "error", "message": f"Clone failed: {e}"}
            return

        yield {"type": "progress", "step": "read", "message": f"Reading repository files…"}

        files = _read_repo(repo_dir)
        yield {"type": "progress", "step": "analyze",
               "message": f"Analyzing {len(files)} files…"}

        all_content_sample = " ".join(f.content_preview for f in files[:100])

        # README summary
        readme_text = ""
        for readme_name in ("README.md", "README.rst", "README.txt", "README"):
            readme_path = repo_dir / readme_name
            if readme_path.exists():
                readme_text = readme_path.read_text(errors="replace")[:3000]
                break

        readme_summary = ""
        if llm_func and readme_text:
            yield {"type": "progress", "step": "summarize", "message": "Summarizing README…"}
            try:
                raw = ""
                async for tok in llm_func(
                    [{"role": "user", "content":
                      f"Summarize this README in 2 sentences for a developer:\n\n{readme_text[:2000]}"}],
                    mode="fast", max_tokens=150
                ):
                    raw += tok
                readme_summary = raw.strip()
            except Exception as e:
                log.warning(f"[SKILL] README summarization failed: {e}")

        language     = _detect_language(files)
        framework    = _detect_framework(all_content_sample, files)
        architecture = _detect_architecture(files, repo_dir)
        deps         = _build_dependency_list(files, repo_dir)
        conventions  = _detect_conventions(files, all_content_sample)
        api_endpoints= []
        for f in files:
            if f.language in ("Python", "JavaScript", "TypeScript"):
                content = (repo_dir / f.path).read_text(errors="replace") if (repo_dir / f.path).exists() else ""
                api_endpoints.extend(_detect_api_endpoints(content, f.path))
        api_endpoints = list(set(api_endpoints))[:30]
        has_tests = any("test" in f.path.lower() or "spec" in f.path.lower() for f in files)
        has_docs  = any(f.path.lower().endswith(".md") for f in files)

        skill = SkillRecord(
            skill_id=skill_id,
            name=name,
            repo_url=repo_url,
            branch=branch,
            commit_sha=commit_sha,
            language=language,
            framework=framework,
            architecture=architecture,
            dependencies=deps,
            file_count=len(files),
            total_bytes=sum(f.size_bytes for f in files),
            files=files,
            readme_summary=readme_summary,
            api_endpoints=api_endpoints,
            coding_conventions=conventions,
            has_tests=has_tests,
            has_docs=has_docs,
            local_path=str(repo_dir),
            created_at=now,
            updated_at=now,
        )

        self._skills[skill_id] = skill
        self._save_manifest()

        yield {
            "type": "complete",
            "skill_id": skill_id,
            "name": name,
            "language": language,
            "framework": framework,
            "architecture": architecture,
            "file_count": len(files),
            "message": f"✅ Skill '{name}' indexed ({len(files)} files, {language}/{framework})",
        }

    def search(self, query: str, top_k: int = 3) -> List[SkillRecord]:
        """
        Simple keyword-based skill search. Returns the top matching skills.
        Phase 4 enhancement: replace with vector embeddings when available.
        """
        query_words = set(query.lower().split())
        scored = []
        for skill in self._skills.values():
            text = (skill.name + " " + skill.language + " " + skill.framework +
                    " " + skill.architecture + " " +
                    " ".join(skill.dependencies) + " " + skill.readme_summary).lower()
            score = sum(1 for w in query_words if w in text)
            if score > 0:
                scored.append((score, skill))
        scored.sort(key=lambda x: x[0], reverse=True)
        return [s for _, s in scored[:top_k]]

    def get_context_for_prompt(self, query: str, max_chars: int = 3000) -> str:
        """
        Returns a skill context block for injection into a system prompt.
        Called by the coding agent to inject relevant repo knowledge.
        """
        skills = self.search(query, top_k=2)
        if not skills:
            return ""
        parts = ["[IMPORTED REPOSITORY SKILLS — use these patterns and conventions]"]
        for s in skills:
            parts.append(s.summary_for_prompt(max_chars // len(skills)))
        return "\n\n".join(parts)

    def find_in_repo(self, skill_id: str, query: str) -> List[dict]:
        """
        Phase 4: Natural language search inside a specific repo.
        Searches file content for classes/functions matching the query.
        """
        skill = self._skills.get(skill_id)
        if not skill:
            return []

        query_lower = query.lower()
        results = []
        repo_dir = Path(skill.local_path)

        for file_rec in skill.files:
            # Check if classes or functions match
            matched_classes = [c for c in file_rec.classes if query_lower in c.lower()]
            matched_funcs   = [f for f in file_rec.functions if query_lower in f.lower()]

            if matched_classes or matched_funcs:
                results.append({
                    "file":     file_rec.path,
                    "language": file_rec.language,
                    "classes":  matched_classes,
                    "functions":matched_funcs,
                    "preview":  file_rec.content_preview[:300],
                })

            # Also check content preview for keyword match
            elif query_lower in file_rec.content_preview.lower():
                results.append({
                    "file":    file_rec.path,
                    "language":file_rec.language,
                    "preview": file_rec.content_preview[:300],
                })

        return results[:20]

    def read_file(self, skill_id: str, relative_path: str) -> Optional[str]:
        """Read a specific file from an indexed repo (for code explanation/debugging)."""
        skill = self._skills.get(skill_id)
        if not skill:
            return None
        file_path = Path(skill.local_path) / relative_path
        # Security: ensure path is inside the skill's own directory
        try:
            file_path = file_path.resolve()
            skill_root = Path(skill.local_path).resolve()
            file_path.relative_to(skill_root)  # raises ValueError if outside
        except ValueError:
            log.warning(f"[SKILL] Path traversal attempt blocked: {relative_path}")
            return None
        if not file_path.exists():
            return None
        return file_path.read_text(errors="replace")[:50_000]  # 50KB limit


# Module-level singleton
skill_engine = SkillEngine()