feat(migration): phase 5b — STRIX_USE_SDK_HARNESS dispatch flag

Adds the env-var gate that lets users opt into the SDK harness without
disturbing the legacy default. Per PLAYBOOK §7.1, this is the cutover
mechanism: STRIX_USE_SDK_HARNESS=1 routes scans through run_strix_scan
(the Phase 5 entry point); anything else continues to use
StrixAgent.execute_scan.

- strix/interface/sdk_dispatch.py:
  - should_use_sdk_harness(): truthy-string parse of the env var.
  - _resolve_sandbox_image(): reads strix_image from Config; falls
    back to "strix-sandbox:latest" with a warning if unset.
  - _resolve_sources_path(): when --local-sources is given, mounts
    its parent so the agent walks down to the source tree; otherwise
    creates a per-run scratch dir under XDG_CACHE_HOME/strix/sources/.
    Phase 6 will replace this with the legacy clone-into-container
    flow once we port that.
  - run_scan_via_sdk(): the adapter — translates the legacy CLI
    (scan_config dict + argparse Namespace + Tracer) into the keyword
    arguments run_strix_scan expects. Returns the SDK RunResult; lets
    failures bubble up.

- strix/interface/cli.py: adds the dispatch branch inside the existing
  Live/status loop. Legacy default unchanged; SDK path is reached only
  when STRIX_USE_SDK_HARNESS is truthy. Two pre-existing lazy imports
  hoisted to module level (cleanup_runtime + sdk_dispatch helpers) so
  ruff is happy.

Pre-existing legacy lint/type issues surfaced when pre-commit checked
the edited cli.py and chased imports — fixed or ignored in passing:
- utils.py:1052 duplicate ``metadata`` annotation removed.
- utils.py:1251 unused ``# type: ignore[import-not-found]`` for yarl.
- main.py:456 ``panel_parts`` inferred type rejected later string
  entries — explicit ``list[Text | str]`` annotation.
- utils.py:resolve_diff_scope_context PLR0912 (16 branches) per-file
  ignore — branches map 1:1 to scope-mode × target-type combinations.

Tests: 18 new tests in tests/interface/test_sdk_dispatch.py — env
flag parsing parametrized over truthy/falsy variants, image lookup
with config hit + miss-with-warning, sources path resolution for
local_sources / alternative key names / scratch-dir creation, and
the adapter's kwarg handoff verified against a patched
run_strix_scan (run_name from args + run_name from scan_config
fallback + failure propagation).

