diff --git a/nexusos_cli/tui_app.py b/nexusos_cli/tui_app.py index c3084ee..54275f7 100644 --- a/nexusos_cli/tui_app.py +++ b/nexusos_cli/tui_app.py @@ -21,6 +21,7 @@ from .monitor import collect_snapshot # Between SSE chunks a silent backend must not pin the UI forever. Connect stays # short; the overall stream may run minutes. _STREAM_TIMEOUT = httpx.Timeout(None, connect=5.0, read=120.0, write=30.0, pool=5.0) +_APPROVAL_TIMEOUT = httpx.Timeout(10.0, connect=5.0) def _require_textual(): @@ -61,6 +62,40 @@ def format_assistant_line(text: str) -> str: return f"[bold blue]nexus>[/] {_escape(text)}" +def _deny_tool_request( + *, + api_url: str, + conversation_id: str, + payload: str, + client_factory=httpx.Client, +) -> list[str]: + """Immediately deny a TUI action request and let the stream resume. + + The web client presents an approval dialog, but the TUI does not yet have + that interaction. Denying with the stream's capability token preserves the + ``ask`` safety boundary without leaving the backend waiting for five minutes. + """ + request = json.loads(payload) + token = request.get("token") or "" + actions = request.get("actions") or [] + names = [ + action.get("name", "") + for action in actions + if isinstance(action, dict) and action.get("name") + ] + if not token or not names: + raise ValueError("invalid tool approval request") + body = { + "conversation_id": conversation_id, + "token": token, + "decisions": {name: False for name in names}, + } + with client_factory(base_url=api_url, timeout=_APPROVAL_TIMEOUT) as client: + response = client.post("/chat/approve", json=body) + response.raise_for_status() + return names + + def _status_line(snap: dict | None = None) -> str: """Format a snapshot. Pass ``snap`` — do not omit it on the UI thread.""" if snap is None: @@ -230,6 +265,10 @@ class NexusTUI: # CancelledError is BaseException; also ignore post-exit races. pass + def _show_error(self, message: str) -> None: + """Write a stream error to the persistent transcript.""" + self.query_one("#log", RichLog).write(message) + def action_quit(self) -> None: self._stop_stream.set() client = self._http_client @@ -339,7 +378,7 @@ class NexusTUI: "utf-8", errors="replace" )[:300] self._call_ui( - live.update, + self._show_error, f"[red]error HTTP {resp.status_code}[/] " f"{_escape(detail)}", ) @@ -359,12 +398,26 @@ class NexusTUI: format_assistant_line(preview), ) elif kind == "tool_request": - self._call_ui( - log.write, - "[yellow]▸ tool approval needed — " - "Approve in the web UI, or set " - "action_tool_policy=allow[/]", - ) + try: + names = _deny_tool_request( + api_url=self.api_url, + conversation_id=self.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( @@ -373,20 +426,20 @@ class NexusTUI: except Exception: detail = payload self._call_ui( - live.update, + self._show_error, f"[red]error:[/] {_escape(str(detail))}", ) elif kind == "done": break except httpx.ConnectError: self._call_ui( - live.update, + self._show_error, f"[red]Backend not reachable at " f"{_escape(self.api_url)}. Start it: nexus start[/]", ) except Exception as exc: self._call_ui( - live.update, + self._show_error, f"[red]{_escape(type(exc).__name__)}:[/] " f"{_escape(str(exc))}", ) diff --git a/tests/test_tui.py b/tests/test_tui.py index 4ca6f8a..4b80423 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -8,6 +8,7 @@ from rich.text import Text from nexusos_cli.tui_app import ( _compact_status, + _deny_tool_request, _escape, _status_line, format_assistant_line, @@ -15,6 +16,28 @@ from nexusos_cli.tui_app import ( ) +class _ApprovalResponse: + def raise_for_status(self): + return None + + +class _ApprovalClient: + calls = [] + + def __init__(self, **kwargs): + self.kwargs = kwargs + + def __enter__(self): + return self + + def __exit__(self, *args): + return None + + def post(self, path, *, json): + self.calls.append((path, json, self.kwargs)) + return _ApprovalResponse() + + def test_status_line_mentions_services(): snap = { "version": "1.0.0", @@ -81,5 +104,43 @@ def test_finish_stream_markup_does_not_wedge_busy(): asyncio.run(_run()) +def test_stream_error_remains_visible_after_finish(): + 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(): + app._busy = True + app._show_error("[red]Backend not reachable[/]") + app._finish_stream("") + log = app.query_one("#log") + assert any("Backend not reachable" in line.text for line in log.lines) + assert app._busy is False + + asyncio.run(_run()) + + +def test_tool_request_is_denied_with_stream_token(): + _ApprovalClient.calls.clear() + names = _deny_tool_request( + api_url="http://localhost:8000", + conversation_id="conversation-1", + payload='{"token":"secret","actions":[{"name":"run_snippet"}]}', + client_factory=_ApprovalClient, + ) + + assert names == ["run_snippet"] + path, body, client_kwargs = _ApprovalClient.calls[-1] + assert path == "/chat/approve" + assert body == { + "conversation_id": "conversation-1", + "token": "secret", + "decisions": {"run_snippet": False}, + } + assert client_kwargs["base_url"] == "http://localhost:8000" + + def test_escape_round_trip_helper(): assert "[" in _escape("x[y]") or "\\[" in _escape("x[y]")