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

refactor: collapse strix/io/, strix/run_config_factory.py, strix/entry.py

Browse Source 
Three top-level files that didn't earn their place:

- ``strix/io/scan_artifacts.py`` had a single consumer (the Tracer);
  collapsing it into ``strix/telemetry/`` puts it next to that consumer.
  ``strix/io/`` is gone.

- ``strix/run_config_factory.py`` held two helpers that didn't earn the
  factoring. ``make_agent_context`` was a 17-line dict-spelling function
  whose argument names were identical to its dict keys — replaced with
  inline dict literals at the two call sites. ``make_run_config`` had
  enough RunConfig assembly logic to justify a helper, but with only
  two callers (root scan + ``create_agent``) inlining is cleaner than
  keeping a top-level file. ``DEFAULT_RETRY`` moves to
  ``strix/llm/retry.py`` next to its other LLM-policy peers; the dead
  ``STRIX_DEFAULT_MAX_TURNS`` constant is dropped.

- ``strix/entry.py`` is a misnomer — it isn't *the* entry point (that's
  ``strix/interface/main.py`` for the CLI), it's the per-scan bring-up
  driver: build the bus, bring up the sandbox, build the root agent +
  child factory, format the scope-context block, register root in bus,
  open SQLiteSession, hand off to ``run_with_continuation``. That all
  lives next to its peers in ``strix/orchestration/`` now, renamed to
  ``scan.py`` so the role is obvious.

