Author SHA1 Message Date
Jon Wingender 35f7461ca4 fix(tui): stop a stream outlived by /new from leaking into the next conversation
package / wheel (pull_request) Waiting to run
_finish_stream appended the completed reply to self.history - whatever
list that name currently pointed at - not to the conversation the
stream was actually answering. /new reassigns self.history to a fresh
list; a stream still running when that happens finished by silently
appending the old conversation's trailing reply onto the new one, which
then rides along in that new conversation's next /chat/stream history
payload. The conversation_id was already captured by closure for this
exact reason (see the tool-denial path); self.history needed the same
treatment.
2026-08-26 14:01:19 -05:00
Athena Kaminsky 9ed2908170 fix(tui): prioritize interrupt and quit keys
package / wheel (pull_request) Waiting to run
Declare Ctrl+C and Ctrl+D as priority Textual bindings so the focused prompt cannot consume them. Drive exit and silent-stream cancellation regressions through Pilot key events instead of calling action handlers directly.
2026-08-26 08:17:31 -05:00
Athena Kaminsky 9ca37057eb 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.
2026-08-26 08:17:31 -05:00
Athena Kaminsky da3509eb04 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.
2026-08-26 08:17:31 -05:00
Athena Kaminsky 6f5094b5fc 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.
2026-08-26 08:17:31 -05:00
Athena Kaminsky 1449280fcd feat(cli): add interactive TUI chat
Add a Textual chat interface with threaded SSE streaming, slash commands, interrupt handling, and bare nexus dispatch. Package it behind the tui extra, document usage, and cover command routing, dependencies, and headless interaction with tests.
2026-08-26 08:17:31 -05:00
8 changed files with 953 additions and 18 deletions
+16 -13
View File
@@ -50,25 +50,28 @@ uvicorn synapse.main:sio_app --host 127.0.0.1 --port 8000 --reload
cd interface/web && npm run dev
```
**Management CLI** (`ncp`) — start/stop services with PID tracking, plus terminal
access to the same features as the web UI (all via the REST API on `:8000`):
**Management CLI** (`nexus` / `ncp`) — start/stop services with PID tracking, plus
terminal access to the same features as the web UI (REST API on `:8000`):
```bash
./management/nexus-cli.sh start # starts backend + frontend
./management/nexus-cli.sh stop
./management/nexus-cli.sh start --backend|-b / --frontend|-f / --memory|-m
# Feature commands (dispatch to nexusos_cli/nexus_api.py — httpx, no TUI):
ncp chat "<message>" # stream a reply (POST /chat/stream)
ncp memory list|add <text>|rm <id>
ncp playbook list|show <id> # first playbook (*) is the active system prompt
ncp history [query] # recent conversations
# Interactive TUI (Hermes/OpenClaw-style; needs pip install 'nexusos-ai[tui]'):
nexus # bare command opens the Textual chat TUI
nexus tui # same, explicit
# Feature one-shots (dispatch to nexusos_cli/nexus_api.py — httpx):
nexus chat send "<message>" # stream a reply (POST /chat/stream)
nexus memory list|add <text>|rm <id>
nexus playbook list|show <id> # first playbook (*) is the active system prompt
nexus history [query] # recent conversations
nexus monitor # ASCII status dashboard (no prompt)
```
The old curses TUIs (`nexus-chat.py`, `nexus-playbook.py`) were removed in favor of
these API-backed subcommands. The CLI covers chat, memory, playbooks, and history;
the web UI and control panel expose the remaining management features.
The CLI itself lives in `nexusos_cli/` (that is what the wheel ships and what
`nexus`/`ncp`/`nexusos` dispatch to); `management/` keeps the desktop-only
pieces — the shell wrappers, the Tk control panel, and the XFCE panel wiring.
The interactive TUI lives in `nexusos_cli/tui_app.py` (Textual, optional extra).
One-shot subcommands and `nexus monitor` remain for scripts. The CLI package is
`nexusos_cli/` (what the wheel ships); `management/` keeps desktop-only pieces —
shell wrappers, Tk control panel, XFCE panel wiring.
`management/controlpanel.py` (tkinter GUI, wired into the XFCE panel via
`bin/panel/nexus-popup.py`) stays.
+3
View File
@@ -37,11 +37,14 @@ and seed playbooks. Extras keep platform-sensitive dependencies optional:
- `desktop`: desktop process support and Windows pywebview
- `search`: DuckDuckGo web search for chat
- `mail`: IMAP mail reading
- `tui`: Textual interactive chat UI (`nexus` with no subcommand)
- `all`: every optional capability at once
## Common commands
```text
nexus Interactive chat TUI (needs nexusos-ai[tui])
nexus tui Same as bare nexus
nexus init Create writable state and seed playbooks
nexus doctor [--fix] [--json] Diagnose the install and provider
nexus paths [--json] Show package, state, and asset locations
+36 -3
View File
@@ -393,6 +393,24 @@ def cmd_monitor(args) -> int:
)
def cmd_tui(args) -> int:
"""Interactive Hermes/OpenClaw-style chat TUI (requires nexusos-ai[tui])."""
if not sys.stdin.isatty() or not sys.stdout.isatty():
print(
"The TUI needs a terminal. Use: nexus chat send \"\"\n"
"Or run `nexus` in an interactive shell.",
file=sys.stderr,
)
return 2
try:
from .tui_app import run_tui
except ImportError as exc:
print(str(exc), file=sys.stderr)
return 2
api = getattr(args, "api_url", None) or settings.api_url
return run_tui(api_url=api)
def _target_flag(target: str | None):
return {
"backend": "--backend",
@@ -689,10 +707,20 @@ def _port(value: str) -> int:
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="nexus", description="NexusOS local AI runtime and API client")
parser = argparse.ArgumentParser(
prog="nexus",
description=(
"NexusOS local AI runtime and API client. "
"With no subcommand, opens the interactive TUI (needs nexusos-ai[tui])."
),
)
parser.add_argument("--version", action="version", version=f"NexusOS {settings.version}")
parser.add_argument("--api-url", help="override the NexusOS backend URL for this command")
sub = parser.add_subparsers(dest="command", required=True)
# Bare `nexus` → TUI. Subcommands remain for scripts and one-shots.
sub = parser.add_subparsers(dest="command", required=False)
p = sub.add_parser("tui", help="interactive chat TUI (default when no subcommand)")
p.set_defaults(fn=cmd_tui)
p = sub.add_parser("init", help="create user state and seed default playbooks"); _add_json(p); p.set_defaults(fn=cmd_init)
p = sub.add_parser("paths", help="show resolved package and writable paths"); _add_json(p); p.set_defaults(fn=cmd_paths)
@@ -793,7 +821,12 @@ def _normalize_legacy_argv(argv) -> list[str]:
def main(argv=None) -> int:
parser = build_parser()
args = parser.parse_args(_normalize_legacy_argv(argv))
argv = _normalize_legacy_argv(argv)
args = parser.parse_args(argv)
if not getattr(args, "command", None):
# Bare `nexus` / `ncp` / `nexusos` → interactive TUI.
args.command = "tui"
args.fn = cmd_tui
if args.command == "config":
if args.action in ("get", "unset") and not args.key:
parser.error(f"config {args.action} requires KEY")
+510
View File
@@ -0,0 +1,510 @@
"""Hermes/OpenClaw-style interactive TUI for NexusOS.
Optional: needs the ``tui`` extra (Textual). Launched by a bare ``nexus`` when
stdin/stdout are a TTY. Classic one-shots (``nexus chat send``, ``nexus monitor``,
``nexus status``, …) stay on the argparse tree.
"""
from __future__ import annotations
import asyncio
import json
import threading
import uuid
from typing import Any
import httpx
from synapse.nexus_config import settings
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():
try:
from textual.app import App
from textual.binding import Binding
from textual.widgets import Footer, Header, Input, RichLog, Static
except ImportError as e: # pragma: no cover - optional extra
raise ImportError(
"The interactive TUI needs the 'tui' extra — "
"pip install 'nexusos-ai[tui]' (or: pip install textual)."
) from e
return App, Binding, Footer, Header, Input, RichLog, Static
def _escape(text: str) -> str:
"""Make model/user text safe for Rich markup widgets.
Rich's own escape is the only version that round-trips. Escaping every
backslash by hand looks equivalent but is not: Rich un-escapes ``\\[`` and
never collapses ``\\\\``, so doubling them puts the doubles on screen -
every Windows path and regex escape in a reply renders wrong.
Imported inside the function so the module still loads without the ``tui``
extra. Rich is not declared in pyproject: Textual depends on it, so it is
present whenever the TUI can run at all, and tests/test_packaging_deps.py
lists it in TRANSITIVE for that reason.
"""
from rich.markup import escape
return escape(text)
def format_user_line(message: str) -> str:
return f"[bold green]you>[/] {_escape(message)}"
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:
snap = collect_snapshot()
svcs = snap.get("services") or {}
api = snap.get("api") or {}
host = snap.get("host") or {}
parts = [f"NexusOS {snap.get('version', '')}"]
for key in ("backend", "memory", "provider"):
info = svcs.get(key) or {}
if key == "provider":
up = bool(info.get("reachable"))
else:
up = bool(info.get("running"))
parts.append(f"{key}={'UP' if up else 'DOWN'}")
if api.get("online"):
parts.append(f"tools={api.get('action_tool_policy') or ''}")
cpu = host.get("cpu_pct")
if cpu is not None:
parts.append(f"cpu={cpu:.0f}%")
chains = snap.get("toolchains") or []
ready = [c["lang"] for c in chains if c.get("ready")]
if ready:
parts.append("run=" + ",".join(ready))
return " · ".join(parts)
def _compact_status(snap: dict | None = None) -> str:
"""One-line strip for the bar under the chat log."""
if snap is None:
snap = collect_snapshot()
host = snap.get("host") or {}
api = snap.get("api") or {}
recent = snap.get("recent_tools") or []
cpu = host.get("cpu_pct")
mem = host.get("mem_pct")
bits = []
if cpu is not None:
bits.append(f"cpu {cpu:.0f}%")
if mem is not None:
bits.append(f"mem {mem:.0f}%")
if api.get("online"):
bits.append(
f"memories={api.get('memories') if api.get('memories') is not None else ''} "
f"chats={api.get('conversations') if api.get('conversations') is not None else ''}"
)
else:
bits.append("api DOWN — nexus start")
if recent:
bits.append("recent " + ", ".join(recent[:4]))
return "".join(bits)
class NexusTUI:
"""Factory so Textual imports stay lazy until run()."""
@staticmethod
def build_app(*, api_url: str | None = None):
App, Binding, Footer, Header, Input, RichLog, Static = _require_textual()
base = (api_url or settings.api_url).rstrip("/")
class AppImpl(App):
CSS = """
Screen { layout: vertical; }
#status {
height: 1;
dock: top;
background: $boost;
color: $text;
padding: 0 1;
}
#strip {
height: 1;
background: $surface;
color: $text-muted;
padding: 0 1;
}
#log {
height: 1fr;
border: tall $accent;
padding: 0 1;
}
#live {
height: auto;
max-height: 8;
padding: 0 1;
color: $text;
}
#prompt { dock: bottom; }
"""
BINDINGS = [
Binding("ctrl+c", "interrupt", "Interrupt", priority=True),
Binding("ctrl+d", "quit", "Quit", priority=True),
]
def __init__(self):
super().__init__()
self.api_url = base
self.conversation_id: str | None = None
self.history: list[dict] = []
self._model: str | None = None
self._busy = False
self._stop_stream = threading.Event()
self._stream_cancel: (
tuple[asyncio.AbstractEventLoop, asyncio.Task] | None
) = None
self._status_lock = threading.Lock()
self._status_pending = False
def compose(self):
# Placeholders only — never collect_snapshot() on the UI thread.
yield Header(show_clock=True)
yield Static("NexusOS …", id="status")
yield RichLog(id="log", highlight=True, markup=True, wrap=True)
yield Static("", id="live")
yield Static("collecting status…", id="strip")
yield Input(
placeholder="Message Nexus… (/help for commands)",
id="prompt",
)
yield Footer()
def on_mount(self) -> None:
self.title = "NexusOS"
self.sub_title = self.api_url
log = self.query_one("#log", RichLog)
log.write("[bold]NexusOS[/] interactive TUI")
log.write(
"Type a message and Enter. "
"Slash: /help /status /new /model /quit"
)
log.write(f"API: {_escape(self.api_url)}")
log.write("")
self._schedule_status_refresh()
self.set_interval(2.0, self._schedule_status_refresh)
self.query_one("#prompt", Input).focus()
def _schedule_status_refresh(self) -> None:
"""Kick a worker; never call collect_snapshot on the event loop."""
with self._status_lock:
if self._status_pending:
return
self._status_pending = True
def worker():
try:
snap = collect_snapshot()
self._call_ui(self._apply_status, snap)
except Exception:
pass
finally:
with self._status_lock:
self._status_pending = False
threading.Thread(target=worker, daemon=True).start()
def _apply_status(self, snap: dict) -> None:
self.query_one("#status", Static).update(_status_line(snap))
self.query_one("#strip", Static).update(_compact_status(snap))
def _call_ui(self, callback, *args) -> None:
"""call_from_thread, but never after quit (avoids CancelledError
traceback garbling the restored shell)."""
if not self.is_running:
return
try:
self.call_from_thread(callback, *args)
except BaseException:
# 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 _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()
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._cancel_stream()
self.query_one("#log", RichLog).write(
"[yellow]▸ interrupt requested[/]"
)
else:
self.exit()
def on_input_submitted(self, event: Input.Submitted) -> None:
text = (event.value or "").strip()
event.input.value = ""
if not text:
return
if text.startswith("/"):
self._handle_slash(text)
return
if self._busy:
self.query_one("#log", RichLog).write(
"[yellow]Still streaming — wait or Ctrl+C to interrupt[/]"
)
return
self._start_chat(text)
def _handle_slash(self, text: str) -> None:
log = self.query_one("#log", RichLog)
cmd, _, rest = text[1:].partition(" ")
cmd = cmd.lower().strip()
rest = rest.strip()
if cmd in ("q", "quit", "exit"):
self.exit()
elif cmd in ("h", "help"):
log.write(
"[bold]/help[/] this list\n"
"[bold]/status[/] refresh service strip\n"
"[bold]/new[/] fresh conversation\n"
"[bold]/model[/] \\[name] pin model for next turns\n"
"[bold]/quit[/] leave the TUI\n"
"One-shot: [dim]nexus chat send \"\"[/]"
)
elif cmd == "status":
self._schedule_status_refresh()
log.write("[dim]refreshing status…[/]")
elif cmd == "new":
self.conversation_id = None
self.history = []
log.write("[bold cyan]— new conversation —[/]")
elif cmd == "model":
if rest:
self._model = rest
log.write(f"[dim]model pinned:[/] {_escape(rest)}")
else:
log.write(
f"[dim]model:[/] {_escape(self._model or '(auto)')}"
)
else:
log.write(
f"[red]unknown command[/] /{_escape(cmd)} — try /help"
)
def _start_chat(self, message: str) -> None:
log = self.query_one("#log", RichLog)
live = self.query_one("#live", Static)
log.write(format_user_line(message))
live.update("[bold blue]nexus>[/] [dim]…[/]")
self._busy = True
self._stop_stream.clear()
if not self.conversation_id:
self.conversation_id = str(uuid.uuid4())
conversation_id = self.conversation_id
# The list object itself, not self.history - /new reassigns
# self.history to a fresh list, and a stream that outlives that
# must keep appending its reply to the conversation it actually
# belongs to, not whatever self.history now points at.
history_ref = self.history
body: dict[str, Any] = {
"message": message,
"conversation_id": conversation_id,
"history": list(history_ref),
}
if self._model:
body["model"] = self._model
history_ref.append({"role": "user", "content": message})
async def stream_worker():
reply_parts: list[str] = []
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:
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 HTTP {resp.status_code}[/] "
f"{_escape(detail)}",
)
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,
f"[red]Backend not reachable at "
f"{_escape(self.api_url)}. Start it: nexus start[/]",
)
except Exception as exc:
self._call_ui(
self._show_error,
f"[red]{_escape(type(exc).__name__)}:[/] "
f"{_escape(str(exc))}",
)
finally:
if self._stream_cancel == (loop, task):
self._stream_cancel = None
text = "".join(reply_parts).strip()
self._call_ui(self._finish_stream, text, history_ref)
threading.Thread(
target=lambda: asyncio.run(stream_worker()), daemon=True
).start()
def _finish_stream(self, text: str, history_ref: list) -> None:
log = self.query_one("#log", RichLog)
live = self.query_one("#live", Static)
try:
if text:
log.write(format_assistant_line(text))
history_ref.append(
{"role": "assistant", "content": text}
)
finally:
# Always clear busy — a MarkupError must not wedge the TUI.
live.update("")
self._busy = False
self._schedule_status_refresh()
return AppImpl()
def run_tui(*, api_url: str | None = None) -> int:
"""Run the Textual app. Returns a process exit code."""
app = NexusTUI.build_app(api_url=api_url)
app.run()
return 0
+2
View File
@@ -42,6 +42,7 @@ mail = ["imap-tools>=1.7,<2"]
# synapse/search.py imports this lazily behind a bare except, so without it
# declared the chat web-search path silently returns nothing.
search = ["duckduckgo-search>=6,<9"]
tui = ["textual>=1.0,<3"]
desktop = [
"psutil>=5.9,<8",
"pywebview>=5,<7; platform_system == 'Windows'",
@@ -64,6 +65,7 @@ all = [
"faster-whisper>=1.1,<2",
"imap-tools>=1.7,<2",
"duckduckgo-search>=6,<9",
"textual>=1.0,<3",
"pywebview>=5,<7; platform_system == 'Windows'",
]
dev = [
+16 -1
View File
@@ -31,8 +31,23 @@ def run_cli(tmp_path: Path, *args: str) -> subprocess.CompletedProcess[str]:
def test_help_exposes_portable_command_tree(tmp_path):
result = run_cli(tmp_path, "--help")
assert result.returncode == 0, result.stderr
for command in ("init", "config", "provider", "doctor", "serve", "models", "chat", "monitor"):
for command in ("init", "config", "provider", "doctor", "serve", "models", "chat", "monitor", "tui"):
assert command in result.stdout
assert "interactive TUI" in result.stdout or "TUI" in result.stdout
def test_bare_nexus_defaults_to_tui_command():
"""No subcommand → TUI entry (Hermes-style). Non-TTY exits 2 without launching."""
from unittest import mock
from nexusos_cli.cli import build_parser, cmd_tui
parser = build_parser()
args = parser.parse_args([])
assert args.command is None # filled in by main()
with mock.patch("sys.stdin.isatty", return_value=False), \
mock.patch("sys.stdout.isatty", return_value=False):
assert cmd_tui(args) == 2
def test_legacy_cli_spellings_remain_compatible():
+2 -1
View File
@@ -29,7 +29,8 @@ DISTRIBUTION_OF = {
}
# Provided by another declared distribution rather than named directly.
TRANSITIVE = {"starlette", "socketio", "engineio"}
# rich: Textual depends on it, so the tui extra already pulls it in.
TRANSITIVE = {"starlette", "socketio", "engineio", "rich"}
# Modules that ship inside this repo.
FIRST_PARTY = {"synapse", "nexusos_cli", "modules", "management", "bin", "tests"}
+368
View File
@@ -0,0 +1,368 @@
"""TUI helpers and headless App.run_test coverage."""
from __future__ import annotations
import asyncio
import threading
import pytest
from rich.text import Text
from nexusos_cli.tui_app import (
_compact_status,
_deny_tool_request,
_escape,
_status_line,
format_assistant_line,
format_user_line,
)
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",
"services": {
"backend": {"running": True},
"memory": {"running": False},
"provider": {"reachable": True},
},
"api": {"online": True, "action_tool_policy": "ask"},
"host": {"cpu_pct": 10.0},
"toolchains": [{"lang": "python", "ready": True}],
}
line = _status_line(snap)
assert "backend=UP" in line
assert "memory=DOWN" in line
assert "provider=UP" in line
assert "tools=ask" in line
assert "run=python" in line
def test_compact_status_handles_api_down():
snap = {
"host": {},
"api": {"online": False},
"recent_tools": [],
}
assert "api DOWN" in _compact_status(snap)
def test_escape_preserves_code_brackets_in_display():
raw = "idx = arr[i] and rng = [a-z]+"
plain = Text.from_markup(format_assistant_line(raw)).plain
assert "arr[i]" in plain
assert "[a-z]+" in plain
# Unescaped markup would drop the bracket contents.
assert plain != "nexus> idx = arr and rng = +"
def test_closing_tag_in_model_output_does_not_raise():
raw = "close with [/] please"
plain = Text.from_markup(format_assistant_line(raw)).plain
assert "[/]" in plain
def test_user_line_escapes_markup():
plain = Text.from_markup(format_user_line("use [bold] please")).plain
assert "[bold]" in plain
def test_finish_stream_markup_does_not_wedge_busy():
"""A stray '[/]' used to raise before _busy=False and lock the TUI forever."""
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._finish_stream("see [/] and arr[i]", app.history)
assert app._busy is False
assert app.history[-1]["content"] == "see [/] and arr[i]"
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("", app.history)
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())
@pytest.mark.parametrize("key", ["ctrl+c", "ctrl+d"])
def test_priority_exit_bindings_reach_app_while_prompt_is_focused(key):
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() as pilot:
assert app.is_running
await pilot.press(key)
await pilot.pause()
assert not app.is_running
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_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
async def __aenter__(self):
return self
async def __aexit__(self, *args):
return None
async def aiter_lines(self):
stream_started.set()
await asyncio.to_thread(release_stream.wait, 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
async def __aenter__(self):
return self
async def __aexit__(self, *args):
return None
def stream(self, *args, **kwargs):
return _StreamResponse()
def _capture_denial(*, conversation_id, **kwargs):
denied_for.append(conversation_id)
return ["run_snippet"]
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")
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_new_mid_stream_does_not_leak_reply_into_next_conversation(monkeypatch):
"""A stream still in flight when /new resets self.history must keep
appending its reply to the conversation it was actually answering, not
whatever self.history now points at - otherwise the old reply's text
silently rides along in the next request's history payload."""
pytest.importorskip("textual")
import nexusos_cli.tui_app as tui_app
stream_started = threading.Event()
release_stream = threading.Event()
class _StreamResponse:
status_code = 200
async def __aenter__(self):
return self
async def __aexit__(self, *args):
return None
async def aiter_lines(self):
stream_started.set()
await asyncio.to_thread(release_stream.wait, 2)
yield 'data: "the old reply"'
yield ""
yield "event: done"
yield "data: {}"
class _StreamClient:
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()
monkeypatch.setattr(tui_app.httpx, "AsyncClient", _StreamClient)
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("first question")
assert await asyncio.to_thread(stream_started.wait, 2)
old_history = app.history
app._handle_slash("/new")
assert app.history is not old_history
release_stream.set()
for _ in range(200):
if not app._busy:
break
await asyncio.sleep(0.01)
assert app._busy is False
# The reply landed on the abandoned conversation's own list...
assert any(m["content"] == "the old reply" for m in old_history)
# ...never on the fresh one /new started.
assert app.history == []
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() as pilot:
app._start_chat("first")
assert await asyncio.to_thread(first_stream_started.wait, 2)
await pilot.press("ctrl+c")
await _wait_until_idle()
log = app.query_one("#log")
assert any("interrupt requested" in line.text for line in log.lines)
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]")