From 129d0035129e3fed971aefa8a1cd3991c369ef2f Mon Sep 17 00:00:00 2001 From: alex s <46074070+bearsyankees@users.noreply.github.com> Date: Sun, 12 Jul 2026 12:05:58 -0400 Subject: [PATCH] Add skill directory registration (#746) --- strix/agents/prompt.py | 6 +- strix/skills/__init__.py | 185 ++++++++++++++++++++++++------ tests/test_skill_dir_extension.py | 120 +++++++++++++++++++ 3 files changed, 273 insertions(+), 38 deletions(-) create mode 100644 tests/test_skill_dir_extension.py diff --git a/strix/agents/prompt.py b/strix/agents/prompt.py index 5d902176..8000eb09 100644 --- a/strix/agents/prompt.py +++ b/strix/agents/prompt.py @@ -7,7 +7,7 @@ from typing import Any from jinja2 import Environment, FileSystemLoader, select_autoescape -from strix.skills import get_available_skills, load_skills +from strix.skills import get_available_skills, load_skills, skill_search_dirs from strix.utils.resource_paths import get_strix_resource_path @@ -69,9 +69,9 @@ def render_system_prompt( """Render the system prompt. Returns empty string on template failure.""" try: prompt_dir = get_strix_resource_path("agents", _PROMPT_DIRNAME) - skills_dir = get_strix_resource_path("skills") + loader_dirs = [prompt_dir, *skill_search_dirs()] env = Environment( - loader=FileSystemLoader([prompt_dir, skills_dir]), + loader=FileSystemLoader(loader_dirs), autoescape=select_autoescape( enabled_extensions=(), default_for_string=False, diff --git a/strix/skills/__init__.py b/strix/skills/__init__.py index 96f94a4b..660aa77a 100644 --- a/strix/skills/__init__.py +++ b/strix/skills/__init__.py @@ -1,6 +1,8 @@ import logging import re +from collections import Counter from collections.abc import Iterator +from pathlib import Path from strix.utils.resource_paths import get_strix_resource_path @@ -10,20 +12,82 @@ logger = logging.getLogger(__name__) _FRONTMATTER_PATTERN = re.compile(r"^---\s*\n.*?\n---\s*\n", re.DOTALL) _INTERNAL_SKILL_CATEGORIES: frozenset[str] = frozenset({"scan_modes", "coordination"}) +_ROOT_SKILL_CATEGORY = "root" + +_EXTRA_SKILL_DIRS: list[Path] = [] + + +def register_skill_dir(path: str | Path) -> None: + """Add a directory searched for skills ahead of the built-in set. + + The directory uses the same layout as the packaged skills + (``//.md``). Skills found in a registered + directory shadow packaged skills with the same relative path, so + callers can both add new skills and override existing ones without + editing the package. The most recently registered directory has the + highest precedence. + """ + resolved = Path(path) + if resolved not in _EXTRA_SKILL_DIRS: + _EXTRA_SKILL_DIRS.append(resolved) + logger.info("Registered extra skill dir: %s", resolved) + + +def registered_skill_dirs() -> tuple[Path, ...]: + """Return registered extra skill directories, highest precedence first.""" + return tuple(reversed(_EXTRA_SKILL_DIRS)) + + +def skill_search_dirs() -> tuple[Path, ...]: + """All existing skill roots, highest precedence first (built-in last).""" + roots = [d for d in registered_skill_dirs() if d.is_dir()] + builtin = get_strix_resource_path("skills") + if builtin.is_dir(): + roots.append(builtin) + return tuple(roots) def _iter_user_skill_files() -> Iterator[tuple[str, str]]: """Yield ``(category_name, skill_name)`` for every user-selectable skill.""" - skills_dir = get_strix_resource_path("skills") - if not skills_dir.exists(): - return - for category_dir in sorted(skills_dir.iterdir()): - if not category_dir.is_dir() or category_dir.name.startswith("__"): - continue - if category_dir.name in _INTERNAL_SKILL_CATEGORIES: - continue - for file_path in sorted(category_dir.glob("*.md")): - yield category_dir.name, file_path.stem + seen: set[tuple[str, str]] = set() + for skills_dir in skill_search_dirs(): + for file_path in sorted(skills_dir.glob("*.md")): + if file_path.name.startswith("__") or file_path.name == "README.md": + continue + key = (_ROOT_SKILL_CATEGORY, file_path.stem) + if key in seen: + continue + seen.add(key) + yield key + + for category_dir in sorted(skills_dir.iterdir()): + if not category_dir.is_dir() or category_dir.name.startswith("__"): + continue + if category_dir.name in _INTERNAL_SKILL_CATEGORIES: + continue + for file_path in sorted(category_dir.glob("*.md")): + key = (category_dir.name, file_path.stem) + if key in seen: + continue + seen.add(key) + yield key + + +def _is_selectable_root_skill_file(file_path: Path) -> bool: + return file_path.suffix == ".md" and not ( + file_path.name.startswith("__") or file_path.name == "README.md" + ) + + +def _qualified_skill_file(skills_dir: Path, category: str, name: str) -> Path | None: + if category == _ROOT_SKILL_CATEGORY: + candidate = skills_dir / f"{name}.md" + if candidate.exists() and _is_selectable_root_skill_file(candidate): + return candidate + return None + + candidate = skills_dir / category / f"{name}.md" + return candidate if candidate.exists() else None def get_all_skill_names() -> set[str]: @@ -31,6 +95,54 @@ def get_all_skill_names() -> set[str]: return {name for _, name in _iter_user_skill_files()} +def _get_all_skill_keys() -> set[str]: + keys: set[str] = set() + for category, name in _iter_user_skill_files(): + keys.add(f"{category}/{name}") + return keys + + +def _get_ambiguous_skill_names() -> set[str]: + counts = Counter(name for _, name in _iter_user_skill_files()) + return {name for name, count in counts.items() if count > 1} + + +def _qualified_skill_files(skill_name: str) -> list[Path]: + 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 [] + + +def _bare_skill_files(skill_name: str) -> list[Path]: + seen: set[tuple[str, str]] = set() + candidates: list[Path] = [] + for skills_dir in skill_search_dirs(): + for category_dir in sorted(skills_dir.iterdir()): + if not category_dir.is_dir() or category_dir.name.startswith("__"): + continue + if category_dir.name in _INTERNAL_SKILL_CATEGORIES: + continue + key = (category_dir.name, skill_name) + if key in seen: + continue + candidate = category_dir / f"{skill_name}.md" + if candidate.exists(): + seen.add(key) + candidates.append(candidate) + + key = (_ROOT_SKILL_CATEGORY, skill_name) + if key in seen: + continue + candidate = _qualified_skill_file(skills_dir, _ROOT_SKILL_CATEGORY, skill_name) + if candidate is not None: + seen.add(key) + candidates.append(candidate) + return candidates + + def get_available_skills() -> dict[str, list[str]]: grouped: dict[str, list[str]] = {} for category, name in _iter_user_skill_files(): @@ -52,48 +164,51 @@ def validate_requested_skills(skill_list: list[str], max_skills: int = 5) -> str if not skill_list: return None available = get_all_skill_names() - invalid = sorted({s for s in skill_list if s not in available}) + available_keys = _get_all_skill_keys() + invalid = sorted({s for s in skill_list if s not in available and s not in available_keys}) if invalid: return f"Invalid skill name(s): {invalid}. Available skills: {sorted(available)}" + ambiguous = sorted({s for s in skill_list if "/" not in s} & _get_ambiguous_skill_names()) + if ambiguous: + return ( + f"Ambiguous skill name(s): {ambiguous}. Use category-qualified names from: " + f"{sorted(available_keys)}" + ) return None +def _candidate_skill_files(skill_name: str) -> list[Path]: + """Resolve *skill_name* to effective matching files.""" + if "/" in skill_name: + return _qualified_skill_files(skill_name) + return _bare_skill_files(skill_name) + + def load_skills(skill_names: list[str]) -> dict[str, str]: """Load skill markdown bodies (frontmatter stripped) by name. - Skill files live at ``strix/skills//.md``. Names - can be ``"name"`` (any category), ``"category/name"``, or a bare - file at the skills root. Missing skills are logged and skipped. + Skill files live at ``strix/skills//.md`` (or any + directory added via :func:`register_skill_dir`, searched first). + Names can be ``"name"`` (any category), ``"category/name"``, or a + bare file at the skills root. Missing skills are logged and skipped. """ - skills_dir = get_strix_resource_path("skills") - if not skills_dir.exists(): + search_dirs = skill_search_dirs() + if not search_dirs: return {} - by_category: dict[str, str] = {} - for category_dir in skills_dir.iterdir(): - if not category_dir.is_dir() or category_dir.name.startswith("__"): - continue - for file_path in category_dir.glob("*.md"): - by_category[file_path.stem] = f"{category_dir.name}/{file_path.stem}.md" - skill_content: dict[str, str] = {} for skill_name in skill_names: - rel_path: str | None - if "/" in skill_name: - rel_path = f"{skill_name}.md" - elif skill_name in by_category: - rel_path = by_category[skill_name] - elif (skills_dir / f"{skill_name}.md").exists(): - rel_path = f"{skill_name}.md" - else: - rel_path = None - - if rel_path is None or not (skills_dir / rel_path).exists(): + candidates = _candidate_skill_files(skill_name) + if not candidates: logger.warning("Skill not found: %s", skill_name) continue + if len(candidates) > 1: + logger.warning("Ambiguous skill name %s; use a category-qualified name", skill_name) + continue + file_path = candidates[0] try: - content = (skills_dir / rel_path).read_text(encoding="utf-8") + content = file_path.read_text(encoding="utf-8") except (OSError, ValueError) as e: logger.warning("Failed to load skill %s: %s", skill_name, e) continue diff --git a/tests/test_skill_dir_extension.py b/tests/test_skill_dir_extension.py new file mode 100644 index 00000000..0c21660c --- /dev/null +++ b/tests/test_skill_dir_extension.py @@ -0,0 +1,120 @@ +from pathlib import Path + +import pytest + +import strix.skills as skills_mod +from strix.skills import ( + get_all_skill_names, + get_available_skills, + load_skills, + register_skill_dir, + registered_skill_dirs, + skill_search_dirs, + validate_requested_skills, +) + + +@pytest.fixture(autouse=True) +def _clear_extra_dirs() -> None: + original = list(skills_mod._EXTRA_SKILL_DIRS) + skills_mod._EXTRA_SKILL_DIRS.clear() + try: + yield + finally: + skills_mod._EXTRA_SKILL_DIRS[:] = original + + +def _write_skill(root: Path, category: str, name: str, body: str) -> None: + category_dir = root / category + category_dir.mkdir(parents=True, exist_ok=True) + (category_dir / f"{name}.md").write_text(body, encoding="utf-8") + + +def _write_root_skill(root: Path, name: str, body: str) -> None: + root.mkdir(parents=True, exist_ok=True) + (root / f"{name}.md").write_text(body, encoding="utf-8") + + +def test_no_registration_leaves_builtin_only() -> None: + assert registered_skill_dirs() == () + builtin = skills_mod.get_strix_resource_path("skills") + assert skill_search_dirs() == (builtin,) + assert {"nmap", "subfinder"}.issubset(get_available_skills()["tooling"]) + + +def test_register_is_idempotent_and_ordered(tmp_path: Path) -> None: + a = tmp_path / "a" + b = tmp_path / "b" + a.mkdir() + b.mkdir() + + register_skill_dir(a) + register_skill_dir(b) + register_skill_dir(a) + + # Most recently registered wins → highest precedence first. + assert registered_skill_dirs() == (b, a) + + +def test_registered_dir_adds_new_skill(tmp_path: Path) -> None: + _write_skill(tmp_path, "extra", "widget", "widget body") + register_skill_dir(tmp_path) + + assert "widget" in get_all_skill_names() + assert get_available_skills()["extra"] == ["widget"] + assert load_skills(["widget"]) == {"widget": "widget body"} + + +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) + + assert "widget" in get_all_skill_names() + assert get_available_skills()["root"] == ["widget"] + assert validate_requested_skills(["widget"]) is None + assert validate_requested_skills(["root/widget"]) is None + assert load_skills(["widget"]) == {"widget": "widget body"} + assert load_skills(["root/widget"]) == {"widget": "widget body"} + + +def test_ambiguous_bare_skill_requires_qualified_name(tmp_path: Path) -> None: + _write_skill(tmp_path, "alpha", "widget", "alpha body") + _write_skill(tmp_path, "beta", "widget", "beta body") + register_skill_dir(tmp_path) + + assert "widget" in get_all_skill_names() + assert get_available_skills()["alpha"] == ["widget"] + assert get_available_skills()["beta"] == ["widget"] + assert validate_requested_skills(["alpha/widget"]) is None + assert validate_requested_skills(["beta/widget"]) is None + + error = validate_requested_skills(["widget"]) + assert error is not None + assert "Ambiguous skill name" in error + assert "alpha/widget" in error + assert "beta/widget" in error + + assert load_skills(["widget"]) == {} + assert load_skills(["alpha/widget"]) == {"widget": "alpha body"} + assert load_skills(["beta/widget"]) == {"widget": "beta body"} + + +def test_registered_dir_overrides_builtin_skill(tmp_path: Path) -> None: + _write_skill(tmp_path, "coordination", "root_agent", "overridden root agent") + register_skill_dir(tmp_path) + + loaded = load_skills(["coordination/root_agent"]) + assert loaded["root_agent"] == "overridden root agent" + + +def test_builtin_skill_still_loads_when_not_overridden(tmp_path: Path) -> None: + _write_skill(tmp_path, "extra", "widget", "widget body") + register_skill_dir(tmp_path) + + # A packaged skill the registered dir does not shadow still resolves. + assert load_skills(["scan_modes/deep"]).get("deep") + + +def test_missing_skill_is_skipped(tmp_path: Path) -> None: + register_skill_dir(tmp_path) + assert load_skills(["does_not_exist"]) == {}