fix(cli): restore enter before webui prompts
This commit is contained in:
+14
-8
@@ -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,
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user