Skip to content
Merged
227 changes: 178 additions & 49 deletions src/lecode/tui/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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[<u\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:
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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="> ",
Expand Down Expand Up @@ -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(
{
Expand All @@ -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[<u" + key.data
return
if xterm_alt:
xterm_alt = False
if len(key.key) == 1 and not key.data:
key = KeyPress(key.key, key.key)
elif key.key == Keys.Escape and key.data.startswith(("\x1b[27;3;", "\x1b[27;4;")):
xterm_alt = True
if key.data.startswith("\x1b[27;2;") and len(key.key) == 1:
key = KeyPress(key.key, key.key)
feed_key(key)

parser._get_match = match_xterm_unicode
parser.feed_key_callback = normalize_xterm_shift
return app

# -- the driver -----------------------------------------------------------

async def run(self, *, input: Input | None = None, output: Output | None = None) -> 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)
Expand All @@ -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()
Expand All @@ -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:
Expand Down
Loading
Loading