fix(cli): keep prompt stable and flush stale arrow-key input
This commit is contained in:
parent
438ec66fd8
commit
8b1ef77970
@ -1,7 +1,10 @@
|
||||
"""CLI commands for nanobot."""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
from pathlib import Path
|
||||
import select
|
||||
import sys
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
@ -18,6 +21,40 @@ app = typer.Typer(
|
||||
console = Console()
|
||||
|
||||
|
||||
def _flush_pending_tty_input() -> None:
|
||||
"""Drop unread keypresses typed while the model was generating output."""
|
||||
try:
|
||||
fd = sys.stdin.fileno()
|
||||
if not os.isatty(fd):
|
||||
return
|
||||
except Exception:
|
||||
return
|
||||
|
||||
try:
|
||||
import termios
|
||||
|
||||
termios.tcflush(fd, termios.TCIFLUSH)
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
while True:
|
||||
ready, _, _ = select.select([fd], [], [], 0)
|
||||
if not ready:
|
||||
break
|
||||
if not os.read(fd, 4096):
|
||||
break
|
||||
except Exception:
|
||||
return
|
||||
|
||||
|
||||
def _read_interactive_input() -> str:
|
||||
"""Read user input with a stable prompt for terminal line editing."""
|
||||
console.print("[bold blue]You:[/bold blue] ", end="")
|
||||
return input()
|
||||
|
||||
|
||||
def version_callback(value: bool):
|
||||
if value:
|
||||
console.print(f"{__logo__} nanobot v{__version__}")
|
||||
@ -318,7 +355,8 @@ def agent(
|
||||
async def run_interactive():
|
||||
while True:
|
||||
try:
|
||||
user_input = console.input("[bold blue]You:[/bold blue] ")
|
||||
_flush_pending_tty_input()
|
||||
user_input = _read_interactive_input()
|
||||
if not user_input.strip():
|
||||
continue
|
||||
|
||||
|
||||
37
tests/test_cli_input_minimal.py
Normal file
37
tests/test_cli_input_minimal.py
Normal file
@ -0,0 +1,37 @@
|
||||
import builtins
|
||||
|
||||
import nanobot.cli.commands as commands
|
||||
|
||||
|
||||
def test_read_interactive_input_uses_plain_input(monkeypatch) -> None:
|
||||
captured: dict[str, object] = {}
|
||||
|
||||
def fake_print(*args, **kwargs):
|
||||
captured["printed"] = args
|
||||
captured["print_kwargs"] = kwargs
|
||||
|
||||
def fake_input(prompt: str = "") -> str:
|
||||
captured["prompt"] = prompt
|
||||
return "hello"
|
||||
|
||||
monkeypatch.setattr(commands.console, "print", fake_print)
|
||||
monkeypatch.setattr(builtins, "input", fake_input)
|
||||
|
||||
value = commands._read_interactive_input()
|
||||
|
||||
assert value == "hello"
|
||||
assert captured["prompt"] == ""
|
||||
assert captured["print_kwargs"] == {"end": ""}
|
||||
assert captured["printed"] == ("[bold blue]You:[/bold blue] ",)
|
||||
|
||||
|
||||
def test_flush_pending_tty_input_skips_non_tty(monkeypatch) -> None:
|
||||
class FakeStdin:
|
||||
def fileno(self) -> int:
|
||||
return 0
|
||||
|
||||
monkeypatch.setattr(commands.sys, "stdin", FakeStdin())
|
||||
monkeypatch.setattr(commands.os, "isatty", lambda _fd: False)
|
||||
|
||||
commands._flush_pending_tty_input()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user