Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 64 additions & 0 deletions cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -872,6 +872,17 @@ def _cleanup_all_browsers(*args, **kwargs):
# 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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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)
# =============================================================================
Expand Down Expand Up @@ -15096,6 +15157,9 @@ def new_event_loop(self):
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
Expand Down
212 changes: 212 additions & 0 deletions tests/cli/test_tui_terminal_reset_on_exit.py
Original file line number Diff line number Diff line change
@@ -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