mirror of
https://github.com/usestrix/strix.git
synced 2026-08-20 18:38:57 +02:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
94bb6ed59a | ||
|
|
f8a063adc8 |
@@ -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, skill_search_dirs
|
from strix.skills import get_available_skills, load_skills
|
||||||
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)
|
||||||
loader_dirs = [prompt_dir, *skill_search_dirs()]
|
skills_dir = get_strix_resource_path("skills")
|
||||||
env = Environment(
|
env = Environment(
|
||||||
loader=FileSystemLoader(loader_dirs),
|
loader=FileSystemLoader([prompt_dir, skills_dir]),
|
||||||
autoescape=select_autoescape(
|
autoescape=select_autoescape(
|
||||||
enabled_extensions=(),
|
enabled_extensions=(),
|
||||||
default_for_string=False,
|
default_for_string=False,
|
||||||
|
|||||||
+68
-2
@@ -14,6 +14,7 @@ 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,
|
||||||
@@ -51,6 +52,52 @@ 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],
|
||||||
@@ -64,8 +111,17 @@ 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]}"
|
||||||
|
|
||||||
@@ -170,6 +226,15 @@ 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",
|
||||||
@@ -179,7 +244,8 @@ 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=scope_context,
|
system_prompt_context=root_context,
|
||||||
|
instructions_override=root_instructions,
|
||||||
)
|
)
|
||||||
|
|
||||||
if not is_resume:
|
if not is_resume:
|
||||||
|
|||||||
+35
-150
@@ -1,8 +1,6 @@
|
|||||||
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
|
||||||
|
|
||||||
@@ -12,82 +10,20 @@ 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."""
|
||||||
seen: set[tuple[str, str]] = set()
|
skills_dir = get_strix_resource_path("skills")
|
||||||
for skills_dir in skill_search_dirs():
|
if not skills_dir.exists():
|
||||||
for file_path in sorted(skills_dir.glob("*.md")):
|
return
|
||||||
if file_path.name.startswith("__") or file_path.name == "README.md":
|
for category_dir in sorted(skills_dir.iterdir()):
|
||||||
continue
|
if not category_dir.is_dir() or category_dir.name.startswith("__"):
|
||||||
key = (_ROOT_SKILL_CATEGORY, file_path.stem)
|
continue
|
||||||
if key in seen:
|
if category_dir.name in _INTERNAL_SKILL_CATEGORIES:
|
||||||
continue
|
continue
|
||||||
seen.add(key)
|
for file_path in sorted(category_dir.glob("*.md")):
|
||||||
yield key
|
yield category_dir.name, file_path.stem
|
||||||
|
|
||||||
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]:
|
||||||
@@ -95,54 +31,6 @@ 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():
|
||||||
@@ -164,51 +52,48 @@ 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()
|
||||||
available_keys = _get_all_skill_keys()
|
invalid = sorted({s for s in skill_list if s not in available})
|
||||||
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`` (or any
|
Skill files live at ``strix/skills/<category>/<name>.md``. Names
|
||||||
directory added via :func:`register_skill_dir`, searched first).
|
can be ``"name"`` (any category), ``"category/name"``, or a bare
|
||||||
Names can be ``"name"`` (any category), ``"category/name"``, or a
|
file at the skills root. Missing skills are logged and skipped.
|
||||||
bare file at the skills root. Missing skills are logged and skipped.
|
|
||||||
"""
|
"""
|
||||||
search_dirs = skill_search_dirs()
|
skills_dir = get_strix_resource_path("skills")
|
||||||
if not search_dirs:
|
if not skills_dir.exists():
|
||||||
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:
|
||||||
candidates = _candidate_skill_files(skill_name)
|
rel_path: str | None
|
||||||
if not candidates:
|
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():
|
||||||
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 = file_path.read_text(encoding="utf-8")
|
content = (skills_dir / rel_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
|
||||||
|
|||||||
@@ -0,0 +1,174 @@
|
|||||||
|
"""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"}
|
||||||
@@ -1,120 +0,0 @@
|
|||||||
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