fix(tui): cancel silent streams promptly

Move chat streaming onto a cancellable async task so Ctrl+C interrupts a pending socket read on macOS instead of waiting for the 120-second read timeout. Add a headless silent-stream regression that verifies prompt recovery and a successful next message.

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
+                cancel = self._stream_cancel
-                if client is not None:
+                if cancel is not None:
-                    try:
+                    loop, task = cancel
-                        client.close()
+                    loop.call_soon_threadsafe(task.cancel)
-                    except Exception:
+
-                        pass
+            def action_quit(self) -> None:
+                self._cancel_stream()
                 self.exit()
 
             def action_interrupt(self) -> None:
                 if self._busy:
-                    self._stop_stream.set()
+                    self._cancel_stream()
-                    client = self._http_client
-                    if client is not None:
-                        try:
-                            client.close()
-                        except Exception:
-                            pass
                     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(
+                    task = asyncio.current_task()
-                        base_url=self.api_url, timeout=_STREAM_TIMEOUT
+                    loop = asyncio.get_running_loop()
-                    )
+                    if task is None:  # pragma: no cover - asyncio guarantees it
-                    self._http_client = client
+                        raise RuntimeError("stream worker has no task")
+                    self._stream_cancel = (loop, task)
                     try:
-                        with client.stream(
+                        if self._stop_stream.is_set():
-                            "POST", "/chat/stream", json=body
+                            raise asyncio.CancelledError
-                        ) as resp:
+                        async with httpx.AsyncClient(
-                            if resp.status_code >= 400:
+                            base_url=self.api_url, timeout=_STREAM_TIMEOUT
-                                detail = resp.read().decode(
+                        ) as client:
-                                    "utf-8", errors="replace"
+                            async with client.stream(
-                                )[:300]
+                                "POST", "/chat/stream", json=body
-                                self._call_ui(
+                            ) as resp:
-                                    self._show_error,
+                                if resp.status_code >= 400:
-                                    f"[red]error HTTP {resp.status_code}[/] "
+                                    detail = (await resp.aread()).decode(
-                                    f"{_escape(detail)}",
+                                        "utf-8", errors="replace"
-                                )
+                                    )[:300]
-                                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
                                     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":
+                                    return
-                                    break
+                                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
+                        if self._stream_cancel == (loop, task):
-                        try:
+                            self._stream_cancel = None
-                            client.close()
-                        except Exception:
-                            pass
                         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)

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]")