fix(tui): retain stream conversation for tool denial

Capture each stream's conversation ID before starting its worker so /new cannot redirect a later action denial. Add a headless regression that mutates the active conversation while a tool request is in flight.

nexusos_cli/tui_app.py

@@ -354,9 +354,10 @@ class NexusTUI:
                 self._stop_stream.clear()
                 if not self.conversation_id:
                     self.conversation_id = str(uuid.uuid4())
+                conversation_id = self.conversation_id
                 body: dict[str, Any] = {
                     "message": message,
-                    "conversation_id": self.conversation_id,
+                    "conversation_id": conversation_id,
                     "history": list(self.history),
                 }
                 if self._model:
@@ -401,7 +402,7 @@ class NexusTUI:
                                     try:
                                         names = _deny_tool_request(
                                             api_url=self.api_url,
-                                            conversation_id=self.conversation_id,
+                                            conversation_id=conversation_id,
                                             payload=payload,
                                         )
                                         shown = ", ".join(names)

tests/test_tui.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import asyncio
+import threading
 
 import pytest
 from rich.text import Text
@@ -142,5 +143,67 @@ def test_tool_request_is_denied_with_stream_token():
     assert client_kwargs["base_url"] == "http://localhost:8000"
 
 
+def test_inflight_tool_denial_uses_original_conversation_id(monkeypatch):
+    pytest.importorskip("textual")
+    import nexusos_cli.tui_app as tui_app
+
+    stream_started = threading.Event()
+    release_stream = threading.Event()
+    denied_for = []
+
+    class _StreamResponse:
+        status_code = 200
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return None
+
+        def iter_lines(self):
+            stream_started.set()
+            release_stream.wait(timeout=2)
+            yield "event: tool_request"
+            yield 'data: {"token":"secret","actions":[{"name":"run_snippet"}]}'
+            yield ""
+            yield "event: done"
+            yield "data: {}"
+
+    class _StreamClient:
+        def __init__(self, **kwargs):
+            pass
+
+        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, "_deny_tool_request", _capture_denial)
+    app = tui_app.NexusTUI.build_app(api_url="http://127.0.0.1:9")
+
+    async def _run():
+        async with app.run_test():
+            app._start_chat("run it")
+            assert await asyncio.to_thread(stream_started.wait, 2)
+            original_id = app.conversation_id
+            app._handle_slash("/new")
+            assert app.conversation_id is None
+            release_stream.set()
+            for _ in range(200):
+                if not app._busy:
+                    break
+                await asyncio.sleep(0.01)
+            assert app._busy is False
+            assert denied_for == [original_id]
+
+    asyncio.run(_run())
+
+
 def test_escape_round_trip_helper():
     assert "[" in _escape("x[y]") or "\\[" in _escape("x[y]")