diff --git a/nexusos_cli/tui_app.py b/nexusos_cli/tui_app.py index 1897bbe..21945df 100644 --- a/nexusos_cli/tui_app.py +++ b/nexusos_cli/tui_app.py @@ -6,6 +6,7 @@ stdin/stdout are a TTY. Classic one-shots (``nexus chat send``, ``nexus monitor` """ from __future__ import annotations +import asyncio import json import threading import uuid @@ -15,7 +16,6 @@ import httpx from synapse.nexus_config import settings -from . import nexus_api from .monitor import collect_snapshot # Between SSE chunks a silent backend must not pin the UI forever. Connect stays @@ -199,7 +199,9 @@ class NexusTUI: self._model: str | None = None self._busy = False self._stop_stream = threading.Event() - self._http_client: httpx.Client | None = None + self._stream_cancel: ( + tuple[asyncio.AbstractEventLoop, asyncio.Task] | None + ) = None self._status_lock = threading.Lock() self._status_pending = False @@ -269,25 +271,26 @@ class NexusTUI: """Write a stream error to the persistent transcript.""" self.query_one("#log", RichLog).write(message) - def action_quit(self) -> None: + def _cancel_stream(self) -> None: + """Cancel the task that owns the socket read. + + Closing a synchronous httpx client from the UI thread does not + reliably unblock its worker-thread read on macOS. Async task + cancellation is delivered to the pending read itself. + """ self._stop_stream.set() - client = self._http_client - if client is not None: - try: - client.close() - except Exception: - pass + cancel = self._stream_cancel + if cancel is not None: + loop, task = cancel + loop.call_soon_threadsafe(task.cancel) + + def action_quit(self) -> None: + self._cancel_stream() self.exit() def action_interrupt(self) -> None: if self._busy: - self._stop_stream.set() - client = self._http_client - if client is not None: - try: - client.close() - except Exception: - pass + self._cancel_stream() self.query_one("#log", RichLog).write( "[yellow]▸ interrupt requested[/]" ) @@ -364,74 +367,96 @@ class NexusTUI: body["model"] = self._model self.history.append({"role": "user", "content": message}) - def worker(): + async def stream_worker(): reply_parts: list[str] = [] - client = httpx.Client( - base_url=self.api_url, timeout=_STREAM_TIMEOUT - ) - self._http_client = client + task = asyncio.current_task() + loop = asyncio.get_running_loop() + if task is None: # pragma: no cover - asyncio guarantees it + raise RuntimeError("stream worker has no task") + self._stream_cancel = (loop, task) try: - with client.stream( - "POST", "/chat/stream", json=body - ) as resp: - if resp.status_code >= 400: - detail = resp.read().decode( - "utf-8", errors="replace" - )[:300] - self._call_ui( - self._show_error, - f"[red]error HTTP {resp.status_code}[/] " - f"{_escape(detail)}", - ) - return - for kind, payload in nexus_api.iter_chunks( - resp.iter_lines() - ): - if self._stop_stream.is_set(): - break - if kind == "chunk": - reply_parts.append(payload) - preview = "".join(reply_parts) - if len(preview) > 4000: - preview = "…" + preview[-4000:] - self._call_ui( - live.update, - format_assistant_line(preview), - ) - elif kind == "tool_request": - try: - names = _deny_tool_request( - api_url=self.api_url, - conversation_id=conversation_id, - payload=payload, - ) - shown = ", ".join(names) - self._call_ui( - log.write, - "[yellow]▸ denied action tool " - f"{_escape(shown)} — interactive approval " - "is not yet available in the TUI[/]", - ) - except Exception as exc: - self._call_ui( - self._show_error, - "[red]tool denial failed:[/] " - f"{_escape(str(exc))}", - ) - return - elif kind == "error": - try: - detail = json.loads(payload).get( - "detail", payload - ) - except Exception: - detail = payload + if self._stop_stream.is_set(): + raise asyncio.CancelledError + async with httpx.AsyncClient( + base_url=self.api_url, timeout=_STREAM_TIMEOUT + ) as client: + async with client.stream( + "POST", "/chat/stream", json=body + ) as resp: + if resp.status_code >= 400: + detail = (await resp.aread()).decode( + "utf-8", errors="replace" + )[:300] self._call_ui( self._show_error, - f"[red]error:[/] {_escape(str(detail))}", + f"[red]error HTTP {resp.status_code}[/] " + f"{_escape(detail)}", ) - elif kind == "done": - break + return + event = "message" + async for line in resp.aiter_lines(): + if self._stop_stream.is_set(): + raise asyncio.CancelledError + if line == "": + event = "message" + continue + if line.startswith("event:"): + event = line[6:].strip() + continue + if not line.startswith("data:"): + continue + payload = line[5:].strip() + kind = event + if kind in ("message", ""): + kind = "chunk" + payload = json.loads(payload) + if kind == "chunk": + reply_parts.append(payload) + preview = "".join(reply_parts) + if len(preview) > 4000: + preview = "…" + preview[-4000:] + self._call_ui( + live.update, + format_assistant_line(preview), + ) + elif kind == "tool_request": + try: + names = _deny_tool_request( + api_url=self.api_url, + conversation_id=conversation_id, + payload=payload, + ) + shown = ", ".join(names) + self._call_ui( + log.write, + "[yellow]▸ denied action tool " + f"{_escape(shown)} — interactive " + "approval is not yet available in " + "the TUI[/]", + ) + except Exception as exc: + self._call_ui( + self._show_error, + "[red]tool denial failed:[/] " + f"{_escape(str(exc))}", + ) + return + elif kind == "error": + try: + detail = json.loads(payload).get( + "detail", payload + ) + except Exception: + detail = payload + self._call_ui( + self._show_error, + f"[red]error:[/] " + f"{_escape(str(detail))}", + ) + elif kind == "done": + break + except asyncio.CancelledError: + pass except httpx.ConnectError: self._call_ui( self._show_error, @@ -445,15 +470,14 @@ class NexusTUI: f"{_escape(str(exc))}", ) finally: - self._http_client = None - try: - client.close() - except Exception: - pass + if self._stream_cancel == (loop, task): + self._stream_cancel = None text = "".join(reply_parts).strip() self._call_ui(self._finish_stream, text) - threading.Thread(target=worker, daemon=True).start() + threading.Thread( + target=lambda: asyncio.run(stream_worker()), daemon=True + ).start() def _finish_stream(self, text: str) -> None: log = self.query_one("#log", RichLog) diff --git a/tests/test_tui.py b/tests/test_tui.py index 7a0ac1e..505c285 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -154,15 +154,15 @@ def test_inflight_tool_denial_uses_original_conversation_id(monkeypatch): class _StreamResponse: status_code = 200 - def __enter__(self): + async def __aenter__(self): return self - def __exit__(self, *args): + async def __aexit__(self, *args): return None - def iter_lines(self): + async def aiter_lines(self): stream_started.set() - release_stream.wait(timeout=2) + await asyncio.to_thread(release_stream.wait, 2) yield "event: tool_request" yield 'data: {"token":"secret","actions":[{"name":"run_snippet"}]}' yield "" @@ -173,17 +173,20 @@ def test_inflight_tool_denial_uses_original_conversation_id(monkeypatch): def __init__(self, **kwargs): pass + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return None + def stream(self, *args, **kwargs): return _StreamResponse() - def close(self): - return None - def _capture_denial(*, conversation_id, **kwargs): denied_for.append(conversation_id) return ["run_snippet"] - monkeypatch.setattr(tui_app.httpx, "Client", _StreamClient) + monkeypatch.setattr(tui_app.httpx, "AsyncClient", _StreamClient) monkeypatch.setattr(tui_app, "_deny_tool_request", _capture_denial) app = tui_app.NexusTUI.build_app(api_url="http://127.0.0.1:9") @@ -205,5 +208,78 @@ def test_inflight_tool_denial_uses_original_conversation_id(monkeypatch): asyncio.run(_run()) +def test_interrupt_cancels_silent_stream_and_accepts_next_message(monkeypatch): + pytest.importorskip("textual") + import nexusos_cli.tui_app as tui_app + + first_stream_started = threading.Event() + + class _StreamResponse: + status_code = 200 + + def __init__(self, call_number): + self.call_number = call_number + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return None + + async def aiter_lines(self): + if self.call_number == 1: + first_stream_started.set() + await asyncio.Event().wait() + yield "data: \"READY\"" + yield "" + yield "event: done" + yield "data: {}" + + class _StreamClient: + calls = 0 + + def __init__(self, **kwargs): + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return None + + def stream(self, *args, **kwargs): + type(self).calls += 1 + return _StreamResponse(type(self).calls) + + monkeypatch.setattr(tui_app.httpx, "AsyncClient", _StreamClient) + app = tui_app.NexusTUI.build_app(api_url="http://127.0.0.1:9") + + async def _wait_until_idle(): + for _ in range(100): + if not app._busy: + return + await asyncio.sleep(0.01) + pytest.fail("stream did not become idle within one second") + + async def _run(): + async with app.run_test(): + app._start_chat("first") + assert await asyncio.to_thread(first_stream_started.wait, 2) + app.action_interrupt() + await _wait_until_idle() + + log = app.query_one("#log") + assert not any("ReadTimeout" in line.text for line in log.lines) + + app._start_chat("second") + await _wait_until_idle() + assert app.history[-1] == { + "role": "assistant", + "content": "READY", + } + + asyncio.run(_run()) + + def test_escape_round_trip_helper(): assert "[" in _escape("x[y]") or "\\[" in _escape("x[y]")