diff --git a/src/lecode/tui/app.py b/src/lecode/tui/app.py index dd32fb6..5fe6be4 100644 --- a/src/lecode/tui/app.py +++ b/src/lecode/tui/app.py @@ -15,11 +15,12 @@ import asyncio import contextlib import os +import re import sys from contextvars import ContextVar from dataclasses import replace from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, TextIO from prompt_toolkit import Application from prompt_toolkit.buffer import Buffer @@ -29,7 +30,8 @@ from prompt_toolkit.formatted_text import ANSI from prompt_toolkit.input import Input from prompt_toolkit.input.ansi_escape_sequences import ANSI_SEQUENCES -from prompt_toolkit.key_binding import KeyBindings +from prompt_toolkit.input.vt100 import Vt100Input +from prompt_toolkit.key_binding import KeyBindings, KeyPress from prompt_toolkit.keys import Keys from prompt_toolkit.layout import Layout from prompt_toolkit.layout.containers import ConditionalContainer, HSplit, Window @@ -180,18 +182,96 @@ class _PromptBlocked(Exception): """Stop a generated sequence after its submission hook denies a prompt.""" -def _register_shift_enter() -> None: - """Map the Shift+Enter escape sequences to Ctrl-J (newline). +def _register_key_sequences() -> None: + """Map modified keys emitted by the enabled terminal modes. - prompt_toolkit 3.0.53 has no ShiftEnter key: the Kitty sequence - ``ESC [ 13 ; 2 u`` is unmapped and the modifyOtherKeys sequence - ``ESC [ 27 ; 2 ; 13 ~`` maps to c-m — i.e. it would *submit*. Both are - (re)mapped to c-j, which the chatbox binds to "insert newline". The - vt100 parser reads ``ANSI_SEQUENCES`` live, so runtime registration is - enough. Idempotent; the dependency is pinned, so this stays in sync. + prompt_toolkit 3.0.53 does not parse Kitty's CSI-u controls. Enabling + disambiguation changes every Ctrl/Alt key and Escape, not just Shift+Enter. + The vt100 parser reads ``ANSI_SEQUENCES`` live; registration is idempotent. """ ANSI_SEQUENCES.setdefault("\x1b[13;2u", Keys.ControlJ) ANSI_SEQUENCES["\x1b[27;2;13~"] = Keys.ControlJ + ANSI_SEQUENCES.setdefault("\x1b[27u", Keys.Escape) + ANSI_SEQUENCES.setdefault("\x1b[9;2u", Keys.BackTab) + for level in range(4): + ANSI_SEQUENCES.setdefault(f"\x1b[>4;{level}m", Keys.Ignore) + + controls = { + ord(char): Keys(f"c-{char.lower()}") + for char in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ" + } + controls.update( + { + 32: Keys.ControlAt, + 50: Keys.ControlAt, + 51: Keys.Escape, + 52: Keys.ControlBackslash, + 53: Keys.ControlSquareClose, + 54: Keys.ControlCircumflex, + 55: Keys.ControlUnderscore, + 56: Keys.ControlH, + 63: Keys.ControlH, + 64: Keys.ControlAt, + 91: Keys.Escape, + 92: Keys.ControlBackslash, + 93: Keys.ControlSquareClose, + 94: Keys.ControlCircumflex, + 95: Keys.ControlUnderscore, + } + ) + for code, key in controls.items(): + for modifier in (5, 6): + ANSI_SEQUENCES.setdefault(f"\x1b[{code};{modifier}u", key) + ANSI_SEQUENCES.setdefault(f"\x1b[27;{modifier};{code}~", key) + for modifier in (7, 8): + ANSI_SEQUENCES.setdefault(f"\x1b[{code};{modifier}u", (Keys.Escape, key)) + ANSI_SEQUENCES.setdefault(f"\x1b[27;{modifier};{code}~", (Keys.Escape, key)) + + for code in range(32, 127): + char = chr(code) + ANSI_SEQUENCES.setdefault(f"\x1b[27;2;{code}~", char) + for modifier in (3, 4): + key = (Keys.Escape, char.upper() if modifier == 4 else char) + ANSI_SEQUENCES.setdefault(f"\x1b[{code};{modifier}u", key) + ANSI_SEQUENCES.setdefault(f"\x1b[27;{modifier};{code}~", key) + + for code, key in ( + (9, Keys.ControlI), + (13, Keys.ControlM), + (27, Keys.Escape), + (127, Keys.ControlH), + ): + for modifier in (3, 4): + ANSI_SEQUENCES.setdefault(f"\x1b[{code};{modifier}u", (Keys.Escape, key)) + ANSI_SEQUENCES.setdefault(f"\x1b[27;{modifier};{code}~", (Keys.Escape, key)) + for modifier in (5, 6): + ANSI_SEQUENCES.setdefault(f"\x1b[{code};{modifier}u", key) + ANSI_SEQUENCES.setdefault(f"\x1b[27;{modifier};{code}~", key) + ANSI_SEQUENCES.setdefault("\x1b[27;2;9~", Keys.BackTab) + ANSI_SEQUENCES.setdefault("\x1b[27;2;127~", Keys.ControlH) + + +#: Mode sequences that make Shift+Enter distinguishable from plain Enter: +#: ``ESC [ > 1 u`` pushes the Kitty keyboard protocol's disambiguate flag +#: (Kitty, Ghostty, WezTerm, foot, Contour) and ``ESC [ > 4 ; 2 m`` sets +#: xterm's modifyOtherKeys level 2. Terminals without support ignore both; +#: _register_key_sequences maps their modified keys to the existing bindings. +_ENABLE_KEY_MODES = "\x1b[>1u\x1b[>4;2m" +_QUERY_XTERM_MODE = "\x1b[?4m" +# ponytail: if xterm does not answer the query, fall back to its configured default. +# A guaranteed restore would require waiting for a response before enabling the mode. +_DISABLE_KEY_MODES = "\x1b[4m" + + +def _set_key_modes(sequence: str, stream: TextIO | None = None) -> None: + """Write a terminal mode sequence to the real terminal, when there is one.""" + if sys.platform == "win32" or os.environ.get("TERM") == "dumb": + return + stream = stream if stream is not None else sys.stdout + if not stream.isatty(): + return + stream.write(sequence) + stream.flush() class TuiApp: @@ -241,6 +321,7 @@ def __init__( self._last_response = "" #: Active worktree isolation (``/worktree``, ``--worktree``). self._worktree: Any | None = None # WorktreeInfo + self._key_mode_restore = _DISABLE_KEY_MODES self._worktree_manager: Any | None = None # WorktreeManager self._original_cwd: Path | None = None #: Background plan-loop task (``/loop``); while it runs, prompts refuse. @@ -971,7 +1052,7 @@ def _alt_enter(event: Any) -> None: def _newline(event: Any) -> None: # Ctrl-J — and Shift-Enter on terminals that report it (the # kitty/modifyOtherKeys sequences are mapped to c-j in - # _register_shift_enter): insert a newline, never submit. + # _register_key_sequences): insert a newline, never submit. event.current_buffer.insert_text("\n") @kb.add("c-c") @@ -1053,10 +1134,15 @@ def _ctrl_g(event: Any) -> None: buf = event.current_buffer async def _edit() -> None: - edited = await open_in_editor(buf.text) + if not os.environ.get("EDITOR"): + self._feed.info("$EDITOR is not set") + return + try: + _set_key_modes(self._key_mode_restore, self._key_mode_stdout) + edited = await open_in_editor(buf.text) + finally: + _set_key_modes(_ENABLE_KEY_MODES, self._key_mode_stdout) if edited is None: - if not os.environ.get("EDITOR"): - self._feed.info("$EDITOR is not set") return buf.text = edited buf.cursor_position = len(edited) @@ -1157,7 +1243,7 @@ def _arg_no_match(self) -> tuple[SlashCommand, str] | None: return command, "no matching options" def _build_app(self, input: Input | None = None, output: Output | None = None) -> Application: - _register_shift_enter() + _register_key_sequences() draft = self._input_history.load_draft() self._input_area = TextArea( prompt="> ", @@ -1313,7 +1399,7 @@ def menu_footer() -> str: ), picker_menu_visible, ) - return Application( + app = Application( layout=Layout(HSplit([live_area, roster_area, self._chatbox, picker_panel, toolbar])), style=Style.from_dict( { @@ -1331,11 +1417,49 @@ def menu_footer() -> str: input=input, output=output, ) + if isinstance(app.input, Vt100Input): + parser = app.input.vt100_parser + get_match = parser._get_match + feed_key = parser.feed_key_callback + xterm_alt = False + + def match_xterm_unicode(sequence: str): + match = get_match(sequence) + if match is not None: + return match + modified = re.fullmatch(r"\x1b\[27;([234]);([0-9]{1,7})~", sequence) + if modified: + code = int(modified[2]) + if 127 < code <= 0x10FFFF and chr(code).isprintable(): + char = chr(code) + return char if modified[1] == "2" else (Keys.Escape, char) + return None + + def normalize_xterm_shift(key: KeyPress) -> None: + nonlocal xterm_alt + if key.key == Keys.Ignore and key.data.startswith("\x1b[>4;"): + self._key_mode_restore = "\x1b[ int: """Run the interactive loop until quit; returns the exit code.""" + self._key_mode_stdout = sys.stdout + self._key_mode_restore = _DISABLE_KEY_MODES self._hydrate_workers() self._status.git = await self._git.get(self._cwd) self._app = self._build_app(input=input, output=output) @@ -1352,6 +1476,8 @@ async def run(self, *, input: Input | None = None, output: Output | None = None) self._file_lister.prefetch() self._spinner_task = asyncio.ensure_future(self._spinner_loop()) try: + _set_key_modes(_QUERY_XTERM_MODE, self._key_mode_stdout) + _set_key_modes(_ENABLE_KEY_MODES, self._key_mode_stdout) with patch_stdout(raw=True): try: await self._app.run_async() @@ -1363,39 +1489,42 @@ async def run(self, *, input: Input | None = None, output: Output | None = None) self._quit = True print(f"lecode: exiting on {type(e).__name__}", file=sys.stderr) finally: - await self._fire_hook(SESSION_END) - self._spinner_task.cancel() - self._approval.cancel() - self._runtime.ctx.approval_callback = None - if self._worker_manager is not None: - self._worker_manager.confirm = None - self._question.cancel() - self._runtime.ctx.question_callback = None - self.cancel_turn() - if self._turn_task is not None: - await asyncio.gather(self._turn_task, return_exceptions=True) - if self._mcp_task is not None: - await asyncio.gather(self._mcp_task, return_exceptions=True) - self._mcp_task = None - lsp = self._runtime.ctx.extras.get("lsp") - if lsp is not None: - await lsp.shutdown() - background = self._runtime.ctx.extras.get(BACKGROUND_EXTRA) - if background is not None: - await background.shutdown() - if self._worker_manager is not None: - await self._worker_manager.shutdown() - mcp = self._runtime.ctx.extras.get(MCP_EXTRA) - if mcp is not None: - await mcp.shutdown() - self._runtime.close() - if self._input_area is not None and self._input_area.text.strip(): - self._input_history.save_draft(self._input_area.text) - self._app = None - if self._session_lock is not None: - self._session_lock.release() - self._session_lock = None - herdr.release() + try: + _set_key_modes(self._key_mode_restore, self._key_mode_stdout) + finally: + await self._fire_hook(SESSION_END) + self._spinner_task.cancel() + self._approval.cancel() + self._runtime.ctx.approval_callback = None + if self._worker_manager is not None: + self._worker_manager.confirm = None + self._question.cancel() + self._runtime.ctx.question_callback = None + self.cancel_turn() + if self._turn_task is not None: + await asyncio.gather(self._turn_task, return_exceptions=True) + if self._mcp_task is not None: + await asyncio.gather(self._mcp_task, return_exceptions=True) + self._mcp_task = None + lsp = self._runtime.ctx.extras.get("lsp") + if lsp is not None: + await lsp.shutdown() + background = self._runtime.ctx.extras.get(BACKGROUND_EXTRA) + if background is not None: + await background.shutdown() + if self._worker_manager is not None: + await self._worker_manager.shutdown() + mcp = self._runtime.ctx.extras.get(MCP_EXTRA) + if mcp is not None: + await mcp.shutdown() + self._runtime.close() + if self._input_area is not None and self._input_area.text.strip(): + self._input_history.save_draft(self._input_area.text) + self._app = None + if self._session_lock is not None: + self._session_lock.release() + self._session_lock = None + herdr.release() try: self.print_totals() finally: diff --git a/tests/test_tui_app.py b/tests/test_tui_app.py index c5c7a1a..aea66f2 100644 --- a/tests/test_tui_app.py +++ b/tests/test_tui_app.py @@ -1286,6 +1286,218 @@ async def test_pipe_shift_enter_and_ctrl_j_insert_newline(tmp_path, monkeypatch) assert contents[1] == "a\nb" +def test_set_key_modes_skips_non_tty_stdout(monkeypatch): + """No mode sequences when stdout is piped (they'd corrupt the stream).""" + from lecode.tui.app import _ENABLE_KEY_MODES, _set_key_modes + + fake = StringIO() # isatty() is False + monkeypatch.setattr("sys.stdout", fake) + monkeypatch.delenv("TERM", raising=False) + _set_key_modes(_ENABLE_KEY_MODES) + assert fake.getvalue() == "" + + +def test_set_key_modes_skips_dumb_terminal(tmp_path, monkeypatch): + """A tty with TERM=dumb (e.g. an Emacs shell) gets no escape sequences.""" + from lecode.tui.app import _ENABLE_KEY_MODES, _set_key_modes + + class TtyStringIO(StringIO): + def isatty(self): + return True + + fake = TtyStringIO() + monkeypatch.setattr("sys.stdout", fake) + monkeypatch.setenv("TERM", "dumb") + _set_key_modes(_ENABLE_KEY_MODES) + assert fake.getvalue() == "" + + +def test_enabled_key_modes_keep_existing_shortcuts(): + from prompt_toolkit.input.vt100_parser import Vt100Parser + from prompt_toolkit.keys import Keys + + from lecode.tui.app import _ENABLE_KEY_MODES, _register_key_sequences + + _register_key_sequences() + assert "\x1b[>4;2m" in _ENABLE_KEY_MODES + cases = ( + ("\x1b[13;2u", (Keys.ControlJ,)), # Shift-Enter + ("\x1b[99;5u", (Keys.ControlC,)), # Ctrl-C + ("\x1b[100;5u", (Keys.ControlD,)), # Ctrl-D + ("\x1b[106;5u", (Keys.ControlJ,)), # Ctrl-J + ("\x1b[103;5u", (Keys.ControlG,)), # Ctrl-G (editor) + ("\x1b[27u", (Keys.Escape,)), + ("\x1b[13;3u", (Keys.Escape, Keys.ControlM)), # Alt-Enter + ("\x1b[98;3u", (Keys.Escape, "b")), # Alt-B + ("\x1b[27;3;13~", (Keys.Escape, Keys.ControlM)), # xterm Alt-Enter + ("\x1b[27;3;98~", (Keys.Escape, "b")), # xterm Alt-B + ("\x1b[27;2;13~", (Keys.ControlJ,)), # xterm Shift-Enter + ("\x1b[27;2;65~", ("A",)), # xterm Shift-A + ("\x1b[27;2;9~", (Keys.BackTab,)), # xterm Shift-Tab + ("\x1b[27;5;99~", (Keys.ControlC,)), # xterm Ctrl-C + ("\x1b[27;6;99~", (Keys.ControlC,)), # xterm Ctrl-Shift-C + ) + for sequence, expected in cases: + events = [] + parser = Vt100Parser(events.append) + parser.feed(sequence) + parser.flush() + assert tuple(press.key for press in events) == expected, repr(sequence) + + +async def test_run_enables_and_restores_key_modification_modes(tmp_path, monkeypatch): + """On a real terminal the TUI pushes the kitty/modifyOtherKeys modes that + make Shift+Enter arrive as a distinct sequence, and restores on exit.""" + from lecode.tui.app import _DISABLE_KEY_MODES, _ENABLE_KEY_MODES + + assert "\x1b[>4m" in _DISABLE_KEY_MODES # configured xterm default, not hard-coded 0 + + class TtyStringIO(StringIO): + def isatty(self): + return True + + app, _, _ = make_app(tmp_path, monkeypatch, []) + fake = TtyStringIO() + monkeypatch.setattr("sys.stdout", fake) + monkeypatch.setenv("TERM", "xterm-256color") + with create_pipe_input() as inp: + task = asyncio.ensure_future(app.run(input=inp, output=DummyOutput())) + await wait_for(lambda: _ENABLE_KEY_MODES in fake.getvalue()) + inp.send_text("/quit\r") + assert await task == 0 + out = fake.getvalue() + assert _DISABLE_KEY_MODES in out + assert out.index(_ENABLE_KEY_MODES) < out.index(_DISABLE_KEY_MODES) + + +@pytest.mark.parametrize("level", [1, 3]) +async def test_run_restores_dynamic_xterm_mode(tmp_path, monkeypatch, level): + from lecode.tui.app import _ENABLE_KEY_MODES + + class TtyStringIO(StringIO): + def isatty(self): + return True + + app, _, _ = make_app(tmp_path, monkeypatch, []) + fake = TtyStringIO() + monkeypatch.setattr("sys.stdout", fake) + monkeypatch.setenv("TERM", "xterm-256color") + with create_pipe_input() as inp: + task = asyncio.create_task(app.run(input=inp, output=DummyOutput())) + await wait_for(lambda: _ENABLE_KEY_MODES in fake.getvalue()) + response = f"\x1b[>4;{level}m" + inp.send_text(response) # response to the preceding xterm mode query + await wait_for(lambda: app._key_mode_restore == "\x1b[ None: + nonlocal pending while not stop_reading: try: chunk = os.read(master, 65536) @@ -160,7 +162,19 @@ def reader_thread() -> None: return if not chunk: return - stream.feed(chunk.decode("utf-8", errors="replace")) + data = pending + chunk.decode("utf-8", errors="replace") + pending = "" + query = "\x1b[?4m" + while query in data: + before, data = data.split(query, 1) + stream.feed(before) + # pyte does not handle xterm's modifier query; emulate its response. + os.write(master, b"\x1b[>4;0m") + for size in range(len(query) - 1, 0, -1): + if data.endswith(query[:size]): + pending, data = data[-size:], data[:-size] + break + stream.feed(data) # Answer cursor-position requests with the real cursor position. for _ in range(chunk.count(b"\x1b[6n")): os.write(