diff --git a/strix/skills/__init__.py b/strix/skills/__init__.py index 227a1b8a..0adf3d99 100644 --- a/strix/skills/__init__.py +++ b/strix/skills/__init__.py @@ -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[A-Za-z_][\w-]*):(?:[ \t]*(?P.*))?$") -_BLOCK_SCALAR_PATTERN = re.compile(r"[|>](?:[1-9][+-]?|[+-][1-9]?)?") +_FRONTMATTER_PATTERN = re.compile(r"^---\s*\n(?P.*?)\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 "", error) + parsed = None + if not _is_frontmatter_mapping(parsed): + logger.warning("Skill frontmatter is not a mapping: %s", source or "") + 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) diff --git a/tests/test_skill_dir_extension.py b/tests/test_skill_dir_extension.py index 2bf4be41..eb28768c 100644 --- a/tests/test_skill_dir_extension.py +++ b/tests/test_skill_dir_extension.py @@ -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)