Harden skill metadata discovery

This commit is contained in:
Alex Schapiro
2026-08-06 23:58:23 +00:00
parent cc292f8139
commit 853265790e
4 changed files with 128 additions and 17 deletions
+1 -1
View File
@@ -492,7 +492,7 @@ On-demand specialist skills. Spawn a specialist via `create_agent(skills=[...])`
{% for category, skills in available_skills | dictsort -%}
{% for skill in skills -%}
- {{ category }}/{{ skill.name }}: {{ skill.description }}
- {{ category }}/{{ skill.name }}{% if skill.description %}: {{ skill.description }}{% endif %}
{% endfor -%}
{% endfor -%}
</available_skills>
+75 -15
View File
@@ -4,6 +4,9 @@ import threading
from collections import Counter
from collections.abc import Iterator
from pathlib import Path
from typing import cast
import yaml
from strix.telemetry import posthog, scarf
from strix.utils.resource_paths import get_strix_resource_path
@@ -17,6 +20,7 @@ _INTERNAL_SKILL_CATEGORIES: frozenset[str] = frozenset({"scan_modes", "coordinat
_ROOT_SKILL_CATEGORY = "root"
_EXTRA_SKILL_DIRS: list[Path] = []
_SKILL_METADATA_CACHE: dict[tuple[Path, int, int], dict[str, str]] = {}
def register_skill_dir(path: str | Path) -> None:
@@ -109,13 +113,18 @@ def _get_ambiguous_skill_names() -> set[str]:
return {name for name, count in counts.items() if count > 1}
def _qualified_skill_files(skill_name: str) -> list[Path]:
def _qualified_skill_file_for_name(skill_name: str) -> Path | None:
category, _, name = skill_name.partition("/")
for skills_dir in skill_search_dirs():
candidate = _qualified_skill_file(skills_dir, category, name)
if candidate is not None:
return [candidate]
return []
return candidate
return None
def _qualified_skill_files(skill_name: str) -> list[Path]:
candidate = _qualified_skill_file_for_name(skill_name)
return [candidate] if candidate is not None else []
def _bare_skill_files(skill_name: str) -> list[Path]:
@@ -145,25 +154,75 @@ def _bare_skill_files(skill_name: str) -> list[Path]:
return candidates
def _skill_description(file_path: Path) -> str:
def _parse_skill_content(content: str) -> 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()
try:
metadata_text = frontmatter.group(0)[4:].rsplit("\n---", 1)[0]
try:
metadata_value: object = yaml.safe_load(metadata_text) or {}
except yaml.YAMLError:
# Preserve the historical support for an unquoted colon in a
# description while accepting full YAML for other frontmatter.
normalized_lines: list[str] = []
for line in metadata_text.splitlines():
if line.startswith("description: ") and not line.startswith(
('description: "', "description: '")
):
value = line.removeprefix("description: ")
value = value.replace("\\", "\\\\").replace('"', '\\"')
normalized_lines.append(f'description: "{value}"')
else:
normalized_lines.append(line)
metadata_value = yaml.safe_load("\n".join(normalized_lines)) or {}
except yaml.YAMLError:
logger.warning("Invalid skill frontmatter")
metadata_value = {}
metadata = (
cast("dict[object, object]", metadata_value) if isinstance(metadata_value, dict) else {}
)
normalized = {str(key): "" if value is None else str(value) for key, value in metadata.items()}
return normalized, content[frontmatter.end() :].lstrip()
def _read_skill_metadata(file_path: Path) -> dict[str, str]:
try:
stat = file_path.stat()
except OSError:
logger.warning("Skill file disappeared while reading metadata: %s", file_path)
return {}
cache_key = (file_path, stat.st_mtime_ns, stat.st_size)
cached = _SKILL_METADATA_CACHE.get(cache_key)
if cached is not None:
return cached
try:
content = file_path.read_text(encoding="utf-8")
except (OSError, ValueError):
return ""
frontmatter = _FRONTMATTER_PATTERN.match(content)
if frontmatter is None:
return ""
match = re.search(r"^description:\s*(.+?)\s*$", frontmatter.group(0), re.MULTILINE)
return match.group(1).strip() if match else ""
logger.warning("Failed to read skill metadata: %s", file_path)
return {}
metadata, _ = _parse_skill_content(content)
_SKILL_METADATA_CACHE[cache_key] = metadata
return metadata
def get_available_skills() -> dict[str, list[dict[str, str]]]:
grouped: dict[str, list[dict[str, str]]] = {}
for category, name in _iter_user_skill_files():
path = _qualified_skill_files(f"{category}/{name}")
grouped.setdefault(category, []).append(
{"name": name, "description": _skill_description(path[0]) if path else ""}
)
file_path = _qualified_skill_file_for_name(f"{category}/{name}")
if file_path is None:
logger.warning(
"Skill disappeared while gathering available skills: %s/%s",
category,
name,
)
continue
metadata = _read_skill_metadata(file_path)
description = " ".join(metadata.get("description", "").split())
grouped.setdefault(category, []).append({"name": name, "description": description})
return grouped
@@ -243,7 +302,8 @@ def load_skills(skill_names: list[str]) -> dict[str, str]:
continue
var_name = skill_name.split("/")[-1]
skill_content[var_name] = _FRONTMATTER_PATTERN.sub("", content).lstrip()
_, skill_body = _parse_skill_content(content)
skill_content[var_name] = skill_body
logger.debug("Loaded skill: %s -> %s", skill_name, var_name)
_track_skill_loaded(var_name, file_path)
-1
View File
@@ -164,7 +164,6 @@ Storage rules use OR-across-matches semantics: a later permissive match can reop
**Tests**
- GET GCS object paths via HTTPS without auth; verify Content-Type and `Content-Disposition: attachment`
- Test the Firebase Storage list/read/write matrix above with each available principal
- Generate and reuse signed URLs across accounts and paths; try case/URL-encoding variants
- Upload HTML/SVG and verify `X-Content-Type-Options: nosniff`; check for script execution
+52
View File
@@ -84,12 +84,64 @@ def test_available_skill_includes_frontmatter_description(tmp_path: Path) -> Non
]
def test_available_skill_supports_colon_in_description(tmp_path: Path) -> None:
_write_skill(
tmp_path,
"extra",
"widget",
"---\nname: widget\ndescription: Useful widget: handles YAML\n---\nwidget body",
)
register_skill_dir(tmp_path)
assert get_available_skills()["extra"] == [
{"name": "widget", "description": "Useful widget: handles YAML"}
]
def test_available_skill_normalizes_quoted_multiline_description(tmp_path: Path) -> None:
_write_skill(
tmp_path,
"extra",
"widget",
'---\nname: widget\ndescription: "Useful: widget guidance"\n---\nwidget body',
)
register_skill_dir(tmp_path)
assert get_available_skills()["extra"] == [
{"name": "widget", "description": "Useful: widget guidance"}
]
def test_available_skill_normalizes_block_description(tmp_path: Path) -> None:
_write_skill(
tmp_path,
"extra",
"widget",
"---\nname: widget\ndescription: |\n First line\n Second line\n---\nwidget body",
)
register_skill_dir(tmp_path)
assert get_available_skills()["extra"] == [
{"name": "widget", "description": "First line Second line"}
]
def test_system_prompt_renders_skill_descriptions() -> None:
prompt = render_system_prompt(scan_mode="quick", is_root=True)
assert "- technologies/firebase: Firebase security testing covering" in prompt
def test_system_prompt_omits_empty_skill_description(tmp_path: Path) -> None:
_write_skill(tmp_path, "extra", "widget", "---\nname: widget\ndescription:\n---\nwidget body")
register_skill_dir(tmp_path)
prompt = render_system_prompt(scan_mode="quick", is_root=True)
assert "- extra/widget\n" in prompt
assert "- extra/widget: " not in prompt
def test_registered_root_skill_is_discoverable_and_valid(tmp_path: Path) -> None:
_write_root_skill(tmp_path, "widget", "widget body")
register_skill_dir(tmp_path)