From a9f58428bf7d26aa7cd91ba9ba72a495825944f8 Mon Sep 17 00:00:00 2001 From: worldmozara Date: Tue, 22 Sep 2026 22:37:42 +0800 Subject: [PATCH] skills: discover roots and parse SKILL.md frontmatter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A skill is a directory with a SKILL.md; roots are ordered by priority (an explicit path, then our own user directory, then the directories other harnesses install into, with the project-scope one behind its own switch). The first copy of a name wins and the ones it shadows stay listed for reporting instead of vanishing. The frontmatter reader covers the subset those files actually use — scalars, booleans, ints, inline and block lists, | and > blocks, folded plain scalars — so no YAML dependency is needed; a nested mapping is skipped and a structurally broken block degrades to "no metadata" with a note, keeping a third-party skill listable rather than dropping it. Metadata is cached against the directory list and its SKILL.md timestamps, so a listing is cheap yet notices an edit. --- src/plyngent/skills/__init__.py | 21 ++ src/plyngent/skills/discovery.py | 300 ++++++++++++++++++++++++++ src/plyngent/skills/frontmatter.py | 196 +++++++++++++++++ tests/test_skills/test_discovery.py | 195 +++++++++++++++++ tests/test_skills/test_frontmatter.py | 114 ++++++++++ 5 files changed, 826 insertions(+) create mode 100644 src/plyngent/skills/__init__.py create mode 100644 src/plyngent/skills/discovery.py create mode 100644 src/plyngent/skills/frontmatter.py create mode 100644 tests/test_skills/test_discovery.py create mode 100644 tests/test_skills/test_frontmatter.py diff --git a/src/plyngent/skills/__init__.py b/src/plyngent/skills/__init__.py new file mode 100644 index 0000000..7a03786 --- /dev/null +++ b/src/plyngent/skills/__init__.py @@ -0,0 +1,21 @@ +"""Skill discovery: roots, ``SKILL.md`` metadata, and the read/write helpers.""" + +from .discovery import SKILL_FILE as SKILL_FILE +from .discovery import Skill as Skill +from .discovery import SkillRoot as SkillRoot +from .discovery import SkillStore as SkillStore +from .discovery import build_roots as build_roots +from .discovery import validate_skill_name as validate_skill_name +from .frontmatter import FrontmatterError as FrontmatterError +from .frontmatter import parse_frontmatter as parse_frontmatter + +__all__ = [ + "SKILL_FILE", + "FrontmatterError", + "Skill", + "SkillRoot", + "SkillStore", + "build_roots", + "parse_frontmatter", + "validate_skill_name", +] diff --git a/src/plyngent/skills/discovery.py b/src/plyngent/skills/discovery.py new file mode 100644 index 0000000..1f2656d --- /dev/null +++ b/src/plyngent/skills/discovery.py @@ -0,0 +1,300 @@ +"""Discover skills under a fixed set of roots. + +A skill is a directory holding a ``SKILL.md``. Roots are ordered by priority +(explicit ``[skills].paths`` first, then the built-in sources), so the first +skill with a given name wins and the copies it shadows stay visible for +reporting rather than disappearing. Parsed metadata is cached against the +directory list and its ``SKILL.md`` timestamps, which makes a listing cheap +while still noticing an edit made outside the chat. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING + +from .frontmatter import FrontmatterError, parse_frontmatter + +if TYPE_CHECKING: + from collections.abc import Mapping, Sequence + + from plyngent.config.models import SkillsConfig + +SKILL_FILE = "SKILL.md" +DEFAULT_SKILL_DIR = "skills" +# Only the head is needed: frontmatter lives at the top, and ``skill_read`` +# reads the file itself when the model actually wants the body. +_METADATA_READ_BYTES = 64 * 1024 +_FALLBACK_DESCRIPTION_CHARS = 200 +_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$") +_FRONTMATTER_NAME = "name" +_FRONTMATTER_DESCRIPTION = "description" + + +@dataclass(frozen=True, slots=True) +class SkillRoot: + """One directory scanned for skills. + + ``origin`` names where the root came from (``config`` for an explicit path, + ``plyngent`` for our own user directory, ``claude`` for another harness); + ``scope`` is ``user`` or ``project``. Together they label a skill's + provenance in listings and in the injected catalog. + """ + + path: Path + origin: str + scope: str + + @property + def label(self) -> str: + """``-``, e.g. ``claude-project``.""" + return f"{self.origin}-{self.scope}" + + +@dataclass(frozen=True, slots=True) +class Skill: + """A discovered skill directory and the metadata read from its ``SKILL.md``.""" + + name: str + description: str + path: Path + root: SkillRoot + metadata: Mapping[str, object] = field(default_factory=dict[str, object]) + # Non-fatal notes (invalid declared name, missing description, …) surfaced to + # the human in listings so a third-party skill is never repaired silently. + issues: tuple[str, ...] = () + + @property + def skill_file(self) -> Path: + return self.path / SKILL_FILE + + @property + def declared_name(self) -> str | None: + """The name the frontmatter asked for (may differ from :attr:`name`).""" + declared = self.metadata.get(_FRONTMATTER_NAME) + return declared.strip() if isinstance(declared, str) and declared.strip() else None + + +def build_roots( + config: SkillsConfig, + *, + workspace: Path, + user_config_dir: Path, + home: Path | None = None, +) -> list[SkillRoot]: + """Resolve ``[skills]`` into ordered roots (highest priority first). + + Explicit ``paths`` win, then the built-in discovery sources in the order the + config lists them. A source contributes only directories that exist *or* are + the default write target — a missing root is not an error, it simply holds + no skills (and may hold the next created one). + """ + roots: list[SkillRoot] = [] + for raw in config.paths: + token = raw.strip() + if token: + roots.append(SkillRoot(path=_expand(token, home=home), origin="config", scope="user")) + for source in config.discover: + if source == "plyngent": + roots.append(SkillRoot(path=user_config_dir / DEFAULT_SKILL_DIR, origin="plyngent", scope="user")) + elif source == "claude": + if home is not None: + roots.append(SkillRoot(path=home / ".claude" / DEFAULT_SKILL_DIR, origin="claude", scope="user")) + if config.include_project_roots: + project = workspace / ".claude" / DEFAULT_SKILL_DIR + roots.append(SkillRoot(path=project, origin="claude", scope="project")) + return roots + + +def _expand(token: str, *, home: Path | None) -> Path: + """Expand a leading ``~`` (against *home* when given) and normalize.""" + if token.startswith("~"): + rest = token[1:].lstrip("/\\") + base = home if home is not None else Path.home() + return (base / rest) if rest else base + return Path(token) + + +@dataclass(frozen=True, slots=True) +class _Entry: + """One candidate skill directory plus the timestamp that invalidates the cache.""" + + root: SkillRoot + directory: Path + mtime_ns: int + + +def validate_skill_name(name: str) -> str | None: + """Return an error message when *name* is not usable as a skill name.""" + token = name.strip() + if not token: + return "skill name must not be empty" + if not _NAME_RE.match(token): + return f"skill name {token!r} must be lowercase letters, digits, and single hyphens (max 64 chars)" + return None + + +class SkillStore: + """Skills discovered under *roots*, highest-priority root first.""" + + _roots: tuple[SkillRoot, ...] + _cache: tuple[tuple[tuple[str, int], ...], list[Skill]] | None + + def __init__(self, roots: Sequence[SkillRoot]) -> None: + self._roots = tuple(roots) + self._cache = None + + @property + def roots(self) -> tuple[SkillRoot, ...]: + return self._roots + + @property + def all_skills(self) -> list[Skill]: + """Every discovered skill, shadowed copies included (priority order).""" + return list(self._load()) + + @property + def skills(self) -> list[Skill]: + """Visible skills by name, sorted for display (shadows removed).""" + by_name: dict[str, Skill] = {} + for skill in self._load(): + _ = by_name.setdefault(skill.name, skill) + return [by_name[name] for name in sorted(by_name)] + + def shadowed(self) -> dict[str, list[Skill]]: + """Skill names found in more than one root → the losing copies in order.""" + found: dict[str, list[Skill]] = {} + for skill in self._load(): + found.setdefault(skill.name, []).append(skill) + return {name: skills for name, skills in sorted(found.items()) if len(skills) > 1} + + def find(self, name: str) -> Skill | None: + """Look up a visible skill by name (exact, then case-insensitive).""" + token = name.strip() + wanted = token.casefold() + fallback: Skill | None = None + for skill in self._load(): + if skill.name == token: + return skill + if fallback is None and skill.name.casefold() == wanted: + fallback = skill + return fallback + + def default_write_root(self) -> SkillRoot | None: + """Root a newly created skill goes into: ours when present, else the first.""" + for root in self._roots: + if root.origin == "plyngent": + return root + return self._roots[0] if self._roots else None + + def reload(self) -> None: + """Drop the cache so the next access rescans the filesystem.""" + self._cache = None + + def _load(self) -> list[Skill]: + entries = self._scan() + signature = tuple((str(entry.directory), entry.mtime_ns) for entry in entries) + if self._cache is not None and self._cache[0] == signature: + return self._cache[1] + skills = [skill for entry in entries if (skill := _read_skill(entry)) is not None] + self._cache = (signature, skills) + return skills + + def _scan(self) -> list[_Entry]: + return [ + _Entry(root=root, directory=directory, mtime_ns=_skill_mtime(directory)) + for root in self._roots + for directory in _skill_dirs(root.path) + ] + + +def _skill_dirs(root: Path) -> list[Path]: + """Skill directories directly under *root* (the root itself may be one).""" + try: + if not root.is_dir(): + return [] + found = [root] if (root / SKILL_FILE).is_file() else [] + children = sorted(child for child in root.iterdir() if child.is_dir()) + except OSError: + return [] + found.extend(child for child in children if (child / SKILL_FILE).is_file()) + return found + + +def _skill_mtime(directory: Path) -> int: + try: + return (directory / SKILL_FILE).stat().st_mtime_ns + except OSError: + return -1 + + +def _read_skill(entry: _Entry) -> Skill | None: + """Read one ``SKILL.md`` head and derive the record; ``None`` when unreadable.""" + issues: list[str] = [] + try: + text = _read_head(entry.directory / SKILL_FILE) + except OSError as exc: + return Skill( + name=entry.directory.name, + description="", + path=entry.directory, + root=entry.root, + issues=(f"cannot read {SKILL_FILE}: {exc}",), + ) + try: + metadata, body = parse_frontmatter(text) + except FrontmatterError as exc: + metadata, body = {}, text + issues.append(f"frontmatter not parsed ({exc}); using the directory name") + + name = entry.directory.name + declared = metadata.get(_FRONTMATTER_NAME) + if isinstance(declared, str) and declared.strip(): + candidate = declared.strip() + if validate_skill_name(candidate) is None: + name = candidate + else: + issues.append(f"declared name {candidate!r} is not usable; using {entry.directory.name!r}") + elif declared is not None: + issues.append(f"declared name is not a string ({type(declared).__name__}); using {entry.directory.name!r}") + + description = metadata.get(_FRONTMATTER_DESCRIPTION) + if isinstance(description, str) and description.strip(): + text_description = description.strip() + else: + text_description = _fallback_description(body) + issues.append("no frontmatter description; using the first paragraph") + return Skill( + name=name, + description=_one_line(text_description), + path=entry.directory, + root=entry.root, + metadata=metadata, + issues=tuple(issues), + ) + + +def _read_head(path: Path) -> str: + """Read the head of a file as UTF-8 (frontmatter is at the top).""" + with path.open("r", encoding="utf-8", errors="replace") as handle: + return handle.read(_METADATA_READ_BYTES) + + +def _fallback_description(body: str) -> str: + """First plain paragraph of the body, for skills without a description.""" + for block in body.split("\n\n"): + lines = [line.strip() for line in block.splitlines() if line.strip()] + if not lines or lines[0].startswith("#"): + continue + return " ".join(lines) + return "" + + +def _one_line(text: str) -> str: + """Collapse whitespace and cap the length of a description line.""" + collapsed = " ".join(text.split()) + if len(collapsed) <= _FALLBACK_DESCRIPTION_CHARS: + return collapsed + return collapsed[:_FALLBACK_DESCRIPTION_CHARS].rstrip() + "…" diff --git a/src/plyngent/skills/frontmatter.py b/src/plyngent/skills/frontmatter.py new file mode 100644 index 0000000..a6b3d5c --- /dev/null +++ b/src/plyngent/skills/frontmatter.py @@ -0,0 +1,196 @@ +"""YAML-subset frontmatter for ``SKILL.md`` files. + +Skills are shared with other harnesses, which write a small YAML block above the +Markdown body. Supporting all of YAML would mean a dependency for a handful of +scalar fields, so this module reads the subset those files actually use: +top-level ``key: value`` pairs, quoted or plain scalars, booleans and integers, +inline ``[a, b]`` lists, block ``- item`` lists, and ``|`` / ``>`` block scalars. +A plain scalar folds the indented lines below it (YAML's multi-line form) and a +nested mapping is skipped rather than guessed at; anything structurally broken +raises :class:`FrontmatterError` so callers can fall back to "no metadata". +""" + +from __future__ import annotations + +import re + +_FENCE = "---" +# Closing markers accepted by the format (``...`` is the YAML document end). +_CLOSERS = frozenset({_FENCE, "..."}) +_BLOCK_SCALAR_MARKERS = frozenset({"|", "|-", "|+", ">", ">-", ">+"}) +_TRUE = frozenset({"true", "yes", "on"}) +_FALSE = frozenset({"false", "no", "off"}) +_INT = re.compile(r"^-?\d+$") +_ITEM = "- " +# A quoted scalar needs at least its two quote characters. +_QUOTED_MIN_LEN = 2 + + +class FrontmatterError(ValueError): + """Raised when a frontmatter block is outside the supported subset.""" + + +def parse_frontmatter(text: str) -> tuple[dict[str, object], str]: + """Return ``(metadata, body)`` for a ``SKILL.md`` document. + + No frontmatter (or an unterminated block) yields empty metadata and the whole + text as the body. + + Raises: + FrontmatterError: The block is outside the supported subset. + """ + block, body = split_frontmatter(text) + if block is None: + return {}, body + return parse_yaml_subset(block), body.removeprefix("\n") + + +def split_frontmatter(text: str) -> tuple[str | None, str]: + """Split *text* into its raw frontmatter block (or ``None``) and the body. + + The block must open with a ``---`` line of its own and close with ``---`` or + ``...``; an unterminated block counts as no frontmatter, so the whole text + stays readable as the body. + """ + normalized = text.replace("\r\n", "\n").replace("\r", "\n") + lines = normalized.split("\n") + if not lines or lines[0].strip() != _FENCE: + return None, normalized + for index, line in enumerate(lines[1:], start=1): + if line.strip() in _CLOSERS: + return "\n".join(lines[1:index]), "\n".join(lines[index + 1 :]) + return None, normalized + + +def parse_yaml_subset(block: str) -> dict[str, object]: + """Parse one frontmatter block into a flat mapping. + + Raises: + FrontmatterError: A line is not a top-level ``key: value`` pair. + """ + metadata: dict[str, object] = {} + lines = block.split("\n") + index = 0 + while index < len(lines): + raw = lines[index] + index += 1 + stripped = raw.strip() + if not stripped or stripped.startswith("#"): + continue + if stripped.startswith(_ITEM): + msg = f"list item without a key: {raw!r}" + raise FrontmatterError(msg) + if _indent_of(raw) > 0: + msg = f"unexpected indentation: {raw!r}" + raise FrontmatterError(msg) + key, sep, value = raw.partition(":") + if not sep or not key.strip(): + msg = f"expected 'key: value', got {raw!r}" + raise FrontmatterError(msg) + name = key.strip() + value = value.strip() + consumed, index = _consume_indented(lines, index) + if value in _BLOCK_SCALAR_MARKERS: + metadata[name] = _join_block_scalar(consumed, folded=value.startswith(">")) + continue + if value: + metadata[name] = _scalar(value, consumed) + continue + items = _list_items(consumed) + if items is not None: + metadata[name] = items + # Empty value with nested mappings: nothing this subset needs to record. + return metadata + + +def _indent_of(line: str) -> int: + return len(line) - len(line.lstrip(" ")) + + +def _consume_indented(lines: list[str], index: int) -> tuple[list[str], int]: + """Consume lines indented deeper than the key line; return them and the next index.""" + consumed: list[str] = [] + while index < len(lines): + line = lines[index] + if line.strip() and _indent_of(line) == 0: + break + consumed.append(line) + index += 1 + return consumed, index + + +def _dedent(lines: list[str]) -> list[str]: + """Strip the common indentation and trim blank edges.""" + indents = [_indent_of(line) for line in lines if line.strip()] + if not indents: + return [] + width = min(indents) + dedented = [line[width:] if len(line) >= width else "" for line in lines] + while dedented and not dedented[0].strip(): + _ = dedented.pop(0) + while dedented and not dedented[-1].strip(): + _ = dedented.pop() + return dedented + + +def _join_block_scalar(lines: list[str], *, folded: bool) -> str: + """Join a ``|`` (literal) or ``>`` (folded) block scalar's content.""" + body = _dedent(lines) + if not body: + return "" + return " ".join(line.strip() for line in body) if folded else "\n".join(body) + + +def _list_items(lines: list[str]) -> list[object] | None: + """Parse a block ``- item`` list; ``None`` when *lines* is not one.""" + meaningful = [line for line in lines if line.strip()] + if not meaningful or any(not line.strip().startswith(_ITEM) for line in meaningful): + return None + items: list[object] = [] + for line in meaningful: + item = line.strip()[len(_ITEM) :].strip() + if item: + items.append(_coerce(item)) + return items + + +def _unquote(text: str) -> str | None: + """Return the contents of a quoted scalar, or ``None`` when it is unquoted.""" + if len(text) >= _QUOTED_MIN_LEN and text[0] == text[-1] and text[0] in {'"', "'"}: + return text[1:-1] + return None + + +def _scalar(value: str, continuation: list[str]) -> object: + """Parse one scalar; a plain scalar folds its indented continuation lines.""" + if value.startswith("[") and value.endswith("]"): + return _inline_list(value) + unquoted = _unquote(value) + if unquoted is not None: + return unquoted + folded = [value] + folded.extend(line.strip() for line in continuation if line.strip()) + return _coerce(" ".join(folded)) + + +def _inline_list(value: str) -> list[object]: + """Parse ``[a, "b", 3]`` (no nested collections in this subset).""" + inner = value[1:-1].strip() + if not inner: + return [] + return [_coerce(part.strip()) for part in inner.split(",") if part.strip()] + + +def _coerce(text: str) -> object: + """Coerce a plain scalar to bool / int / str (YAML's implicit typing).""" + unquoted = _unquote(text) + if unquoted is not None: + return unquoted + lowered = text.lower() + if lowered in _TRUE: + return True + if lowered in _FALSE: + return False + if _INT.match(text): + return int(text) + return text diff --git a/tests/test_skills/test_discovery.py b/tests/test_skills/test_discovery.py new file mode 100644 index 0000000..ca20df9 --- /dev/null +++ b/tests/test_skills/test_discovery.py @@ -0,0 +1,195 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from plyngent.config.models import SkillsConfig +from plyngent.skills import SKILL_FILE, Skill, SkillStore, build_roots, validate_skill_name + +if TYPE_CHECKING: + from pathlib import Path + + +def _write_skill(root: Path, name: str, *, description: str | None = "A test skill", body: str = "Body.") -> Path: + directory = root / name + directory.mkdir(parents=True, exist_ok=True) + front = "---\n" + if description is not None: + front += f"name: {name}\ndescription: {description}\n" + front += "---\n\n" + _ = (directory / SKILL_FILE).write_text(front + body + "\n", encoding="utf-8") + return directory + + +def test_build_roots_orders_explicit_paths_first(tmp_path: Path) -> None: + config = SkillsConfig(paths=["~/team-skills", "/srv/skills"]) + roots = build_roots( + config, + workspace=tmp_path / "proj", + user_config_dir=tmp_path / "config", + home=tmp_path / "home", + ) + assert [(r.origin, r.scope, str(r.path)) for r in roots] == [ + ("config", "user", str(tmp_path / "home" / "team-skills")), + ("config", "user", "/srv/skills"), + ("plyngent", "user", str(tmp_path / "config" / "skills")), + ("claude", "user", str(tmp_path / "home" / ".claude" / "skills")), + ("claude", "project", str(tmp_path / "proj" / ".claude" / "skills")), + ] + + +def test_build_roots_respects_discover_and_project_switch(tmp_path: Path) -> None: + config = SkillsConfig(discover=["plyngent"], include_project_roots=False) + roots = build_roots(config, workspace=tmp_path, user_config_dir=tmp_path / "cfg", home=tmp_path) + assert [r.origin for r in roots] == ["plyngent"] + + +def test_store_discovers_skills_and_labels_their_root(tmp_path: Path) -> None: + user = tmp_path / "config" / "skills" + _ = _write_skill(user, "pdf-processing", description="Extract text from PDFs") + store = SkillStore( + build_roots(SkillsConfig(discover=["plyngent"]), workspace=tmp_path, user_config_dir=tmp_path / "config") + ) + skills = store.skills + assert [(s.name, s.description, s.root.label) for s in skills] == [ + ("pdf-processing", "Extract text from PDFs", "plyngent-user"), + ] + assert skills[0].skill_file.name == SKILL_FILE + + +def test_store_lists_skills_from_several_roots_sorted(tmp_path: Path) -> None: + config_dir = tmp_path / "config" + _ = _write_skill(config_dir / "skills", "zeta") + _ = _write_skill(tmp_path / "home" / ".claude" / "skills", "alpha") + store = SkillStore( + build_roots(SkillsConfig(), workspace=tmp_path, user_config_dir=config_dir, home=tmp_path / "home") + ) + assert [s.name for s in store.skills] == ["alpha", "zeta"] + assert [s.root.origin for s in store.skills] == ["claude", "plyngent"] + + +def test_explicit_path_wins_and_shadowed_copy_stays_listed(tmp_path: Path) -> None: + explicit = tmp_path / "team" + config_dir = tmp_path / "config" + _ = _write_skill(explicit, "shared", description="Team copy") + _ = _write_skill(config_dir / "skills", "shared", description="User copy") + config = SkillsConfig(paths=[str(explicit)], discover=["plyngent"]) + store = SkillStore(build_roots(config, workspace=tmp_path, user_config_dir=config_dir)) + + found = store.find("shared") + assert found is not None + assert found.description == "Team copy" + assert [s.description for s in store.shadowed()["shared"]] == ["Team copy", "User copy"] + assert len(store.all_skills) == 2 + + +def test_find_is_case_insensitive_fallback(tmp_path: Path) -> None: + _ = _write_skill(tmp_path / "skills", "pdf-processing") + store = SkillStore(build_roots(SkillsConfig(discover=["plyngent"]), workspace=tmp_path, user_config_dir=tmp_path)) + assert store.find("PDF-Processing") is not None + assert store.find("nope") is None + + +def test_root_itself_may_be_a_skill(tmp_path: Path) -> None: + single = tmp_path / "my-skill" + single.mkdir() + _ = (single / SKILL_FILE).write_text("---\nname: my-skill\ndescription: One-off\n---\n\nBody.\n", encoding="utf-8") + store = SkillStore(build_roots(SkillsConfig(paths=[str(single)]), workspace=tmp_path, user_config_dir=tmp_path)) + assert [s.name for s in store.skills] == ["my-skill"] + + +def test_missing_roots_are_not_an_error(tmp_path: Path) -> None: + store = SkillStore( + build_roots(SkillsConfig(), workspace=tmp_path, user_config_dir=tmp_path / "nope", home=tmp_path) + ) + assert store.skills == [] + assert store.shadowed() == {} + + +def test_description_falls_back_to_the_first_paragraph(tmp_path: Path) -> None: + directory = tmp_path / "skills" / "no-desc" + directory.mkdir(parents=True) + _ = (directory / SKILL_FILE).write_text( + "# Heading\n\nFirst paragraph of the skill.\n\nSecond paragraph.\n", + encoding="utf-8", + ) + store = SkillStore(build_roots(SkillsConfig(discover=["plyngent"]), workspace=tmp_path, user_config_dir=tmp_path)) + skill = store.skills[0] + assert skill.description == "First paragraph of the skill." + assert skill.declared_name is None + assert any("no frontmatter description" in issue for issue in skill.issues) + + +def test_unusable_declared_name_falls_back_to_the_directory(tmp_path: Path) -> None: + directory = tmp_path / "skills" / "dir-name" + directory.mkdir(parents=True) + _ = (directory / SKILL_FILE).write_text( + '---\nname: "PDF Skills"\ndescription: Mixed case\n---\n\nBody.\n', + encoding="utf-8", + ) + store = SkillStore(build_roots(SkillsConfig(discover=["plyngent"]), workspace=tmp_path, user_config_dir=tmp_path)) + skill = store.skills[0] + assert skill.name == "dir-name" + assert skill.declared_name == "PDF Skills" + assert any("not usable" in issue for issue in skill.issues) + + +def test_broken_frontmatter_still_yields_a_listable_skill(tmp_path: Path) -> None: + directory = tmp_path / "skills" / "broken" + directory.mkdir(parents=True) + _ = (directory / SKILL_FILE).write_text("---\nname: broken\nbroken line\n---\n\nBody text.\n", encoding="utf-8") + store = SkillStore(build_roots(SkillsConfig(discover=["plyngent"]), workspace=tmp_path, user_config_dir=tmp_path)) + skill = store.skills[0] + assert skill.name == "broken" + assert skill.description # falls back to the body + assert any("frontmatter not parsed" in issue for issue in skill.issues) + + +def test_cache_picks_up_an_edit_and_reload_forces_a_rescan(tmp_path: Path) -> None: + root = tmp_path / "skills" + directory = _write_skill(root, "cached", description="Before") + store = SkillStore(build_roots(SkillsConfig(discover=["plyngent"]), workspace=tmp_path, user_config_dir=tmp_path)) + assert store.skills[0].description == "Before" + + _ = (directory / SKILL_FILE).write_text( + "---\nname: cached\ndescription: After\n---\n\nBody.\n", + encoding="utf-8", + ) + assert store.skills[0].description == "After" + + _ = _write_skill(root, "added") + store.reload() + assert [s.name for s in store.skills] == ["added", "cached"] + + +def test_default_write_root_prefers_our_own_directory(tmp_path: Path) -> None: + config_dir = tmp_path / "config" + store = SkillStore(build_roots(SkillsConfig(), workspace=tmp_path, user_config_dir=config_dir, home=tmp_path)) + root = store.default_write_root() + assert root is not None + assert root.origin == "plyngent" + assert root.path == config_dir / "skills" + + explicit = SkillStore( + build_roots(SkillsConfig(paths=["/srv/skills"], discover=[]), workspace=tmp_path, user_config_dir=config_dir) + ) + fallback = explicit.default_write_root() + assert fallback is not None and fallback.origin == "config" + assert SkillStore([]).default_write_root() is None + + +def test_validate_skill_name() -> None: + assert validate_skill_name("pdf-processing") is None + assert validate_skill_name("a1") is None + assert validate_skill_name("") is not None + assert validate_skill_name("PDF") is not None + assert validate_skill_name("with_underscore") is not None + assert validate_skill_name("x" * 65) is not None + + +def test_skill_is_a_frozen_record(tmp_path: Path) -> None: + directory = _write_skill(tmp_path / "skills", "frozen") + store = SkillStore(build_roots(SkillsConfig(discover=["plyngent"]), workspace=tmp_path, user_config_dir=tmp_path)) + skill = store.skills[0] + assert isinstance(skill, Skill) + assert skill.path == directory + assert skill.metadata["name"] == "frozen" diff --git a/tests/test_skills/test_frontmatter.py b/tests/test_skills/test_frontmatter.py new file mode 100644 index 0000000..ed3f0db --- /dev/null +++ b/tests/test_skills/test_frontmatter.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import pytest + +from plyngent.skills.frontmatter import ( + FrontmatterError, + parse_frontmatter, + parse_yaml_subset, + split_frontmatter, +) + + +def test_parse_frontmatter_basic() -> None: + text = ( + "---\n" + "name: pdf-processing\n" + 'description: "Extract text and tables from PDFs"\n' + "version: 2\n" + "license: MIT\n" + "---\n" + "\n" + "# PDF processing\n" + "\n" + "Use pdftotext first.\n" + ) + metadata, body = parse_frontmatter(text) + assert metadata == { + "name": "pdf-processing", + "description": "Extract text and tables from PDFs", + "version": 2, + "license": "MIT", + } + assert body.startswith("# PDF processing") + assert "pdftotext" in body + + +def test_parse_frontmatter_without_block_keeps_the_whole_text() -> None: + text = "# Just a skill\n\nNo frontmatter here.\n" + metadata, body = parse_frontmatter(text) + assert metadata == {} + assert body == text + + +def test_parse_frontmatter_unterminated_block_is_not_frontmatter() -> None: + text = "---\nname: x\n\nbody without a closing fence\n" + metadata, body = parse_frontmatter(text) + assert metadata == {} + assert body == text + + +def test_frontmatter_closing_dots_are_accepted() -> None: + block, body = split_frontmatter("---\nname: x\n...\nbody\n") + assert block == "name: x" + assert body == "body\n" + + +def test_yaml_subset_types_lists_and_booleans() -> None: + block = "\n".join( + [ + "name: acme", + "read_only: true", + "confirm: no", + "retries: 3", + "allowed-tools: [read_file, grep_files]", + "tags:", + " - alpha", + " - beta", + ] + ) + assert parse_yaml_subset(block) == { + "name": "acme", + "read_only": True, + "confirm": False, + "retries": 3, + "allowed-tools": ["read_file", "grep_files"], + "tags": ["alpha", "beta"], + } + + +def test_yaml_subset_plain_scalar_folds_continuation_lines() -> None: + block = "description: A long description\n that continues on the next line\n" + assert parse_yaml_subset(block) == { + "description": "A long description that continues on the next line", + } + + +def test_yaml_subset_block_scalars() -> None: + literal = "description: |\n first\n second\n" + folded = "description: >\n first\n second\n" + assert parse_yaml_subset(literal) == {"description": "first\nsecond"} + assert parse_yaml_subset(folded) == {"description": "first second"} + + +def test_yaml_subset_nested_mapping_is_skipped() -> None: + """Unsupported nesting must not lose the flat fields around it.""" + block = "name: acme\nmetadata:\n author: someone\n homepage: https://example.com\nlicense: MIT\n" + assert parse_yaml_subset(block) == {"name": "acme", "license": "MIT"} + + +def test_yaml_subset_comments_and_blank_lines() -> None: + block = "# a comment\n\nname: acme\n" + assert parse_yaml_subset(block) == {"name": "acme"} + + +def test_yaml_subset_rejects_broken_lines() -> None: + with pytest.raises(FrontmatterError): + _ = parse_yaml_subset("name: acme\nnot a pair\n") + with pytest.raises(FrontmatterError): + _ = parse_yaml_subset("- orphan item\n") + + +def test_yaml_subset_rejects_indented_top_level() -> None: + with pytest.raises(FrontmatterError): + _ = parse_yaml_subset(" name: acme\n")