fix(gateway): restore tty signal mode for ctrl-c
This commit is contained in:
@@ -80,6 +80,29 @@ 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."""
|
||||
try:
|
||||
fd = sys.stdin.fileno()
|
||||
if not os.isatty(fd):
|
||||
return
|
||||
except Exception:
|
||||
return
|
||||
|
||||
with suppress(Exception):
|
||||
import termios
|
||||
|
||||
attrs = termios.tcgetattr(fd)
|
||||
lflag = attrs[3]
|
||||
required = termios.ISIG | termios.ICANON | termios.ECHO
|
||||
if (lflag & required) == required:
|
||||
return
|
||||
attrs[3] = lflag | required
|
||||
termios.tcsetattr(fd, termios.TCSANOW, attrs)
|
||||
termios.tcflush(fd, termios.TCIFLUSH)
|
||||
logger.debug("Restored foreground gateway TTY signal mode")
|
||||
|
||||
|
||||
def _install_gateway_shutdown_handlers(
|
||||
loop: asyncio.AbstractEventLoop,
|
||||
shutdown_event: asyncio.Event,
|
||||
@@ -1166,6 +1189,7 @@ def _run_gateway(
|
||||
runtime_tasks: asyncio.Future | None = None
|
||||
runtime_tasks_drained = False
|
||||
shutdown_event = asyncio.Event()
|
||||
_ensure_gateway_tty_signal_mode()
|
||||
restore_shutdown_handlers = _install_gateway_shutdown_handlers(
|
||||
asyncio.get_running_loop(),
|
||||
shutdown_event,
|
||||
|
||||
@@ -109,6 +109,37 @@ 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:
|
||||
try:
|
||||
import os
|
||||
import pty
|
||||
import termios
|
||||
except ImportError: # pragma: no cover - platform without POSIX termios
|
||||
pytest.skip("termios unavailable")
|
||||
|
||||
master_fd, slave_fd = pty.openpty()
|
||||
|
||||
class _Stdin:
|
||||
def fileno(self) -> int:
|
||||
return slave_fd
|
||||
|
||||
try:
|
||||
attrs = termios.tcgetattr(slave_fd)
|
||||
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()
|
||||
|
||||
restored = termios.tcgetattr(slave_fd)
|
||||
assert restored[3] & termios.ISIG
|
||||
assert restored[3] & termios.ICANON
|
||||
assert restored[3] & termios.ECHO
|
||||
finally:
|
||||
os.close(master_fd)
|
||||
os.close(slave_fd)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_paths():
|
||||
"""Mock config/workspace paths for test isolation."""
|
||||
|
||||
Reference in New Issue
Block a user