cluster-knaster/core/src/knaster/skills.py

"""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 reload(self, directory: str | Path) -> None:
        """Lädt die Dateien neu; das Objekt bleibt, damit alle Referenzen (Router, Agent) gültig bleiben."""
        self._skills = SkillLibrary.from_dir(directory)._skills

    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,
        )