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
104 changes: 88 additions & 16 deletions hermes_cli/curses_ui.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,78 @@ def flush_stdin() -> None:
pass


# Normalized menu actions returned by ``read_menu_key``. Using sentinels keeps
# every menu's key-handling branch identical and free of raw escape-byte logic.
NAV_UP = "up"
NAV_DOWN = "down"
NAV_SELECT = "select"
NAV_TOGGLE = "toggle"
NAV_CANCEL = "cancel"
NAV_NONE = "none"


def read_menu_key(stdscr) -> str:
"""Read one keypress and normalize it to a menu action.

Decodes raw arrow-key escape sequences in addition to the translated
``curses.KEY_*`` values. Even with ``keypad(True)`` (which
``curses.wrapper`` sets), some terminals/terminfo entries deliver cursor
keys as raw CSI/SS3 byte sequences — ``getch()`` then returns ``27`` (ESC)
followed by e.g. ``[`` ``A``. Treating that leading ``27`` as a cancel is
what made the setup wizard's provider/model pickers bail to the numbered
fallback the moment a user pressed up/down.

Returns one of the ``NAV_*`` constants. A lone ESC (no continuation byte
within a short window) is the only thing that maps to ``NAV_CANCEL`` via
the escape path; ``q`` also cancels. Unknown sequences map to
``NAV_NONE`` so the caller simply ignores them rather than misfiring.
"""
import curses

key = stdscr.getch()

if key in (curses.KEY_UP, ord("k")):
return NAV_UP
if key in (curses.KEY_DOWN, ord("j")):
return NAV_DOWN
if key in (curses.KEY_ENTER, 10, 13):
return NAV_SELECT
if key == ord(" "):
return NAV_TOGGLE
if key == ord("q"):
return NAV_CANCEL

if key == 27: # ESC — could be a lone ESC (cancel) or an escape sequence.
# Wait briefly for a continuation byte. On slow PTYs (SSH/tmux) the
# bytes of an arrow key can arrive across separate reads, so a tiny
# timeout avoids misreading a split sequence as a bare ESC.
try:
stdscr.timeout(60)
nxt = stdscr.getch()
finally:
stdscr.timeout(-1) # restore blocking mode

if nxt == -1:
return NAV_CANCEL # genuine lone ESC

if nxt in (ord("["), ord("O")): # CSI / SS3 introducer
final = stdscr.getch()
if final in (ord("A"), ord("k")):
return NAV_UP
if final in (ord("B"), ord("j")):
return NAV_DOWN
# Consume the tail of any other CSI sequence (e.g. ``[3~`` Delete,
# ``[H`` Home) up to its terminator so stray bytes don't leak into
# the next input() and corrupt it.
while 0x20 <= final <= 0x3F: # CSI parameter/intermediate bytes
final = stdscr.getch()
return NAV_NONE
# ESC followed by some other byte we don't handle — swallow it.
return NAV_NONE

return NAV_NONE


