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.
This commit is contained in:
kent 2026-07-23 17:40:48 +07:00
parent b04bfa9860
commit e190683359
3 changed files with 151 additions and 4 deletions

View file

@ -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(

View file

@ -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

71
tests/test_event_pump.py Normal file
View file

@ -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