No behavior change. Net -125 LoC.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
0xallam
2026-04-25 18:54:46 -07:00
parent 5253332906
commit 72d932f6c4
11 changed files with 130 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+2 -2
View File
@@ -211,7 +211,7 @@ ignore = [
"strix/runtime/backends.py" = ["PLC0415"]
# The vulnerability MD renderer is a long monolithic format function;
# splitting per-section would obscure the structure without simplifying.
"strix/io/scan_artifacts.py" = ["PLR0912", "PLR0915", "PERF401"]
"strix/telemetry/scan_artifacts.py" = ["PLR0912", "PLR0915", "PERF401"]
"strix/runtime/docker_client.py" = [
"TC002", # Manifest, Container imported for annotations
"TC003", # uuid imported for annotation
@@ -234,7 +234,7 @@ ignore = [
# resolution past where mypy needs it. ``_build_root_task`` legitimately
# walks every supported target type — splitting it into per-type
# helpers would add indirection without simplifying anything.
"strix/entry.py" = ["TC003", "PLR0912"]
"strix/orchestration/scan.py" = ["TC003", "PLR0912"]
# Tracer carries a long event surface and a runtime ``Callable``
# annotation on ``vulnerability_found_callback``.
"strix/telemetry/tracer.py" = ["TC003", "PLR0912", "PLR0915", "E501"]
strix/interface/cli.py
+1 -1
View File
@@ -14,7 +14,7 @@ from rich.panel import Panel
from rich.text import Text
from strix.config import load_settings
from strix.entry import run_strix_scan
from strix.orchestration.scan import run_strix_scan
from strix.runtime import session_manager
from strix.telemetry.tracer import Tracer, set_global_tracer
strix/interface/tui.py
+1 -1
View File
@@ -32,11 +32,11 @@ from textual.widgets import Button, Label, Static, TextArea, Tree
from textual.widgets.tree import TreeNode
from strix.config import load_settings
from strix.entry import run_strix_scan
from strix.interface.tool_components.agent_message_renderer import AgentMessageRenderer
from strix.interface.tool_components.registry import get_tool_renderer
from strix.interface.tool_components.user_message_renderer import UserMessageRenderer
from strix.interface.utils import build_tui_stats_text
from strix.orchestration.scan import run_strix_scan
from strix.runtime import session_manager
from strix.telemetry.tracer import Tracer, set_global_tracer
strix/io/__init__.py
-6
View File
@@ -1,6 +0,0 @@
"""Strix I/O — disk artifact writers.
- :class:`ScanArtifactWriter` — writes vulnerability MD/CSV plus the
final penetration-test report under ``strix_runs/<run>/``. Used by
the tracer; could be used directly by post-run tooling.
"""
strix/llm/dedupe.py
+1 -1
View File
@@ -18,7 +18,7 @@ from openai.types.responses import ResponseOutputMessage
from strix.config import load_settings
from strix.llm.multi_provider_setup import build_multi_provider
from strix.run_config_factory import DEFAULT_RETRY
from strix.llm.retry import DEFAULT_RETRY
if TYPE_CHECKING:
strix/llm/retry.py
+30
View File
@@ -0,0 +1,30 @@
"""Shared model-retry policy used across every Strix LLM call."""
from __future__ import annotations
from agents.retry import (
ModelRetryBackoffSettings,
ModelRetrySettings,
retry_policies,
)
# Retry: 5 attempts with ``min(90, 2*2^n)`` backoff. 4xx auth/validation
# errors are excluded from the retryable status list — they can't be
# fixed by retrying and should fail fast. Used by every ``RunConfig``
# Strix builds, plus the dedupe path's one-shot LLM call outside
# ``Runner.run``.
DEFAULT_RETRY = ModelRetrySettings(
max_retries=5,
backoff=ModelRetryBackoffSettings(
initial_delay=2.0,
max_delay=90.0,
multiplier=2.0,
jitter=False,
),
policy=retry_policies.any(
retry_policies.provider_suggested(),
retry_policies.network_error(),
retry_policies.http_status((429, 500, 502, 503, 504)),
),
)
strix/entry.py → strix/orchestration/scan.py
+47 -27
View File
@@ -20,21 +20,27 @@ import uuid
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal
from agents import RunConfig
from agents.memory import SQLiteSession
from agents.model_settings import ModelSettings
from agents.sandbox import SandboxRunConfig
from openai.types.shared import Reasoning
from strix.agents.factory import build_strix_agent, make_child_factory
from strix.config import load_settings
from strix.llm.multi_provider_setup import build_multi_provider
from strix.llm.retry import DEFAULT_RETRY
from strix.orchestration.bus import AgentMessageBus
from strix.orchestration.filter import inject_messages_filter
from strix.orchestration.hooks import StrixOrchestrationHooks
from strix.orchestration.run_loop import run_with_continuation
from strix.run_config_factory import (
STRIX_DEFAULT_MAX_TURNS,
make_agent_context,
make_run_config,
)
from strix.runtime import session_manager
#: Default ``max_turns`` budget passed to ``Runner.run``.
_MAX_TURNS = 300
if TYPE_CHECKING:
from agents.result import RunResultBase
@@ -160,7 +166,7 @@ async def run_strix_scan(
tracer: Any | None = None,
bus: AgentMessageBus | None = None,
interactive: bool = False,
max_turns: int = STRIX_DEFAULT_MAX_TURNS,
max_turns: int = _MAX_TURNS,
model: str | None = None,
cleanup_on_exit: bool = True,
) -> RunResultBase:
@@ -215,7 +221,7 @@ async def run_strix_scan(
)
try:
# Lazy: ``strix.interface`` pulls cli→tui→entry which would cycle.
# Lazy: ``strix.interface`` pulls cli→tui→scan which would cycle.
from strix.interface.utils import is_whitebox_scan
scan_mode = str(scan_config.get("scan_mode") or "deep")
@@ -245,31 +251,45 @@ async def run_strix_scan(
system_prompt_context=scope_context,
)
context = make_agent_context(
bus=bus,
sandbox_session=bundle["session"],
sandbox_client=bundle["client"],
caido_client=bundle["caido_client"],
agent_id=root_id,
parent_id=None,
tracer=tracer,
model=resolved_model,
max_turns=max_turns,
is_whitebox=is_whitebox,
interactive=interactive,
diff_scope=diff_scope,
run_id=run_id,
agent_factory=agent_factory,
)
context: dict[str, Any] = {
"bus": bus,
"sandbox_session": bundle["session"],
"sandbox_client": bundle["client"],
"caido_client": bundle["caido_client"],
"agent_id": root_id,
"parent_id": None,
"tracer": tracer,
"model": resolved_model,
"model_settings": None,
"max_turns": max_turns,
"agent_finish_called": False,
"is_whitebox": is_whitebox,
"interactive": interactive,
"diff_scope": diff_scope,
"run_id": run_id,
"agent_factory": agent_factory,
}
reasoning_effort: Literal["low", "medium", "high"] | None = (
load_settings().llm.reasoning_effort
)
run_config = make_run_config(
sandbox_session=bundle["session"],
sandbox_client=bundle["client"],
model_settings = ModelSettings(
parallel_tool_calls=False,
tool_choice="required",
retry=DEFAULT_RETRY,
)
if reasoning_effort is not None:
model_settings = model_settings.resolve(
ModelSettings(reasoning=Reasoning(effort=reasoning_effort)),
)
run_config = RunConfig(
model=resolved_model,
reasoning_effort=reasoning_effort,
model_provider=build_multi_provider(),
model_settings=model_settings,
sandbox=SandboxRunConfig(client=bundle["client"], session=bundle["session"]),
call_model_input_filter=inject_messages_filter,
tracing_disabled=False,
trace_include_sensitive_data=False,
)
# Native SDK session: persists conversation history to
strix/run_config_factory.py
-159
View File
@@ -1,159 +0,0 @@
"""``make_run_config`` — assemble a Strix-flavored ``RunConfig`` for ``Runner.run``.
Every scan goes through here so defaults apply uniformly. Per-call
overrides land via ``model_settings_override``.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Literal
from agents import RunConfig
from agents.model_settings import ModelSettings
from agents.retry import (
ModelRetryBackoffSettings,
ModelRetrySettings,
retry_policies,
)
from agents.sandbox import SandboxRunConfig
from openai.types.shared import Reasoning
from strix.llm.multi_provider_setup import build_multi_provider
from strix.orchestration.filter import inject_messages_filter
if TYPE_CHECKING:
from agents.sandbox.session.base_sandbox_session import BaseSandboxSession
from strix.orchestration.bus import AgentMessageBus
#: Default ``max_turns`` callers should pass to ``Runner.run``.
STRIX_DEFAULT_MAX_TURNS = 300
# Retry: 5 attempts with ``min(90, 2*2^n)`` backoff. 4xx auth/validation
# errors are excluded from the retryable status list — they can't be
# fixed by retrying and should fail fast. Public so the dedupe path
# (and any other one-shot LLM call outside ``Runner.run``) reuses the
# same policy.
DEFAULT_RETRY = ModelRetrySettings(
max_retries=5,
backoff=ModelRetryBackoffSettings(
initial_delay=2.0,
max_delay=90.0,
multiplier=2.0,
jitter=False,
),
policy=retry_policies.any(
retry_policies.provider_suggested(),
retry_policies.network_error(),
retry_policies.http_status((429, 500, 502, 503, 504)),
),
)
def make_run_config(
*,
sandbox_session: BaseSandboxSession | None,
model: str,
reasoning_effort: Literal["low", "medium", "high"] | None = None,
model_settings_override: ModelSettings | None = None,
sandbox_client: Any | None = None,
) -> RunConfig:
"""Build a ``RunConfig`` with Strix defaults.
Note: ``max_turns`` is not a ``RunConfig`` field — pass it directly
to ``Runner.run``. ``STRIX_DEFAULT_MAX_TURNS`` is the budget Strix
uses.
Args:
sandbox_session: Live sandbox session shared by every agent in
this scan (one container per scan; see
:mod:`strix.runtime.session_manager`). ``None`` is allowed
for unit tests and dry runs.
model: Litellm model alias passed to ``MultiProvider``. Caller
resolves from :attr:`Settings.llm.model`.
reasoning_effort: ``"low" | "medium" | "high"``; routes to
``ModelSettings.reasoning``.
model_settings_override: Optional per-run ``ModelSettings``
merged over factory defaults.
sandbox_client: Optional pre-built sandbox client (Strix Docker
subclass). The SDK instantiates its built-in if a session is
supplied without a client.
"""
base_settings = ModelSettings(
parallel_tool_calls=False,
tool_choice="required",
retry=DEFAULT_RETRY,
)
if reasoning_effort is not None:
base_settings = base_settings.resolve(
ModelSettings(reasoning=Reasoning(effort=reasoning_effort)),
)
if model_settings_override is not None:
base_settings = base_settings.resolve(model_settings_override)
sandbox_config = (
SandboxRunConfig(client=sandbox_client, session=sandbox_session)
if sandbox_session is not None
else None
)
return RunConfig(
model=model,
model_provider=build_multi_provider(),
model_settings=base_settings,
sandbox=sandbox_config,
call_model_input_filter=inject_messages_filter,
tracing_disabled=False,
trace_include_sensitive_data=False,
)
def make_agent_context(
*,
bus: AgentMessageBus,
sandbox_session: BaseSandboxSession | None,
agent_id: str,
parent_id: str | None,
tracer: Any | None,
model: str,
model_settings: ModelSettings | None = None,
max_turns: int = 300,
is_whitebox: bool = False,
interactive: bool = False,
diff_scope: dict[str, Any] | None = None,
run_id: str | None = None,
sandbox_client: Any | None = None,
agent_factory: Any | None = None,
caido_client: Any | None = None,
) -> dict[str, Any]:
"""Build the per-agent ``context`` dict passed to ``Runner.run(context=...)``.
The canonical place where bus, sandbox handles, identity, tracer
reference, and per-agent toggles live. Tools, hooks, and
``inject_messages_filter`` reach in via ``ctx.context.get(...)``.
``agent_factory`` is a callable ``(name, skills) -> agents.Agent`` —
the ``create_agent`` graph tool uses it to spin up children that
inherit the same wiring. ``sandbox_client`` is the host-side Docker
subclass, reused across child runs.
"""
return {
"bus": bus,
"sandbox_session": sandbox_session,
"sandbox_client": sandbox_client,
"caido_client": caido_client,
"agent_id": agent_id,
"parent_id": parent_id,
"tracer": tracer,
"model": model,
"model_settings": model_settings,
"max_turns": max_turns,
"agent_finish_called": False,
"is_whitebox": is_whitebox,
"interactive": interactive,
"diff_scope": diff_scope,
"run_id": run_id,
"agent_factory": agent_factory,
}
strix/io/scan_artifacts.py → strix/telemetry/scan_artifacts.py
View File
strix/telemetry/tracer.py
+1 -1
View File
@@ -5,8 +5,8 @@ from pathlib import Path
from typing import Any, Optional
from uuid import uuid4
from strix.io.scan_artifacts import ScanArtifactWriter
from strix.telemetry import posthog
from strix.telemetry.scan_artifacts import ScanArtifactWriter
logger = logging.getLogger(__name__)
strix/tools/agents_graph/tools.py
+47 -27
View File
@@ -22,12 +22,16 @@ import uuid
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any, Literal
from agents import RunContextWrapper, function_tool
from agents import RunConfig, RunContextWrapper, function_tool
from agents.items import TResponseInputItem
from agents.model_settings import ModelSettings
from agents.sandbox import SandboxRunConfig
from strix.llm.multi_provider_setup import build_multi_provider
from strix.llm.retry import DEFAULT_RETRY
from strix.orchestration.filter import inject_messages_filter
from strix.orchestration.hooks import StrixOrchestrationHooks
from strix.orchestration.run_loop import run_with_continuation
from strix.run_config_factory import make_agent_context, make_run_config
if TYPE_CHECKING:
@@ -428,7 +432,7 @@ async def create_agent(
"success": False,
"error": (
"No agent_factory in context. "
"The root assembly must inject one via make_agent_context."
"The root assembly must inject one when building the run context."
),
},
ensure_ascii=False,
@@ -485,32 +489,48 @@ async def create_agent(
)
initial_input.append({"role": "user", "content": task})
child_ctx = make_agent_context(
bus=bus,
sandbox_session=inner.get("sandbox_session"),
sandbox_client=inner.get("sandbox_client"),
caido_client=inner.get("caido_client"),
agent_id=child_id,
parent_id=parent_id,
tracer=inner.get("tracer"),
model=inner["model"],
model_settings=inner.get("model_settings"),
max_turns=int(inner.get("max_turns", 300)),
is_whitebox=bool(inner.get("is_whitebox", False)),
interactive=bool(inner.get("interactive", False)),
diff_scope=inner.get("diff_scope"),
run_id=inner.get("run_id"),
agent_factory=factory,
)
# Stash the task string for ``agent_finish`` to echo back in its
# XML completion report.
child_ctx["task"] = task
child_ctx: dict[str, Any] = {
"bus": bus,
"sandbox_session": inner.get("sandbox_session"),
"sandbox_client": inner.get("sandbox_client"),
"caido_client": inner.get("caido_client"),
"agent_id": child_id,
"parent_id": parent_id,
"tracer": inner.get("tracer"),
"model": inner["model"],
"model_settings": inner.get("model_settings"),
"max_turns": int(inner.get("max_turns", 300)),
"agent_finish_called": False,
"is_whitebox": bool(inner.get("is_whitebox", False)),
"interactive": bool(inner.get("interactive", False)),
"diff_scope": inner.get("diff_scope"),
"run_id": inner.get("run_id"),
"agent_factory": factory,
# Stashed for ``agent_finish`` to echo back in its completion report.
"task": task,
}
child_run_config = make_run_config(
sandbox_session=inner.get("sandbox_session"),
sandbox_client=inner.get("sandbox_client"),
child_model_settings = ModelSettings(
parallel_tool_calls=False,
tool_choice="required",
retry=DEFAULT_RETRY,
)
override = inner.get("model_settings")
if override is not None:
child_model_settings = child_model_settings.resolve(override)
sandbox_session = inner.get("sandbox_session")
child_run_config = RunConfig(
model=inner["model"],
model_settings_override=inner.get("model_settings"),
model_provider=build_multi_provider(),
model_settings=child_model_settings,
sandbox=(
SandboxRunConfig(client=inner.get("sandbox_client"), session=sandbox_session)
if sandbox_session is not None
else None
),
call_model_input_filter=inject_messages_filter,
tracing_disabled=False,
trace_include_sensitive_data=False,
)
task_handle = asyncio.create_task(