115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
"""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 / "platform_adapters" / "antigravity" / "agent" / "skills"
|
|
LEGACY_AGENT_SKILLS_DIR = REPO_ROOT / ".agent" / "skills"
|
|
LEGACY_AGENTS_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 = []
|
|
if self.canonical_dir == CANONICAL_SKILLS_DIR and PROJECT_SKILLS_DIR.exists():
|
|
roots.append(PROJECT_SKILLS_DIR)
|
|
roots.append(self.canonical_dir)
|
|
if self.canonical_dir == CANONICAL_SKILLS_DIR and LEGACY_AGENT_SKILLS_DIR.exists():
|
|
roots.append(LEGACY_AGENT_SKILLS_DIR)
|
|
if self.canonical_dir == CANONICAL_SKILLS_DIR and LEGACY_AGENTS_SKILLS_DIR.exists():
|
|
roots.append(LEGACY_AGENTS_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 / "platform_adapters" / "antigravity" / "agent" / "skills",
|
|
]
|