AthenaandClaude Sonnet 5
1183ab5282
Adds synapse/slash_commands.py: a chat message that's nothing but /tool_name(arg=val, arg=val) dispatches straight through tools.dispatch(), skipping model selection, RAG/playbook context assembly, and the ask-policy approval round-trip entirely. A human typing this IS the approval - there's no one else to ask - so it's a deliberate, reviewed bypass of the approval step specifically, not of anything a tool validates internally (path boundaries, size caps, Curry's own sandbox checks all still run). Argument values parse via ast.literal_eval only: strings/numbers/bools/None/literal containers, no names, no calls, no attribute access - a malformed or hostile-looking argument fails to parse rather than executing anything. Wired into chat_stream_endpoint (main.py) as an early short-circuit, before any of the RAG/model-selection work that a slash-command doesn't need. Web needed no changes (it already forwards raw text unchanged); the TUI previously swallowed every leading "/" locally and never reached the backend with it, so tui_app.py's _handle_slash now falls through to _start_chat for anything shaped like a tool call while still handling its own local meta-commands (/help, /model, /new, ...) exactly as before. Also finally wires Curry in as ten real tools (curry_declare_constant, curry_get_constant/_latest, curry_list_constants, curry_retire_constant, curry_declare_function, curry_get_function, curry_list_functions, curry_call_function, curry_retire_function) - deferred from the vendoring pass. The five write/execute ones are ACTION tools in the same always-ask-regardless-of-global-policy floor as edit_source (ALWAYS_ASK_ACTION_TOOLS, generalized in tools.py from the old self_edit-only ALWAYS_ASK_TOOLS so future tool families share one place to register into). curry_call_function is gated as an action for the same reason run_snippet is: it executes code, even sandboxed. Fixed a real bug surfaced while wiring this up: curry_db is a long-lived singleton holding one sqlite3 connection (unlike NexusOS's own memory store, which opens/closes a fresh connection per call specifically to dodge this), and sqlite3 forbids using a connection from a different thread than created it. That's a non-issue in production (uvicorn's single event-loop thread), but Starlette's TestClient runs the ASGI app through an anyio portal thread, so it broke immediately under test. Fixed at the source (curry_core.py, Curry.__init__) with check_same_thread=False, documented as a second deliberate vendoring deviation alongside the PR #4 sandbox fix - there was never real concurrent access here, just an overly strict same-thread assertion tripping on a thread-identity change with only one logical caller. Verified: 244 backend tests pass (18 new for the parser + endpoint wiring + curry tool registration, 4 new for the TUI passthrough); the 12 pre-existing C/C++/Rust toolchain failures are unrelated and unchanged. Confirmed by hand over the real HTTP endpoint: successful dispatch, zero tool_request events (approval bypass working as designed), a format()-dunder exploit attempt still rejected by the vendored sandbox fix even through the new tool registration, malformed arguments rejected before ever reaching dispatch, and an unknown tool name rejected cleanly. Wheel rebuilt and content-checked (bin/check.sh's gate now also asserts slash_commands.py ships). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
520 lines
22 KiB
Python
520 lines
22 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 synapse.slash_commands import parse_slash_command
|
|
|
|
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)')}"
|
|
)
|
|
elif parse_slash_command(text) is not None:
|
|
# Shaped like /tool_name(arg=val, ...) rather than one of
|
|
# the local meta-commands above — not handled here, sent
|
|
# to the backend as-is. chat_stream_endpoint recognizes
|
|
# and dispatches it directly (see synapse/slash_commands.py);
|
|
# a malformed one still goes through so the user sees the
|
|
# backend's own error, with full context, in one place.
|
|
if self._busy:
|
|
log.write(
|
|
"[yellow]Still streaming — wait or Ctrl+C to interrupt[/]"
|
|
)
|
|
else:
|
|
self._start_chat(text)
|
|
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
|
|
body: dict[str, Any] = {
|
|
"message": message,
|
|
"conversation_id": conversation_id,
|
|
"history": list(self.history),
|
|
}
|
|
if self._model:
|
|
body["model"] = self._model
|
|
self.history.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)
|
|
|
|
threading.Thread(
|
|
target=lambda: asyncio.run(stream_worker()), daemon=True
|
|
).start()
|
|
|
|
def _finish_stream(self, text: str) -> None:
|
|
log = self.query_one("#log", RichLog)
|
|
live = self.query_one("#live", Static)
|
|
try:
|
|
if text:
|
|
log.write(format_assistant_line(text))
|
|
self.history.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