From 75a72a8b24ed166bd3b6a72c218c4aabc0ef22fd Mon Sep 17 00:00:00 2001 From: Zhengshou Lai Date: Sat, 15 Aug 2026 22:38:15 +0800 Subject: [PATCH] =?UTF-8?q?myclaude=20-r:=20=E4=BF=AE=E5=BF=AB=E9=80=9F?= =?UTF-8?q?=E8=BF=9E=E6=8C=89=E6=96=B9=E5=90=91=E9=94=AE=E8=A2=AB=E9=80=80?= =?UTF-8?q?=E5=87=BA=20+=20=E9=9A=90=E8=97=8F=E5=85=89=E6=A0=87=20+=20?= =?UTF-8?q?=E9=80=80=E5=87=BA=E5=88=A0=E9=99=A4=E8=A1=8C=E4=B8=8D=E7=95=99?= =?UTF-8?q?=E7=A9=BA=E8=A1=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _read_key 改按需读 CSI 序列（方向键读到终止符即返回），连按 \x1b[B 不再被吞进同一序列误判为 ESC 退出 - picker 全程隐藏光标（\x1b[?25l），退出恢复（\x1b[?25h），末尾不再突兀闪现光标 - 退出改用 Delete Line（\x1b[{n}M）删除块区域，替代逐行清空，不再留下 N 个空行 - tests: 新增快速连按方向键回归测试 --- myagents/launcher.py | 69 +++++++++++++++++++++++++-------- tests/test_launcher_sessions.py | 15 +++++++ 2 files changed, 67 insertions(+), 17 deletions(-) diff --git a/myagents/launcher.py b/myagents/launcher.py index c0611d4..e7dba1e 100644 --- a/myagents/launcher.py +++ b/myagents/launcher.py @@ -829,6 +829,12 @@ def _read_key(fd: int, timeout: float = 0.3) -> str: Handles single bytes (q, digits, Enter) and multi-byte escape sequences (arrow keys, PageUp/PageDown). Returns "timeout" when idle. + + Escape sequences are consumed greedily but only up to the terminating + byte (``[A``/``[B``… or ``[5~``/``[6~``) — any further bytes belong to the + next keypress and are left in the buffer. This keeps rapid arrow-key + repeats from being merged into one unknown sequence (which would read as + ESC and quit the picker). """ readable, _, _ = select.select([fd], [], [], timeout) if not readable: @@ -838,20 +844,7 @@ def _read_key(fd: int, timeout: float = 0.3) -> str: except OSError: return "quit" if first == b"\x1b": - seq = first - for _ in range(8): - readable, _, _ = select.select([fd], [], [], 0.1) - if not readable: - break - seq += os.read(fd, 1) - return { - b"\x1b[A": "up", - b"\x1b[B": "down", - b"\x1b[C": "right", - b"\x1b[D": "left", - b"\x1b[5~": "pgup", - b"\x1b[6~": "pgdown", - }.get(seq, "esc") + return _read_escape_seq(fd) if first in (b"\r", b"\n"): return "enter" if first in (b"q", b"Q"): @@ -865,6 +858,41 @@ def _read_key(fd: int, timeout: float = 0.3) -> str: return "other" +def _read_escape_seq(fd: int) -> str: + """Read one CSI escape sequence (``ESC [`` …), returning a symbolic key. + + Reads only the bytes of one sequence: arrow keys are ``ESC [ A-D`` (3 + bytes), PgUp/PgDn are ``ESC [ 5~`` / ``ESC [ 6~`` (4 bytes). It never + reads past the terminator, so rapid back-to-back arrow keys stay distinct + and don't merge into one unknown sequence (which would read as ESC). + """ + bracket = _read_byte_or_none(fd, 0.05) + if bracket != b"[": + return "esc" # bare ESC or unknown prefix + nxt = _read_byte_or_none(fd, 0.05) + if nxt is None: + return "esc" + if nxt in (b"A", b"B", b"C", b"D"): + return {"A": "up", "B": "down", "C": "right", "D": "left"}[nxt.decode()] + if nxt in (b"5", b"6"): + term = _read_byte_or_none(fd, 0.05) + if term == b"~": + return "pgup" if nxt == b"5" else "pgdown" + return "esc" + return "esc" + + +def _read_byte_or_none(fd: int, timeout: float) -> bytes | None: + """Read one byte within ``timeout``, or None if nothing arrives.""" + readable, _, _ = select.select([fd], [], [], timeout) + if not readable: + return None + try: + return os.read(fd, 1) + except OSError: + return None + + def _picker_advance( state: _PickerState, key: str ) -> tuple[_PickerState, str, ClaudeSessionRow | str | None]: @@ -1016,11 +1044,14 @@ def _draw_picker_block(lines: list[str], prev_height: int) -> None: def _clear_picker_block(prev_height: int) -> None: - """Blank the picker block lines before exiting so no residue remains.""" + """Delete the picker block lines before exiting so no residue remains. + + Uses Delete Line (``CSI n M``) rather than blanking lines, so the block + disappears entirely instead of leaving N empty rows below the prompt. + """ if prev_height: sys.stdout.write(f"\x1b[{prev_height}A") - for _ in range(prev_height): - sys.stdout.write("\r\x1b[2K\n") + sys.stdout.write(f"\x1b[{prev_height}M") sys.stdout.flush() @@ -1038,6 +1069,8 @@ def _pick_session( prev_height = 0 try: tty.setraw(fd) + sys.stdout.write("\x1b[?25l") # hide cursor while the picker owns the block + sys.stdout.flush() while True: lines = _picker_block(state, chat_cwd, message) _draw_picker_block(lines, prev_height) @@ -1055,6 +1088,8 @@ def _pick_session( _clear_picker_block(prev_height) return None finally: + sys.stdout.write("\x1b[?25h") # restore cursor on the way out + sys.stdout.flush() termios.tcsetattr(fd, termios.TCSADRAIN, old) diff --git a/tests/test_launcher_sessions.py b/tests/test_launcher_sessions.py index 3ede9dd..aad02ac 100644 --- a/tests/test_launcher_sessions.py +++ b/tests/test_launcher_sessions.py @@ -340,3 +340,18 @@ class TestReadKey: def test_escape(self) -> None: assert _read_key_from_bytes(b"\x1b") == "esc" + + def test_rapid_arrows_not_merged(self) -> None: + """Back-to-back arrow keys stay distinct (no bleed into one ESC).""" + r, w = os.pipe() + try: + # 4 rapid down presses back-to-back, exactly as the terminal sends them + os.write(w, b"\x1b[B" * 4) + keys = [L._read_key(r) for _ in range(4)] + assert keys == ["down"] * 4 + # bare ESC still reads as its own key afterwards + os.write(w, b"\x1b") + assert L._read_key(r) == "esc" + finally: + os.close(r) + os.close(w)