diff --git a/cli.py b/cli.py index f3b601d88c..aaab6c76bf 100644 --- a/cli.py +++ b/cli.py @@ -1540,9 +1540,30 @@ def _strip_leaked_bracketed_paste_wrappers(text: str) -> str: # that appears when the ESC byte was stripped by a prior filter. _DSR_CPR_ESC_RE = re.compile(r"\x1b\[\d+;\d+R") _DSR_CPR_VISIBLE_RE = re.compile(r"\^\[\[\d+;\d+R") +_SGR_MOUSE_ESC_RE = re.compile(r"\x1b\[<\d+;\d+;\d+[Mm]") +_SGR_MOUSE_VISIBLE_RE = re.compile(r"\^\[\[<\d+;\d+;\d+[Mm]") +# Some terminals/filters can drop ESC and literal "^[[", leaving only +# "4m" # reset modifyOtherKeys + "\x1b[0m" # reset text attributes + "\x1b[?25h" # ensure cursor visible +) -def _strip_leaked_terminal_responses(text: str) -> str: +def _strip_leaked_terminal_responses_with_meta(text: str) -> tuple[str, bool]: """Strip leaked terminal control-response sequences from user input. Covers Cursor Position Report (CPR / DSR) responses — ``ESC[;R`` @@ -1552,12 +1573,39 @@ def _strip_leaked_terminal_responses(text: str) -> str: (resize storms, multiplexer focus changes, slow PTYs) the response lands in the input buffer as literal text and corrupts what the user typed. + + Also strips leaked SGR mouse-report fragments (``ESC[<...M/m`` and + degraded visible forms). Returns ``(cleaned_text, had_mouse_reports)`` + so callers can trigger an in-place terminal mode recovery when needed. """ if not text: - return text - text = _DSR_CPR_ESC_RE.sub("", text) - text = _DSR_CPR_VISIBLE_RE.sub("", text) - return text + return text, False + if not any(marker in text for marker in _TERMINAL_RESPONSE_SENTINELS): + return text, False + + had_mouse_reports = False + + if "\x1b[" in text: + text = _DSR_CPR_ESC_RE.sub("", text) + text, count = _SGR_MOUSE_ESC_RE.subn("", text) + had_mouse_reports = had_mouse_reports or count > 0 + + if "^[" in text: + text = _DSR_CPR_VISIBLE_RE.sub("", text) + text, count = _SGR_MOUSE_VISIBLE_RE.subn("", text) + had_mouse_reports = had_mouse_reports or count > 0 + + if "<" in text: + text, count = _SGR_MOUSE_BARE_RE.subn("", text) + had_mouse_reports = had_mouse_reports or count > 0 + + return text, had_mouse_reports + + +def _strip_leaked_terminal_responses(text: str) -> str: + """Compatibility wrapper returning only cleaned text.""" + cleaned, _ = _strip_leaked_terminal_responses_with_meta(text) + return cleaned def _collect_query_images(query: str | None, image_arg: str | None = None) -> tuple[str, list[Path]]: @@ -1931,6 +1979,8 @@ class HermesCLI: self._stream_box_opened = False # True once the response box header is printed self._reasoning_preview_buf = "" # Coalesce tiny reasoning chunks for [thinking] output self._pending_edit_snapshots = {} + self._last_input_mode_recovery = 0.0 + self._input_mode_recovery_notice_shown = False # Configuration - priority: CLI args > env vars > config file # Model comes from: CLI arg or config.yaml (single source of truth). @@ -4122,6 +4172,37 @@ class HermesCLI: sys.stdout.write(seq) sys.stdout.flush() + def _recover_terminal_input_modes(self, *, reason: str) -> None: + """Best-effort reset when leaked mouse reports indicate mode drift.""" + now = time.monotonic() + # Rate-limit to avoid thrashing if a terminal floods reports. + if now - self._last_input_mode_recovery < 0.5: + return + self._last_input_mode_recovery = now + + out = getattr(self, "_app", None) + output = getattr(out, "output", None) if out else None + try: + if output and hasattr(output, "write_raw"): + output.write_raw(_TERMINAL_INPUT_MODE_RESET_SEQ) + output.flush() + elif output and hasattr(output, "write"): + output.write(_TERMINAL_INPUT_MODE_RESET_SEQ) + output.flush() + else: + sys.stdout.write(_TERMINAL_INPUT_MODE_RESET_SEQ) + sys.stdout.flush() + except Exception: + return + + logger.warning("Recovered terminal input modes after leak: %s", reason) + if not self._input_mode_recovery_notice_shown: + self._input_mode_recovery_notice_shown = True + _cprint( + f" {_DIM}Recovered terminal input modes after leaked mouse reports. " + f"If this repeats, run /new or restart this tab.{_RST}" + ) + def _handle_copy_command(self, cmd_original: str) -> None: """Handle /copy [number] — copy assistant output to clipboard.""" parts = cmd_original.split(maxsplit=1) @@ -10195,7 +10276,9 @@ class HermesCLI: # so the 5-line collapse threshold and display are consistent. pasted_text = pasted_text.replace('\r\n', '\n').replace('\r', '\n') pasted_text = _strip_leaked_bracketed_paste_wrappers(pasted_text) - pasted_text = _strip_leaked_terminal_responses(pasted_text) + pasted_text, _had_mouse_reports = _strip_leaked_terminal_responses_with_meta(pasted_text) + if _had_mouse_reports: + self._recover_terminal_input_modes(reason="mouse reports leaked into bracketed paste payload") if _should_auto_attach_clipboard_image_on_paste(pasted_text) and self._try_attach_clipboard_image(): event.app.invalidate() if pasted_text: @@ -10349,7 +10432,9 @@ class HermesCLI: event so it never triggers this. """ text = _strip_leaked_bracketed_paste_wrappers(buf.text) - text = _strip_leaked_terminal_responses(text) + text, _had_mouse_reports = _strip_leaked_terminal_responses_with_meta(text) + if _had_mouse_reports: + self._recover_terminal_input_modes(reason="mouse reports leaked into prompt buffer") if text != buf.text: cursor = min(buf.cursor_position, len(text)) _paste_just_collapsed[0] = True @@ -11102,7 +11187,9 @@ class HermesCLI: if isinstance(user_input, str): user_input = _strip_leaked_bracketed_paste_wrappers(user_input) - user_input = _strip_leaked_terminal_responses(user_input) + user_input, _had_mouse_reports = _strip_leaked_terminal_responses_with_meta(user_input) + if _had_mouse_reports: + self._recover_terminal_input_modes(reason="mouse reports leaked into submitted input") # Check for commands — but detect dragged/pasted file paths first. # See _detect_file_drop() for details. diff --git a/tests/cli/test_cli_terminal_response_sanitizer.py b/tests/cli/test_cli_terminal_response_sanitizer.py index 469c48edb9..1db16df90b 100644 --- a/tests/cli/test_cli_terminal_response_sanitizer.py +++ b/tests/cli/test_cli_terminal_response_sanitizer.py @@ -55,3 +55,27 @@ class TestStripLeakedTerminalResponses: def test_preserves_multiline_content(self): text = "line 1\n\x1b[53;1Rline 2" assert _strip_leaked_terminal_responses(text) == "line 1\nline 2" + + def test_strips_sgr_mouse_report_esc_form(self): + text = "abc\x1b[<65;1;49Mdef" + assert _strip_leaked_terminal_responses(text) == "abcdef" + + def test_strips_sgr_mouse_report_visible_form(self): + text = "abc^[[<65;1;49Mdef" + assert _strip_leaked_terminal_responses(text) == "abcdef" + + def test_strips_sgr_mouse_report_bare_form(self): + text = "abc<65;1;49Mdef" + assert _strip_leaked_terminal_responses(text) == "abcdef" + + def test_strips_sgr_mouse_report_with_large_coordinates(self): + text = "abc\x1b[<10000;12345;98765Mdef" + assert _strip_leaked_terminal_responses(text) == "abcdef" + + def test_strips_multiple_concatenated_sgr_mouse_reports(self): + text = "<65;1;49M<35;1;42Mhello<64;1;40m" + assert _strip_leaked_terminal_responses(text) == "hello" + + def test_does_not_strip_regular_angle_bracket_text(self): + text = "render
literal" + assert _strip_leaked_terminal_responses(text) == text diff --git a/ui-tui/src/__tests__/terminalModes.test.ts b/ui-tui/src/__tests__/terminalModes.test.ts new file mode 100644 index 0000000000..38ad8fe6a2 --- /dev/null +++ b/ui-tui/src/__tests__/terminalModes.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it, vi } from 'vitest' + +import { resetTerminalModes, TERMINAL_MODE_RESET } from '../lib/terminalModes.js' + +describe('terminal mode reset', () => { + it('includes the sticky input modes Hermes enables', () => { + expect(TERMINAL_MODE_RESET).toContain('\x1b[?1006l') + expect(TERMINAL_MODE_RESET).toContain('\x1b[?1003l') + expect(TERMINAL_MODE_RESET).toContain('\x1b[?1002l') + expect(TERMINAL_MODE_RESET).toContain('\x1b[?1000l') + expect(TERMINAL_MODE_RESET).toContain('\x1b[?1004l') + expect(TERMINAL_MODE_RESET).toContain('\x1b[?2004l') + expect(TERMINAL_MODE_RESET).toContain('\x1b[?1049l') + expect(TERMINAL_MODE_RESET).toContain('\x1b[4m') + }) + + it('writes reset sequence to TTY streams without fds', () => { + const write = vi.fn() + + expect(resetTerminalModes({ isTTY: true, write } as unknown as NodeJS.WriteStream)).toBe(true) + expect(write).toHaveBeenCalledWith(TERMINAL_MODE_RESET) + }) + + it('skips non-TTY streams', () => { + const write = vi.fn() + + expect(resetTerminalModes({ isTTY: false, write } as unknown as NodeJS.WriteStream)).toBe(false) + expect(write).not.toHaveBeenCalled() + }) +}) diff --git a/ui-tui/src/entry.tsx b/ui-tui/src/entry.tsx index bd56c7f0f8..31111d5468 100644 --- a/ui-tui/src/entry.tsx +++ b/ui-tui/src/entry.tsx @@ -9,12 +9,17 @@ import { GatewayClient } from './gatewayClient.js' import { setupGracefulExit } from './lib/gracefulExit.js' import { formatBytes, type HeapDumpResult, performHeapDump } from './lib/memory.js' import { type MemorySnapshot, startMemoryMonitor } from './lib/memoryMonitor.js' +import { resetTerminalModes } from './lib/terminalModes.js' if (!process.stdin.isTTY) { console.log('hermes-tui: no TTY') process.exit(0) } +// Start from a clean slate. If a previous TUI crashed or was kill -9'd, the +// terminal tab can still have mouse/focus/paste modes enabled. +resetTerminalModes() + const gw = new GatewayClient() gw.start() @@ -23,17 +28,27 @@ const dumpNotice = (snap: MemorySnapshot, dump: HeapDumpResult | null) => `hermes-tui: ${snap.level} memory (${formatBytes(snap.heapUsed)}) — auto heap dump → ${dump?.heapPath ?? '(failed)'}\n` setupGracefulExit({ - cleanups: [() => gw.kill()], + cleanups: [ + () => { + resetTerminalModes() + + return gw.kill() + } + ], onError: (scope, err) => { const message = err instanceof Error ? `${err.name}: ${err.message}` : String(err) process.stderr.write(`hermes-tui ${scope}: ${message.slice(0, 2000)}\n`) }, - onSignal: signal => process.stderr.write(`hermes-tui: received ${signal}\n`) + onSignal: signal => { + resetTerminalModes() + process.stderr.write(`hermes-tui: received ${signal}\n`) + } }) const stopMemoryMonitor = startMemoryMonitor({ onCritical: (snap, dump) => { + resetTerminalModes() process.stderr.write(dumpNotice(snap, dump)) process.stderr.write('hermes-tui: exiting to avoid OOM; restart to recover\n') process.exit(137) diff --git a/ui-tui/src/lib/terminalModes.ts b/ui-tui/src/lib/terminalModes.ts new file mode 100644 index 0000000000..7add599892 --- /dev/null +++ b/ui-tui/src/lib/terminalModes.ts @@ -0,0 +1,43 @@ +import { writeSync } from 'node:fs' + +export const TERMINAL_MODE_RESET = + '\x1b[?1006l' + // SGR mouse + '\x1b[?1003l' + // any-motion mouse + '\x1b[?1002l' + // button-motion mouse + '\x1b[?1000l' + // click mouse + '\x1b[?1004l' + // focus events + '\x1b[?2004l' + // bracketed paste + '\x1b[?1049l' + // alternate screen + '\x1b[4m' + // modifyOtherKeys + '\x1b[0m' + // attributes + '\x1b[?25h' // cursor visible + +type ResettableStream = Pick & { + fd?: number +} + +export function resetTerminalModes(stream: ResettableStream = process.stdout): boolean { + if (!stream.isTTY) { + return false + } + + const fd = typeof stream.fd === 'number' ? stream.fd : stream === process.stdout ? 1 : undefined + if (fd !== undefined) { + try { + writeSync(fd, TERMINAL_MODE_RESET) + + return true + } catch { + // Fall through to stream.write for mocked or unusual TTY streams. + } + } + + try { + stream.write(TERMINAL_MODE_RESET) + + return true + } catch { + return false + } +}