Jon Wingender
35f7461ca4
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.
511 lines
21 KiB
Python
511 lines
21 KiB
Python
"""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
|
GitNexus · part of the Nexus stack