mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-06-09 08:21:50 +00:00
Merge pull request #36864 from maxmilian/fix/tui-reset-terminal-input-modes-on-exit
fix(cli): reset terminal input modes on TUI exit to stop focus/mouse leaks
This commit is contained in:
commit
a6b6afdff4
2 changed files with 276 additions and 0 deletions
64
cli.py
64
cli.py
|
|
@ -872,6 +872,17 @@ _cleanup_done = False
|
|||
# Weak reference to the active AIAgent for memory provider shutdown at exit
|
||||
_active_agent_ref = None
|
||||
_deferred_agent_startup_done = False
|
||||
# Set True once the TUI's prompt_toolkit app starts (which enables focus
|
||||
# reporting + mouse tracking). Gates the on-exit terminal reset so non-TUI
|
||||
# one-shot CLI runs — which also register _run_cleanup via atexit — don't emit
|
||||
# escape codes for modes they never enabled (#36823).
|
||||
_tui_input_modes_active = False
|
||||
|
||||
|
||||
def _mark_tui_input_modes_active() -> None:
|
||||
"""Record that the TUI app started, so _run_cleanup resets input modes."""
|
||||
global _tui_input_modes_active
|
||||
_tui_input_modes_active = True
|
||||
|
||||
|
||||
def _prepare_deferred_agent_startup() -> None:
|
||||
|
|
@ -927,6 +938,12 @@ def _run_cleanup():
|
|||
return
|
||||
_cleanup_done = True
|
||||
|
||||
# Reset terminal input modes first, before the slower resource teardown
|
||||
# below (MCP / browser / memory shutdown can take seconds). On Ctrl+C the
|
||||
# user's terminal becomes usable immediately, and a later step raising
|
||||
# can't skip the reset (#36823). No-op unless the TUI actually ran.
|
||||
_reset_terminal_input_modes_on_exit()
|
||||
|
||||
try:
|
||||
_cleanup_all_terminals()
|
||||
except Exception:
|
||||
|
|
@ -972,6 +989,50 @@ def _run_cleanup():
|
|||
pass
|
||||
|
||||
|
||||
def _reset_terminal_input_modes_on_exit() -> None:
|
||||
"""Best-effort: disable focus reporting + mouse tracking on TUI exit so they
|
||||
don't leak into the next shell session sharing the tab.
|
||||
|
||||
prompt_toolkit restores these on a clean teardown, but Ctrl+C, SIGTERM /
|
||||
SIGHUP and crashes can bypass its unwind, leaving the modes enabled. The
|
||||
terminal then emits raw ``ESC[I`` / ``ESC[O`` focus events and fragmented
|
||||
SGR mouse reports as visible text in whatever runs next in the same tab
|
||||
(#36823). Called from ``_run_cleanup`` (atexit-registered + invoked on the
|
||||
normal / EOF / interrupt exit paths) this covers normal quit, Ctrl+C and
|
||||
SIGTERM/SIGHUP. ``kill -9`` is uncatchable, and the kanban worker's
|
||||
``os._exit(0)`` path bypasses ``atexit``; neither runs this — but both are
|
||||
non-TTY / non-TUI, so there is nothing to reset there.
|
||||
|
||||
Gated on ``_tui_input_modes_active`` so one-shot non-TUI CLI runs (which
|
||||
share ``_run_cleanup`` via ``atexit``) never emit these codes. Writes to the
|
||||
controlling terminal directly: by exit, prompt_toolkit's own output is torn
|
||||
down, so ``sys.stdout`` is the real fd; falls back to ``/dev/tty`` when
|
||||
stdout is redirected away from the terminal.
|
||||
"""
|
||||
global _tui_input_modes_active
|
||||
if not _tui_input_modes_active:
|
||||
return
|
||||
# About to disable the modes — clear the flag so a re-armed _run_cleanup (or
|
||||
# a long-lived process that reuses it) doesn't re-emit them.
|
||||
_tui_input_modes_active = False
|
||||
# Prefer stdout when it's the terminal; otherwise the TUI may have driven
|
||||
# /dev/tty while stdout was redirected — reset there instead of nowhere.
|
||||
try:
|
||||
stream = sys.stdout
|
||||
if stream is not None and stream.isatty():
|
||||
stream.write(_TERMINAL_INPUT_MODE_RESET_SEQ)
|
||||
stream.flush()
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
with open("/dev/tty", "w", encoding="ascii") as tty:
|
||||
tty.write(_TERMINAL_INPUT_MODE_RESET_SEQ)
|
||||
tty.flush()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Git Worktree Isolation (#652)
|
||||
# =============================================================================
|
||||
|
|
@ -15135,6 +15196,9 @@ class HermesCLI:
|
|||
pass # No running loop -- nothing to patch
|
||||
except Exception:
|
||||
pass
|
||||
# The app enables focus reporting + mouse tracking; record that
|
||||
# so _run_cleanup resets them on exit (#36823).
|
||||
_mark_tui_input_modes_active()
|
||||
app.run()
|
||||
except (EOFError, KeyboardInterrupt, BrokenPipeError):
|
||||
pass
|
||||
|
|
|
|||
212
tests/cli/test_tui_terminal_reset_on_exit.py
Normal file
212
tests/cli/test_tui_terminal_reset_on_exit.py
Normal file
|
|
@ -0,0 +1,212 @@
|
|||
"""Regression tests for GitHub #36823 — the TUI must reset terminal input
|
||||
modes on exit so focus-reporting / mouse-tracking escape sequences don't leak
|
||||
into the next shell session sharing the tab.
|
||||
|
||||
prompt_toolkit restores these on a clean teardown, but Ctrl+C, SIGTERM/SIGHUP
|
||||
and crashes can bypass its unwind. ``_run_cleanup`` (the once-only cleanup that
|
||||
runs on every catchable exit path, including ``atexit``) now emits the disable
|
||||
sequence as its first step via ``_reset_terminal_input_modes_on_exit`` — gated
|
||||
on ``_tui_input_modes_active`` so non-TUI one-shot CLI runs (which share
|
||||
``_run_cleanup`` via ``atexit``) don't emit codes for modes they never set.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
from unittest.mock import mock_open, patch
|
||||
|
||||
|
||||
def _import_cli():
|
||||
import hermes_cli.config as config_mod
|
||||
|
||||
if not hasattr(config_mod, "save_env_value_secure"):
|
||||
config_mod.save_env_value_secure = lambda key, value: {
|
||||
"success": True,
|
||||
"stored_as": key,
|
||||
"validated": False,
|
||||
}
|
||||
|
||||
import cli as cli_mod
|
||||
|
||||
return cli_mod
|
||||
|
||||
|
||||
class _FakeStream:
|
||||
def __init__(self, isatty: bool = True):
|
||||
self._isatty = isatty
|
||||
self.written: list[str] = []
|
||||
self.flushed = 0
|
||||
|
||||
def isatty(self) -> bool:
|
||||
return self._isatty
|
||||
|
||||
def write(self, s: str) -> int:
|
||||
self.written.append(s)
|
||||
return len(s)
|
||||
|
||||
def flush(self) -> None:
|
||||
self.flushed += 1
|
||||
|
||||
|
||||
class TestResetTerminalInputModes(unittest.TestCase):
|
||||
def test_emits_reset_seq_on_tty_when_tui_ran(self):
|
||||
cli_mod = _import_cli()
|
||||
fake = _FakeStream(isatty=True)
|
||||
with (
|
||||
patch.object(cli_mod, "_tui_input_modes_active", True),
|
||||
patch.object(cli_mod.sys, "stdout", fake),
|
||||
):
|
||||
cli_mod._reset_terminal_input_modes_on_exit()
|
||||
|
||||
written = "".join(fake.written)
|
||||
self.assertEqual(written, cli_mod._TERMINAL_INPUT_MODE_RESET_SEQ)
|
||||
self.assertGreaterEqual(fake.flushed, 1)
|
||||
# The focus-reporting disable is the specific leak the issue reports.
|
||||
self.assertIn("\x1b[?1004l", written)
|
||||
|
||||
def test_noop_when_tui_never_ran(self):
|
||||
"""Non-TUI one-shot CLI runs share _run_cleanup via atexit — they must
|
||||
not emit terminal escape codes they never needed (review finding #1)."""
|
||||
cli_mod = _import_cli()
|
||||
fake = _FakeStream(isatty=True)
|
||||
with (
|
||||
patch.object(cli_mod, "_tui_input_modes_active", False),
|
||||
patch.object(cli_mod.sys, "stdout", fake),
|
||||
# Guard: must not touch the real /dev/tty either.
|
||||
patch("builtins.open", mock_open()) as m_open,
|
||||
):
|
||||
cli_mod._reset_terminal_input_modes_on_exit()
|
||||
|
||||
self.assertEqual(fake.written, [])
|
||||
m_open.assert_not_called()
|
||||
|
||||
def test_noop_when_not_a_tty_and_no_dev_tty(self):
|
||||
"""stdout redirected and /dev/tty unavailable → nothing written, no raise."""
|
||||
cli_mod = _import_cli()
|
||||
fake = _FakeStream(isatty=False)
|
||||
with (
|
||||
patch.object(cli_mod, "_tui_input_modes_active", True),
|
||||
patch.object(cli_mod.sys, "stdout", fake),
|
||||
patch("builtins.open", side_effect=OSError("no /dev/tty")),
|
||||
):
|
||||
cli_mod._reset_terminal_input_modes_on_exit()
|
||||
|
||||
self.assertEqual(fake.written, [], "must not pollute the redirected stream")
|
||||
|
||||
def test_falls_back_to_dev_tty_when_stdout_redirected(self):
|
||||
"""When stdout isn't the terminal, reset via /dev/tty (issue's own
|
||||
suggestion) so a TUI that drove /dev/tty still gets cleaned up."""
|
||||
cli_mod = _import_cli()
|
||||
fake = _FakeStream(isatty=False)
|
||||
m_open = mock_open()
|
||||
with (
|
||||
patch.object(cli_mod, "_tui_input_modes_active", True),
|
||||
patch.object(cli_mod.sys, "stdout", fake),
|
||||
patch("builtins.open", m_open),
|
||||
):
|
||||
cli_mod._reset_terminal_input_modes_on_exit()
|
||||
|
||||
self.assertEqual(fake.written, [])
|
||||
m_open.assert_called_once_with("/dev/tty", "w", encoding="ascii")
|
||||
m_open().write.assert_called_once_with(cli_mod._TERMINAL_INPUT_MODE_RESET_SEQ)
|
||||
|
||||
def test_swallows_stdout_errors(self):
|
||||
cli_mod = _import_cli()
|
||||
|
||||
class _Boom:
|
||||
def isatty(self):
|
||||
raise OSError("stdout closed")
|
||||
|
||||
with (
|
||||
patch.object(cli_mod, "_tui_input_modes_active", True),
|
||||
patch.object(cli_mod.sys, "stdout", _Boom()),
|
||||
patch("builtins.open", side_effect=OSError("no /dev/tty")),
|
||||
):
|
||||
# Cleanup runs at process teardown — it must never raise.
|
||||
cli_mod._reset_terminal_input_modes_on_exit()
|
||||
|
||||
def test_mark_tui_input_modes_active_sets_flag(self):
|
||||
cli_mod = _import_cli()
|
||||
original = cli_mod._tui_input_modes_active
|
||||
cli_mod._tui_input_modes_active = False
|
||||
try:
|
||||
cli_mod._mark_tui_input_modes_active()
|
||||
self.assertTrue(cli_mod._tui_input_modes_active)
|
||||
finally:
|
||||
cli_mod._tui_input_modes_active = original
|
||||
|
||||
def test_flag_cleared_after_reset(self):
|
||||
"""Once the modes are disabled they are no longer active — the flag must
|
||||
flip back so a re-armed cleanup doesn't re-emit the sequence."""
|
||||
cli_mod = _import_cli()
|
||||
fake = _FakeStream(isatty=True)
|
||||
original = cli_mod._tui_input_modes_active
|
||||
cli_mod._tui_input_modes_active = True
|
||||
try:
|
||||
with patch.object(cli_mod.sys, "stdout", fake):
|
||||
cli_mod._reset_terminal_input_modes_on_exit()
|
||||
self.assertIn("\x1b[?1004l", "".join(fake.written))
|
||||
self.assertFalse(
|
||||
cli_mod._tui_input_modes_active, "flag must clear after reset"
|
||||
)
|
||||
finally:
|
||||
cli_mod._tui_input_modes_active = original
|
||||
|
||||
|
||||
class TestRunCleanupWiring(unittest.TestCase):
|
||||
"""_run_cleanup must call the reset, as its first step, on every invocation
|
||||
— even if a later cleanup step raises."""
|
||||
|
||||
def _run_cleanup_isolated(self, cli_mod, **extra_patches):
|
||||
"""Invoke _run_cleanup with heavy/real teardown steps stubbed out so the
|
||||
test is hermetic (review finding #5)."""
|
||||
original_done = cli_mod._cleanup_done
|
||||
cli_mod._cleanup_done = False
|
||||
patches = {
|
||||
"_cleanup_all_terminals": lambda: None,
|
||||
"_cleanup_all_browsers": lambda: None,
|
||||
}
|
||||
try:
|
||||
with (
|
||||
patch.object(
|
||||
cli_mod, "_reset_terminal_input_modes_on_exit"
|
||||
) as mock_reset,
|
||||
patch.object(
|
||||
cli_mod, "_cleanup_all_terminals", patches["_cleanup_all_terminals"]
|
||||
),
|
||||
patch.object(
|
||||
cli_mod, "_cleanup_all_browsers", patches["_cleanup_all_browsers"]
|
||||
),
|
||||
patch("tools.mcp_tool.shutdown_mcp_servers", lambda *a, **k: None),
|
||||
patch(
|
||||
"agent.auxiliary_client.shutdown_cached_clients",
|
||||
lambda *a, **k: None,
|
||||
),
|
||||
patch("hermes_cli.plugins.invoke_hook", lambda *a, **k: None),
|
||||
):
|
||||
if extra_patches.get("terminals_raise"):
|
||||
with patch.object(
|
||||
cli_mod,
|
||||
"_cleanup_all_terminals",
|
||||
side_effect=RuntimeError("boom"),
|
||||
):
|
||||
cli_mod._run_cleanup()
|
||||
else:
|
||||
cli_mod._run_cleanup()
|
||||
return mock_reset
|
||||
finally:
|
||||
cli_mod._cleanup_done = original_done
|
||||
|
||||
def test_run_cleanup_calls_reset(self):
|
||||
cli_mod = _import_cli()
|
||||
mock_reset = self._run_cleanup_isolated(cli_mod)
|
||||
mock_reset.assert_called_once()
|
||||
|
||||
def test_reset_runs_even_when_a_cleanup_step_raises(self):
|
||||
"""The reset is the first step, so a failing teardown step can't skip
|
||||
it — covering the Ctrl+C / crash paths the issue is about."""
|
||||
cli_mod = _import_cli()
|
||||
mock_reset = self._run_cleanup_isolated(cli_mod, terminals_raise=True)
|
||||
mock_reset.assert_called_once()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue