#!/usr/bin/env python3 """ Aria Skill-Index (KAR-686, AgentScope-Pattern #3) ================================================= AgentScope behandelt Skills als Dataclass (name/description/dir/updated_at) mit async Discovery. Aria hat 77 Skills als Verzeichnisse mit SKILL.md, aber keinen maschinen-lesbaren Index. Dieses Script scannt die Skill-Dirs, parst SKILL.md-Frontmatter und emittiert einen Index (JSON), nutzbar für brain-search-Integration und als Curator-Input (stale via mtime). Modi: (default) JSON-Index nach state/skill-index.json schreiben + Kurz-Summary --search Q Substring/Keyword-Suche über name+description (Discovery-Foundation) --check Skills ohne name/description flaggen + stale (mtime) auflisten --stale-days N Schwelle für "stale" (default 120) """ from __future__ import annotations import argparse import json import re import sys import time from pathlib import Path try: import yaml except ImportError: yaml = None SKILLS_DIR = Path("/root/.claude/skills") INDEX_PATH = Path("/root/aria/state/skill-index.json") def _tolerant_extract(block: str) -> dict: """Robuster Fallback: zieht name + description auch wenn YAML scheitert. Arias Frontmatter enthält real ASCII-Quotes IN quotierten Strings und [[wikilinks]], woran strict-YAML zerbricht. Hier: top-level `key:` Zeilen erkennen, description ggf. mehrzeilig (folded >, literal |, oder gequotet) bis zum nächsten Key sammeln. """ lines = block.splitlines() key_re = re.compile(r"^([A-Za-z_][\w-]*):\s?(.*)$") out: dict = {} i = 0 while i < len(lines): m = key_re.match(lines[i]) if not m: i += 1 continue key, val = m.group(1), m.group(2) # mehrzeilig sammeln, wenn folded/literal/leer oder offener Quote collected = [val] if val in (">", "|", ">-", "|-", "") or (val[:1] in "\"'" and not _quote_closed(val)): i += 1 while i < len(lines) and not key_re.match(lines[i]): collected.append(lines[i].strip()) i += 1 else: i += 1 joined = " ".join(c for c in collected if c).strip() out[key] = _strip_scalar(joined) return out def _quote_closed(s: str) -> bool: q = s[0] return len(s) >= 2 and s.rstrip().endswith(q) def _strip_scalar(s: str) -> str: s = s.strip() if len(s) >= 2 and s[0] in "\"'" and s[-1] == s[0]: s = s[1:-1] return s.strip() def parse_frontmatter(text: str) -> dict: """Extrahiert den YAML-Frontmatter-Block (zwischen den ersten beiden ---). {} wenn keiner. Strategie: erst strict-YAML (sauber bei wohlgeformten Files), bei Fehler/Non-Dict toleranter Fallback der mind. name + description rettet. """ if not text.startswith("---"): return {} end = text.find("\n---", 3) if end == -1: return {} block = text[3:end].strip("\n") if yaml: try: data = yaml.safe_load(block) if isinstance(data, dict) and data.get("name") and data.get("description"): return data except yaml.YAMLError: pass return _tolerant_extract(block) def parse_skill(skill_md: Path) -> dict | None: """Eine SKILL.md -> Index-Record. None wenn unlesbar.""" try: text = skill_md.read_text(encoding="utf-8") except OSError: return None fm = parse_frontmatter(text) desc = fm.get("description") or "" if isinstance(desc, str): desc = " ".join(desc.split()) # folded/multiline -> single line st = skill_md.stat() return { "name": fm.get("name") or skill_md.parent.name, "dir": str(skill_md.parent), "slug": skill_md.parent.name, "description": desc, "has_name": bool(fm.get("name")), "has_description": bool(fm.get("description")), "mtime": st.st_mtime, "size_bytes": st.st_size, } def build_index(skills_dir: Path) -> list[dict]: """Pure: scannt alle */SKILL.md -> sortierte Record-Liste.""" records = [] for sub in sorted(skills_dir.iterdir()): md = sub / "SKILL.md" if md.is_file(): rec = parse_skill(md) if rec: records.append(rec) return records def search(index: list[dict], query: str) -> list[dict]: """Einfache Keyword-Suche über name+description (alle Terme müssen vorkommen).""" terms = [t.lower() for t in query.split()] hits = [] for r in index: hay = (r["name"] + " " + r["description"]).lower() if all(t in hay for t in terms): hits.append(r) return hits def main(argv: list[str]) -> int: ap = argparse.ArgumentParser() ap.add_argument("--search", metavar="Q") ap.add_argument("--check", action="store_true") ap.add_argument("--stale-days", type=int, default=120) ap.add_argument("--skills-dir", default=str(SKILLS_DIR)) args = ap.parse_args(argv) sd = Path(args.skills_dir) if not sd.is_dir(): print(f"[skill-index] Skills-Dir nicht gefunden: {sd}", file=sys.stderr) return 2 index = build_index(sd) if args.search: hits = search(index, args.search) print(f"[skill-index] {len(hits)} Treffer für '{args.search}':") for r in hits: print(f" • {r['name']}: {r['description'][:90]}") return 0 if args.check: no_name = [r["slug"] for r in index if not r["has_name"]] no_desc = [r["slug"] for r in index if not r["has_description"]] now = time.time() stale = sorted( ((r["slug"], int((now - r["mtime"]) / 86400)) for r in index if (now - r["mtime"]) > args.stale_days * 86400), key=lambda x: -x[1], ) print(f"[skill-index] {len(index)} Skills.") if no_name: print(f" ⚠ ohne name: {no_name}") if no_desc: print(f" ⚠ ohne description: {no_desc}") if stale: print(f" ⚠ stale (>{args.stale_days}d, Curator-Kandidaten): " + ", ".join(f"{s}({d}d)" for s, d in stale[:15])) if not (no_name or no_desc): print(" ✓ alle Skills haben name + description.") return 0 # Default: Index schreiben INDEX_PATH.parent.mkdir(parents=True, exist_ok=True) INDEX_PATH.write_text(json.dumps(index, indent=2, ensure_ascii=False)) print(f"[skill-index] {len(index)} Skills indexiert → {INDEX_PATH}") return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))