client/src/toolstore/skill_discovery.py

"""
Recursive skill discovery engine.

Walks a directory tree and discovers all skills (directories containing
SKILL.md), parses them, and returns structured results including
hierarchy/category information derived from path structure.

Supports three layout patterns:

    Case A — single skill (root itself is a skill):
        /some/skill-name/SKILL.md → one skill

    Case B — flat skill collection:
        /some/skills/
            ├── pdf-processing/SKILL.md
            ├── code-review/SKILL.md
            └── webapp-testing/SKILL.md

    Case C — nested / categorized skill collection:
        /some/skills/
            ├── debug/SKILL.md
            ├── design/SKILL.md
            ├── development/
            │   ├── agent-api/SKILL.md
            │   ├── mcp-builder/SKILL.md
            │   └── webapp-testing/SKILL.md
            └── documents/SKILL.md
"""

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Dict, List, Optional

if TYPE_CHECKING:
    from toolstore.skill_manager import SkillDefinition


@dataclass
class DiscoveredSkill:
    """A single skill discovered in a directory tree."""

    skill_def: "SkillDefinition"
    rel_path: Path   # relative path from the scan root
    category: str    # derived from parent directory hierarchy ("" for root-level)

    @property
    def name(self) -> str:
        return self.skill_def.name

    @property
    def description(self) -> str:
        return self.skill_def.description

    @property
    def is_valid(self) -> bool:
        return len(self.skill_def.errors) == 0

    @property
    def errors(self) -> List[str]:
        return self.skill_def.errors


@dataclass
class DiscoveryResult:
    """Complete result of a skill discovery scan."""

    root_path: Path
    skills: List[DiscoveredSkill] = field(default_factory=list)
    scan_errors: List[str] = field(default_factory=list)

    # ------------------------------------------------------------------
    # convenience accessors
    # ------------------------------------------------------------------

    @property
    def valid_skills(self) -> List[DiscoveredSkill]:
        return [s for s in self.skills if s.is_valid]

    @property
    def invalid_skills(self) -> List[DiscoveredSkill]:
        return [s for s in self.skills if not s.is_valid]

    @property
    def total(self) -> int:
        return len(self.skills)

    @property
    def valid_count(self) -> int:
        return len(self.valid_skills)

    @property
    def invalid_count(self) -> int:
        return len(self.invalid_skills)

    # ------------------------------------------------------------------
    # grouping
    # ------------------------------------------------------------------

    def by_category(self) -> Dict[str, List[DiscoveredSkill]]:
        """Group skills by their derived category tag.

        Returns a dict mapping category → skills (preserves insertion
        order from Python 3.7+).
        """
        cats: Dict[str, List[DiscoveredSkill]] = {}
        for ds in self.skills:
            cat = ds.category or "(root)"
            cats.setdefault(cat, []).append(ds)
        return cats

    # ------------------------------------------------------------------
    # formatting
    # ------------------------------------------------------------------

    def summary(self) -> str:
        """One-line summary suitable for progress output."""
        return (
            f"Discovered {self.total} skill(s) — "
            f"{self.valid_count} valid, "
            f"{self.invalid_count} invalid"
        )

    def tree(self) -> str:
        """Pretty-printed tree of discovered skills grouped by category."""
        lines = [f"📁 {self.root_path}", f"   {self.summary()}", ""]
        by_cat = self.by_category()
        for cat, skills in by_cat.items():
            cat_icon = "📂" if cat == "(root)" else "📂"
            lines.append(f"  {cat_icon} {cat}/")
            for ds in skills:
                icon = "✅" if ds.is_valid else "⚠️"
                lines.append(f"    {icon} {ds.name}")
                if not ds.is_valid:
                    for err in ds.errors:
                        lines.append(f"       ❌ {err}")
        return "\n".join(lines)

    def report(self) -> str:
        """Full report with validation details and error messages."""
        parts = [f"## Skill Discovery Report", f"", f"Scan root: `{self.root_path}`", f""]
        parts.append(f"**{self.summary()}**")
        parts.append("")

        # Valid skills table
        if self.valid_skills:
            parts.append("### ✅ Valid Skills")
            parts.append("")
            parts.append("| Name | Category | Description |")
            parts.append("|------|----------|-------------|")
            for ds in self.valid_skills:
                desc = ds.description[:80].replace("\n", " ")
                cat = ds.category or "—"
                parts.append(f"| `{ds.name}` | `{cat}` | {desc} |")
            parts.append("")

        # Invalid skills
        if self.invalid_skills:
            parts.append("### ⚠️ Invalid Skills")
            parts.append("")
            for ds in self.invalid_skills:
                parts.append(f"**`{ds.name or ds.rel_path}`**")
                for err in ds.errors:
                    parts.append(f"- ❌ {err}")
                parts.append("")

        # Scan errors (permission denied, etc.)
        if self.scan_errors:
            parts.append("### 🔴 Scan Errors")
            parts.append("")
            for err in self.scan_errors:
                parts.append(f"- {err}")
            parts.append("")

        return "\n".join(parts)


