From e1906833593c3c560487bd03f249140e4eb9f920 Mon Sep 17 00:00:00 2001 From: kent Date: Thu, 23 Jul 2026 17:40:48 +0700 Subject: [PATCH] fix(tui): keep the UI responsive during scans via a batched event pump The scan thread delivered every SDK event to the Textual UI with a per-event blocking App.call_from_thread, which blocked the scan thread and flooded the UI message pump during streaming, starving input handling and making the TUI feel hung. Add interface/tui/event_pump.drain_queue and a thread-safe queue: the scan thread now enqueues events non-blocking and a UI-loop timer drains bounded FIFO batches (lossless, error-resilient). Also surface a hint when a message is typed with no agent selected instead of dropping it silently. Adds tests/test_event_pump.py. Stdlib-only. --- strix/interface/tui/app.py | 29 +++++++++++-- strix/interface/tui/event_pump.py | 55 ++++++++++++++++++++++++ tests/test_event_pump.py | 71 +++++++++++++++++++++++++++++++ 3 files changed, 151 insertions(+), 4 deletions(-) create mode 100644 strix/interface/tui/event_pump.py create mode 100644 tests/test_event_pump.py diff --git a/strix/interface/tui/app.py b/strix/interface/tui/app.py index 5f350898..f3e28c3d 100644 --- a/strix/interface/tui/app.py +++ b/strix/interface/tui/app.py @@ -3,6 +3,7 @@ import asyncio import atexit import contextlib import logging +import queue import signal import sys import threading @@ -35,6 +36,7 @@ from strix.config import load_settings from strix.config.models import is_recommended_or_frontier_model from strix.core.hooks import BudgetExceededError from strix.core.runner import run_strix_scan +from strix.interface.tui.event_pump import DEFAULT_MAX_DRAIN, drain_queue from strix.interface.tui.live_view import TuiLiveView from strix.interface.tui.messages import send_user_message_to_agent from strix.interface.tui.renderers import render_tool_widget @@ -796,6 +798,10 @@ class StrixTUIApp(App): # type: ignore[misc] self._displayed_events: list[str] = [] self._scan_thread: threading.Thread | None = None + # Scan thread enqueues SDK events here (non-blocking); a UI-loop timer + # drains them in bounded batches so a high event rate during streaming + # cannot block the scan thread or starve UI input handling. + self._sdk_event_queue: queue.Queue[tuple[str, Any]] = queue.Queue() self._viewer_httpd: Any = None self._viewer_url: str | None = None self._scan_loop: asyncio.AbstractEventLoop | None = None @@ -966,6 +972,9 @@ class StrixTUIApp(App): # type: ignore[misc] self._start_scan_thread() self.set_interval(0.5, self._update_ui) + # Fast, bounded drain of the SDK event queue keeps the UI responsive to + # input even under a high scan event rate. + self.set_interval(0.05, self._drain_sdk_events) def _update_ui(self) -> None: if self.show_splash: @@ -1508,10 +1517,18 @@ class StrixTUIApp(App): # type: ignore[misc] self._scan_thread.start() def _capture_sdk_event(self, agent_id: str, event: Any) -> None: - try: - self.call_from_thread(self._record_sdk_event, agent_id, event) - except RuntimeError: - self._record_sdk_event(agent_id, event) + # Non-blocking hand-off: enqueue and return immediately so the scan + # thread never blocks on the UI thread per event. The UI-loop timer + # (_drain_sdk_events) ingests batches in order. Replaces the previous + # per-event blocking call_from_thread that flooded the UI pump. + self._sdk_event_queue.put_nowait((agent_id, event)) + + def _drain_sdk_events(self) -> None: + drain_queue( + self._sdk_event_queue, + lambda item: self._record_sdk_event(item[0], item[1]), + max_items=DEFAULT_MAX_DRAIN, + ) def _record_sdk_event(self, agent_id: str, event: Any) -> None: self.live_view.ingest_sdk_event(agent_id, event) @@ -1688,6 +1705,10 @@ class StrixTUIApp(App): # type: ignore[misc] def _send_user_message(self, message: str) -> None: if not self.selected_agent_id: + self.notify( + "Select an agent in the tree first, then send your message.", + severity="warning", + ) return logger.info( diff --git a/strix/interface/tui/event_pump.py b/strix/interface/tui/event_pump.py new file mode 100644 index 00000000..339a0fac --- /dev/null +++ b/strix/interface/tui/event_pump.py @@ -0,0 +1,55 @@ +"""Thread-safe SDK-event pump for the TUI. + +The scan runs on a separate thread and produces SDK events at a high rate during +LLM streaming. Delivering each event to the Textual UI with a blocking +``App.call_from_thread`` per event blocks the scan thread and floods the UI +message pump, starving input handling. Instead the scan thread enqueues events +(non-blocking) and the UI loop drains them in bounded batches via +:func:`drain_queue`, keeping the UI responsive. + +``drain_queue`` is a pure function (no Textual/UI dependency) so it can be unit +tested without a live terminal app. +""" + +from __future__ import annotations + +import logging +import queue +from typing import TYPE_CHECKING + + +if TYPE_CHECKING: + from collections.abc import Callable + + +logger = logging.getLogger(__name__) + +# Default number of events ingested per UI tick. Bounds UI work per frame so a +# burst of events cannot monopolize the event loop. +DEFAULT_MAX_DRAIN = 100 + + +def drain_queue[T]( + q: queue.Queue[T], + sink: Callable[[T], None], + *, + max_items: int = DEFAULT_MAX_DRAIN, +) -> int: + """Drain up to ``max_items`` items from ``q`` (FIFO) into ``sink``. + + Returns the number of items processed. Remaining items are left on the queue + for the next call (never dropped). A ``sink`` error on one item is logged and + does not stop the drain. + """ + processed = 0 + for _ in range(max_items): + try: + item = q.get_nowait() + except queue.Empty: + break + try: + sink(item) + except Exception: + logger.exception("SDK event ingest failed for one item; continuing") + processed += 1 + return processed diff --git a/tests/test_event_pump.py b/tests/test_event_pump.py new file mode 100644 index 00000000..b69fabd9 --- /dev/null +++ b/tests/test_event_pump.py @@ -0,0 +1,71 @@ +"""Tests for the TUI SDK-event pump (bounded, ordered, lossless drain).""" + +from __future__ import annotations + +import queue + +from strix.interface.tui.event_pump import DEFAULT_MAX_DRAIN, drain_queue + + +def test_drain_empty_queue_returns_zero() -> None: + q: queue.Queue[int] = queue.Queue() + sink: list[int] = [] + assert drain_queue(q, sink.append) == 0 + assert sink == [] + + +def test_drain_preserves_fifo_order() -> None: + q: queue.Queue[int] = queue.Queue() + for i in range(5): + q.put(i) + sink: list[int] = [] + processed = drain_queue(q, sink.append) + assert processed == 5 + assert sink == [0, 1, 2, 3, 4] + assert q.empty() + + +def test_drain_bounded_defers_remaining() -> None: + q: queue.Queue[int] = queue.Queue() + for i in range(10): + q.put(i) + sink: list[int] = [] + + first = drain_queue(q, sink.append, max_items=4) + assert first == 4 + assert sink == [0, 1, 2, 3] + # Remaining items are NOT dropped — they stay for the next drain. + assert q.qsize() == 6 + + second = drain_queue(q, sink.append, max_items=4) + assert second == 4 + assert sink == [0, 1, 2, 3, 4, 5, 6, 7] + assert q.qsize() == 2 + + +def test_drain_continues_after_sink_error() -> None: + q: queue.Queue[int] = queue.Queue() + for i in range(3): + q.put(i) + seen: list[int] = [] + + def sink(item: int) -> None: + if item == 1: + raise ValueError("boom") + seen.append(item) + + processed = drain_queue(q, sink) + # All three are consumed (error on item 1 is swallowed), not left stuck. + assert processed == 3 + assert seen == [0, 2] + assert q.empty() + + +def test_default_max_drain_is_bounded() -> None: + q: queue.Queue[int] = queue.Queue() + for i in range(DEFAULT_MAX_DRAIN + 25): + q.put(i) + sink: list[int] = [] + processed = drain_queue(q, sink.append) + assert processed == DEFAULT_MAX_DRAIN + assert q.qsize() == 25