mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 17:27:26 +02:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ecbafb2592 |
@@ -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,
|
||||
|
||||
+150
-35
@@ -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
|
||||
(``<root>/<category>/<name>.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/<category>/<name>.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/<category>/<name>.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
|
||||
|
||||
@@ -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"]) == {}
|
||||
Reference in New Issue
Block a user