fix(tui): preserve errors and deny gated tools safely

Keep stream failures in the persistent transcript instead of clearing them with the live preview. Use each tool request's capability token to deny actions immediately until the TUI has an interactive approval flow, and cover both behaviors with focused regressions.
This commit is contained in:
Athena Kaminsky
2026-08-26 08:17:31 -05:00
committed by Athena
parent 1449280fcd
commit 6f5094b5fc
2 changed files with 124 additions and 10 deletions
+61
View File
@@ -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]")