mortal/strix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(migration): phase 5 — root agent factory + entry point

Browse Source 
Three new modules that wire Phases 0-4 into a runnable Strix scan:

- strix/agents/sdk_prompt.py: standalone Jinja-based system prompt
  renderer. Reuses the existing strix/agents/StrixAgent/system_prompt.
  jinja template (508 lines, the actual production prompt) so behavior
  parity with the legacy LLM._load_system_prompt is byte-identical.
  Skill resolution mirrors LLM._get_skills_to_load (caller skills →
  scan_modes/<mode> → whitebox pair, deduped). Fail-soft: template
  errors return empty string and log; agent construction must never
  blow up on prompt load.

- strix/agents/sdk_factory.py: build_strix_agent(name, skills, is_root)
  assembles an agents.Agent. Root carries finish_scan and stops there;
  child carries agent_finish and stops there (C4). Caido tools come
  from CaidoCapability automatically — we don't include them in
  _BASE_TOOLS to avoid double-registration when the SDK runtime merges
  capability tools. model=None so RunConfig drives the model alias
  through MultiProvider rather than the SDK default. make_child_factory
  returns a closure over scan-level config (scan_mode, is_whitebox,
  interactive, scope context) for ctx.context['agent_factory'] — the
  Phase 3 create_agent tool calls it with (name, skills) per child.

- strix/sdk_entry.py: run_strix_scan() — the top-level coroutine.
  Builds the bus, brings up (or reuses) a sandbox session via
  session_manager, builds the root Agent and the child factory, builds
  the per-agent context dict, registers the root in the bus, builds
  the RunConfig, calls Runner.run, and cleans up the session in a
  finally. Cancels descendants before re-raising any exception (C9).
  cleanup_on_exit toggle preserves the cached session for resume
  scenarios. _build_root_task and _build_scope_context preserve the
  legacy StrixAgent.execute_scan task formatting + scope context shape
  so the prompt template sees identical inputs.

Tests: 21 new tests (10 for factory + prompt, 11 for entry point).
Factory: root vs child tool list parity, finish_scan/agent_finish
placement, tool_use_behavior dict shape, Caido absence (capability-
provided), make_child_factory closure semantics. Entry point (all
mocked, no real Docker/LLM): wiring shape verification — context dict
carries every field downstream consumers read, session manager called
with correct scan_id, cleanup runs even on Runner.run failure,
cleanup skipped when disabled, scan_id auto-generation, scan-level
config (scan_mode, is_whitebox) flows into the factory. Task and scope
builders verified against the same shape as legacy.

