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).
This commit is contained in:
0xallam
2026-04-25 08:03:00 -07:00
parent f9fcfd4edf
commit 0339ba85ba
6 changed files with 376 additions and 19 deletions
+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")