feat(cli): rewrite input layer with prompt_toolkit and polish UI
- Replaces fragile input() hacks with robust prompt_toolkit.PromptSession - Native support for multiline paste, history, and clean display - Restores animated spinner in _thinking_ctx (now safe) - Replaces boxed Panel with clean header for easier copying - Adds prompt-toolkit dependency - Adds new unit tests for input layer
This commit is contained in:
张涔熙
@@ -1,7 +1,6 @@
|
||||
"""CLI commands for nanobot."""
|
||||
|
||||
import asyncio
|
||||
import atexit
|
||||
import os
|
||||
import signal
|
||||
from pathlib import Path
|
||||
@@ -15,6 +14,11 @@ from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
from rich.text import Text
|
||||
|
||||
from prompt_toolkit import PromptSession
|
||||
from prompt_toolkit.formatted_text import HTML
|
||||
from prompt_toolkit.history import FileHistory
|
||||
from prompt_toolkit.patch_stdout import patch_stdout
|
||||
|
||||
from nanobot import __version__, __logo__
|
||||
|
||||
app = typer.Typer(
|
||||
@@ -27,13 +31,10 @@ console = Console()
|
||||
EXIT_COMMANDS = {"exit", "quit", "/exit", "/quit", ":q"}
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Lightweight CLI input: readline for arrow keys / history, termios for flush
|
||||
# CLI input: prompt_toolkit for editing, paste, history, and display
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_READLINE = None
|
||||
_HISTORY_FILE: Path | None = None
|
||||
_HISTORY_HOOK_REGISTERED = False
|
||||
_USING_LIBEDIT = False
|
||||
_PROMPT_SESSION: PromptSession | None = None
|
||||
_SAVED_TERM_ATTRS = None # original termios settings, restored on exit
|
||||
|
||||
|
||||
@@ -64,15 +65,6 @@ def _flush_pending_tty_input() -> None:
|
||||
return
|
||||
|
||||
|
||||
def _save_history() -> None:
|
||||
if _READLINE is None or _HISTORY_FILE is None:
|
||||
return
|
||||
try:
|
||||
_READLINE.write_history_file(str(_HISTORY_FILE))
|
||||
except Exception:
|
||||
return
|
||||
|
||||
|
||||
def _restore_terminal() -> None:
|
||||
"""Restore terminal to its original state (echo, line buffering, etc.)."""
|
||||
if _SAVED_TERM_ATTRS is None:
|
||||
@@ -84,11 +76,11 @@ def _restore_terminal() -> None:
|
||||
pass
|
||||
|
||||
|
||||
def _enable_line_editing() -> None:
|
||||
"""Enable readline for arrow keys, line editing, and persistent history."""
|
||||
global _READLINE, _HISTORY_FILE, _HISTORY_HOOK_REGISTERED, _USING_LIBEDIT, _SAVED_TERM_ATTRS
|
||||
def _init_prompt_session() -> None:
|
||||
"""Create the prompt_toolkit session with persistent file history."""
|
||||
global _PROMPT_SESSION, _SAVED_TERM_ATTRS
|
||||
|
||||
# Save terminal state before readline touches it
|
||||
# Save terminal state so we can restore it on exit
|
||||
try:
|
||||
import termios
|
||||
_SAVED_TERM_ATTRS = termios.tcgetattr(sys.stdin.fileno())
|
||||
@@ -97,59 +89,22 @@ def _enable_line_editing() -> None:
|
||||
|
||||
history_file = Path.home() / ".nanobot" / "history" / "cli_history"
|
||||
history_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
_HISTORY_FILE = history_file
|
||||
|
||||
try:
|
||||
import readline
|
||||
except ImportError:
|
||||
return
|
||||
|
||||
_READLINE = readline
|
||||
_USING_LIBEDIT = "libedit" in (readline.__doc__ or "").lower()
|
||||
|
||||
try:
|
||||
if _USING_LIBEDIT:
|
||||
readline.parse_and_bind("bind ^I rl_complete")
|
||||
else:
|
||||
readline.parse_and_bind("tab: complete")
|
||||
readline.parse_and_bind("set editing-mode emacs")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
readline.read_history_file(str(history_file))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not _HISTORY_HOOK_REGISTERED:
|
||||
atexit.register(_save_history)
|
||||
_HISTORY_HOOK_REGISTERED = True
|
||||
|
||||
|
||||
def _prompt_text() -> str:
|
||||
"""Build a readline-friendly colored prompt."""
|
||||
if _READLINE is None:
|
||||
return "You: "
|
||||
# libedit on macOS does not honor GNU readline non-printing markers.
|
||||
if _USING_LIBEDIT:
|
||||
return "\033[1;34mYou:\033[0m "
|
||||
return "\001\033[1;34m\002You:\001\033[0m\002 "
|
||||
_PROMPT_SESSION = PromptSession(
|
||||
history=FileHistory(str(history_file)),
|
||||
enable_open_in_editor=False,
|
||||
multiline=False, # Enter submits (single line mode)
|
||||
)
|
||||
|
||||
|
||||
def _print_agent_response(response: str, render_markdown: bool) -> None:
|
||||
"""Render assistant response with consistent terminal styling."""
|
||||
"""Render assistant response with clean, copy-friendly header."""
|
||||
content = response or ""
|
||||
body = Markdown(content) if render_markdown else Text(content)
|
||||
console.print()
|
||||
console.print(
|
||||
Panel(
|
||||
body,
|
||||
title=f"{__logo__} nanobot",
|
||||
title_align="left",
|
||||
border_style="cyan",
|
||||
padding=(0, 1),
|
||||
)
|
||||
)
|
||||
# Use a simple header instead of a Panel box, making it easier to copy text
|
||||
console.print(f"{__logo__} [bold cyan]nanobot[/bold cyan]")
|
||||
console.print(body)
|
||||
console.print()
|
||||
|
||||
|
||||
@@ -159,13 +114,25 @@ def _is_exit_command(command: str) -> bool:
|
||||
|
||||
|
||||
async def _read_interactive_input_async() -> str:
|
||||
"""Read user input with arrow keys and history (runs input() in a thread)."""
|
||||
"""Read user input using prompt_toolkit (handles paste, history, display).
|
||||
|
||||
prompt_toolkit natively handles:
|
||||
- Multiline paste (bracketed paste mode)
|
||||
- History navigation (up/down arrows)
|
||||
- Clean display (no ghost characters or artifacts)
|
||||
"""
|
||||
if _PROMPT_SESSION is None:
|
||||
raise RuntimeError("Call _init_prompt_session() first")
|
||||
try:
|
||||
return await asyncio.to_thread(input, _prompt_text())
|
||||
with patch_stdout():
|
||||
return await _PROMPT_SESSION.prompt_async(
|
||||
HTML("<b fg='ansiblue'>You:</b> "),
|
||||
)
|
||||
except EOFError as exc:
|
||||
raise KeyboardInterrupt from exc
|
||||
|
||||
|
||||
|
||||
def version_callback(value: bool):
|
||||
if value:
|
||||
console.print(f"{__logo__} nanobot v{__version__}")
|
||||
@@ -473,6 +440,7 @@ def agent(
|
||||
if logs:
|
||||
from contextlib import nullcontext
|
||||
return nullcontext()
|
||||
# Animated spinner is safe to use with prompt_toolkit input handling
|
||||
return console.status("[dim]nanobot is thinking...[/dim]", spinner="dots")
|
||||
|
||||
if message:
|
||||
@@ -485,13 +453,10 @@ def agent(
|
||||
asyncio.run(run_once())
|
||||
else:
|
||||
# Interactive mode
|
||||
_enable_line_editing()
|
||||
_init_prompt_session()
|
||||
console.print(f"{__logo__} Interactive mode (type [bold]exit[/bold] or [bold]Ctrl+C[/bold] to quit)\n")
|
||||
|
||||
# input() runs in a worker thread that can't be cancelled.
|
||||
# Without this handler, asyncio.run() would hang waiting for it.
|
||||
def _exit_on_sigint(signum, frame):
|
||||
_save_history()
|
||||
_restore_terminal()
|
||||
console.print("\nGoodbye!")
|
||||
os._exit(0)
|
||||
@@ -508,7 +473,6 @@ def agent(
|
||||
continue
|
||||
|
||||
if _is_exit_command(command):
|
||||
_save_history()
|
||||
_restore_terminal()
|
||||
console.print("\nGoodbye!")
|
||||
break
|
||||
@@ -517,12 +481,10 @@ def agent(
|
||||
response = await agent_loop.process_direct(user_input, session_id)
|
||||
_print_agent_response(response, render_markdown=markdown)
|
||||
except KeyboardInterrupt:
|
||||
_save_history()
|
||||
_restore_terminal()
|
||||
console.print("\nGoodbye!")
|
||||
break
|
||||
except EOFError:
|
||||
_save_history()
|
||||
_restore_terminal()
|
||||
console.print("\nGoodbye!")
|
||||
break
|
||||
|
||||
Reference in New Issue
Block a user