fix(tui): prioritize interrupt and quit keys
package / wheel (pull_request) Waiting to run

Declare Ctrl+C and Ctrl+D as priority Textual bindings so the focused prompt cannot consume them. Drive exit and silent-stream cancellation regressions through Pilot key events instead of calling action handlers directly.
This commit is contained in:
Athena Kaminsky
2026-08-26 08:17:31 -05:00
committed by Athena
parent 9ca37057eb
commit 9ed2908170
2 changed files with 25 additions and 6 deletions
+5 -4
View File
@@ -27,13 +27,14 @@ _APPROVAL_TIMEOUT = httpx.Timeout(10.0, connect=5.0)
def _require_textual(): def _require_textual():
try: try:
from textual.app import App from textual.app import App
from textual.binding import Binding
from textual.widgets import Footer, Header, Input, RichLog, Static from textual.widgets import Footer, Header, Input, RichLog, Static
except ImportError as e: # pragma: no cover - optional extra except ImportError as e: # pragma: no cover - optional extra
raise ImportError( raise ImportError(
"The interactive TUI needs the 'tui' extra — " "The interactive TUI needs the 'tui' extra — "
"pip install 'nexusos-ai[tui]' (or: pip install textual)." "pip install 'nexusos-ai[tui]' (or: pip install textual)."
) from e ) from e
return App, Footer, Header, Input, RichLog, Static return App, Binding, Footer, Header, Input, RichLog, Static
def _escape(text: str) -> str: def _escape(text: str) -> str:
@@ -154,7 +155,7 @@ class NexusTUI:
@staticmethod @staticmethod
def build_app(*, api_url: str | None = None): def build_app(*, api_url: str | None = None):
App, Footer, Header, Input, RichLog, Static = _require_textual() App, Binding, Footer, Header, Input, RichLog, Static = _require_textual()
base = (api_url or settings.api_url).rstrip("/") base = (api_url or settings.api_url).rstrip("/")
class AppImpl(App): class AppImpl(App):
@@ -187,8 +188,8 @@ class NexusTUI:
#prompt { dock: bottom; } #prompt { dock: bottom; }
""" """
BINDINGS = [ BINDINGS = [
("ctrl+c", "interrupt", "Interrupt"), Binding("ctrl+c", "interrupt", "Interrupt", priority=True),
("ctrl+d", "quit", "Quit"), Binding("ctrl+d", "quit", "Quit", priority=True),
] ]
def __init__(self): def __init__(self):
+20 -2
View File
@@ -123,6 +123,23 @@ def test_stream_error_remains_visible_after_finish():
asyncio.run(_run()) asyncio.run(_run())
@pytest.mark.parametrize("key", ["ctrl+c", "ctrl+d"])
def test_priority_exit_bindings_reach_app_while_prompt_is_focused(key):
pytest.importorskip("textual")
from nexusos_cli.tui_app import NexusTUI
app = NexusTUI.build_app(api_url="http://127.0.0.1:9")
async def _run():
async with app.run_test() as pilot:
assert app.is_running
await pilot.press(key)
await pilot.pause()
assert not app.is_running
asyncio.run(_run())
def test_tool_request_is_denied_with_stream_token(): def test_tool_request_is_denied_with_stream_token():
_ApprovalClient.calls.clear() _ApprovalClient.calls.clear()
names = _deny_tool_request( names = _deny_tool_request(
@@ -262,13 +279,14 @@ def test_interrupt_cancels_silent_stream_and_accepts_next_message(monkeypatch):
pytest.fail("stream did not become idle within one second") pytest.fail("stream did not become idle within one second")
async def _run(): async def _run():
async with app.run_test(): async with app.run_test() as pilot:
app._start_chat("first") app._start_chat("first")
assert await asyncio.to_thread(first_stream_started.wait, 2) assert await asyncio.to_thread(first_stream_started.wait, 2)
app.action_interrupt() await pilot.press("ctrl+c")
await _wait_until_idle() await _wait_until_idle()
log = app.query_one("#log") log = app.query_one("#log")
assert any("interrupt requested" in line.text for line in log.lines)
assert not any("ReadTimeout" in line.text for line in log.lines) assert not any("ReadTimeout" in line.text for line in log.lines)
app._start_chat("second") app._start_chat("second")