v0.20 alpha skill-driven python core

scripts/runtime/skills.py

@@ -0,0 +1,107 @@
+"""Canonical skill registry and adapter sync helpers."""
+
+from __future__ import annotations
+
+import shutil
+from dataclasses import dataclass
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+CANONICAL_SKILLS_DIR = REPO_ROOT / ".agents" / "skills"
+PROJECT_SKILLS_DIR = REPO_ROOT / "skills"
+REQUIRED_SKILLS = {
+    "search-strategy",
+    "search-gateway",
+    "source-quality",
+    "length-budget",
+    "evidence-table",
+    "citation-manager",
+    "mckinsey-method",
+}
+
+
+@dataclass(frozen=True)
+class SkillInfo:
+    name: str
+    path: Path
+
+
+class SkillRegistry:
+    """Reads skills from the canonical cross-adapter registry."""
+
+    def __init__(self, canonical_dir: Path | None = None) -> None:
+        self.canonical_dir = canonical_dir or CANONICAL_SKILLS_DIR
+
+    def roots(self) -> list[Path]:
+        roots = [self.canonical_dir]
+        if self.canonical_dir == CANONICAL_SKILLS_DIR and PROJECT_SKILLS_DIR.exists():
+            roots.append(PROJECT_SKILLS_DIR)
+        return roots
+
+    def list(self) -> list[SkillInfo]:
+        seen: set[str] = set()
+        out: list[SkillInfo] = []
+        for root in self.roots():
+            if not root.exists():
+                continue
+            for path in sorted(root.glob("*/SKILL.md")):
+                name = path.parent.name
+                if name in seen:
+                    continue
+                seen.add(name)
+                out.append(SkillInfo(name=name, path=path))
+        return out
+
+    def list_names(self) -> list[str]:
+        return [item.name for item in self.list()]
+
+    def read(self, name: str) -> str:
+        for root in self.roots():
+            path = root / name / "SKILL.md"
+            if path.exists():
+                return path.read_text(encoding="utf-8")
+        raise FileNotFoundError(f"skill not found: {name}")
+
+    def validate(self, required: set[str] | None = None) -> dict[str, object]:
+        names = set(self.list_names())
+        required_names = required or REQUIRED_SKILLS
+        missing = sorted(required_names - names)
+        malformed: list[str] = []
+        for item in self.list():
+            text = item.path.read_text(encoding="utf-8")
+            if "name:" not in text[:300]:
+                malformed.append(item.name)
+        return {
+            "ok": not missing and not malformed,
+            "canonical_dir": str(self.canonical_dir),
+            "count": len(names),
+            "missing": missing,
+            "malformed": malformed,
+        }
+
+    def sync_to(self, targets: list[Path], *, force: bool = True) -> int:
+        """Copy canonical skills into adapter skill directories.
+
+        Returns the number of skill directories copied across all targets.
+        """
+        copied = 0
+        for target in targets:
+            if target.resolve() == self.canonical_dir.resolve():
+                continue
+            target.mkdir(parents=True, exist_ok=True)
+            for item in self.list():
+                dst = target / item.name
+                if dst.exists() and force:
+                    shutil.rmtree(dst)
+                if not dst.exists():
+                    shutil.copytree(item.path.parent, dst)
+                    copied += 1
+        return copied
+
+
+def default_adapter_skill_dirs() -> list[Path]:
+    return [
+        REPO_ROOT / ".opencode" / "skills",
+        REPO_ROOT / ".agents" / "skills",
+    ]