mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
skills: discover roots and parse SKILL.md frontmatter
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.
This commit is contained in:
@@ -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",
|
||||
]
|
||||
@@ -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:
|
||||
"""``<origin>-<scope>``, 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() + "…"
|
||||
@@ -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
|
||||
@@ -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"
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user