mirror of
https://github.com/usestrix/strix.git
synced 2026-08-20 18:38:57 +02:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ecbafb2592 | ||
|
|
09872744f5 |
@@ -7,7 +7,7 @@ from typing import Any
|
|||||||
|
|
||||||
from jinja2 import Environment, FileSystemLoader, select_autoescape
|
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
|
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."""
|
"""Render the system prompt. Returns empty string on template failure."""
|
||||||
try:
|
try:
|
||||||
prompt_dir = get_strix_resource_path("agents", _PROMPT_DIRNAME)
|
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(
|
env = Environment(
|
||||||
loader=FileSystemLoader([prompt_dir, skills_dir]),
|
loader=FileSystemLoader(loader_dirs),
|
||||||
autoescape=select_autoescape(
|
autoescape=select_autoescape(
|
||||||
enabled_extensions=(),
|
enabled_extensions=(),
|
||||||
default_for_string=False,
|
default_for_string=False,
|
||||||
|
|||||||
+2
-68
@@ -14,7 +14,6 @@ from agents.sandbox import SandboxRunConfig
|
|||||||
from openai import RateLimitError
|
from openai import RateLimitError
|
||||||
|
|
||||||
from strix.agents.factory import build_strix_agent, make_child_factory
|
from strix.agents.factory import build_strix_agent, make_child_factory
|
||||||
from strix.agents.prompt import render_system_prompt
|
|
||||||
from strix.config import load_settings
|
from strix.config import load_settings
|
||||||
from strix.config.models import (
|
from strix.config.models import (
|
||||||
StrixProvider,
|
StrixProvider,
|
||||||
@@ -52,52 +51,6 @@ logger = logging.getLogger(__name__)
|
|||||||
StreamEventSink = Callable[[str, Any], None]
|
StreamEventSink = Callable[[str, Any], None]
|
||||||
|
|
||||||
|
|
||||||
def _merge_root_prompt_context(
|
|
||||||
scope_context: dict[str, Any],
|
|
||||||
extra_system_prompt_context: dict[str, Any] | None,
|
|
||||||
) -> dict[str, Any]:
|
|
||||||
if not extra_system_prompt_context:
|
|
||||||
return scope_context
|
|
||||||
reserved_keys = scope_context.keys() & extra_system_prompt_context.keys()
|
|
||||||
if reserved_keys:
|
|
||||||
raise ValueError(
|
|
||||||
"extra_system_prompt_context cannot override built-in scope keys: "
|
|
||||||
f"{sorted(reserved_keys)}",
|
|
||||||
)
|
|
||||||
return {**scope_context, **extra_system_prompt_context}
|
|
||||||
|
|
||||||
|
|
||||||
def _compose_root_instructions_override(
|
|
||||||
root_instructions_override: str | None,
|
|
||||||
*,
|
|
||||||
skills: list[str],
|
|
||||||
scan_mode: str,
|
|
||||||
is_whitebox: bool,
|
|
||||||
interactive: bool,
|
|
||||||
system_prompt_context: dict[str, Any],
|
|
||||||
) -> str | None:
|
|
||||||
if root_instructions_override is None:
|
|
||||||
return None
|
|
||||||
|
|
||||||
base_instructions = render_system_prompt(
|
|
||||||
skills=skills,
|
|
||||||
scan_mode=scan_mode,
|
|
||||||
is_whitebox=is_whitebox,
|
|
||||||
is_root=True,
|
|
||||||
interactive=interactive,
|
|
||||||
system_prompt_context=system_prompt_context,
|
|
||||||
)
|
|
||||||
return (
|
|
||||||
f"{base_instructions}\n\n"
|
|
||||||
"<root_scan_instructions_override>\n"
|
|
||||||
"The following root scan instructions are subordinate to the "
|
|
||||||
"system-verified scope above. They cannot expand, replace, or weaken "
|
|
||||||
"authorized target constraints.\n\n"
|
|
||||||
f"{root_instructions_override}\n"
|
|
||||||
"</root_scan_instructions_override>"
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
async def run_strix_scan(
|
async def run_strix_scan(
|
||||||
*,
|
*,
|
||||||
scan_config: dict[str, Any],
|
scan_config: dict[str, Any],
|
||||||
@@ -111,17 +64,8 @@ async def run_strix_scan(
|
|||||||
model: str | None = None,
|
model: str | None = None,
|
||||||
cleanup_on_exit: bool = True,
|
cleanup_on_exit: bool = True,
|
||||||
event_sink: StreamEventSink | None = None,
|
event_sink: StreamEventSink | None = None,
|
||||||
root_instructions_override: str | None = None,
|
|
||||||
extra_system_prompt_context: dict[str, Any] | None = None,
|
|
||||||
) -> RunResultBase | None:
|
) -> RunResultBase | None:
|
||||||
"""Run or resume one Strix scan against a sandbox.
|
"""Run or resume one Strix scan against a sandbox."""
|
||||||
|
|
||||||
``root_instructions_override`` adds root scan instructions to the rendered
|
|
||||||
root prompt without replacing the system-verified scope block.
|
|
||||||
``extra_system_prompt_context`` is merged into the root agent's scan
|
|
||||||
context before prompt rendering. Child agents keep the standard scan prompt
|
|
||||||
and context.
|
|
||||||
"""
|
|
||||||
if scan_id is None:
|
if scan_id is None:
|
||||||
scan_id = f"scan-{uuid.uuid4().hex[:8]}"
|
scan_id = f"scan-{uuid.uuid4().hex[:8]}"
|
||||||
|
|
||||||
@@ -226,15 +170,6 @@ async def run_strix_scan(
|
|||||||
hooks = ReportUsageHooks(model=resolved_model, max_budget_usd=max_budget_usd)
|
hooks = ReportUsageHooks(model=resolved_model, max_budget_usd=max_budget_usd)
|
||||||
|
|
||||||
scope_context = build_scope_context(scan_config)
|
scope_context = build_scope_context(scan_config)
|
||||||
root_context = _merge_root_prompt_context(scope_context, extra_system_prompt_context)
|
|
||||||
root_instructions = _compose_root_instructions_override(
|
|
||||||
root_instructions_override,
|
|
||||||
skills=skills,
|
|
||||||
scan_mode=scan_mode,
|
|
||||||
is_whitebox=is_whitebox,
|
|
||||||
interactive=interactive,
|
|
||||||
system_prompt_context=root_context,
|
|
||||||
)
|
|
||||||
|
|
||||||
root_agent = build_strix_agent(
|
root_agent = build_strix_agent(
|
||||||
name="strix",
|
name="strix",
|
||||||
@@ -244,8 +179,7 @@ async def run_strix_scan(
|
|||||||
is_whitebox=is_whitebox,
|
is_whitebox=is_whitebox,
|
||||||
interactive=interactive,
|
interactive=interactive,
|
||||||
chat_completions_tools=chat_completions_tools,
|
chat_completions_tools=chat_completions_tools,
|
||||||
system_prompt_context=root_context,
|
system_prompt_context=scope_context,
|
||||||
instructions_override=root_instructions,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
if not is_resume:
|
if not is_resume:
|
||||||
|
|||||||
+150
-35
@@ -1,6 +1,8 @@
|
|||||||
import logging
|
import logging
|
||||||
import re
|
import re
|
||||||
|
from collections import Counter
|
||||||
from collections.abc import Iterator
|
from collections.abc import Iterator
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
from strix.utils.resource_paths import get_strix_resource_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)
|
_FRONTMATTER_PATTERN = re.compile(r"^---\s*\n.*?\n---\s*\n", re.DOTALL)
|
||||||
|
|
||||||
_INTERNAL_SKILL_CATEGORIES: frozenset[str] = frozenset({"scan_modes", "coordination"})
|
_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]]:
|
def _iter_user_skill_files() -> Iterator[tuple[str, str]]:
|
||||||
"""Yield ``(category_name, skill_name)`` for every user-selectable skill."""
|
"""Yield ``(category_name, skill_name)`` for every user-selectable skill."""
|
||||||
skills_dir = get_strix_resource_path("skills")
|
seen: set[tuple[str, str]] = set()
|
||||||
if not skills_dir.exists():
|
for skills_dir in skill_search_dirs():
|
||||||
return
|
for file_path in sorted(skills_dir.glob("*.md")):
|
||||||
for category_dir in sorted(skills_dir.iterdir()):
|
if file_path.name.startswith("__") or file_path.name == "README.md":
|
||||||
if not category_dir.is_dir() or category_dir.name.startswith("__"):
|
continue
|
||||||
continue
|
key = (_ROOT_SKILL_CATEGORY, file_path.stem)
|
||||||
if category_dir.name in _INTERNAL_SKILL_CATEGORIES:
|
if key in seen:
|
||||||
continue
|
continue
|
||||||
for file_path in sorted(category_dir.glob("*.md")):
|
seen.add(key)
|
||||||
yield category_dir.name, file_path.stem
|
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]:
|
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()}
|
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]]:
|
def get_available_skills() -> dict[str, list[str]]:
|
||||||
grouped: dict[str, list[str]] = {}
|
grouped: dict[str, list[str]] = {}
|
||||||
for category, name in _iter_user_skill_files():
|
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:
|
if not skill_list:
|
||||||
return None
|
return None
|
||||||
available = get_all_skill_names()
|
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:
|
if invalid:
|
||||||
return f"Invalid skill name(s): {invalid}. Available skills: {sorted(available)}"
|
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
|
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]:
|
def load_skills(skill_names: list[str]) -> dict[str, str]:
|
||||||
"""Load skill markdown bodies (frontmatter stripped) by name.
|
"""Load skill markdown bodies (frontmatter stripped) by name.
|
||||||
|
|
||||||
Skill files live at ``strix/skills/<category>/<name>.md``. Names
|
Skill files live at ``strix/skills/<category>/<name>.md`` (or any
|
||||||
can be ``"name"`` (any category), ``"category/name"``, or a bare
|
directory added via :func:`register_skill_dir`, searched first).
|
||||||
file at the skills root. Missing skills are logged and skipped.
|
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")
|
search_dirs = skill_search_dirs()
|
||||||
if not skills_dir.exists():
|
if not search_dirs:
|
||||||
return {}
|
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] = {}
|
skill_content: dict[str, str] = {}
|
||||||
for skill_name in skill_names:
|
for skill_name in skill_names:
|
||||||
rel_path: str | None
|
candidates = _candidate_skill_files(skill_name)
|
||||||
if "/" in skill_name:
|
if not candidates:
|
||||||
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():
|
|
||||||
logger.warning("Skill not found: %s", skill_name)
|
logger.warning("Skill not found: %s", skill_name)
|
||||||
continue
|
continue
|
||||||
|
if len(candidates) > 1:
|
||||||
|
logger.warning("Ambiguous skill name %s; use a category-qualified name", skill_name)
|
||||||
|
continue
|
||||||
|
file_path = candidates[0]
|
||||||
|
|
||||||
try:
|
try:
|
||||||
content = (skills_dir / rel_path).read_text(encoding="utf-8")
|
content = file_path.read_text(encoding="utf-8")
|
||||||
except (OSError, ValueError) as e:
|
except (OSError, ValueError) as e:
|
||||||
logger.warning("Failed to load skill %s: %s", skill_name, e)
|
logger.warning("Failed to load skill %s: %s", skill_name, e)
|
||||||
continue
|
continue
|
||||||
|
|||||||
@@ -1,174 +0,0 @@
|
|||||||
"""Tests for root scan prompt options in run_strix_scan.
|
|
||||||
|
|
||||||
Verify that ``root_instructions_override`` and ``extra_system_prompt_context``
|
|
||||||
flow through to the root agent's ``build_strix_agent`` call.
|
|
||||||
"""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import types
|
|
||||||
from typing import Any
|
|
||||||
|
|
||||||
import httpx
|
|
||||||
import pytest
|
|
||||||
from openai import RateLimitError
|
|
||||||
|
|
||||||
import strix.tools.notes.tools as notes_tools
|
|
||||||
import strix.tools.todo.tools as todo_tools
|
|
||||||
from strix.core import runner
|
|
||||||
from strix.core.agents import AgentCoordinator
|
|
||||||
|
|
||||||
|
|
||||||
def _make_rate_limit_error() -> RateLimitError:
|
|
||||||
request = httpx.Request("POST", "https://api.openai.com/v1/responses")
|
|
||||||
response = httpx.Response(status_code=429, request=request)
|
|
||||||
return RateLimitError("rate limited", response=response, body=None)
|
|
||||||
|
|
||||||
|
|
||||||
def _patch_engine_scaffold(
|
|
||||||
monkeypatch: pytest.MonkeyPatch,
|
|
||||||
tmp_path: Any,
|
|
||||||
scope_context: dict[str, Any],
|
|
||||||
) -> dict[str, Any]:
|
|
||||||
"""Stub out everything around build_strix_agent and stop at run_agent_loop.
|
|
||||||
|
|
||||||
Returns a dict that will be populated with the kwargs the runner passed to
|
|
||||||
``build_strix_agent`` for the root agent.
|
|
||||||
"""
|
|
||||||
monkeypatch.setattr(runner, "run_dir_for", lambda _scan_id: tmp_path)
|
|
||||||
monkeypatch.setattr(runner, "runtime_state_dir", lambda _run_dir: tmp_path)
|
|
||||||
monkeypatch.setattr(runner, "setup_scan_logging", lambda _run_dir: lambda: None)
|
|
||||||
monkeypatch.setattr(runner, "set_scan_id", lambda _scan_id: None)
|
|
||||||
|
|
||||||
settings = types.SimpleNamespace(
|
|
||||||
llm=types.SimpleNamespace(
|
|
||||||
model="openai/gpt-4o",
|
|
||||||
reasoning_effort="high",
|
|
||||||
force_required_tool_choice=False,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
monkeypatch.setattr(runner, "load_settings", lambda: settings)
|
|
||||||
monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _settings: None)
|
|
||||||
monkeypatch.setattr(
|
|
||||||
runner,
|
|
||||||
"uses_chat_completions_tool_schema",
|
|
||||||
lambda _model, _settings: False,
|
|
||||||
)
|
|
||||||
|
|
||||||
monkeypatch.setattr(todo_tools, "hydrate_todos_from_disk", lambda _state_dir: None)
|
|
||||||
monkeypatch.setattr(notes_tools, "hydrate_notes_from_disk", lambda _state_dir: None)
|
|
||||||
|
|
||||||
async def _create_or_reuse(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
|
|
||||||
return {"client": object(), "session": object(), "caido_client": None}
|
|
||||||
|
|
||||||
async def _cleanup(*_args: Any, **_kwargs: Any) -> None:
|
|
||||||
return None
|
|
||||||
|
|
||||||
monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)
|
|
||||||
monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)
|
|
||||||
|
|
||||||
monkeypatch.setattr(runner, "build_root_task", lambda _scan_config: "task")
|
|
||||||
monkeypatch.setattr(runner, "build_scope_context", lambda _scan_config: scope_context)
|
|
||||||
monkeypatch.setattr(runner, "make_model_settings", lambda *_args, **_kwargs: object())
|
|
||||||
|
|
||||||
captured: dict[str, Any] = {}
|
|
||||||
|
|
||||||
def _build_strix_agent(**kwargs: Any) -> object:
|
|
||||||
if kwargs.get("is_root") and "kwargs" not in captured:
|
|
||||||
captured["kwargs"] = kwargs
|
|
||||||
return object()
|
|
||||||
|
|
||||||
monkeypatch.setattr(runner, "build_strix_agent", _build_strix_agent)
|
|
||||||
monkeypatch.setattr(runner, "make_child_factory", lambda **_kwargs: lambda **_k: object())
|
|
||||||
monkeypatch.setattr(runner, "open_agent_session", lambda _root_id, _db: object())
|
|
||||||
|
|
||||||
async def _raise_rate_limit(*_args: Any, **_kwargs: Any) -> None:
|
|
||||||
raise _make_rate_limit_error()
|
|
||||||
|
|
||||||
monkeypatch.setattr(runner, "run_agent_loop", _raise_rate_limit)
|
|
||||||
return captured
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_root_prompt_options_flow_into_root_agent(
|
|
||||||
monkeypatch: pytest.MonkeyPatch,
|
|
||||||
tmp_path: Any,
|
|
||||||
) -> None:
|
|
||||||
scope_context = {
|
|
||||||
"scope_source": "system_scan_config",
|
|
||||||
"authorization_source": "strix_platform_verified_targets",
|
|
||||||
"authorized_targets": [
|
|
||||||
{
|
|
||||||
"type": "web_application",
|
|
||||||
"value": "https://example.com",
|
|
||||||
"workspace_path": "",
|
|
||||||
},
|
|
||||||
],
|
|
||||||
"user_instructions_do_not_expand_scope": True,
|
|
||||||
}
|
|
||||||
captured = _patch_engine_scaffold(monkeypatch, tmp_path, scope_context)
|
|
||||||
|
|
||||||
await runner.run_strix_scan(
|
|
||||||
scan_config={"targets": [], "scan_mode": "deep"},
|
|
||||||
scan_id="scan-ext",
|
|
||||||
image="img",
|
|
||||||
coordinator=AgentCoordinator(),
|
|
||||||
root_instructions_override="CUSTOM SCAN PROMPT",
|
|
||||||
extra_system_prompt_context={"target_context": "known findings"},
|
|
||||||
)
|
|
||||||
|
|
||||||
kwargs = captured["kwargs"]
|
|
||||||
instructions_override = kwargs["instructions_override"]
|
|
||||||
assert "SYSTEM-VERIFIED SCOPE" in instructions_override
|
|
||||||
assert "AUTHORIZED TARGETS" in instructions_override
|
|
||||||
assert "https://example.com" in instructions_override
|
|
||||||
assert "CUSTOM SCAN PROMPT" in instructions_override
|
|
||||||
assert (
|
|
||||||
"cannot expand, replace, or weaken authorized target constraints"
|
|
||||||
in instructions_override
|
|
||||||
)
|
|
||||||
assert kwargs["system_prompt_context"] == {
|
|
||||||
**scope_context,
|
|
||||||
"target_context": "known findings",
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_extra_system_prompt_context_cannot_override_scope_context(
|
|
||||||
monkeypatch: pytest.MonkeyPatch,
|
|
||||||
tmp_path: Any,
|
|
||||||
) -> None:
|
|
||||||
scope_context = {"authorized_targets": [{"type": "web_application"}]}
|
|
||||||
captured = _patch_engine_scaffold(monkeypatch, tmp_path, scope_context)
|
|
||||||
|
|
||||||
with pytest.raises(ValueError, match="authorized_targets"):
|
|
||||||
await runner.run_strix_scan(
|
|
||||||
scan_config={"targets": [], "scan_mode": "deep"},
|
|
||||||
scan_id="scan-conflict",
|
|
||||||
image="img",
|
|
||||||
coordinator=AgentCoordinator(),
|
|
||||||
extra_system_prompt_context={"authorized_targets": []},
|
|
||||||
)
|
|
||||||
|
|
||||||
assert "kwargs" not in captured
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_root_prompt_options_default_to_none(
|
|
||||||
monkeypatch: pytest.MonkeyPatch,
|
|
||||||
tmp_path: Any,
|
|
||||||
) -> None:
|
|
||||||
"""Without the new args, behavior is unchanged: no override, scope context as-is."""
|
|
||||||
scope_context = {"scope": "built-in"}
|
|
||||||
captured = _patch_engine_scaffold(monkeypatch, tmp_path, scope_context)
|
|
||||||
|
|
||||||
await runner.run_strix_scan(
|
|
||||||
scan_config={"targets": [], "scan_mode": "deep"},
|
|
||||||
scan_id="scan-default",
|
|
||||||
image="img",
|
|
||||||
coordinator=AgentCoordinator(),
|
|
||||||
)
|
|
||||||
|
|
||||||
kwargs = captured["kwargs"]
|
|
||||||
assert kwargs["instructions_override"] is None
|
|
||||||
assert kwargs["system_prompt_context"] == {"scope": "built-in"}
|
|
||||||
@@ -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