# ======================================================================
# Public API
# ======================================================================

def discover_skills(root: str | Path, *, recursive: bool = True) -> DiscoveryResult:
    """Walk a directory tree and discover all skills.

    A *skill* is any directory that contains a ``SKILL.md`` file.  The
    scan automatically detects whether *root* itself is a skill (Case A)
    or a container of multiple skills (Cases B / C).

    Parameters
    ----------
    root:
        Path to scan.  May be a single skill directory or a tree
        containing many skills.
    recursive:
        When ``True`` (the default) sub-directories are walked
        recursively.  Set to ``False`` for a shallow (single-level)
        scan.

    Returns
    -------
    DiscoveryResult
        Structured result containing every discovered skill and any
        errors encountered during the scan.
    """
    from toolstore.skill_manager import SkillDefinition

    root_path = Path(root).expanduser().resolve()
    result = DiscoveryResult(root_path=root_path)

    if not root_path.is_dir():
        result.scan_errors.append(f"Path does not exist or is not a directory: {root_path}")
        return result

    # --- Case A: root is itself a skill -----------------------------------
    if (root_path / "SKILL.md").exists():
        sd = SkillDefinition(root_path)
        sd.load()
        result.skills.append(DiscoveredSkill(
            skill_def=sd,
            rel_path=Path(root_path.name),
            category="",
        ))
        return result

    # --- Cases B / C: root is a container ---------------------------------
    _walk_for_skills(root=root_path, current=root_path,
                     result=result, recursive=recursive)

    return result


def discover_skills_batch(
    *paths: str | Path,
    recursive: bool = True,
) -> Dict[str, DiscoveryResult]:
    """Run :func:`discover_skills` against multiple paths.

    Returns a dict mapping each path → its ``DiscoveryResult``.
    """
    return {str(p): discover_skills(p, recursive=recursive) for p in paths}


# ======================================================================
# Internal helpers
# ======================================================================

def _walk_for_skills(
    *,
    root: Path,
    current: Path,
    result: DiscoveryResult,
    recursive: bool,
) -> None:
    """Recursively walk *current*, appending discovered skills to *result*."""
    try:
        entries = sorted(current.iterdir())
    except PermissionError:
        result.scan_errors.append(f"Permission denied: {current}")
        return

    subdirs = [e for e in entries if e.is_dir() and not e.name.startswith(".")]

    for subdir in subdirs:
        if (subdir / "SKILL.md").exists():
            _ingest_skill_dir(subdir, root, result)

        if recursive:
            _walk_for_skills(root=root, current=subdir,
                             result=result, recursive=True)


def _ingest_skill_dir(skill_dir: Path, root: Path, result: DiscoveryResult) -> None:
    """Parse *skill_dir* as a skill and append to *result*."""
    from toolstore.skill_manager import SkillDefinition

    sd = SkillDefinition(skill_dir)
    sd.load()

    rel = skill_dir.relative_to(root)
    category = str(rel.parent) if str(rel.parent) != "." else ""

    result.skills.append(DiscoveredSkill(
        skill_def=sd,
        rel_path=rel,
        category=category,
    ))