Refs: PLAYBOOK.md §7.1 (cutover), §7.2 (rollback).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
0xallam
2026-04-25 08:03:00 -07:00
parent f0e254c1fd
commit 4e0d0f35d9
6 changed files with 376 additions and 19 deletions
+4
View File
@@ -312,6 +312,10 @@ ignore = [
# walks every supported target type — splitting it into per-type
# helpers would add indirection without simplifying anything.
"strix/sdk_entry.py" = ["TC003", "PLR0912"]
# Legacy interface utility with intentionally many branches per supported
# scope-mode / target-type combination; refactor would obscure the
# decision tree without simplifying it.
"strix/interface/utils.py" = ["PLR0912"]
# Sandbox dispatch helper has many short-circuit error returns (auth fail,
# size cap, decode fail, etc). Each is a distinct, documented failure mode
# the model needs to see verbatim — collapsing them harms readability.
+24 -13
View File
@@ -11,7 +11,9 @@ from rich.panel import Panel
from rich.text import Text
from strix.agents.StrixAgent import StrixAgent
from strix.interface.sdk_dispatch import run_scan_via_sdk, should_use_sdk_harness
from strix.llm.config import LLMConfig
from strix.runtime import cleanup_runtime
from strix.telemetry.tracer import Tracer, set_global_tracer
from .utils import (
@@ -109,8 +111,6 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
tracer.vulnerability_found_callback = display_vulnerability
def cleanup_on_exit() -> None:
from strix.runtime import cleanup_runtime
tracer.cleanup()
cleanup_runtime()
@@ -163,18 +163,29 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
update_thread.start()
try:
agent = StrixAgent(agent_config)
result = await agent.execute_scan(scan_config)
if should_use_sdk_harness():
# SDK harness opt-in (PLAYBOOK §7.1). Returns a
# RunResult, not the legacy success-dict shape, so
# we skip the legacy error-extraction block —
# failures inside run_strix_scan raise instead.
await run_scan_via_sdk(
scan_config=scan_config,
args=args,
tracer=tracer,
)
else:
agent = StrixAgent(agent_config)
result = await agent.execute_scan(scan_config)
if isinstance(result, dict) and not result.get("success", True):
error_msg = result.get("error", "Unknown error")
error_details = result.get("details")
console.print()
console.print(f"[bold red]Penetration test failed:[/] {error_msg}")
if error_details:
console.print(f"[dim]{error_details}[/]")
console.print()
sys.exit(1)
if isinstance(result, dict) and not result.get("success", True):
error_msg = result.get("error", "Unknown error")
error_details = result.get("details")
console.print()
console.print(f"[bold red]Penetration test failed:[/] {error_msg}")
if error_details:
console.print(f"[dim]{error_details}[/]")
console.print()
sys.exit(1)
finally:
stop_updates.set()
update_thread.join(timeout=1)
+1 -1
View File
@@ -453,7 +453,7 @@ def display_completion_message(args: argparse.Namespace, results_path: Path) ->
stats_text = build_final_stats_text(tracer)
panel_parts = [completion_text, "\n\n", target_text]
panel_parts: list[Text | str] = [completion_text, "\n\n", target_text]
if stats_text.plain:
panel_parts.extend(["\n", stats_text])
+143
View File
@@ -0,0 +1,143 @@
"""STRIX_USE_SDK_HARNESS dispatch — selects legacy vs SDK harness at run-time.
Phase 5b cutover gate. The legacy CLI (``strix.interface.cli``) calls
``StrixAgent(...).execute_scan(scan_config)`` directly. To roll out the
SDK migration safely we want a single env-var-gated branch:
STRIX_USE_SDK_HARNESS=1 → await run_strix_scan(...)
STRIX_USE_SDK_HARNESS=0 → await StrixAgent(...).execute_scan(...)
This module is a thin adapter: it reads the env var, and when the SDK
path is active, translates the legacy ``scan_config`` + ``args`` pair
into the keyword arguments :func:`run_strix_scan` expects.
Per PLAYBOOK §7.1: the legacy default stays in place until end-to-end
validation against a stable target succeeds; the env flag is the
opt-in. Removal of the legacy branch happens one release after cutover.
References:
- PLAYBOOK.md §7.1 (cutover strategy)
- PLAYBOOK.md §7.2 (rollback procedure)
"""
from __future__ import annotations
import logging
import os
from pathlib import Path
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from agents.result import RunResult
logger = logging.getLogger(__name__)
_ENV_FLAG = "STRIX_USE_SDK_HARNESS"
def should_use_sdk_harness() -> bool:
"""Return True iff ``STRIX_USE_SDK_HARNESS`` is truthy in the env.
Truthy values: ``"1"``, ``"true"``, ``"yes"`` (case-insensitive).
Anything else — including unset — returns False so the default
deployed posture stays the legacy harness.
"""
raw = os.environ.get(_ENV_FLAG, "")
return raw.strip().lower() in {"1", "true", "yes"}
def _resolve_sandbox_image() -> str:
"""Read the sandbox image tag from Strix config.
Falls back to ``"strix-sandbox:latest"`` if unset — same behavior
the legacy ``DockerRuntime`` would surface as a config error.
"""
from strix.config import Config
image = Config.get("strix_image")
if not image:
logger.warning(
"strix_image not configured; falling back to strix-sandbox:latest. "
"Set this in ~/.strix/cli-config.json for production use.",
)
return "strix-sandbox:latest"
return str(image)
def _resolve_sources_path(args: Any) -> Path:
"""Pick the host directory to mount into ``/workspace/sources``.
- When ``--local-sources`` was passed, use the parent of the first
source's ``host_path`` (the legacy harness then copies each
individual source under ``/workspace/<subdir>``; we mount the
parent and let the agent walk down).
- Otherwise, use a per-run scratch directory under
``$XDG_CACHE_HOME/strix`` (or ``~/.cache/strix``) — the legacy
flow eventually populates ``/workspace`` via post-create copies,
which the SDK session manager doesn't replicate yet (Phase 6
will bring that in).
"""
local_sources: list[dict[str, str]] | None = getattr(args, "local_sources", None)
if local_sources:
first = local_sources[0]
host_path = first.get("host_path") or first.get("source_path") or first.get("path")
if host_path:
return Path(host_path).expanduser().resolve().parent
cache_root = os.environ.get("XDG_CACHE_HOME") or str(Path.home() / ".cache")
run_name = getattr(args, "run_name", "default") or "default"
sources = Path(cache_root) / "strix" / "sources" / str(run_name)
sources.mkdir(parents=True, exist_ok=True)
return sources
async def run_scan_via_sdk(
*,
scan_config: dict[str, Any],
args: Any,
tracer: Any,
) -> RunResult:
"""Translate legacy CLI args into ``run_strix_scan`` kwargs.
Args:
scan_config: The same dict the legacy ``StrixAgent.execute_scan``
accepts. Forwarded verbatim to ``run_strix_scan``; the
entry point reads ``targets``, ``user_instructions``,
``diff_scope``, ``scan_mode``, ``is_whitebox``, ``skills``
from it.
args: argparse Namespace from ``strix.interface.cli``. We read
``run_name``, ``local_sources``, ``scan_mode`` from it.
tracer: Live ``Tracer`` instance — flows through context so
tools (``create_vulnerability_report``, ``finish_scan``)
persist into the same on-disk run directory the legacy
path uses.
Returns the SDK ``RunResult``. Raises whatever ``run_strix_scan``
raises (sandbox bring-up failure, LLM error, etc.).
"""
from strix.sdk_entry import run_strix_scan
run_name = getattr(args, "run_name", None) or scan_config.get("run_name")
image = _resolve_sandbox_image()
sources_path = _resolve_sources_path(args)
interactive = bool(getattr(args, "interactive", False))
logger.info(
"STRIX_USE_SDK_HARNESS active; dispatching scan %s via run_strix_scan "
"(image=%s, sources=%s)",
run_name,
image,
sources_path,
)
return await run_strix_scan(
scan_config=scan_config,
scan_id=run_name,
image=image,
sources_path=sources_path,
tracer=tracer,
interactive=interactive,
)
+5 -5
View File
@@ -55,7 +55,7 @@ def get_cvss_color(cvss_score: float) -> str:
return "#6b7280"
def format_vulnerability_report(report: dict[str, Any]) -> Text: # noqa: PLR0912, PLR0915
def format_vulnerability_report(report: dict[str, Any]) -> Text: # noqa: PLR0915
"""Format a vulnerability report for CLI display with all rich fields."""
field_style = "bold #4ade80"
@@ -823,7 +823,7 @@ def _truncate_file_list(
return files[:max_files], True
def build_diff_scope_instruction(scopes: list[RepoDiffScope]) -> str: # noqa: PLR0912
def build_diff_scope_instruction(scopes: list[RepoDiffScope]) -> str:
lines = [
"The user is requesting a review of a Pull Request.",
"Instruction: Direct your analysis primarily at the changes in the listed files. "
@@ -1049,7 +1049,7 @@ def resolve_diff_scope_context(
)
instruction_block = build_diff_scope_instruction(repo_scopes)
metadata: dict[str, Any] = {
metadata = {
"active": True,
"mode": scope_mode,
"repos": [scope.to_metadata() for scope in repo_scopes],
@@ -1082,7 +1082,7 @@ def _is_http_git_repo(url: str) -> bool:
return False
def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR0911, PLR0912
def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR0911
if not target or not isinstance(target, str):
raise ValueError("Target must be a non-empty string")
@@ -1248,7 +1248,7 @@ def _is_localhost_host(host: str) -> bool:
def rewrite_localhost_targets(targets_info: list[dict[str, Any]], host_gateway: str) -> None:
from yarl import URL # type: ignore[import-not-found]
from yarl import URL
for target_info in targets_info:
target_type = target_info.get("type")
+199
View File
@@ -0,0 +1,199 @@
"""Phase 5b tests for the STRIX_USE_SDK_HARNESS dispatch.
Covers the env-flag reader, source-path resolution, sandbox image
lookup, and the adapter that translates legacy CLI args into
``run_strix_scan`` kwargs.
We never call ``run_strix_scan`` for real — that requires a live
Docker daemon + LLM. The tests patch it and verify the kwargs handoff.
"""
from __future__ import annotations
import logging
from pathlib import Path
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from strix.interface.sdk_dispatch import (
_resolve_sandbox_image,
_resolve_sources_path,
run_scan_via_sdk,
should_use_sdk_harness,
)
# --- env flag reader ----------------------------------------------------
@pytest.mark.parametrize(
("value", "expected"),
[
("1", True),
("true", True),
("True", True),
("YES", True),
("0", False),
("false", False),
("no", False),
("", False),
("anything-else", False),
],
)
def test_should_use_sdk_harness_parses_env(
monkeypatch: pytest.MonkeyPatch,
value: str,
expected: bool,
) -> None:
monkeypatch.setenv("STRIX_USE_SDK_HARNESS", value)
assert should_use_sdk_harness() is expected
def test_should_use_sdk_harness_defaults_false_when_unset(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("STRIX_USE_SDK_HARNESS", raising=False)
assert should_use_sdk_harness() is False
# --- image lookup -------------------------------------------------------
def test_resolve_sandbox_image_uses_config_value() -> None:
with patch(
"strix.config.Config.get",
return_value="strix-sandbox:0.1.13",
):
assert _resolve_sandbox_image() == "strix-sandbox:0.1.13"
def test_resolve_sandbox_image_falls_back_when_unset(
caplog: pytest.LogCaptureFixture,
) -> None:
with (
patch("strix.config.Config.get", return_value=None),
caplog.at_level(logging.WARNING, logger="strix.interface.sdk_dispatch"),
):
out = _resolve_sandbox_image()
assert out == "strix-sandbox:latest"
assert any("strix_image not configured" in r.message for r in caplog.records)
# --- sources path -------------------------------------------------------
def test_resolve_sources_path_uses_local_sources_parent(tmp_path: Path) -> None:
"""When --local-sources is given, mount that path's parent so the
agent can walk down into the actual source directory tree."""
src_dir = tmp_path / "my-project"
src_dir.mkdir()
args = SimpleNamespace(
local_sources=[{"host_path": str(src_dir)}],
run_name="run-1",
)
assert _resolve_sources_path(args) == tmp_path
def test_resolve_sources_path_handles_alternative_keys(tmp_path: Path) -> None:
"""Some legacy paths use 'source_path' or 'path' instead of
'host_path' — we accept all three."""
src_dir = tmp_path / "alt"
src_dir.mkdir()
args = SimpleNamespace(
local_sources=[{"path": str(src_dir)}],
run_name="run-2",
)
assert _resolve_sources_path(args) == tmp_path
def test_resolve_sources_path_creates_scratch_dir_when_absent(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path))
args = SimpleNamespace(local_sources=None, run_name="scan-x")
out = _resolve_sources_path(args)
assert out == tmp_path / "strix" / "sources" / "scan-x"
assert out.exists()
assert out.is_dir()
# --- adapter -----------------------------------------------------------
@pytest.mark.asyncio
async def test_run_scan_via_sdk_translates_args_to_kwargs(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Verify every kwarg the entry point reads is forwarded correctly."""
monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path))
scan_config = {"targets": [], "scan_mode": "deep"}
args = SimpleNamespace(
run_name="scan-42",
local_sources=None,
interactive=True,
)
fake_tracer = MagicMock(name="tracer")
fake_run = AsyncMock(return_value=MagicMock(name="run_result"))
with (
patch("strix.config.Config.get", return_value="strix-sandbox:test"),
patch("strix.sdk_entry.run_strix_scan", new=fake_run),
):
await run_scan_via_sdk(scan_config=scan_config, args=args, tracer=fake_tracer)
fake_run.assert_awaited_once()
assert fake_run.await_args is not None
kwargs = fake_run.await_args.kwargs
assert kwargs["scan_config"] is scan_config
assert kwargs["scan_id"] == "scan-42"
assert kwargs["image"] == "strix-sandbox:test"
assert kwargs["sources_path"] == tmp_path / "strix" / "sources" / "scan-42"
assert kwargs["tracer"] is fake_tracer
assert kwargs["interactive"] is True
@pytest.mark.asyncio
async def test_run_scan_via_sdk_falls_back_to_scan_config_run_name(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""If args has no run_name, scan_config['run_name'] should be used."""
monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path))
scan_config = {"targets": [], "run_name": "from-config"}
args = SimpleNamespace(local_sources=None)
fake_run = AsyncMock(return_value=MagicMock())
with (
patch("strix.config.Config.get", return_value="img:1"),
patch("strix.sdk_entry.run_strix_scan", new=fake_run),
):
await run_scan_via_sdk(scan_config=scan_config, args=args, tracer=None)
assert fake_run.await_args is not None
assert fake_run.await_args.kwargs["scan_id"] == "from-config"
@pytest.mark.asyncio
async def test_run_scan_via_sdk_propagates_run_failure(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A failure inside run_strix_scan should bubble up to the caller —
the legacy CLI relies on raised exceptions for the SDK path."""
monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path))
scan_config: dict[str, Any] = {"targets": []}
args = SimpleNamespace(run_name="r", local_sources=None)
fake_run = AsyncMock(side_effect=RuntimeError("boom"))
with (
patch("strix.config.Config.get", return_value="img"),
patch("strix.sdk_entry.run_strix_scan", new=fake_run),
pytest.raises(RuntimeError, match="boom"),
):
await run_scan_via_sdk(scan_config=scan_config, args=args, tracer=None)