mirror of
https://github.com/usestrix/strix.git
synced 2026-08-20 18:38:57 +02:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d98fb167eb |
+11
-64
@@ -55,7 +55,7 @@ from strix.tools.web_search.tool import web_search
|
|||||||
|
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from collections.abc import Awaitable, Callable, Sequence
|
from collections.abc import Awaitable, Callable
|
||||||
|
|
||||||
from agents import RunContextWrapper
|
from agents import RunContextWrapper
|
||||||
from agents.tool import FunctionToolResult
|
from agents.tool import FunctionToolResult
|
||||||
@@ -349,48 +349,6 @@ _BASE_TOOLS: tuple[Tool, ...] = (
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
# Extra tools registered for scan agents. Mirrors
|
|
||||||
# ``strix.runtime.backends.register_backend``: register before the first
|
|
||||||
# ``build_strix_agent`` call and every agent (root + children) gets them.
|
|
||||||
_EXTRA_TOOLS: list[Tool] = []
|
|
||||||
|
|
||||||
|
|
||||||
def _ensure_unique_tool_names(tools: Sequence[Tool]) -> None:
|
|
||||||
seen: set[str] = set()
|
|
||||||
duplicates: set[str] = set()
|
|
||||||
for tool in tools:
|
|
||||||
if tool.name in seen:
|
|
||||||
duplicates.add(tool.name)
|
|
||||||
seen.add(tool.name)
|
|
||||||
if duplicates:
|
|
||||||
msg = f"Agent tools must have unique names: {sorted(duplicates)}"
|
|
||||||
raise ValueError(msg)
|
|
||||||
|
|
||||||
|
|
||||||
def register_agent_tools(*tools: Tool) -> None:
|
|
||||||
"""Register tools for every scan agent built afterwards.
|
|
||||||
|
|
||||||
Tools are added to both root and child agents, after the base set and
|
|
||||||
before the lifecycle tool (``finish_scan`` / ``agent_finish``). Duplicate
|
|
||||||
tool objects are ignored so repeated imports don't double-register.
|
|
||||||
"""
|
|
||||||
new_tools: list[Tool] = []
|
|
||||||
for tool in tools:
|
|
||||||
if tool not in _EXTRA_TOOLS and tool not in new_tools:
|
|
||||||
new_tools.append(tool)
|
|
||||||
|
|
||||||
_ensure_unique_tool_names([*_BASE_TOOLS, *_EXTRA_TOOLS, *new_tools, finish_scan, agent_finish])
|
|
||||||
|
|
||||||
for tool in new_tools:
|
|
||||||
_EXTRA_TOOLS.append(tool)
|
|
||||||
logger.info("Registered extra agent tool: %s", getattr(tool, "name", tool))
|
|
||||||
|
|
||||||
|
|
||||||
def registered_agent_tools() -> tuple[Tool, ...]:
|
|
||||||
"""Return the currently registered scan-agent tools."""
|
|
||||||
return tuple(_EXTRA_TOOLS)
|
|
||||||
|
|
||||||
|
|
||||||
def build_strix_agent(
|
def build_strix_agent(
|
||||||
*,
|
*,
|
||||||
name: str = "strix",
|
name: str = "strix",
|
||||||
@@ -401,37 +359,26 @@ def build_strix_agent(
|
|||||||
interactive: bool = False,
|
interactive: bool = False,
|
||||||
chat_completions_tools: bool = False,
|
chat_completions_tools: bool = False,
|
||||||
system_prompt_context: dict[str, Any] | None = None,
|
system_prompt_context: dict[str, Any] | None = None,
|
||||||
extra_tools: Sequence[Tool] | None = None,
|
|
||||||
instructions_override: str | None = None,
|
|
||||||
) -> SandboxAgent[Any]:
|
) -> SandboxAgent[Any]:
|
||||||
"""Build a SandboxAgent for either root or child use.
|
"""Build a SandboxAgent for either root or child use.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
chat_completions_tools: Wrap SDK custom tools as function tools
|
chat_completions_tools: Wrap SDK custom tools as function tools
|
||||||
when the selected backend cannot accept Responses custom tools.
|
when the selected backend cannot accept Responses custom tools.
|
||||||
extra_tools: Additional tools for this scan agent only, on top of any
|
|
||||||
registered via ``register_agent_tools``.
|
|
||||||
instructions_override: Use this verbatim as the system prompt instead
|
|
||||||
of rendering the built-in scan prompt.
|
|
||||||
"""
|
"""
|
||||||
if instructions_override is not None:
|
instructions = render_system_prompt(
|
||||||
instructions = instructions_override
|
skills=skills,
|
||||||
else:
|
scan_mode=scan_mode,
|
||||||
instructions = render_system_prompt(
|
is_whitebox=is_whitebox,
|
||||||
skills=skills,
|
is_root=is_root,
|
||||||
scan_mode=scan_mode,
|
interactive=interactive,
|
||||||
is_whitebox=is_whitebox,
|
system_prompt_context=system_prompt_context,
|
||||||
is_root=is_root,
|
)
|
||||||
interactive=interactive,
|
|
||||||
system_prompt_context=system_prompt_context,
|
|
||||||
)
|
|
||||||
|
|
||||||
agent_tools = [*_EXTRA_TOOLS, *(extra_tools or [])]
|
|
||||||
if is_root:
|
if is_root:
|
||||||
tools: list[Tool] = [*_BASE_TOOLS, *agent_tools, finish_scan]
|
tools: list[Tool] = [*_BASE_TOOLS, finish_scan]
|
||||||
else:
|
else:
|
||||||
tools = [*_BASE_TOOLS, *agent_tools, agent_finish]
|
tools = [*_BASE_TOOLS, agent_finish]
|
||||||
_ensure_unique_tool_names(tools)
|
|
||||||
|
|
||||||
logger.info(
|
logger.info(
|
||||||
"Built %s agent '%s' (skills=%d, tools=%d, scan_mode=%s, whitebox=%s)",
|
"Built %s agent '%s' (skills=%d, tools=%d, scan_mode=%s, whitebox=%s)",
|
||||||
|
|||||||
@@ -24,10 +24,6 @@ DEFAULT_MAX_TURNS = 500
|
|||||||
|
|
||||||
def _accepts_required_tool_choice(model_name: str | None) -> bool:
|
def _accepts_required_tool_choice(model_name: str | None) -> bool:
|
||||||
name = (model_name or "").strip().lower()
|
name = (model_name or "").strip().lower()
|
||||||
for prefix in ("litellm/", "any-llm/"):
|
|
||||||
if name.startswith(prefix):
|
|
||||||
name = name[len(prefix) :]
|
|
||||||
break
|
|
||||||
return name.startswith("openai/") or is_known_openai_bare_model(name)
|
return name.startswith("openai/") or is_known_openai_bare_model(name)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+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:
|
||||||
|
|||||||
@@ -1,88 +0,0 @@
|
|||||||
"""Tests for scan-agent tool registration in factory."""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
from agents.tool import FunctionTool
|
|
||||||
|
|
||||||
from strix.agents import factory
|
|
||||||
|
|
||||||
|
|
||||||
def _tool(name: str) -> FunctionTool:
|
|
||||||
return FunctionTool(
|
|
||||||
name=name,
|
|
||||||
description="test tool",
|
|
||||||
params_json_schema={"type": "object", "properties": {}, "additionalProperties": False},
|
|
||||||
on_invoke_tool=lambda _ctx, _inp: "ok",
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
|
||||||
def _reset_registry() -> object:
|
|
||||||
saved = list(factory._EXTRA_TOOLS)
|
|
||||||
factory._EXTRA_TOOLS.clear()
|
|
||||||
try:
|
|
||||||
yield
|
|
||||||
finally:
|
|
||||||
factory._EXTRA_TOOLS[:] = saved
|
|
||||||
|
|
||||||
|
|
||||||
def test_register_agent_tools_is_deduped() -> None:
|
|
||||||
tool = _tool("dup")
|
|
||||||
factory.register_agent_tools(tool)
|
|
||||||
factory.register_agent_tools(tool)
|
|
||||||
assert factory.registered_agent_tools() == (tool,)
|
|
||||||
|
|
||||||
|
|
||||||
def test_registered_tools_appear_before_lifecycle_tool() -> None:
|
|
||||||
tool = _tool("extra")
|
|
||||||
factory.register_agent_tools(tool)
|
|
||||||
|
|
||||||
root = factory.build_strix_agent(is_root=True)
|
|
||||||
child = factory.build_strix_agent(is_root=False)
|
|
||||||
|
|
||||||
root_names = [t.name for t in root.tools]
|
|
||||||
child_names = [t.name for t in child.tools]
|
|
||||||
|
|
||||||
assert root_names[-2:] == ["extra", "finish_scan"]
|
|
||||||
assert child_names[-2:] == ["extra", "agent_finish"]
|
|
||||||
|
|
||||||
|
|
||||||
def test_per_call_extra_tools_stack_with_registry() -> None:
|
|
||||||
factory.register_agent_tools(_tool("registered"))
|
|
||||||
|
|
||||||
agent = factory.build_strix_agent(is_root=True, extra_tools=[_tool("per_call")])
|
|
||||||
names = [t.name for t in agent.tools]
|
|
||||||
|
|
||||||
assert "registered" in names
|
|
||||||
assert "per_call" in names
|
|
||||||
assert names[-1] == "finish_scan"
|
|
||||||
|
|
||||||
|
|
||||||
def test_register_agent_tools_rejects_duplicate_names() -> None:
|
|
||||||
factory.register_agent_tools(_tool("same_name"))
|
|
||||||
|
|
||||||
with pytest.raises(ValueError, match="same_name"):
|
|
||||||
factory.register_agent_tools(_tool("same_name"))
|
|
||||||
|
|
||||||
|
|
||||||
def test_per_call_extra_tools_reject_duplicate_registered_names() -> None:
|
|
||||||
factory.register_agent_tools(_tool("same_name"))
|
|
||||||
|
|
||||||
with pytest.raises(ValueError, match="same_name"):
|
|
||||||
factory.build_strix_agent(is_root=True, extra_tools=[_tool("same_name")])
|
|
||||||
|
|
||||||
|
|
||||||
def test_instructions_override_is_used_verbatim() -> None:
|
|
||||||
custom = "You are a scan agent. Follow the provided scope."
|
|
||||||
|
|
||||||
agent = factory.build_strix_agent(is_root=True, instructions_override=custom)
|
|
||||||
|
|
||||||
assert agent.instructions == custom
|
|
||||||
|
|
||||||
|
|
||||||
def test_no_override_renders_builtin_prompt() -> None:
|
|
||||||
agent = factory.build_strix_agent(is_root=True)
|
|
||||||
|
|
||||||
assert isinstance(agent.instructions, str)
|
|
||||||
assert agent.instructions != ""
|
|
||||||
@@ -135,23 +135,3 @@ def test_make_model_settings_skips_required_tool_choice_for_non_openai_models()
|
|||||||
)
|
)
|
||||||
|
|
||||||
assert settings.tool_choice is None
|
assert settings.tool_choice is None
|
||||||
|
|
||||||
|
|
||||||
def test_make_model_settings_forces_required_for_routed_openai_model() -> None:
|
|
||||||
settings = make_model_settings(
|
|
||||||
None,
|
|
||||||
model_name="litellm/openai/gpt-4o",
|
|
||||||
force_required_tool_choice=True,
|
|
||||||
)
|
|
||||||
|
|
||||||
assert settings.tool_choice == "required"
|
|
||||||
|
|
||||||
|
|
||||||
def test_make_model_settings_forces_required_for_anyllm_routed_openai_model() -> None:
|
|
||||||
settings = make_model_settings(
|
|
||||||
None,
|
|
||||||
model_name="any-llm/openai/gpt-4o",
|
|
||||||
force_required_tool_choice=True,
|
|
||||||
)
|
|
||||||
|
|
||||||
assert settings.tool_choice == "required"
|
|
||||||
|
|||||||
@@ -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"}
|
|
||||||
Reference in New Issue
Block a user