Compare commits

..
Author SHA1 Message Date
bearsyankees d98fb167eb feat(settings): add force_required_tool_choice to LlmSettings
feat(inputs): implement logic for required tool choice based on model

test(inputs): add tests for force_required_tool_choice behavior

test(runner): update tests to include force_required_tool_choice in settings
2026-07-10 18:31:37 -04:00
6 changed files with 13 additions and 418 deletions
+11 -64
View File
@@ -55,7 +55,7 @@ from strix.tools.web_search.tool import web_search
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable, Sequence
from collections.abc import Awaitable, Callable
from agents import RunContextWrapper
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(
*,
name: str = "strix",
@@ -401,37 +359,26 @@ def build_strix_agent(
interactive: bool = False,
chat_completions_tools: bool = False,
system_prompt_context: dict[str, Any] | None = None,
extra_tools: Sequence[Tool] | None = None,
instructions_override: str | None = None,
) -> SandboxAgent[Any]:
"""Build a SandboxAgent for either root or child use.
Args:
chat_completions_tools: Wrap SDK custom tools as function 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 = instructions_override
else:
instructions = render_system_prompt(
skills=skills,
scan_mode=scan_mode,
is_whitebox=is_whitebox,
is_root=is_root,
interactive=interactive,
system_prompt_context=system_prompt_context,
)
instructions = render_system_prompt(
skills=skills,
scan_mode=scan_mode,
is_whitebox=is_whitebox,
is_root=is_root,
interactive=interactive,
system_prompt_context=system_prompt_context,
)
agent_tools = [*_EXTRA_TOOLS, *(extra_tools or [])]
if is_root:
tools: list[Tool] = [*_BASE_TOOLS, *agent_tools, finish_scan]
tools: list[Tool] = [*_BASE_TOOLS, finish_scan]
else:
tools = [*_BASE_TOOLS, *agent_tools, agent_finish]
_ensure_unique_tool_names(tools)
tools = [*_BASE_TOOLS, agent_finish]
logger.info(
"Built %s agent '%s' (skills=%d, tools=%d, scan_mode=%s, whitebox=%s)",
-4
View File
@@ -24,10 +24,6 @@ DEFAULT_MAX_TURNS = 500
def _accepts_required_tool_choice(model_name: str | None) -> bool:
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)
+2 -68
View File
@@ -14,7 +14,6 @@ from agents.sandbox import SandboxRunConfig
from openai import RateLimitError
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.models import (
StrixProvider,
@@ -52,52 +51,6 @@ logger = logging.getLogger(__name__)
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(
*,
scan_config: dict[str, Any],
@@ -111,17 +64,8 @@ async def run_strix_scan(
model: str | None = None,
cleanup_on_exit: bool = True,
event_sink: StreamEventSink | None = None,
root_instructions_override: str | None = None,
extra_system_prompt_context: dict[str, Any] | None = None,
) -> RunResultBase | None:
"""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.
"""
"""Run or resume one Strix scan against a sandbox."""
if scan_id is None:
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)
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(
name="strix",
@@ -244,8 +179,7 @@ async def run_strix_scan(
is_whitebox=is_whitebox,
interactive=interactive,
chat_completions_tools=chat_completions_tools,
system_prompt_context=root_context,
instructions_override=root_instructions,
system_prompt_context=scope_context,
)
if not is_resume:
-88
View File
@@ -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 != ""
-20
View File
@@ -135,23 +135,3 @@ def test_make_model_settings_skips_required_tool_choice_for_non_openai_models()
)
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"
-174
View File
@@ -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"}