From 2c789767280482f38667044f8a3be5102c71dd26 Mon Sep 17 00:00:00 2001 From: Xubin Ren <52506698+Re-bin@users.noreply.github.com> Date: Mon, 13 Jul 2026 13:32:29 +0800 Subject: [PATCH] fix(cli): restore enter before webui prompts --- nanobot/cli/commands.py | 22 ++++++++++++++-------- tests/cli/test_commands.py | 37 +++++++++++++++++++++++++++++++++++-- 2 files changed, 49 insertions(+), 10 deletions(-) diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py index 8ca3bb23..04f81b33 100644 --- a/nanobot/cli/commands.py +++ b/nanobot/cli/commands.py @@ -109,8 +109,8 @@ def _signal_name(signum: int) -> str: return f"signal {signum}" -def _ensure_gateway_tty_signal_mode() -> None: - """Keep foreground gateway Ctrl+C usable even after a raw-mode TTY leak.""" +def _ensure_interactive_tty_mode() -> None: + """Restore interactive line input after a raw-mode TTY leak.""" try: fd = sys.stdin.fileno() if not os.isatty(fd): @@ -122,14 +122,19 @@ def _ensure_gateway_tty_signal_mode() -> None: import termios attrs = termios.tcgetattr(fd) - lflag = attrs[3] - required = termios.ISIG | termios.ICANON | termios.ECHO - if (lflag & required) == required: + required_lflag = termios.ISIG | termios.ICANON | termios.ECHO + blocked_input_flags = getattr(termios, "IGNCR", 0) | getattr(termios, "INLCR", 0) + if ( + (attrs[3] & required_lflag) == required_lflag + and attrs[0] & termios.ICRNL + and not attrs[0] & blocked_input_flags + ): return - attrs[3] = lflag | required + attrs[0] = (attrs[0] | termios.ICRNL) & ~blocked_input_flags + attrs[3] |= required_lflag termios.tcsetattr(fd, termios.TCSANOW, attrs) termios.tcflush(fd, termios.TCIFLUSH) - logger.debug("Restored foreground gateway TTY signal mode") + logger.debug("Restored foreground gateway TTY mode") def _install_gateway_shutdown_handlers( @@ -1406,6 +1411,7 @@ def webui( from nanobot.config.loader import save_config from nanobot.gateway import GatewayRuntime, GatewayRuntimePaths, GatewayStartOptions + _ensure_interactive_tty_mode() config_path = _resolve_webui_config_path(config) created_config = not config_path.exists() if created_config: @@ -2011,7 +2017,7 @@ def _run_gateway( runtime_tasks: asyncio.Future | None = None runtime_tasks_drained = False shutdown_event = asyncio.Event() - _ensure_gateway_tty_signal_mode() + _ensure_interactive_tty_mode() restore_shutdown_handlers = _install_gateway_shutdown_handlers( asyncio.get_running_loop(), shutdown_event, diff --git a/tests/cli/test_commands.py b/tests/cli/test_commands.py index ffe2d16f..04b5efdd 100644 --- a/tests/cli/test_commands.py +++ b/tests/cli/test_commands.py @@ -112,7 +112,7 @@ def test_gateway_signal_handler_first_signal_stops_and_second_forces() -> None: asyncio.run(_run()) -def test_gateway_tty_signal_mode_restores_ctrl_c(monkeypatch) -> None: +def test_interactive_tty_mode_restores_line_input(monkeypatch) -> None: try: import os import pty @@ -128,13 +128,17 @@ def test_gateway_tty_signal_mode_restores_ctrl_c(monkeypatch) -> None: try: attrs = termios.tcgetattr(slave_fd) + attrs[0] &= ~termios.ICRNL + attrs[0] |= termios.IGNCR attrs[3] &= ~(termios.ISIG | termios.ICANON | termios.ECHO) termios.tcsetattr(slave_fd, termios.TCSANOW, attrs) monkeypatch.setattr(cli_commands.sys, "stdin", _Stdin()) - cli_commands._ensure_gateway_tty_signal_mode() + cli_commands._ensure_interactive_tty_mode() restored = termios.tcgetattr(slave_fd) + assert restored[0] & termios.ICRNL + assert not restored[0] & termios.IGNCR assert restored[3] & termios.ISIG assert restored[3] & termios.ICANON assert restored[3] & termios.ECHO @@ -143,6 +147,35 @@ def test_gateway_tty_signal_mode_restores_ctrl_c(monkeypatch) -> None: os.close(slave_fd) +def test_webui_restores_tty_before_loading_config(monkeypatch, tmp_path: Path) -> None: + config_file = tmp_path / "config.json" + config_file.write_text("{}", encoding="utf-8") + calls: list[str] = [] + original_resolve = cli_commands._resolve_webui_config_path + + monkeypatch.setattr( + cli_commands, + "_ensure_interactive_tty_mode", + lambda: calls.append("tty"), + ) + monkeypatch.setattr( + cli_commands, + "_resolve_webui_config_path", + lambda path: calls.append("config") or original_resolve(path), + ) + _patch_webui_provider_ready(monkeypatch) + monkeypatch.setattr(cli_commands, "sync_workspace_templates", lambda _path: None) + monkeypatch.setattr(cli_commands, "_gateway_health_ready", lambda *_args, **_kwargs: False) + monkeypatch.setattr(cli_commands, "_webui_endpoint_reachable", lambda *_args, **_kwargs: False) + monkeypatch.setattr(cli_commands, "_tcp_endpoint_reachable", lambda *_args, **_kwargs: False) + monkeypatch.setattr(cli_commands, "_run_gateway", lambda *_args, **_kwargs: None) + + result = runner.invoke(app, ["webui", "--config", str(config_file), "--yes", "--no-open"]) + + assert result.exit_code == 0 + assert calls[:2] == ["tty", "config"] + + def test_disabled_dream_cursor_only_advances_when_behind(tmp_path) -> None: store = MemoryStore(tmp_path) store.append_history("first")