core: Registry, Skills, Router, Executor, Verifier, Runner mit Tests

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

core/src/knaster/skills.py

@@ -0,0 +1,161 @@
+"""SKILL.md-Bibliothek (agentskills-/Hermes-kompatibel, Zusatzfelder unter metadata.knaster)."""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import Any, Literal
+
+import yaml
+from pydantic import BaseModel, Field
+
+from .store import Store
+
+_FRONTMATTER = re.compile(r"^---\s*\n(.*?)\n---\s*\n?(.*)$", re.DOTALL)
+_OPTIONAL_HEADING = re.compile(r"^##\s+Beispiele\b", re.IGNORECASE)
+
+SkillStatus = Literal["draft", "active", "demoted", "retired"]
+
+
+class OutputSpec(BaseModel):
+    format: Literal["text", "json"] = "text"
+    schema_: dict[str, Any] | None = Field(default=None, alias="schema")
+
+    model_config = {"populate_by_name": True}
+
+
+class KnasterMeta(BaseModel):
+    version: int = 1
+    status: SkillStatus = "draft"
+    stage: int = Field(default=0, ge=0, le=3)
+    execution: Literal["llm", "program", "classifier"] = "llm"
+    llm_fallback: bool = True
+    slot: str = "tier1-general"
+    cluster: str | None = None
+    adapter: Literal["auto", "none"] = "auto"
+    threshold: float = 0.80
+    triggers: list[str] = Field(default_factory=list)
+    output: OutputSpec = Field(default_factory=OutputSpec)
+    verifier: list[dict[str, Any]] = Field(default_factory=list)
+    max_output_tokens: int = 1024
+    program: str | None = None
+    classifier: str | None = None
+    min_confidence: float = 0.6
+
+
+class Skill(BaseModel):
+    name: str
+    description: str
+    body: str
+    meta: KnasterMeta
+    path: str | None = None
+
+    @property
+    def cluster(self) -> str:
+        return self.meta.cluster or self.name
+
+    @property
+    def routable(self) -> bool:
+        return self.meta.status in {"active", "demoted"}
+
+    def prompt(self, variant: Literal["full", "compact"]) -> str:
+        if variant == "full":
+            return self.body.strip()
+        return compact_body(self.body)
+
+    def index_texts(self) -> list[str]:
+        return [self.description, *self.meta.triggers]
+
+
+def compact_body(body: str) -> str:
+    """Entfernt '## Beispiele…'-Abschnitte bis zur nächsten Überschrift gleicher/höherer Ebene."""
+    out: list[str] = []
+    skipping = False
+    for line in body.splitlines():
+        if _OPTIONAL_HEADING.match(line):
+            skipping = True
+            continue
+        if skipping and re.match(r"^#{1,2}\s", line):
+            skipping = False
+        if not skipping:
+            out.append(line)
+    return "\n".join(out).strip()
+
+
+def parse_skill(text: str, path: str | None = None) -> Skill:
+    match = _FRONTMATTER.match(text)
+    if not match:
+        raise ValueError(f"{path or 'Skill'}: Frontmatter fehlt")
+    front = yaml.safe_load(match.group(1)) or {}
+    for key in ("name", "description"):
+        if not front.get(key):
+            raise ValueError(f"{path or 'Skill'}: Feld '{key}' fehlt")
+    knaster = ((front.get("metadata") or {}).get("knaster")) or {}
+    return Skill(
+        name=str(front["name"]),
+        description=str(front["description"]).strip(),
+        body=match.group(2),
+        meta=KnasterMeta(**knaster),
+        path=path,
+    )
+
+
+class SkillLibrary:
+    """Lädt Skills aus einem Verzeichnis (<dir>/<name>/SKILL.md) und überlagert Laufzeitzustand."""
+
+    def __init__(self, skills: list[Skill], store: Store | None = None):
+        self._skills = {s.name: s for s in skills}
+        if len(self._skills) != len(skills):
+            raise ValueError("Doppelte Skill-Namen")
+        self.store = store
+
+    @classmethod
+    def from_dir(cls, directory: str | Path, store: Store | None = None) -> SkillLibrary:
+        skills = []
+        for file in sorted(Path(directory).glob("*/SKILL.md")):
+            skills.append(parse_skill(file.read_text(encoding="utf-8"), str(file)))
+        return cls(skills, store)
+
+    def _states(self) -> dict[str, dict[str, Any]]:
+        if not self.store:
+            return {}
+        return {row["name"]: row for row in self.store.all_skill_states()}
+
+    @staticmethod
+    def _apply_state(skill: Skill, states: dict[str, dict[str, Any]]) -> Skill:
+        row = states.get(skill.name)
+        if not row:
+            return skill
+        meta = skill.meta.model_copy()
+        if row.get("status"):
+            meta.status = row["status"]
+        if row.get("stage") is not None:
+            meta.stage = row["stage"]
+        return skill.model_copy(update={"meta": meta})
+
+    def get(self, name: str) -> Skill:
+        return self._apply_state(self._skills[name], self._states())
+
+    def has(self, name: str) -> bool:
+        return name in self._skills
+
+    def all(self) -> list[Skill]:
+        states = self._states()
+        return [self._apply_state(s, states) for s in self._skills.values()]
+
+    def routable(self) -> list[Skill]:
+        return [s for s in self.all() if s.routable]
+
+    def set_state(self, name: str, *, status: str | None = None, stage: int | None = None,
+                  reason: str = "") -> None:
+        if name not in self._skills:
+            raise KeyError(name)
+        if not self.store:
+            raise RuntimeError("SkillLibrary ohne Store kann keinen Zustand speichern")
+        current = self.get(name)
+        self.store.set_skill_state(
+            name,
+            status=status or current.meta.status,
+            stage=current.meta.stage if stage is None else stage,
+            reason=reason,
+        )