mirror of
https://github.com/usestrix/strix.git
synced 2026-08-18 17:52:32 +02:00
Parse skill frontmatter with YAML
This commit is contained in:
+20
-47
@@ -4,6 +4,9 @@ import threading
|
||||
from collections import Counter
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
from typing import TypeGuard
|
||||
|
||||
import yaml
|
||||
|
||||
from strix.telemetry import posthog, scarf
|
||||
from strix.utils.resource_paths import get_strix_resource_path
|
||||
@@ -11,9 +14,7 @@ from strix.utils.resource_paths import get_strix_resource_path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_FRONTMATTER_PATTERN = re.compile(r"^---\s*\n.*?\n---\s*\n", re.DOTALL)
|
||||
_FRONTMATTER_LINE_PATTERN = re.compile(r"^(?P<key>[A-Za-z_][\w-]*):(?:[ \t]*(?P<value>.*))?$")
|
||||
_BLOCK_SCALAR_PATTERN = re.compile(r"[|>](?:[1-9][+-]?|[+-][1-9]?)?")
|
||||
_FRONTMATTER_PATTERN = re.compile(r"^---\s*\n(?P<body>.*?)\n---\s*\n", re.DOTALL)
|
||||
|
||||
_INTERNAL_SKILL_CATEGORIES: frozenset[str] = frozenset({"scan_modes", "coordination"})
|
||||
_ROOT_SKILL_CATEGORY = "root"
|
||||
@@ -22,6 +23,10 @@ _EXTRA_SKILL_DIRS: list[Path] = []
|
||||
_SKILL_METADATA_CACHE: dict[tuple[Path, int, int], dict[str, str]] = {}
|
||||
|
||||
|
||||
def _is_frontmatter_mapping(value: object) -> TypeGuard[dict[object, object]]:
|
||||
return isinstance(value, dict)
|
||||
|
||||
|
||||
def register_skill_dir(path: str | Path) -> None:
|
||||
"""Add a directory searched for skills ahead of the built-in set.
|
||||
|
||||
@@ -153,57 +158,25 @@ def _bare_skill_files(skill_name: str) -> list[Path]:
|
||||
return candidates
|
||||
|
||||
|
||||
def _parse_skill_content(content: str) -> tuple[dict[str, str], str]:
|
||||
def _parse_skill_content(content: str, source: Path | None = None) -> tuple[dict[str, str], str]:
|
||||
"""Parse skill frontmatter once and return metadata plus markdown body."""
|
||||
frontmatter = _FRONTMATTER_PATTERN.match(content)
|
||||
if frontmatter is None:
|
||||
return {}, content.lstrip()
|
||||
|
||||
metadata: dict[str, str] = {}
|
||||
lines = frontmatter.group(0).splitlines()[1:]
|
||||
line_index = 0
|
||||
while line_index < len(lines):
|
||||
line = lines[line_index]
|
||||
match = _FRONTMATTER_LINE_PATTERN.match(line)
|
||||
if match is None:
|
||||
line_index += 1
|
||||
continue
|
||||
value = match.group("value") or ""
|
||||
is_quoted = len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'"
|
||||
line_indent = len(line) - len(line.lstrip())
|
||||
continuation, next_index = _consume_frontmatter_continuation(
|
||||
lines, line_index + 1, line_indent
|
||||
)
|
||||
try:
|
||||
parsed: object = yaml.safe_load(frontmatter.group("body"))
|
||||
except yaml.YAMLError as error:
|
||||
logger.warning("Failed to parse skill frontmatter %s: %s", source or "<content>", error)
|
||||
parsed = None
|
||||
if not _is_frontmatter_mapping(parsed):
|
||||
logger.warning("Skill frontmatter is not a mapping: %s", source or "<content>")
|
||||
return {}, content[frontmatter.end() :].lstrip()
|
||||
|
||||
if _BLOCK_SCALAR_PATTERN.fullmatch(value):
|
||||
value = " ".join(continuation)
|
||||
elif value and not is_quoted and continuation:
|
||||
value = " ".join([value, *continuation])
|
||||
if is_quoted:
|
||||
value = value[1:-1]
|
||||
metadata[match.group("key")] = value
|
||||
line_index = next_index
|
||||
metadata = {str(key): "" if value is None else str(value) for key, value in parsed.items()}
|
||||
return metadata, content[frontmatter.end() :].lstrip()
|
||||
|
||||
|
||||
def _consume_frontmatter_continuation(
|
||||
lines: list[str], start_index: int, base_indent: int
|
||||
) -> tuple[list[str], int]:
|
||||
continuation: list[str] = []
|
||||
index = start_index
|
||||
while index < len(lines):
|
||||
line = lines[index]
|
||||
if not line.strip():
|
||||
index += 1
|
||||
continue
|
||||
indent = len(line) - len(line.lstrip())
|
||||
if indent <= base_indent:
|
||||
break
|
||||
continuation.append(line.strip())
|
||||
index += 1
|
||||
return continuation, index
|
||||
|
||||
|
||||
def _read_skill_metadata(file_path: Path) -> dict[str, str]:
|
||||
try:
|
||||
stat = file_path.stat()
|
||||
@@ -219,7 +192,7 @@ def _read_skill_metadata(file_path: Path) -> dict[str, str]:
|
||||
except (OSError, ValueError):
|
||||
logger.warning("Failed to read skill metadata: %s", file_path)
|
||||
return {}
|
||||
metadata, _ = _parse_skill_content(content)
|
||||
metadata, _ = _parse_skill_content(content, file_path)
|
||||
_SKILL_METADATA_CACHE[cache_key] = metadata
|
||||
return metadata
|
||||
|
||||
@@ -317,7 +290,7 @@ def load_skills(skill_names: list[str]) -> dict[str, str]:
|
||||
continue
|
||||
|
||||
var_name = skill_name.split("/")[-1]
|
||||
_, skill_body = _parse_skill_content(content)
|
||||
_, skill_body = _parse_skill_content(content, file_path)
|
||||
skill_content[var_name] = skill_body
|
||||
logger.debug("Loaded skill: %s -> %s", skill_name, var_name)
|
||||
_track_skill_loaded(var_name, file_path)
|
||||
|
||||
@@ -89,7 +89,7 @@ def test_available_skill_supports_colon_in_description(tmp_path: Path) -> None:
|
||||
tmp_path,
|
||||
"extra",
|
||||
"widget",
|
||||
"---\nname: widget\ndescription: Useful widget: handles YAML\n---\nwidget body",
|
||||
'---\nname: widget\ndescription: "Useful widget: handles YAML"\n---\nwidget body',
|
||||
)
|
||||
register_skill_dir(tmp_path)
|
||||
|
||||
@@ -135,6 +135,34 @@ def test_available_skill_normalizes_multiline_descriptions(tmp_path: Path) -> No
|
||||
}
|
||||
|
||||
|
||||
def test_available_skill_supports_block_scalar_trailing_comment(tmp_path: Path) -> None:
|
||||
_write_skill(
|
||||
tmp_path,
|
||||
"extra",
|
||||
"commented",
|
||||
"---\nname: commented\ndescription: | # paragraph\n"
|
||||
" First line\n Second line\n---\ncommented body",
|
||||
)
|
||||
register_skill_dir(tmp_path)
|
||||
|
||||
assert get_available_skills()["extra"] == [
|
||||
{"name": "commented", "description": "First line Second line"}
|
||||
]
|
||||
|
||||
|
||||
def test_malformed_frontmatter_keeps_skill_body(tmp_path: Path) -> None:
|
||||
_write_skill(
|
||||
tmp_path,
|
||||
"extra",
|
||||
"broken",
|
||||
"---\nname: [broken\ndescription: should be empty\n---\nbroken body",
|
||||
)
|
||||
register_skill_dir(tmp_path)
|
||||
|
||||
assert get_available_skills()["extra"] == [{"name": "broken", "description": ""}]
|
||||
assert load_skills(["extra/broken"]) == {"broken": "broken body"}
|
||||
|
||||
|
||||
def test_system_prompt_renders_skill_descriptions() -> None:
|
||||
prompt = render_system_prompt(scan_mode="quick", is_root=True)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user