Per-file ruff ignores added: TC002 on sdk_factory (Tool used at
runtime in _BASE_TOOLS tuple), TC003 + PLR0912 on sdk_entry (Path
runtime-imported; _build_root_task's per-target-type branches are
intentional and well-bounded).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
0xallam
2026-04-25 00:58:32 -07:00
parent 1d86e4506a
commit f0e254c1fd
6 changed files with 1162 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/agents/test_sdk_factory.py
+201
View File
@@ -0,0 +1,201 @@
"""Phase 5 tests for the SDK agent factory + prompt renderer.
These two modules are the keystone wiring between Phases 2-4 and an
actual ``Runner.run`` invocation. The tests verify:
- The prompt renderer reuses the existing Jinja template (parity with
legacy LLM._load_system_prompt) and degrades gracefully when the
template isn't available.
- ``build_strix_agent(is_root=True)`` carries ``finish_scan`` and
stops on it; child agents carry ``agent_finish`` and stop on it.
- ``make_child_factory`` snapshots scan-level config into a closure
so each spawned child inherits the right scan_mode / is_whitebox /
prompt context without create_agent having to re-derive it.
"""
from __future__ import annotations
from typing import Any
from unittest.mock import patch
from agents import Agent
from agents.tool import FunctionTool
from strix.agents.sdk_factory import build_strix_agent, make_child_factory
from strix.agents.sdk_prompt import _resolve_skills, render_system_prompt
# --- prompt renderer ----------------------------------------------------
def test_resolve_skills_deduplicates_and_orders() -> None:
out = _resolve_skills(
requested=["recon", "xss", "recon"],
scan_mode="deep",
is_whitebox=False,
)
assert out == ["recon", "xss", "scan_modes/deep"]
def test_resolve_skills_adds_whitebox_pair() -> None:
out = _resolve_skills(requested=None, scan_mode="fast", is_whitebox=True)
# The whitebox pair sits at the tail; scan_modes goes in the middle
# because callers can append more skills after it via the requested arg.
assert out == [
"scan_modes/fast",
"coordination/source_aware_whitebox",
"custom/source_aware_sast",
]
def test_render_system_prompt_returns_string() -> None:
"""Smoke: the StrixAgent template is on disk and renders to non-empty."""
out = render_system_prompt(skills=[], scan_mode="deep")
assert isinstance(out, str)
# The first line of the template starts with 'You are Strix'.
assert out.startswith("You are Strix")
def test_render_system_prompt_swallows_template_errors() -> None:
"""If the template path can't be resolved, return an empty string
(not raise) — agent construction must never blow up on prompt load."""
with patch(
"strix.agents.sdk_prompt.get_strix_resource_path",
side_effect=RuntimeError("missing"),
):
out = render_system_prompt(skills=[])
assert out == ""
# --- factory: shape + tools --------------------------------------------
def test_root_agent_carries_finish_scan_and_stops_there() -> None:
agent = build_strix_agent(name="strix", is_root=True)
assert isinstance(agent, Agent)
tool_names = {t.name for t in agent.tools if isinstance(t, FunctionTool)}
assert "finish_scan" in tool_names
assert "agent_finish" not in tool_names
behavior = agent.tool_use_behavior
# StopAtTools is a TypedDict at runtime → behavior is a dict.
assert isinstance(behavior, dict)
assert behavior["stop_at_tool_names"] == ["finish_scan"]
def test_child_agent_carries_agent_finish_and_stops_there() -> None:
agent = build_strix_agent(name="recon-bot", is_root=False)
tool_names = {t.name for t in agent.tools if isinstance(t, FunctionTool)}
assert "agent_finish" in tool_names
assert "finish_scan" not in tool_names
behavior = agent.tool_use_behavior
assert isinstance(behavior, dict)
assert behavior["stop_at_tool_names"] == ["agent_finish"]
def test_root_and_child_share_base_tool_set() -> None:
"""The base tool set (think/todo/notes/file_edit/web_search/etc) is
identical between root and child — only the terminator differs."""
root = build_strix_agent(is_root=True)
child = build_strix_agent(is_root=False)
root_names = {t.name for t in root.tools if isinstance(t, FunctionTool)}
child_names = {t.name for t in child.tools if isinstance(t, FunctionTool)}
# Drop the terminators and compare.
assert root_names - {"finish_scan"} == child_names - {"agent_finish"}
def test_agent_includes_graph_and_sandbox_tools() -> None:
"""The graph + sandbox tool families are required for parity with
legacy. Spot-check the ones most likely to be forgotten in a refactor."""
agent = build_strix_agent(is_root=True)
names = {t.name for t in agent.tools if isinstance(t, FunctionTool)}
expected = {
"think",
"create_todo",
"create_note",
"web_search",
"str_replace_editor",
"create_vulnerability_report",
"load_skill",
"browser_action",
"terminal_execute",
"python_action",
"view_agent_graph",
"agent_status",
"send_message_to_agent",
"wait_for_message",
"create_agent",
}
missing = expected - names
assert not missing, f"missing tools: {missing}"
def test_agent_does_not_include_caido_tools() -> None:
"""Caido tools come from CaidoCapability.tools(); the agent doesn't
declare them directly to avoid double-registration when the SDK
runtime merges capability tools."""
agent = build_strix_agent(is_root=True)
names = {t.name for t in agent.tools if isinstance(t, FunctionTool)}
caido = {
"list_requests",
"view_request",
"send_request",
"repeat_request",
"scope_rules",
"list_sitemap",
"view_sitemap_entry",
}
overlap = names & caido
assert overlap == set(), f"unexpected Caido tools in agent.tools: {overlap}"
def test_agent_uses_run_config_model() -> None:
"""``model=None`` so the RunConfig drives the model alias through
MultiProvider rather than an SDK default like gpt-4.1."""
agent = build_strix_agent(is_root=True)
assert agent.model is None
def test_agent_instructions_contain_rendered_prompt() -> None:
"""The factory must wire the rendered prompt into ``instructions``."""
agent = build_strix_agent(is_root=True, scan_mode="deep")
assert isinstance(agent.instructions, str)
assert agent.instructions.startswith("You are Strix")
# --- child factory ------------------------------------------------------
def test_make_child_factory_returns_callable_that_builds_child() -> None:
factory = make_child_factory(scan_mode="deep", is_whitebox=False)
assert callable(factory)
child = factory(name="sub-1", skills=["recon"])
assert isinstance(child, Agent)
assert child.name == "sub-1"
behavior = child.tool_use_behavior
assert isinstance(behavior, dict)
assert behavior["stop_at_tool_names"] == ["agent_finish"]
def test_make_child_factory_passes_scan_level_config() -> None:
"""Verify scan_mode + is_whitebox flow into the rendered prompt
via the closure rather than the create_agent call site."""
captured: dict[str, Any] = {}
def fake_render(**kwargs: Any) -> str:
captured.update(kwargs)
return "stub-prompt"
factory = make_child_factory(
scan_mode="fast",
is_whitebox=True,
interactive=True,
system_prompt_context={"scope_source": "test"},
)
with patch("strix.agents.sdk_factory.render_system_prompt", side_effect=fake_render):
factory(name="child", skills=["xss"])
assert captured["scan_mode"] == "fast"
assert captured["is_whitebox"] is True
assert captured["interactive"] is True
assert captured["system_prompt_context"] == {"scope_source": "test"}
assert captured["skills"] == ["xss"]
tests/test_sdk_entry.py
+319
View File
@@ -0,0 +1,319 @@
"""Phase 5 tests for the top-level SDK scan entry point.
We never spin up a real Docker container or hit a real LLM here. The
tests patch ``session_manager.create_or_reuse``, ``Runner.run``, and
the agent factory so we can verify the wiring shape:
- The bus is registered with a root agent before Runner.run.
- The context dict carries every field downstream code (tools, hooks,
filter) reads.
- The session manager's bundle flows through to the context (host
ports, bearer, sandbox session/client).
- ``cleanup_on_exit=True`` always cleans up, even when Runner.run
raises.
- ``cleanup_on_exit=False`` preserves the cached session.
- Cancellation propagates: if Runner.run raises, descendants are
cancelled before re-raising.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from strix.orchestration.bus import AgentMessageBus
from strix.sdk_entry import _build_root_task, _build_scope_context, run_strix_scan
# --- helpers ------------------------------------------------------------
def _bundle_for_test() -> dict[str, Any]:
return {
"client": MagicMock(name="docker_client"),
"session": MagicMock(name="sandbox_session"),
"capability": MagicMock(),
"tool_server_host_port": 12001,
"caido_host_port": 12002,
"bearer": "test-bearer-token-1234567890",
}
def _scan_config(**overrides: Any) -> dict[str, Any]:
base: dict[str, Any] = {
"targets": [
{
"type": "web_application",
"details": {"target_url": "https://example.com"},
},
],
"user_instructions": "find xss",
"scan_mode": "deep",
"is_whitebox": False,
}
base.update(overrides)
return base
# --- task / scope builders ---------------------------------------------
def test_build_root_task_groups_targets_and_appends_instructions() -> None:
config = _scan_config(
targets=[
{
"type": "repository",
"details": {
"target_repo": "https://github.com/x/y",
"cloned_repo_path": "/tmp/y",
"workspace_subdir": "y",
},
},
{
"type": "ip_address",
"details": {"target_ip": "10.0.0.1"},
},
],
user_instructions="report only critical issues",
)
task = _build_root_task(config)
assert "Repositories:" in task
assert "https://github.com/x/y (available at: /workspace/y)" in task
assert "IP Addresses:" in task
assert "10.0.0.1" in task
assert "Special instructions: report only critical issues" in task
def test_build_root_task_renders_diff_scope_block() -> None:
config = _scan_config(
diff_scope={
"active": True,
"repos": [
{
"workspace_subdir": "service-x",
"analyzable_files_count": 7,
"deleted_files_count": 2,
},
],
},
)
task = _build_root_task(config)
assert "Scope Constraints:" in task
assert "service-x: 7 changed file(s)" in task
assert "service-x: 2 deleted file(s)" in task
def test_build_scope_context_marks_authorization_source() -> None:
config = _scan_config(
targets=[
{
"type": "web_application",
"details": {"target_url": "https://target.test"},
},
],
)
ctx = _build_scope_context(config)
assert ctx["scope_source"] == "system_scan_config"
assert ctx["authorization_source"] == "strix_platform_verified_targets"
assert ctx["user_instructions_do_not_expand_scope"] is True
assert ctx["authorized_targets"] == [
{"type": "web_application", "value": "https://target.test", "workspace_path": ""},
]
# --- run_strix_scan wiring ---------------------------------------------
@pytest.mark.asyncio
async def test_run_strix_scan_wires_context_and_calls_runner(tmp_path: Path) -> None:
"""End-to-end (mocked) — assert every downstream consumer of context
sees the bundle's bearer + host ports."""
bundle = _bundle_for_test()
captured_context: dict[str, Any] = {}
async def fake_runner_run(*args: Any, **kwargs: Any) -> Any:
captured_context.update(kwargs.get("context", {}))
return MagicMock(name="run_result")
with (
patch(
"strix.sdk_entry.session_manager.create_or_reuse",
new=AsyncMock(return_value=bundle),
) as create_mock,
patch(
"strix.sdk_entry.session_manager.cleanup",
new=AsyncMock(),
) as cleanup_mock,
patch("strix.sdk_entry.Runner.run", side_effect=fake_runner_run) as runner_mock,
# Stub the factory to avoid rendering the 158k-char prompt for
# every test (it's covered by sdk_prompt tests).
patch(
"strix.sdk_entry.build_strix_agent",
return_value=MagicMock(name="root_agent"),
) as factory_mock,
):
await run_strix_scan(
scan_config=_scan_config(),
scan_id="scan-test",
image="strix-sandbox:test",
sources_path=tmp_path,
)
# Session manager calls.
create_mock.assert_awaited_once()
create_args = create_mock.await_args
assert create_args is not None
assert create_args.args == ("scan-test",)
assert create_args.kwargs["image"] == "strix-sandbox:test"
assert create_args.kwargs["sources_path"] == tmp_path
cleanup_mock.assert_awaited_once_with("scan-test")
# Factory called with is_root=True.
factory_mock.assert_called_once()
assert factory_mock.call_args.kwargs["is_root"] is True
# Runner.run called once with the root agent.
assert runner_mock.call_count == 1
# Context shape passed into Runner.run.
assert captured_context["sandbox_session"] is bundle["session"]
assert captured_context["sandbox_client"] is bundle["client"]
assert captured_context["sandbox_token"] == bundle["bearer"]
assert captured_context["tool_server_host_port"] == bundle["tool_server_host_port"]
assert captured_context["caido_host_port"] == bundle["caido_host_port"]
# Bus is registered and root agent_id is populated.
bus = captured_context["bus"]
assert isinstance(bus, AgentMessageBus)
assert captured_context["agent_id"] in bus.statuses
assert bus.parent_of[captured_context["agent_id"]] is None
# Child factory wired through so create_agent works.
assert callable(captured_context["agent_factory"])
@pytest.mark.asyncio
async def test_run_strix_scan_cleans_up_on_runner_failure(tmp_path: Path) -> None:
"""If Runner.run raises, cleanup must still fire (the finally branch)."""
bundle = _bundle_for_test()
with (
patch(
"strix.sdk_entry.session_manager.create_or_reuse",
new=AsyncMock(return_value=bundle),
),
patch(
"strix.sdk_entry.session_manager.cleanup",
new=AsyncMock(),
) as cleanup_mock,
patch(
"strix.sdk_entry.Runner.run",
side_effect=RuntimeError("simulated LLM blow-up"),
),
patch("strix.sdk_entry.build_strix_agent", return_value=MagicMock()),
pytest.raises(RuntimeError, match="simulated LLM"),
):
await run_strix_scan(
scan_config=_scan_config(),
scan_id="scan-fail",
image="i",
sources_path=tmp_path,
)
cleanup_mock.assert_awaited_once_with("scan-fail")
@pytest.mark.asyncio
async def test_run_strix_scan_skips_cleanup_when_disabled(tmp_path: Path) -> None:
bundle = _bundle_for_test()
async def fake_runner_run(*args: Any, **kwargs: Any) -> Any:
return MagicMock()
with (
patch(
"strix.sdk_entry.session_manager.create_or_reuse",
new=AsyncMock(return_value=bundle),
),
patch(
"strix.sdk_entry.session_manager.cleanup",
new=AsyncMock(),
) as cleanup_mock,
patch("strix.sdk_entry.Runner.run", side_effect=fake_runner_run),
patch("strix.sdk_entry.build_strix_agent", return_value=MagicMock()),
):
await run_strix_scan(
scan_config=_scan_config(),
scan_id="scan-keep",
image="i",
sources_path=tmp_path,
cleanup_on_exit=False,
)
cleanup_mock.assert_not_called()
@pytest.mark.asyncio
async def test_run_strix_scan_auto_generates_scan_id(tmp_path: Path) -> None:
"""A caller without a stable id should still get a valid scan_id
flowing into create_or_reuse."""
bundle = _bundle_for_test()
captured_scan_id: list[str] = []
async def fake_create(scan_id: str, **_kwargs: Any) -> Any:
captured_scan_id.append(scan_id)
return bundle
with (
patch(
"strix.sdk_entry.session_manager.create_or_reuse",
new=AsyncMock(side_effect=fake_create),
),
patch("strix.sdk_entry.session_manager.cleanup", new=AsyncMock()),
patch("strix.sdk_entry.Runner.run", new=AsyncMock(return_value=MagicMock())),
patch("strix.sdk_entry.build_strix_agent", return_value=MagicMock()),
):
await run_strix_scan(
scan_config=_scan_config(),
image="i",
sources_path=tmp_path,
)
assert len(captured_scan_id) == 1
assert captured_scan_id[0].startswith("scan-")
assert len(captured_scan_id[0]) > len("scan-")
@pytest.mark.asyncio
async def test_run_strix_scan_passes_scan_level_config_into_factory(
tmp_path: Path,
) -> None:
"""scan_mode / is_whitebox flow from scan_config into both the
root factory call and the child factory closure."""
bundle = _bundle_for_test()
factory_calls: list[dict[str, Any]] = []
def fake_factory(**kwargs: Any) -> Any:
factory_calls.append(kwargs)
return MagicMock()
with (
patch(
"strix.sdk_entry.session_manager.create_or_reuse",
new=AsyncMock(return_value=bundle),
),
patch("strix.sdk_entry.session_manager.cleanup", new=AsyncMock()),
patch("strix.sdk_entry.Runner.run", new=AsyncMock(return_value=MagicMock())),
patch("strix.sdk_entry.build_strix_agent", side_effect=fake_factory),
):
await run_strix_scan(
scan_config=_scan_config(scan_mode="fast", is_whitebox=True),
scan_id="s",
image="i",
sources_path=tmp_path,
)
assert factory_calls[0]["scan_mode"] == "fast"
assert factory_calls[0]["is_whitebox"] is True
assert factory_calls[0]["is_root"] is True