refactor: nuke legacy harness, drop sdk_ prefixes

The SDK harness is the only path now; legacy host-side code is gone.
File names no longer carry the ``sdk_`` distinction.

Deleted legacy host-side modules:
- strix/agents/StrixAgent/ (template moved to strix/agents/prompts/)
- strix/agents/base_agent.py, state.py
- strix/llm/llm.py, config.py
- strix/runtime/docker_runtime.py, runtime.py
- strix/tools/executor.py, agents_graph/agents_graph_actions.py
- strix/interface/sdk_dispatch.py + the env-flag dispatch in cli.py

Renamed (drop ``sdk_`` prefix):
- strix/sdk_entry.py → strix/entry.py
- strix/agents/sdk_factory.py → strix/agents/factory.py
- strix/agents/sdk_prompt.py → strix/agents/prompt.py
- strix/tools/<x>/<x>_sdk_tool[s].py → strix/tools/<x>/tool[s].py
- strix/tools/_legacy_adapter.py → strix/tools/_state_adapter.py
- ``_legacy`` aliases inside the wrappers → ``_impl``

CLI + TUI now call ``run_strix_scan`` directly — they build the
sandbox image / sources_path locally and rely on
``session_manager.cleanup`` (called inside ``run_strix_scan``'s finally)
for teardown. Three TUI handlers that reached into legacy multi-agent
globals (``_agent_instances``, ``send_user_message_to_agent``,
``stop_agent``) are now no-ops with a TODO; reconnecting them to the
``AgentMessageBus`` is a follow-up.

Tracer.get_total_llm_stats no longer reaches into the deleted
``agents_graph_actions`` globals — the orchestration hooks now feed the
tracer via ``Tracer.record_llm_usage`` (live + completed buckets).
finish_scan's ``_check_active_agents`` and load_skill's runtime
``_agent_instances`` reach-in are no-op stubs; the
``AgentMessageBus`` is the source of truth post-migration.

llm/utils.py rewritten to keep only the streaming-parser helpers
(``normalize_tool_format``, ``parse_tool_invocations``,
``fix_incomplete_tool_call``, ``format_tool_call``, ``clean_content``).
``STRIX_MODEL_MAP`` moved to ``llm/multi_provider_setup.py`` (its only
remaining caller).

Per-file ruff ignores added for legacy interface modules (TUI / main /
CLI / utils / streaming_parser / tool_components) and tracer.py —
pre-existing PLC0415/BLE001/PLR0915 patterns are out of scope.

Tests: 287/287 passing. Renamed test files to drop ``sdk_`` prefix.
``test_tracer.py::test_get_total_llm_stats_aggregates_live_and_completed``
rewritten to feed ``Tracer.record_llm_usage`` instead of legacy globals.
Test file annotations added so pre-commit's strict mypy passes.
This commit is contained in:
0xallam
2026-04-25 09:30:23 -07:00
parent 0339ba85ba
commit 5606504563
69 changed files with 646 additions and 4537 deletions
+55 -43
View File
@@ -1,8 +1,11 @@
import atexit
import contextlib
import os
import signal
import sys
import threading
import time
from pathlib import Path
from typing import Any
from rich.console import Console
@@ -10,10 +13,9 @@ from rich.live import Live
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.config import Config
from strix.entry import run_strix_scan
from strix.sandbox import session_manager
from strix.telemetry.tracer import Tracer, set_global_tracer
from .utils import (
@@ -22,6 +24,35 @@ from .utils import (
)
def _resolve_sandbox_image() -> str:
image = Config.get("strix_image")
if not image:
raise RuntimeError(
"strix_image is not configured. Set it in ~/.strix/cli-config.json.",
)
return str(image)
def _resolve_sources_path(args: Any) -> Path:
"""Pick the host directory to mount into ``/workspace/sources``.
- With ``--local-sources``, mount the parent of the first source so
the agent can walk down into the actual tree.
- Otherwise, a per-run scratch dir under ``$XDG_CACHE_HOME/strix``.
"""
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")
sources = Path(cache_root) / "strix" / "sources" / str(args.run_name)
sources.mkdir(parents=True, exist_ok=True)
return sources
async def run_cli(args: Any) -> None: # noqa: PLR0915
console = Console()
@@ -68,27 +99,18 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
console.print()
scan_mode = getattr(args, "scan_mode", "deep")
is_whitebox = bool(getattr(args, "local_sources", []))
scan_config = {
scan_config: dict[str, Any] = {
"scan_id": args.run_name,
"targets": args.targets_info,
"user_instructions": args.instruction or "",
"run_name": args.run_name,
"diff_scope": getattr(args, "diff_scope", {"active": False}),
"scan_mode": scan_mode,
"is_whitebox": is_whitebox,
}
llm_config = LLMConfig(
scan_mode=scan_mode,
is_whitebox=bool(getattr(args, "local_sources", [])),
)
agent_config = {
"llm_config": llm_config,
"max_iterations": 300,
}
if getattr(args, "local_sources", None):
agent_config["local_sources"] = args.local_sources
tracer = Tracer(args.run_name)
tracer.set_scan_config(scan_config)
@@ -112,7 +134,6 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
def cleanup_on_exit() -> None:
tracer.cleanup()
cleanup_runtime()
def signal_handler(_signum: int, _frame: Any) -> None:
tracer.cleanup()
@@ -131,7 +152,7 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
status_text.append("Penetration test in progress", style="bold #22c55e")
status_text.append("\n\n")
stats_text = build_live_stats_text(tracer, agent_config)
stats_text = build_live_stats_text(tracer)
if stats_text:
status_text.append(stats_text)
@@ -156,39 +177,30 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
try:
live.update(create_live_status())
time.sleep(2)
except Exception: # noqa: BLE001
except Exception:
break
update_thread = threading.Thread(target=update_status, daemon=True)
update_thread.start()
try:
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)
await run_strix_scan(
scan_config=scan_config,
scan_id=args.run_name,
image=_resolve_sandbox_image(),
sources_path=_resolve_sources_path(args),
tracer=tracer,
interactive=bool(getattr(args, "interactive", False)),
)
finally:
stop_updates.set()
update_thread.join(timeout=1)
# Best-effort: tear down the sandbox session even if the
# run raised. ``run_strix_scan`` already does this in its
# own ``finally``, but call here too in case the failure
# was during early setup.
with contextlib.suppress(Exception):
await session_manager.cleanup(args.run_name)
except Exception as e:
console.print(f"[bold red]Error during penetration test:[/] {e}")
+16 -9
View File
@@ -20,7 +20,7 @@ from rich.text import Text
from strix.config import Config, apply_saved_config, save_current_config
from strix.config.config import resolve_llm_config
from strix.llm.utils import resolve_strix_model
from strix.llm.multi_provider_setup import STRIX_MODEL_MAP
apply_saved_config()
@@ -42,7 +42,9 @@ from strix.interface.utils import ( # noqa: E402
validate_config_file,
validate_llm_response,
)
from strix.runtime.docker_runtime import HOST_GATEWAY_HOSTNAME # noqa: E402
HOST_GATEWAY_HOSTNAME = "host.docker.internal"
from strix.telemetry import posthog # noqa: E402
from strix.telemetry.tracer import get_global_tracer # noqa: E402
@@ -50,7 +52,7 @@ from strix.telemetry.tracer import get_global_tracer # noqa: E402
logging.getLogger().setLevel(logging.ERROR)
def validate_environment() -> None: # noqa: PLR0912, PLR0915
def validate_environment() -> None:
console = Console()
missing_required_vars = []
missing_optional_vars = []
@@ -209,8 +211,13 @@ async def warm_up_llm() -> None:
try:
model_name, api_key, api_base = resolve_llm_config()
litellm_model, _ = resolve_strix_model(model_name)
litellm_model = litellm_model or model_name
# ``strix/<alias>`` is routed through the Strix proxy (OpenAI-compatible);
# everything else is sent as-is.
litellm_model: str | None = model_name
if model_name and model_name.startswith("strix/"):
base = model_name[len("strix/") :]
if base in STRIX_MODEL_MAP:
litellm_model = f"openai/{base}"
test_messages = [
{"role": "system", "content": "You are a helpful assistant."},
@@ -233,7 +240,7 @@ async def warm_up_llm() -> None:
validate_llm_response(response)
except Exception as e: # noqa: BLE001
except Exception as e:
error_text = Text()
error_text.append("LLM CONNECTION FAILED", style="bold red")
error_text.append("\n\n", style="white")
@@ -260,7 +267,7 @@ def get_version() -> str:
from importlib.metadata import version
return version("strix-agent")
except Exception: # noqa: BLE001
except Exception:
return "unknown"
@@ -401,7 +408,7 @@ Examples:
args.instruction = f.read().strip()
if not args.instruction:
parser.error(f"Instruction file '{instruction_path}' is empty")
except Exception as e: # noqa: BLE001
except Exception as e:
parser.error(f"Failed to read instruction file '{instruction_path}': {e}")
args.targets_info = []
@@ -544,7 +551,7 @@ def persist_config() -> None:
save_current_config()
def main() -> None: # noqa: PLR0912, PLR0915
def main() -> None:
if sys.platform == "win32":
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
-143
View File
@@ -1,143 +0,0 @@
"""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,
)
+57 -66
View File
@@ -1,13 +1,16 @@
import argparse
import asyncio
import atexit
import contextlib
import logging
import os
import signal
import sys
import threading
from collections.abc import Callable
from importlib.metadata import PackageNotFoundError
from importlib.metadata import version as pkg_version
from pathlib import Path
from typing import TYPE_CHECKING, Any, ClassVar
@@ -28,13 +31,14 @@ from textual.screen import ModalScreen
from textual.widgets import Button, Label, Static, TextArea, Tree
from textual.widgets.tree import TreeNode
from strix.agents.StrixAgent import StrixAgent
from strix.config import Config
from strix.entry import run_strix_scan
from strix.interface.streaming_parser import parse_streaming_content
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.llm.config import LLMConfig
from strix.sandbox import session_manager
from strix.telemetry.tracer import Tracer, set_global_tracer
@@ -329,7 +333,7 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
else:
return text
def _render_vulnerability(self) -> Text: # noqa: PLR0912, PLR0915
def _render_vulnerability(self) -> Text:
vuln = self.vulnerability
text = Text()
@@ -455,7 +459,7 @@ class VulnerabilityDetailScreen(ModalScreen): # type: ignore[misc]
return text
def _get_markdown_report(self) -> str: # noqa: PLR0912, PLR0915
def _get_markdown_report(self) -> str:
"""Get Markdown version of vulnerability report for clipboard."""
vuln = self.vulnerability
lines: list[str] = []
@@ -702,7 +706,6 @@ class StrixTUIApp(App): # type: ignore[misc]
super().__init__()
self.args = args
self.scan_config = self._build_scan_config(args)
self.agent_config = self._build_agent_config(args)
self.tracer = Tracer(self.scan_config["run_name"])
self.tracer.set_scan_config(self.scan_config)
@@ -736,6 +739,18 @@ class StrixTUIApp(App): # type: ignore[misc]
self._setup_cleanup_handlers()
def _resolve_sources_path(self) -> Path:
local_sources = getattr(self.args, "local_sources", None) or []
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")
sources = Path(cache_root) / "strix" / "sources" / str(self.args.run_name)
sources.mkdir(parents=True, exist_ok=True)
return sources
def _build_scan_config(self, args: argparse.Namespace) -> dict[str, Any]:
return {
"scan_id": args.run_name,
@@ -743,32 +758,13 @@ class StrixTUIApp(App): # type: ignore[misc]
"user_instructions": args.instruction or "",
"run_name": args.run_name,
"diff_scope": getattr(args, "diff_scope", {"active": False}),
"scan_mode": getattr(args, "scan_mode", "deep"),
"is_whitebox": bool(getattr(args, "local_sources", [])),
}
def _build_agent_config(self, args: argparse.Namespace) -> dict[str, Any]:
scan_mode = getattr(args, "scan_mode", "deep")
llm_config = LLMConfig(
scan_mode=scan_mode,
interactive=True,
is_whitebox=bool(getattr(args, "local_sources", [])),
)
config = {
"llm_config": llm_config,
"max_iterations": 300,
}
if getattr(args, "local_sources", None):
config["local_sources"] = args.local_sources
return config
def _setup_cleanup_handlers(self) -> None:
def cleanup_on_exit() -> None:
from strix.runtime import cleanup_runtime
self.tracer.cleanup()
cleanup_runtime()
def signal_handler(_signum: int, _frame: Any) -> None:
self.tracer.cleanup()
@@ -1305,7 +1301,7 @@ class StrixTUIApp(App): # type: ignore[misc]
stats_content = Text()
stats_text = build_tui_stats_text(self.tracer, self.agent_config)
stats_text = build_tui_stats_text(self.tracer)
if stats_text:
stats_content.append(stats_text)
@@ -1502,10 +1498,19 @@ class StrixTUIApp(App): # type: ignore[misc]
asyncio.set_event_loop(loop)
try:
agent = StrixAgent(self.agent_config)
if not self._scan_stop_event.is_set():
loop.run_until_complete(agent.execute_scan(self.scan_config))
image = Config.get("strix_image") or "strix-sandbox:latest"
sources_path = self._resolve_sources_path()
loop.run_until_complete(
run_strix_scan(
scan_config=self.scan_config,
scan_id=self.scan_config["run_name"],
image=str(image),
sources_path=sources_path,
tracer=self.tracer,
interactive=True,
),
)
except (KeyboardInterrupt, asyncio.CancelledError):
logging.info("Scan interrupted by user")
@@ -1516,6 +1521,12 @@ class StrixTUIApp(App): # type: ignore[misc]
except Exception:
logging.exception("Unexpected error during scan")
finally:
# Best-effort sandbox teardown if early setup failed
# before run_strix_scan's own ``finally`` ran.
with contextlib.suppress(Exception):
loop.run_until_complete(
session_manager.cleanup(self.scan_config["run_name"]),
)
loop.close()
self._scan_completed.set()
@@ -1816,15 +1827,9 @@ class StrixTUIApp(App): # type: ignore[misc]
metadata={"interrupted": True},
)
try:
from strix.tools.agents_graph.agents_graph_actions import _agent_instances
if self.selected_agent_id in _agent_instances:
agent_instance = _agent_instances[self.selected_agent_id]
if hasattr(agent_instance, "cancel_current_execution"):
agent_instance.cancel_current_execution()
except (ImportError, AttributeError, KeyError):
pass
# TODO: route user→agent messages through the AgentMessageBus
# once the TUI has a handle on it. The bus currently lives
# inside ``run_strix_scan`` scope only.
if self.tracer:
self.tracer.log_chat_message(
@@ -1833,15 +1838,11 @@ class StrixTUIApp(App): # type: ignore[misc]
agent_id=self.selected_agent_id,
)
try:
from strix.tools.agents_graph.agents_graph_actions import send_user_message_to_agent
send_user_message_to_agent(self.selected_agent_id, message)
except (ImportError, AttributeError) as e:
import logging
logging.warning(f"Failed to send message to agent {self.selected_agent_id}: {e}")
logging.warning(
"User-message-to-agent dispatch is not wired post-migration; "
"message %r logged to tracer but not delivered.",
message,
)
self._displayed_events.clear()
self._update_chat_view()
@@ -1940,22 +1941,12 @@ class StrixTUIApp(App): # type: ignore[misc]
return agent_name, False
def action_confirm_stop_agent(self, agent_id: str) -> None:
try:
from strix.tools.agents_graph.agents_graph_actions import stop_agent
result = stop_agent(agent_id)
import logging
if result.get("success"):
logging.info(f"Stop request sent to agent: {result.get('message', 'Unknown')}")
else:
logging.warning(f"Failed to stop agent: {result.get('error', 'Unknown error')}")
except Exception:
import logging
logging.exception(f"Failed to stop agent {agent_id}")
# TODO: route to ``bus.cancel_descendants(agent_id)`` once the TUI
# has a handle on the AgentMessageBus.
logging.warning(
"Stop-agent dispatch is not wired post-migration; agent %s left running.",
agent_id,
)
def action_custom_quit(self) -> None:
if self._scan_thread and self._scan_thread.is_alive():
@@ -2071,7 +2062,7 @@ class StrixTUIApp(App): # type: ignore[misc]
cleaned = self._clean_copied_text(selected)
self.copy_to_clipboard(cleaned if cleaned.strip() else selected)
copied = True
except Exception: # noqa: BLE001
except Exception:
logger.debug("Failed to copy screen selection", exc_info=True)
if not copied:
@@ -2082,7 +2073,7 @@ class StrixTUIApp(App): # type: ignore[misc]
self.copy_to_clipboard(selected)
chat_input.move_cursor(chat_input.cursor_location)
copied = True
except Exception: # noqa: BLE001
except Exception:
logger.debug("Failed to copy chat input selection", exc_info=True)
if copied:
+12 -14
View File
@@ -20,6 +20,8 @@ from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from strix.config import Config
# Token formatting utilities
def format_token_count(count: float) -> str:
@@ -297,17 +299,15 @@ def build_final_stats_text(tracer: Any) -> Text:
return stats_text
def build_live_stats_text(tracer: Any, agent_config: dict[str, Any] | None = None) -> Text:
def build_live_stats_text(tracer: Any) -> Text:
stats_text = Text()
if not tracer:
return stats_text
if agent_config:
llm_config = agent_config["llm_config"]
model = getattr(llm_config, "model_name", "Unknown")
stats_text.append("Model ", style="dim")
stats_text.append(model, style="white")
stats_text.append("\n")
model = Config.get("strix_llm") or "unknown"
stats_text.append("Model ", style="dim")
stats_text.append(str(model), style="white")
stats_text.append("\n")
vuln_count = len(tracer.vulnerability_reports)
tool_count = tracer.get_real_tool_count()
@@ -370,15 +370,13 @@ def build_live_stats_text(tracer: Any, agent_config: dict[str, Any] | None = Non
return stats_text
def build_tui_stats_text(tracer: Any, agent_config: dict[str, Any] | None = None) -> Text:
def build_tui_stats_text(tracer: Any) -> Text:
stats_text = Text()
if not tracer:
return stats_text
if agent_config:
llm_config = agent_config["llm_config"]
model = getattr(llm_config, "model_name", "Unknown")
stats_text.append(model, style="white")
model = Config.get("strix_llm") or "unknown"
stats_text.append(str(model), style="white")
llm_stats = tracer.get_total_llm_stats()
total_stats = llm_stats["total"]
@@ -427,7 +425,7 @@ def _derive_target_label_for_run_name(targets_info: list[dict[str, Any]] | None)
try:
parsed = urlparse(url)
return str(parsed.netloc or parsed.path or url)
except Exception: # noqa: BLE001
except Exception:
return str(url)
if target_type == "repository":
@@ -443,7 +441,7 @@ def _derive_target_label_for_run_name(targets_info: list[dict[str, Any]] | None)
path_str = details.get("target_path", original)
try:
return str(Path(path_str).name or path_str)
except Exception: # noqa: BLE001
except Exception:
return str(path_str)
if target_type == "ip_address":