diff --git a/strix/interface/tui/app.py b/strix/interface/tui/app.py index cc5a36c4f..776980b3c 100644 --- a/strix/interface/tui/app.py +++ b/strix/interface/tui/app.py @@ -36,7 +36,7 @@ from strix.config.models import is_recommended_or_frontier_model from strix.core.hooks import BudgetExceededError from strix.core.inputs import DEFAULT_MAX_TURNS from strix.core.runner import run_strix_scan -from strix.interface.tui.live_view import TuiLiveView +from strix.interface.tui.live_view import SCAN_FAILURE_AGENT_ID, TuiLiveView from strix.interface.tui.messages import send_user_message_to_agent from strix.interface.tui.renderers import render_tool_widget from strix.interface.tui.renderers.agent_message_renderer import AgentMessageRenderer @@ -814,6 +814,7 @@ class StrixTUIApp(App): # type: ignore[misc] self._scan_stop_event = threading.Event() self._scan_completed = threading.Event() self._scan_error: BaseException | None = None + self._scan_error_noted = False self._error_noted_agents: set[str] = set() self._budget_pause_notified = False @@ -1000,6 +1001,8 @@ class StrixTUIApp(App): # type: ignore[misc] except (ValueError, Exception): return + self._record_scan_error_if_needed() + self._sync_agent_graph() for agent_id, agent_data in list(self.live_view.agents.items()): @@ -1017,6 +1020,16 @@ class StrixTUIApp(App): # type: ignore[misc] self._update_vulnerabilities_panel() + def _record_scan_error_if_needed(self) -> None: + if self._scan_error is None or self._scan_error_noted: + return + error_message = str(self._scan_error).strip() or type(self._scan_error).__name__ + self.live_view.record_scan_error(error_message) + self._scan_error_noted = True + if not self.selected_agent_id: + self.selected_agent_id = SCAN_FAILURE_AGENT_ID + self._displayed_events.clear() + def _sync_agent_graph(self) -> None: future = self._agent_graph_sync_future if future is not None: diff --git a/strix/interface/tui/live_view.py b/strix/interface/tui/live_view.py index ec4847dc1..c08dc90a1 100644 --- a/strix/interface/tui/live_view.py +++ b/strix/interface/tui/live_view.py @@ -14,6 +14,12 @@ from strix.core.paths import runtime_state_dir from strix.interface.tui.history import load_session_history +SCAN_FAILURE_AGENT_ID = "scan" +SCAN_FAILURE_AGENT_NAME = "Scan" +SCAN_FAILURE_MESSAGE_PREFIX = "Scan failed" +SCAN_FAILURE_RECOVERY_HINT = "Fix the cause and start a new scan." + + class TuiLiveView: def __init__(self) -> None: self.agents: dict[str, dict[str, Any]] = {} @@ -97,6 +103,23 @@ class TuiLiveView: }, ) + def record_scan_error(self, error: str) -> None: + self.upsert_agent( + SCAN_FAILURE_AGENT_ID, + name=SCAN_FAILURE_AGENT_NAME, + status="failed", + error_message=error, + ) + self._append_event( + SCAN_FAILURE_AGENT_ID, + "chat", + { + "role": "assistant", + "content": f"{SCAN_FAILURE_MESSAGE_PREFIX}: {error}\n{SCAN_FAILURE_RECOVERY_HINT}", + "metadata": {"source": "scan_error"}, + }, + ) + def record_user_message(self, agent_id: str, content: str) -> None: self._append_event( agent_id, diff --git a/tests/test_tui_scan_failure.py b/tests/test_tui_scan_failure.py new file mode 100644 index 000000000..4da1c9df9 --- /dev/null +++ b/tests/test_tui_scan_failure.py @@ -0,0 +1,55 @@ +"""Tests for surfacing scan-level failures in the TUI.""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import TYPE_CHECKING + +from strix.interface.tui.app import StrixTUIApp + + +if TYPE_CHECKING: + from pathlib import Path + + import pytest + + +SCAN_FAILURE_TEXT = "model context input is too large" + + +def test_tui_records_scan_failure_once( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.chdir(tmp_path) + app = StrixTUIApp( + SimpleNamespace( + run_name="scan-error-test", + targets_info=[], + instruction=None, + diff_scope={"active": False}, + scan_mode="deep", + non_interactive=False, + local_sources=[], + scope_mode="auto", + diff_base=None, + user_explicit_instruction=None, + ) + ) + app._scan_error = RuntimeError(SCAN_FAILURE_TEXT) + app._scan_error_noted = False + app.selected_agent_id = None + app._displayed_events = [] + + app._record_scan_error_if_needed() + app._record_scan_error_if_needed() + + scan_agent = app.live_view.agents["scan"] + events = app.live_view.events_for_agent("scan") + + assert scan_agent["status"] == "failed" + assert scan_agent["error_message"] == SCAN_FAILURE_TEXT + assert app.selected_agent_id == "scan" + assert len(events) == 1 + assert "Scan failed" in events[0]["data"]["content"] + assert SCAN_FAILURE_TEXT in events[0]["data"]["content"]