def curses_checklist(
title: str,
items: List[str],
Expand Down Expand Up @@ -137,18 +209,18 @@ def _draw(stdscr):
pass

stdscr.refresh()
key = stdscr.getch()
action = read_menu_key(stdscr)

if key in {curses.KEY_UP, ord("k")}:
if action == NAV_UP:
cursor = (cursor - 1) % len(items)
elif key in {curses.KEY_DOWN, ord("j")}:
elif action == NAV_DOWN:
cursor = (cursor + 1) % len(items)
elif key == ord(" "):
elif action == NAV_TOGGLE:
chosen.symmetric_difference_update({cursor})
elif key in {curses.KEY_ENTER, 10, 13}:
elif action == NAV_SELECT:
result_holder[0] = set(chosen)
return
elif key in {27, ord("q")}:
elif action == NAV_CANCEL:
result_holder[0] = cancel_returns
return

Expand Down Expand Up @@ -263,16 +335,16 @@ def _draw(stdscr):
pass

stdscr.refresh()
key = stdscr.getch()
action = read_menu_key(stdscr)

if key in {curses.KEY_UP, ord("k")}:
if action == NAV_UP:
cursor = (cursor - 1) % len(items)
elif key in {curses.KEY_DOWN, ord("j")}:
elif action == NAV_DOWN:
cursor = (cursor + 1) % len(items)
elif key in {ord(" "), curses.KEY_ENTER, 10, 13}:
elif action in (NAV_SELECT, NAV_TOGGLE):
result_holder[0] = cursor
return
elif key in {27, ord("q")}:
elif action == NAV_CANCEL:
result_holder[0] = cancel_returns
return

Expand Down Expand Up @@ -386,16 +458,16 @@ def _draw(stdscr):
pass

stdscr.refresh()
key = stdscr.getch()
action = read_menu_key(stdscr)

if key in {curses.KEY_UP, ord("k")}:
if action == NAV_UP:
cursor = (cursor - 1) % len(all_items)
elif key in {curses.KEY_DOWN, ord("j")}:
elif action == NAV_DOWN:
cursor = (cursor + 1) % len(all_items)
elif key in {curses.KEY_ENTER, 10, 13}:
elif action == NAV_SELECT:
result_holder[0] = cursor
return
elif key in {27, ord("q")}:
elif action == NAV_CANCEL:
result_holder[0] = None
return

Expand Down
102 changes: 102 additions & 0 deletions tests/hermes_cli/test_curses_arrow_keys.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
"""Regression tests for arrow-key decoding in the curses menus.

Root cause these guard against: on many terminals/terminfo entries, cursor
keys are delivered to ``getch()`` as raw CSI/SS3 escape byte sequences
(``27, 91, 66`` for arrow-down) even when ``keypad(True)`` is set. The menus
used to treat the leading ``27`` as ESC/cancel, which dumped the setup wizard's
provider/model picker into its numbered "Select [1-N]" fallback the instant a
user pressed up or down.
"""
import curses

from hermes_cli.curses_ui import (
NAV_CANCEL,
NAV_DOWN,
NAV_NONE,
NAV_SELECT,
NAV_UP,
read_menu_key,
)


class FakeStdscr:
"""Minimal stdscr stand-in that replays a queue of getch() byte returns.

``getch`` pops from ``keys``; an empty queue yields ``-1`` (matching curses
non-blocking behavior). ``timeout`` is recorded but otherwise inert.
"""

def __init__(self, keys):
self.keys = list(keys)
self.timeouts = []

def getch(self):
return self.keys.pop(0) if self.keys else -1

def timeout(self, ms):
self.timeouts.append(ms)


def test_raw_csi_arrow_down_decodes_to_down():
# ESC [ B -> down, NOT cancel
assert read_menu_key(FakeStdscr([27, ord("["), ord("B")])) == NAV_DOWN


def test_raw_csi_arrow_up_decodes_to_up():
# ESC [ A -> up
assert read_menu_key(FakeStdscr([27, ord("["), ord("A")])) == NAV_UP


def test_raw_ss3_arrow_keys_decode():
# Application cursor mode: ESC O B / ESC O A
assert read_menu_key(FakeStdscr([27, ord("O"), ord("B")])) == NAV_DOWN
assert read_menu_key(FakeStdscr([27, ord("O"), ord("A")])) == NAV_UP


def test_translated_key_constants_still_work():
assert read_menu_key(FakeStdscr([curses.KEY_DOWN])) == NAV_DOWN
assert read_menu_key(FakeStdscr([curses.KEY_UP])) == NAV_UP


def test_vim_keys():
assert read_menu_key(FakeStdscr([ord("j")])) == NAV_DOWN
assert read_menu_key(FakeStdscr([ord("k")])) == NAV_UP


def test_lone_escape_is_cancel():
# ESC with no continuation byte (getch returns -1) -> genuine cancel.
assert read_menu_key(FakeStdscr([27])) == NAV_CANCEL


def test_q_is_cancel():
assert read_menu_key(FakeStdscr([ord("q")])) == NAV_CANCEL


def test_enter_variants_select():
assert read_menu_key(FakeStdscr([10])) == NAV_SELECT
assert read_menu_key(FakeStdscr([13])) == NAV_SELECT
assert read_menu_key(FakeStdscr([curses.KEY_ENTER])) == NAV_SELECT


def test_unhandled_csi_sequence_is_consumed_and_ignored():
# Delete key (ESC [ 3 ~): must be swallowed whole and map to NAV_NONE so
# its tail bytes don't leak into a subsequent input() call.
fake = FakeStdscr([27, ord("["), ord("3"), ord("~"), ord("X")])
assert read_menu_key(fake) == NAV_NONE
# The trailing 'X' (a genuinely separate keypress) must remain unconsumed.
assert fake.keys == [ord("X")]


def test_home_end_csi_sequences_ignored():
# ESC [ H (Home) and ESC [ F (End) -> NAV_NONE, fully consumed.
assert read_menu_key(FakeStdscr([27, ord("["), ord("H")])) == NAV_NONE
assert read_menu_key(FakeStdscr([27, ord("["), ord("F")])) == NAV_NONE


def test_escape_uses_short_timeout_then_restores_blocking():
fake = FakeStdscr([27, ord("["), ord("B")])
read_menu_key(fake)
# A short positive timeout is set to wait for the continuation byte, then
# blocking mode (-1) is restored.
assert fake.timeouts[0] > 0
assert fake.timeouts[-1] == -1
Loading