fix(cli): support reliable setup menu navigation
Decode Ghostty/Kitty enhanced selection and cancellation keys, make setup cancellation terminal, and add cross-terminal previous-step navigation to setup and model flows. Refs #92833
This commit is contained in:
+341
-45
@@ -5,8 +5,10 @@ Provides a curses multi-select with keyboard navigation, plus a
|
||||
text-based numbered fallback for terminals without curses support.
|
||||
"""
|
||||
import sys
|
||||
from contextvars import ContextVar, Token
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable, List, Optional, Sequence, Set, Tuple, Union
|
||||
from enum import Enum
|
||||
from typing import Callable, List, Optional, Protocol, Sequence, Set, Tuple, Union
|
||||
|
||||
from hermes_cli.colors import Colors, color
|
||||
|
||||
@@ -15,6 +17,113 @@ from hermes_cli.colors import Colors, color
|
||||
RadioItem = Union[str, Sequence[Tuple[str, Optional[str]]]]
|
||||
|
||||
|
||||
_NO_REPLAY = object()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MenuNavigationStart:
|
||||
"""Navigation instructions returned when a scoped menu begins."""
|
||||
|
||||
allow_back: bool = False
|
||||
replay_value: object = _NO_REPLAY
|
||||
|
||||
@property
|
||||
def should_replay(self) -> bool:
|
||||
return self.replay_value is not _NO_REPLAY
|
||||
|
||||
|
||||
class MenuNavigationEvent(str, Enum):
|
||||
BEGIN = "begin"
|
||||
RESOLVE = "resolve"
|
||||
CANCEL = "cancel"
|
||||
BACK = "back"
|
||||
|
||||
|
||||
class MenuNavigationHandler(Protocol):
|
||||
"""Typed contract between shared menus and a scoped flow controller."""
|
||||
|
||||
def __call__(
|
||||
self,
|
||||
event: MenuNavigationEvent,
|
||||
value: object = None,
|
||||
) -> MenuNavigationStart | None: ...
|
||||
|
||||
|
||||
_MENU_NAVIGATION_HANDLER: ContextVar[MenuNavigationHandler | None] = ContextVar(
|
||||
"hermes_menu_navigation_handler", default=None
|
||||
)
|
||||
_NUMBERED_BACK_ENABLED: ContextVar[bool] = ContextVar(
|
||||
"hermes_numbered_back_enabled", default=False
|
||||
)
|
||||
|
||||
|
||||
def set_menu_navigation_handler(
|
||||
handler: MenuNavigationHandler,
|
||||
) -> Token[MenuNavigationHandler | None]:
|
||||
"""Scope setup-style cancel/back behavior to the current CLI invocation."""
|
||||
return _MENU_NAVIGATION_HANDLER.set(handler)
|
||||
|
||||
|
||||
def reset_menu_navigation_handler(token: Token[MenuNavigationHandler | None]) -> None:
|
||||
"""Restore the menu navigation handler active before ``token``."""
|
||||
_MENU_NAVIGATION_HANDLER.reset(token)
|
||||
|
||||
|
||||
def _cancel_scoped_navigation() -> None:
|
||||
"""Notify an active menu flow that a text fallback was interrupted."""
|
||||
handler = _MENU_NAVIGATION_HANDLER.get()
|
||||
if handler is not None:
|
||||
handler(MenuNavigationEvent.CANCEL)
|
||||
|
||||
|
||||
def _back_scoped_navigation() -> None:
|
||||
"""Notify an active menu flow that its text fallback requested back."""
|
||||
handler = _MENU_NAVIGATION_HANDLER.get()
|
||||
if handler is not None:
|
||||
handler(MenuNavigationEvent.BACK)
|
||||
|
||||
|
||||
class _NumberedNavigation(Enum):
|
||||
CANCEL = "cancel"
|
||||
BACK = "back"
|
||||
|
||||
|
||||
def _read_numbered_input(prompt_text: str) -> str | _NumberedNavigation:
|
||||
"""Read a numbered fallback choice with setup navigation key bindings.
|
||||
|
||||
Ordinary numbered menus retain their historical ``input()`` behavior.
|
||||
During setup/model flows, prompt_toolkit supplies portable Escape, Ctrl+C,
|
||||
and Left bindings on POSIX and native Windows when curses is unavailable.
|
||||
"""
|
||||
if _MENU_NAVIGATION_HANDLER.get() is None:
|
||||
return input(prompt_text)
|
||||
|
||||
from prompt_toolkit import PromptSession
|
||||
from prompt_toolkit.formatted_text import ANSI
|
||||
from prompt_toolkit.key_binding import KeyBindings
|
||||
from prompt_toolkit.keys import Keys
|
||||
|
||||
# Setup can be invoked without importing the classic CLI, which normally
|
||||
# installs Ghostty/Kitty CSI-u aliases at process startup.
|
||||
from hermes_cli.pt_input_extras import install_modify_other_keys_aliases
|
||||
|
||||
install_modify_other_keys_aliases()
|
||||
bindings = KeyBindings()
|
||||
|
||||
@bindings.add(Keys.Escape)
|
||||
@bindings.add(Keys.ControlC)
|
||||
def _cancel(event) -> None:
|
||||
event.app.exit(result=_NumberedNavigation.CANCEL)
|
||||
|
||||
if _NUMBERED_BACK_ENABLED.get():
|
||||
|
||||
@bindings.add(Keys.Left)
|
||||
def _back(event) -> None:
|
||||
event.app.exit(result=_NumberedNavigation.BACK)
|
||||
|
||||
return PromptSession().prompt(ANSI(prompt_text), key_bindings=bindings)
|
||||
|
||||
|
||||
def radio_item_plain(item: RadioItem) -> str:
|
||||
"""Flatten a radiolist item to searchable/plain display text."""
|
||||
if isinstance(item, str):
|
||||
@@ -362,9 +471,11 @@ def flush_stdin() -> None:
|
||||
# every menu's key-handling branch identical and free of raw escape-byte logic.
|
||||
NAV_UP = "up"
|
||||
NAV_DOWN = "down"
|
||||
NAV_BACK = "back"
|
||||
NAV_SELECT = "select"
|
||||
NAV_TOGGLE = "toggle"
|
||||
NAV_CANCEL = "cancel"
|
||||
NAV_INTERRUPT = "interrupt"
|
||||
NAV_NONE = "none"
|
||||
|
||||
|
||||
@@ -387,6 +498,76 @@ def read_menu_key(stdscr) -> str:
|
||||
return _decode_menu_key(stdscr, stdscr.getch())
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _EnhancedKey:
|
||||
codepoint: int
|
||||
modifier: int = 1
|
||||
event_type: int = 1
|
||||
|
||||
|
||||
def _parse_int(value: str, default: int = 0) -> int:
|
||||
try:
|
||||
return int(value)
|
||||
except ValueError:
|
||||
return default
|
||||
|
||||
|
||||
def _parse_csi_u_key(raw: str) -> _EnhancedKey | None:
|
||||
"""Parse a Kitty/CSI-u key, preserving its press/repeat/release type."""
|
||||
parts = raw.split(";")
|
||||
codepoint = _parse_int(parts[0].split(":", 1)[0]) if parts else 0
|
||||
if not codepoint:
|
||||
return None
|
||||
|
||||
modifier = 1
|
||||
event_type = 1
|
||||
if len(parts) > 1:
|
||||
modifier_parts = parts[1].split(":")
|
||||
modifier = _parse_int(modifier_parts[0], 1)
|
||||
if len(modifier_parts) > 1:
|
||||
event_type = _parse_int(modifier_parts[1], 1)
|
||||
return _EnhancedKey(codepoint, modifier, event_type)
|
||||
|
||||
|
||||
def _parse_csi_numbers(raw: str) -> list[int]:
|
||||
"""Parse semicolon-delimited CSI numbers for modifyOtherKeys."""
|
||||
return [_parse_int(part.split(":", 1)[0]) for part in raw.split(";")]
|
||||
|
||||
|
||||
def _enhanced_key_action(codepoint: int, modifier: int = 1) -> str:
|
||||
"""Map CSI-u/modifyOtherKeys codepoints to setup menu actions."""
|
||||
if codepoint in (10, 13):
|
||||
return NAV_SELECT
|
||||
if codepoint == 27:
|
||||
return NAV_CANCEL
|
||||
if codepoint == 32:
|
||||
return NAV_TOGGLE
|
||||
|
||||
# CSI-u encodes Ctrl+C as codepoint `c` plus the Ctrl modifier. Lock-state
|
||||
# bits may be added to the modifier, so inspect the Ctrl bit rather than
|
||||
# matching only the canonical value 5.
|
||||
has_ctrl = bool((max(1, modifier) - 1) & 4)
|
||||
if codepoint == 3 or (codepoint in (ord("c"), ord("C")) and has_ctrl):
|
||||
return NAV_INTERRUPT
|
||||
return NAV_NONE
|
||||
|
||||
|
||||
def _read_csi_tail(stdscr) -> tuple[str, int | None]:
|
||||
"""Read CSI/SS3 parameter bytes through the final byte."""
|
||||
raw: list[str] = []
|
||||
for _ in range(32):
|
||||
value = stdscr.getch()
|
||||
if value == -1:
|
||||
return "".join(raw), None
|
||||
if 0x40 <= value <= 0x7E:
|
||||
return "".join(raw), value
|
||||
if 0x20 <= value <= 0x3F:
|
||||
raw.append(chr(value))
|
||||
continue
|
||||
return "".join(raw), None
|
||||
return "".join(raw), None
|
||||
|
||||
|
||||
def _decode_menu_key(stdscr, key: int) -> str:
|
||||
"""Normalize an already-read keypress to a menu action.
|
||||
|
||||
@@ -399,6 +580,10 @@ def _decode_menu_key(stdscr, key: int) -> str:
|
||||
return NAV_UP
|
||||
if key in (curses.KEY_DOWN, ord("j")):
|
||||
return NAV_DOWN
|
||||
if key == curses.KEY_LEFT:
|
||||
return NAV_BACK
|
||||
if key == 3: # Ctrl+C in curses raw/cbreak mode.
|
||||
return NAV_INTERRUPT
|
||||
if key in (curses.KEY_ENTER, 10, 13):
|
||||
return NAV_SELECT
|
||||
if key == ord(" "):
|
||||
@@ -413,27 +598,35 @@ def _decode_menu_key(stdscr, key: int) -> str:
|
||||
try:
|
||||
stdscr.timeout(60)
|
||||
nxt = stdscr.getch()
|
||||
if nxt == -1:
|
||||
return NAV_CANCEL # genuine lone ESC
|
||||
|
||||
if nxt in (ord("["), ord("O")): # CSI / SS3 introducer
|
||||
raw_params, final = _read_csi_tail(stdscr)
|
||||
if final in (ord("A"), ord("k")):
|
||||
return NAV_UP
|
||||
if final in (ord("B"), ord("j")):
|
||||
return NAV_DOWN
|
||||
if final == ord("D"):
|
||||
return NAV_BACK
|
||||
if final == ord("u"):
|
||||
enhanced = _parse_csi_u_key(raw_params)
|
||||
if enhanced is not None:
|
||||
if enhanced.event_type == 3: # key release
|
||||
return NAV_NONE
|
||||
return _enhanced_key_action(
|
||||
enhanced.codepoint, enhanced.modifier
|
||||
)
|
||||
if final == ord("~"):
|
||||
params = _parse_csi_numbers(raw_params)
|
||||
if len(params) >= 3 and params[0] == 27:
|
||||
return _enhanced_key_action(params[2], params[1])
|
||||
return NAV_NONE
|
||||
# ESC followed by some other byte we don't handle — swallow it.
|
||||
return NAV_NONE
|
||||
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
|
||||
|
||||
|
||||
@@ -477,7 +670,7 @@ def _run_curses_menu(
|
||||
Draw one item row. ``idx`` is always the ORIGINAL item index, so
|
||||
per-menu rendering is unchanged whether or not a filter is active.
|
||||
on_action(action, cursor) -> value
|
||||
Reducer for SELECT/TOGGLE/CANCEL. Return ``_KEEP`` to continue the
|
||||
Reducer for SELECT/TOGGLE/CANCEL/BACK. Return ``_KEEP`` to continue the
|
||||
loop; return anything else to resolve the menu with that value.
|
||||
(UP/DOWN cursor movement is handled by the driver itself.)
|
||||
reserve_bottom: number of bottom screen rows kept clear of items
|
||||
@@ -494,6 +687,25 @@ def _run_curses_menu(
|
||||
search_labels: per-item text used for filtering (required when
|
||||
``searchable`` is true; length must equal ``item_count``).
|
||||
"""
|
||||
navigation_handler = _MENU_NAVIGATION_HANDLER.get()
|
||||
navigation_start = (
|
||||
navigation_handler(MenuNavigationEvent.BEGIN)
|
||||
if navigation_handler is not None
|
||||
else None
|
||||
)
|
||||
if navigation_start is not None and not isinstance(
|
||||
navigation_start, MenuNavigationStart
|
||||
):
|
||||
raise TypeError("menu navigation 'begin' must return MenuNavigationStart")
|
||||
context_back = bool(navigation_start and navigation_start.allow_back)
|
||||
if navigation_start is not None and navigation_start.should_replay:
|
||||
if navigation_handler is not None:
|
||||
navigation_handler(
|
||||
MenuNavigationEvent.RESOLVE, navigation_start.replay_value
|
||||
)
|
||||
return navigation_start.replay_value
|
||||
effective_allow_back = context_back
|
||||
|
||||
# Non-TTY (piped/redirected stdin): curses and input() both hang or spin,
|
||||
# so return the cancel value directly — matching the pre-refactor guard in
|
||||
# each menu (the numbered fallback is only for curses errors on a real TTY).
|
||||
@@ -502,8 +714,22 @@ def _run_curses_menu(
|
||||
|
||||
use_search = searchable and search_labels is not None and len(search_labels) == item_count
|
||||
|
||||
def _run_fallback():
|
||||
back_token = _NUMBERED_BACK_ENABLED.set(effective_allow_back)
|
||||
try:
|
||||
result = fallback()
|
||||
finally:
|
||||
_NUMBERED_BACK_ENABLED.reset(back_token)
|
||||
if navigation_handler is not None:
|
||||
navigation_handler(MenuNavigationEvent.RESOLVE, result)
|
||||
return result
|
||||
|
||||
try:
|
||||
import curses
|
||||
except ImportError:
|
||||
return _run_fallback()
|
||||
|
||||
try:
|
||||
result_holder = [_KEEP]
|
||||
|
||||
def _draw(stdscr):
|
||||
@@ -537,12 +763,13 @@ def _run_curses_menu(
|
||||
)
|
||||
cursor, cursor_pos = _reconcile_cursor(filtered, cursor)
|
||||
|
||||
# draw_header accepts an optional `search` kwarg when the menu
|
||||
# wants to render the live filter; tolerate headers that don't.
|
||||
try:
|
||||
items_start = draw_header(stdscr, max_y, max_x, search=search)
|
||||
except TypeError:
|
||||
items_start = draw_header(stdscr, max_y, max_x)
|
||||
items_start = draw_header(
|
||||
stdscr,
|
||||
max_y,
|
||||
max_x,
|
||||
search=search,
|
||||
back_enabled=effective_allow_back,
|
||||
)
|
||||
|
||||
visible_rows = max(1, max_y - items_start - reserve_bottom)
|
||||
scroll_offset = _scroll_for_cursor(
|
||||
@@ -572,7 +799,20 @@ def _run_curses_menu(
|
||||
if use_search:
|
||||
key = stdscr.getch()
|
||||
|
||||
if search.active:
|
||||
if search.active and key == 27:
|
||||
# Ghostty/Kitty enhanced keys also begin with ESC.
|
||||
# Decode the full sequence before treating a genuine
|
||||
# Escape as "stop search"; otherwise Enter/Left/Ctrl+C
|
||||
# lose their tail while the search prompt is active.
|
||||
action = _decode_menu_key(stdscr, key)
|
||||
if action == NAV_CANCEL:
|
||||
search.active = False
|
||||
search.query = ""
|
||||
scroll_offset = 0
|
||||
continue
|
||||
if action == NAV_NONE:
|
||||
continue
|
||||
elif search.active:
|
||||
# Active search consumes query-editing keys; nav keys
|
||||
# fall through to be decoded below.
|
||||
handled, confirm, changed = _handle_active_search_key(
|
||||
@@ -587,6 +827,10 @@ def _run_curses_menu(
|
||||
if filtered:
|
||||
outcome = on_action(NAV_SELECT, cursor)
|
||||
if outcome is not _KEEP:
|
||||
if navigation_handler is not None:
|
||||
navigation_handler(
|
||||
MenuNavigationEvent.RESOLVE, outcome
|
||||
)
|
||||
result_holder[0] = outcome
|
||||
return
|
||||
continue
|
||||
@@ -605,11 +849,25 @@ def _run_curses_menu(
|
||||
cursor = _move_filtered_cursor(filtered, cursor, cursor_pos, -1)
|
||||
elif action == NAV_DOWN:
|
||||
cursor = _move_filtered_cursor(filtered, cursor, cursor_pos, 1)
|
||||
elif action in (NAV_SELECT, NAV_TOGGLE, NAV_CANCEL):
|
||||
elif action in (
|
||||
NAV_SELECT,
|
||||
NAV_TOGGLE,
|
||||
NAV_CANCEL,
|
||||
NAV_INTERRUPT,
|
||||
) or (
|
||||
action == NAV_BACK and effective_allow_back
|
||||
):
|
||||
if action == NAV_SELECT and use_search and not filtered:
|
||||
continue
|
||||
if navigation_handler is not None:
|
||||
if action in (NAV_CANCEL, NAV_INTERRUPT):
|
||||
navigation_handler(MenuNavigationEvent.CANCEL)
|
||||
elif action == NAV_BACK and context_back:
|
||||
navigation_handler(MenuNavigationEvent.BACK)
|
||||
outcome = on_action(action, cursor)
|
||||
if outcome is not _KEEP:
|
||||
if navigation_handler is not None:
|
||||
navigation_handler(MenuNavigationEvent.RESOLVE, outcome)
|
||||
result_holder[0] = outcome
|
||||
return
|
||||
|
||||
@@ -618,9 +876,11 @@ def _run_curses_menu(
|
||||
return result_holder[0] if result_holder[0] is not _KEEP else cancel_value
|
||||
|
||||
except KeyboardInterrupt:
|
||||
if navigation_handler is not None:
|
||||
navigation_handler(MenuNavigationEvent.CANCEL)
|
||||
return cancel_value
|
||||
except Exception:
|
||||
return fallback()
|
||||
except curses.error:
|
||||
return _run_fallback()
|
||||
|
||||
|
||||
def curses_checklist(
|
||||
@@ -647,18 +907,17 @@ def curses_checklist(
|
||||
|
||||
chosen = set(selected)
|
||||
|
||||
def _draw_header(stdscr, max_y, max_x):
|
||||
def _draw_header(stdscr, max_y, max_x, search=None, back_enabled=False):
|
||||
import curses
|
||||
try:
|
||||
hattr = curses.A_BOLD
|
||||
if curses.has_colors():
|
||||
hattr |= curses.color_pair(2)
|
||||
stdscr.addnstr(0, 0, title, max_x - 1, hattr)
|
||||
stdscr.addnstr(
|
||||
1, 0,
|
||||
" ↑↓ navigate SPACE toggle ENTER confirm ESC cancel",
|
||||
max_x - 1, curses.A_DIM,
|
||||
)
|
||||
hint = " ↑↓ navigate SPACE toggle ENTER confirm ESC cancel"
|
||||
if back_enabled:
|
||||
hint += " ← previous"
|
||||
stdscr.addnstr(1, 0, hint, max_x - 1, curses.A_DIM)
|
||||
except curses.error:
|
||||
pass
|
||||
return 3
|
||||
@@ -753,7 +1012,7 @@ def curses_radiolist(
|
||||
|
||||
plain_labels = [radio_item_plain(item) for item in items]
|
||||
|
||||
def _draw_header(stdscr, max_y, max_x, search=None):
|
||||
def _draw_header(stdscr, max_y, max_x, search=None, back_enabled=False):
|
||||
import curses
|
||||
row = 0
|
||||
try:
|
||||
@@ -776,6 +1035,8 @@ def curses_radiolist(
|
||||
hint = " \u2191\u2193 navigate ENTER/SPACE select / search ESC cancel"
|
||||
else:
|
||||
hint = " \u2191\u2193 navigate ENTER/SPACE select ESC cancel"
|
||||
if back_enabled:
|
||||
hint += " \u2190 previous"
|
||||
stdscr.addnstr(row, 0, hint, max_x - 1, curses.A_DIM)
|
||||
row += 1
|
||||
except curses.error:
|
||||
@@ -852,14 +1113,26 @@ def _radio_numbered_fallback(
|
||||
print(f" {marker} {i + 1:>2}. {format_radio_item_ansi(label)}")
|
||||
print()
|
||||
try:
|
||||
val = input(color(f" Choice [default {selected + 1}]: ", Colors.DIM)).strip()
|
||||
val = _read_numbered_input(
|
||||
color(f" Choice [default {selected + 1}]: ", Colors.DIM)
|
||||
)
|
||||
if val is _NumberedNavigation.BACK:
|
||||
_back_scoped_navigation()
|
||||
return cancel_returns
|
||||
if val is _NumberedNavigation.CANCEL:
|
||||
_cancel_scoped_navigation()
|
||||
return cancel_returns
|
||||
val = val.strip()
|
||||
if not val:
|
||||
return selected
|
||||
idx = int(val) - 1
|
||||
if 0 <= idx < len(items):
|
||||
return idx
|
||||
return selected
|
||||
except (ValueError, KeyboardInterrupt, EOFError):
|
||||
except ValueError:
|
||||
return cancel_returns
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
_cancel_scoped_navigation()
|
||||
return cancel_returns
|
||||
|
||||
|
||||
@@ -881,7 +1154,7 @@ def curses_single_select(
|
||||
all_items = list(items) + [cancel_label]
|
||||
cancel_idx = len(items)
|
||||
|
||||
def _draw_header(stdscr, max_y, max_x, search=None):
|
||||
def _draw_header(stdscr, max_y, max_x, search=None, back_enabled=False):
|
||||
import curses
|
||||
try:
|
||||
hattr = curses.A_BOLD
|
||||
@@ -894,6 +1167,8 @@ def curses_single_select(
|
||||
hint = " ↑↓ navigate ENTER confirm / search ESC/q cancel"
|
||||
else:
|
||||
hint = " ↑↓ navigate ENTER confirm ESC/q cancel"
|
||||
if back_enabled:
|
||||
hint += " ← previous"
|
||||
stdscr.addnstr(1, 0, hint, max_x - 1, curses.A_DIM)
|
||||
except curses.error:
|
||||
pass
|
||||
@@ -918,7 +1193,7 @@ def curses_single_select(
|
||||
# Selecting the synthetic cancel row resolves to None, mirroring
|
||||
# the old post-loop ``>= cancel_idx`` guard.
|
||||
return None if cursor >= cancel_idx else cursor
|
||||
if action == NAV_CANCEL:
|
||||
if action in (NAV_CANCEL, NAV_INTERRUPT):
|
||||
return None
|
||||
return _KEEP # NAV_TOGGLE — no-op for this menu
|
||||
|
||||
@@ -947,7 +1222,14 @@ def _numbered_single_fallback(
|
||||
print(f" {i}. {label}")
|
||||
print()
|
||||
try:
|
||||
val = input(f" Choice [1-{len(items)}]: ").strip()
|
||||
val = _read_numbered_input(f" Choice [1-{len(items)}]: ")
|
||||
if val is _NumberedNavigation.BACK:
|
||||
_back_scoped_navigation()
|
||||
return None
|
||||
if val is _NumberedNavigation.CANCEL:
|
||||
_cancel_scoped_navigation()
|
||||
return None
|
||||
val = val.strip()
|
||||
if not val:
|
||||
return None
|
||||
idx = int(val) - 1
|
||||
@@ -955,8 +1237,10 @@ def _numbered_single_fallback(
|
||||
return idx
|
||||
if idx == cancel_idx:
|
||||
return None
|
||||
except (ValueError, KeyboardInterrupt, EOFError):
|
||||
except ValueError:
|
||||
pass
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
_cancel_scoped_navigation()
|
||||
return None
|
||||
|
||||
|
||||
@@ -982,13 +1266,25 @@ def _numbered_fallback(
|
||||
print(color(f"\n {status_text}", Colors.DIM))
|
||||
print()
|
||||
try:
|
||||
val = input(color(" Toggle # (or Enter to confirm): ", Colors.DIM)).strip()
|
||||
val = _read_numbered_input(
|
||||
color(" Toggle # (or Enter to confirm): ", Colors.DIM)
|
||||
)
|
||||
if val is _NumberedNavigation.BACK:
|
||||
_back_scoped_navigation()
|
||||
return cancel_returns
|
||||
if val is _NumberedNavigation.CANCEL:
|
||||
_cancel_scoped_navigation()
|
||||
return cancel_returns
|
||||
val = val.strip()
|
||||
if not val:
|
||||
break
|
||||
idx = int(val) - 1
|
||||
if 0 <= idx < len(items):
|
||||
chosen.symmetric_difference_update({idx})
|
||||
except (ValueError, KeyboardInterrupt, EOFError):
|
||||
except ValueError:
|
||||
return cancel_returns
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
_cancel_scoped_navigation()
|
||||
return cancel_returns
|
||||
print()
|
||||
|
||||
|
||||
+7
-1
@@ -3579,7 +3579,13 @@ def cmd_model(args):
|
||||
print(" Cleared model picker cache.")
|
||||
except Exception:
|
||||
pass
|
||||
select_provider_and_model(args=args)
|
||||
from hermes_cli.setup import run_setup_action_with_navigation
|
||||
|
||||
run_setup_action_with_navigation(
|
||||
"Model & Provider",
|
||||
lambda: select_provider_and_model(args=args),
|
||||
cancelled_message="No change.",
|
||||
)
|
||||
|
||||
|
||||
def _is_profile_api_key_provider(provider_id: str) -> bool:
|
||||
|
||||
+285
-50
@@ -19,9 +19,12 @@ import re
|
||||
import shutil
|
||||
import sys
|
||||
import copy
|
||||
from contextlib import contextmanager
|
||||
from contextvars import ContextVar
|
||||
from pathlib import Path
|
||||
from typing import Optional, Dict, Any
|
||||
from typing import Optional, Dict, Any, Callable
|
||||
|
||||
from hermes_cli.curses_ui import MenuNavigationEvent, MenuNavigationStart
|
||||
from hermes_cli.nous_subscription import get_nous_subscription_features
|
||||
from tools.tool_backend_helpers import managed_nous_tools_enabled
|
||||
from hermes_constants import get_optional_skills_dir
|
||||
@@ -222,6 +225,84 @@ def prompt(question: str, default: str = None, password: bool = False) -> str:
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
class _SetupControlFlow(BaseException):
|
||||
"""Bypass provider error handlers that intentionally catch ``Exception``.
|
||||
|
||||
Provider setup contains broad compatibility boundaries around network,
|
||||
plugin, and credential integrations. Navigation must cross those layers
|
||||
unchanged so the outer setup state machine can replay the prior prompt.
|
||||
"""
|
||||
|
||||
|
||||
class _SetupCancelled(_SetupControlFlow):
|
||||
"""Internal control flow for cancelling the interactive setup wizard."""
|
||||
|
||||
|
||||
class _SetupGoBack(_SetupControlFlow):
|
||||
"""Internal control flow for returning to an earlier setup choice."""
|
||||
|
||||
def __init__(self, prompt_index: int):
|
||||
super().__init__(prompt_index)
|
||||
self.prompt_index = prompt_index
|
||||
|
||||
|
||||
class _SetupNavigationState:
|
||||
"""Per-invocation navigation state for the synchronous setup wizard."""
|
||||
|
||||
def __init__(self, *, section_index: int = -1, prompt_index: int = 0):
|
||||
self.section_index = section_index
|
||||
self.prompt_index = prompt_index
|
||||
self.active_prompt_index = -1
|
||||
self.resolved_choices: list[object] = []
|
||||
self.replay_choices: list[object] = []
|
||||
|
||||
|
||||
_SETUP_NAVIGATION: ContextVar[_SetupNavigationState | None] = ContextVar(
|
||||
"hermes_setup_navigation", default=None
|
||||
)
|
||||
|
||||
|
||||
def _handle_setup_menu_navigation(
|
||||
event: MenuNavigationEvent,
|
||||
value: object = None,
|
||||
) -> MenuNavigationStart | None:
|
||||
"""Translate shared curses menu events into setup control flow."""
|
||||
state = _SETUP_NAVIGATION.get()
|
||||
if state is None:
|
||||
return None
|
||||
if event is MenuNavigationEvent.BEGIN:
|
||||
if state.section_index < 0:
|
||||
state.active_prompt_index = -1
|
||||
return MenuNavigationStart()
|
||||
state.active_prompt_index = state.prompt_index
|
||||
state.prompt_index += 1
|
||||
allow_back = state.section_index > 0 or state.active_prompt_index > 0
|
||||
if state.active_prompt_index < len(state.replay_choices):
|
||||
return MenuNavigationStart(
|
||||
allow_back=allow_back,
|
||||
replay_value=copy.deepcopy(
|
||||
state.replay_choices[state.active_prompt_index]
|
||||
),
|
||||
)
|
||||
return MenuNavigationStart(allow_back=allow_back)
|
||||
if event is MenuNavigationEvent.RESOLVE:
|
||||
prompt_index = state.active_prompt_index
|
||||
if prompt_index < 0:
|
||||
return None
|
||||
resolved = copy.deepcopy(value)
|
||||
if prompt_index < len(state.resolved_choices):
|
||||
state.resolved_choices[prompt_index] = resolved
|
||||
del state.resolved_choices[prompt_index + 1 :]
|
||||
else:
|
||||
state.resolved_choices.append(resolved)
|
||||
return None
|
||||
if event is MenuNavigationEvent.CANCEL:
|
||||
raise _SetupCancelled()
|
||||
if event is MenuNavigationEvent.BACK:
|
||||
raise _SetupGoBack(state.active_prompt_index)
|
||||
return None
|
||||
|
||||
|
||||
_BRACKETED_PASTE_PATTERN = re.compile(r"\x1b\[\s*200~|\x1b\[\s*201~")
|
||||
|
||||
|
||||
@@ -235,14 +316,22 @@ def _sanitize_pasted_input(value: str) -> str:
|
||||
def _curses_prompt_choice(question: str, choices: list, default: int = 0, description: str | None = None) -> int:
|
||||
"""Single-select menu using curses. Delegates to curses_radiolist."""
|
||||
from hermes_cli.curses_ui import curses_radiolist
|
||||
return curses_radiolist(question, choices, selected=default, cancel_returns=-1, description=description)
|
||||
return curses_radiolist(
|
||||
question,
|
||||
choices,
|
||||
selected=default,
|
||||
cancel_returns=-1,
|
||||
description=description,
|
||||
)
|
||||
|
||||
|
||||
|
||||
def prompt_choice(question: str, choices: list, default: int = 0, description: str | None = None) -> int:
|
||||
"""Prompt for a choice from a list with arrow key navigation.
|
||||
|
||||
Escape keeps the current default (skips the question).
|
||||
Escape cancels an active setup wizard. Outside setup it keeps the current
|
||||
default. The curses component owns its own numbered fallback, so a cancel
|
||||
result must never be mistaken for a request to open another prompt.
|
||||
Ctrl+C exits the wizard.
|
||||
"""
|
||||
idx = _curses_prompt_choice(question, choices, default, description=description)
|
||||
@@ -254,32 +343,7 @@ def prompt_choice(question: str, choices: list, default: int = 0, description: s
|
||||
print()
|
||||
return idx
|
||||
|
||||
print(color(question, Colors.YELLOW))
|
||||
for i, choice in enumerate(choices):
|
||||
marker = "●" if i == default else "○"
|
||||
if i == default:
|
||||
print(color(f" {marker} {choice}", Colors.GREEN))
|
||||
else:
|
||||
print(f" {marker} {choice}")
|
||||
|
||||
print_info(f" Enter for default ({default + 1}) Ctrl+C to exit")
|
||||
|
||||
while True:
|
||||
try:
|
||||
value = input(
|
||||
color(f" Select [1-{len(choices)}] ({default + 1}): ", Colors.DIM)
|
||||
)
|
||||
if not value:
|
||||
return default
|
||||
idx = int(value) - 1
|
||||
if 0 <= idx < len(choices):
|
||||
return idx
|
||||
print_error(f"Please enter a number between 1 and {len(choices)}")
|
||||
except ValueError:
|
||||
print_error("Please enter a number")
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
print()
|
||||
sys.exit(1)
|
||||
return default
|
||||
|
||||
|
||||
def is_noninteractive() -> bool:
|
||||
@@ -311,6 +375,17 @@ def prompt_yes_no(question: str, default: bool = True) -> bool:
|
||||
if is_noninteractive():
|
||||
return default
|
||||
|
||||
# Setup owns a scoped curses navigation handler. Route binary selections
|
||||
# through the same menu surface so ESC and left-arrow work consistently,
|
||||
# while preserving the traditional line prompt for every other caller.
|
||||
if _SETUP_NAVIGATION.get() is not None:
|
||||
default_index = 0 if default else 1
|
||||
return _curses_prompt_choice(
|
||||
question,
|
||||
["Yes", "No"],
|
||||
default_index,
|
||||
) == 0
|
||||
|
||||
default_str = "Y/n" if default else "y/N"
|
||||
|
||||
while True:
|
||||
@@ -2843,7 +2918,122 @@ def _run_portal_one_shot(config: dict) -> None:
|
||||
print_info(" Run `hermes` to start chatting.")
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _setup_navigation_scope():
|
||||
"""Install and reliably restore the setup menu navigation context."""
|
||||
from hermes_cli.curses_ui import (
|
||||
reset_menu_navigation_handler,
|
||||
set_menu_navigation_handler,
|
||||
)
|
||||
|
||||
token = _SETUP_NAVIGATION.set(_SetupNavigationState())
|
||||
menu_token = set_menu_navigation_handler(_handle_setup_menu_navigation)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
reset_menu_navigation_handler(menu_token)
|
||||
_SETUP_NAVIGATION.reset(token)
|
||||
|
||||
|
||||
def run_setup_wizard(args):
|
||||
"""Run setup with navigation control scoped to this invocation."""
|
||||
with _setup_navigation_scope():
|
||||
try:
|
||||
return _run_setup_wizard_impl(args)
|
||||
except _SetupCancelled:
|
||||
print()
|
||||
print_info("Setup cancelled. Remaining sections were not changed.")
|
||||
return None
|
||||
|
||||
|
||||
def _run_setup_steps(
|
||||
steps: list[tuple[str, Callable[[], None]]],
|
||||
) -> None:
|
||||
"""Run setup sections with left-arrow navigation between choices.
|
||||
|
||||
Left arrow at a section's first choice returns to the previous section.
|
||||
From a later, nested choice it replays earlier selections invisibly and
|
||||
reopens only the immediately preceding prompt.
|
||||
"""
|
||||
state = _SETUP_NAVIGATION.get()
|
||||
section_index = 0
|
||||
answers_by_section: dict[int, list[object]] = {}
|
||||
replay_by_section: dict[int, list[object]] = {}
|
||||
try:
|
||||
while section_index < len(steps):
|
||||
label, action = steps[section_index]
|
||||
if state is not None:
|
||||
state.section_index = section_index
|
||||
state.prompt_index = 0
|
||||
state.active_prompt_index = -1
|
||||
state.resolved_choices = []
|
||||
state.replay_choices = copy.deepcopy(
|
||||
replay_by_section.pop(section_index, [])
|
||||
)
|
||||
try:
|
||||
action()
|
||||
except _SetupGoBack as navigation:
|
||||
if state is not None:
|
||||
answers_by_section[section_index] = copy.deepcopy(
|
||||
state.resolved_choices
|
||||
)
|
||||
if navigation.prompt_index > 0:
|
||||
previous_index = section_index
|
||||
target_prompt = navigation.prompt_index - 1
|
||||
replay_by_section[previous_index] = copy.deepcopy(
|
||||
answers_by_section.get(previous_index, [])[:target_prompt]
|
||||
)
|
||||
else:
|
||||
previous_index = max(0, section_index - 1)
|
||||
previous_answers = answers_by_section.get(previous_index, [])
|
||||
target_prompt = max(0, len(previous_answers) - 1)
|
||||
replay_by_section[previous_index] = copy.deepcopy(
|
||||
previous_answers[:target_prompt]
|
||||
)
|
||||
previous_label = steps[previous_index][0]
|
||||
print()
|
||||
if previous_index == section_index:
|
||||
print_info(f"Returning to the previous choice in {label}...")
|
||||
else:
|
||||
print_info(f"Returning to {previous_label}...")
|
||||
section_index = previous_index
|
||||
continue
|
||||
if state is not None:
|
||||
answers_by_section[section_index] = copy.deepcopy(
|
||||
state.resolved_choices
|
||||
)
|
||||
section_index += 1
|
||||
finally:
|
||||
if state is not None:
|
||||
state.section_index = -1
|
||||
state.prompt_index = 0
|
||||
state.active_prompt_index = -1
|
||||
state.resolved_choices = []
|
||||
state.replay_choices = []
|
||||
|
||||
|
||||
def run_setup_action_with_navigation(
|
||||
label: str,
|
||||
action: Callable[[], None],
|
||||
*,
|
||||
cancelled_message: str = "Setup cancelled.",
|
||||
) -> None:
|
||||
"""Run a setup-style menu flow with Escape and nested Left navigation.
|
||||
|
||||
Shared commands such as ``hermes model`` use the same provider/model
|
||||
pickers as the setup wizard, but run outside ``run_setup_wizard``. This
|
||||
installs the setup navigation context for that standalone command and
|
||||
reuses the same prompt replay state machine.
|
||||
"""
|
||||
with _setup_navigation_scope():
|
||||
try:
|
||||
_run_setup_steps([(label, action)])
|
||||
except _SetupCancelled:
|
||||
print()
|
||||
print_info(cancelled_message)
|
||||
|
||||
|
||||
def _run_setup_wizard_impl(args):
|
||||
"""Run the interactive setup wizard.
|
||||
|
||||
Supports full, quick, and section-specific setup:
|
||||
@@ -2923,7 +3113,9 @@ def run_setup_wizard(args):
|
||||
Colors.MAGENTA,
|
||||
)
|
||||
)
|
||||
func(config)
|
||||
_run_setup_steps(
|
||||
[(label, lambda setup_func=func: setup_func(config))]
|
||||
)
|
||||
save_config(config)
|
||||
print()
|
||||
print_success(f"{label} configuration complete!")
|
||||
@@ -2987,7 +3179,9 @@ def run_setup_wizard(args):
|
||||
# missing items" flow (useful after a partial OpenClaw migration
|
||||
# or when a required API key got cleared).
|
||||
if quick_requested:
|
||||
_run_quick_setup(config, hermes_home)
|
||||
_run_setup_steps(
|
||||
[("Quick Setup", lambda: _run_quick_setup(config, hermes_home))]
|
||||
)
|
||||
return
|
||||
|
||||
print()
|
||||
@@ -3027,10 +3221,28 @@ def run_setup_wizard(args):
|
||||
)
|
||||
|
||||
if setup_mode == 0:
|
||||
_run_first_time_quick_setup(config, hermes_home, is_existing)
|
||||
_run_setup_steps(
|
||||
[
|
||||
(
|
||||
"Quick Setup",
|
||||
lambda: _run_first_time_quick_setup(
|
||||
config, hermes_home, is_existing
|
||||
),
|
||||
)
|
||||
]
|
||||
)
|
||||
return
|
||||
if setup_mode == 2:
|
||||
_run_blank_slate_setup(config, hermes_home, is_existing)
|
||||
_run_setup_steps(
|
||||
[
|
||||
(
|
||||
"Blank Slate",
|
||||
lambda: _run_blank_slate_setup(
|
||||
config, hermes_home, is_existing
|
||||
),
|
||||
)
|
||||
]
|
||||
)
|
||||
return
|
||||
|
||||
# ── Full Setup — run all sections ──
|
||||
@@ -3048,32 +3260,55 @@ def run_setup_wizard(args):
|
||||
print_info("Each section below will show what was imported — press Enter to keep,")
|
||||
print_info("or choose to reconfigure if needed.")
|
||||
|
||||
# Section 1: Model & Provider
|
||||
if not (migration_ran and _skip_configured_section(config, "model", "Model & Provider")):
|
||||
setup_model_provider(config)
|
||||
|
||||
# Section 2: Terminal Backend
|
||||
if not (migration_ran and _skip_configured_section(config, "terminal", "Terminal Backend")):
|
||||
setup_terminal_backend(config)
|
||||
|
||||
# Section 3: Agent Settings — no longer prompted. First installs get the
|
||||
# recommended defaults silently; existing installs keep whatever they have.
|
||||
# Tune later with `hermes setup agent`.
|
||||
if not is_existing:
|
||||
_apply_default_agent_settings(config)
|
||||
|
||||
# Section 4: Messaging Platforms
|
||||
if not (migration_ran and _skip_configured_section(config, "gateway", "Messaging Platforms")):
|
||||
setup_gateway(config)
|
||||
else:
|
||||
# Section skipped (migrated config) — still make sure the gateway
|
||||
# service exists so cron jobs and migrated platforms actually run.
|
||||
def _model_step() -> None:
|
||||
if not (
|
||||
migration_ran
|
||||
and _skip_configured_section(config, "model", "Model & Provider")
|
||||
):
|
||||
setup_model_provider(config)
|
||||
|
||||
def _terminal_step() -> None:
|
||||
if not (
|
||||
migration_ran
|
||||
and _skip_configured_section(config, "terminal", "Terminal Backend")
|
||||
):
|
||||
setup_terminal_backend(config)
|
||||
|
||||
def _gateway_step() -> None:
|
||||
if not (
|
||||
migration_ran
|
||||
and _skip_configured_section(config, "gateway", "Messaging Platforms")
|
||||
):
|
||||
setup_gateway(config)
|
||||
return
|
||||
|
||||
# A migrated gateway section can be skipped, but its service still
|
||||
# needs to exist so imported platforms and cron jobs become active.
|
||||
from hermes_cli.gateway import ensure_gateway_service
|
||||
|
||||
ensure_gateway_service(context="setup")
|
||||
|
||||
# Section 5: Tools
|
||||
if not (migration_ran and _skip_configured_section(config, "tools", "Tools")):
|
||||
setup_tools(config, first_install=not is_existing)
|
||||
def _tools_step() -> None:
|
||||
if not (
|
||||
migration_ran
|
||||
and _skip_configured_section(config, "tools", "Tools")
|
||||
):
|
||||
setup_tools(config, first_install=not is_existing)
|
||||
|
||||
_run_setup_steps(
|
||||
[
|
||||
("Model & Provider", _model_step),
|
||||
("Terminal Backend", _terminal_step),
|
||||
("Messaging Platforms", _gateway_step),
|
||||
("Tools", _tools_step),
|
||||
]
|
||||
)
|
||||
|
||||
# Save and show summary
|
||||
save_config(config)
|
||||
|
||||
@@ -8,6 +8,7 @@ provider/model picker into its numbered "Select [1-N]" fallback the instant a
|
||||
user pressed up or down.
|
||||
"""
|
||||
import sys
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -17,12 +18,20 @@ if sys.platform == "win32":
|
||||
import curses
|
||||
|
||||
from hermes_cli.curses_ui import (
|
||||
NAV_BACK,
|
||||
NAV_CANCEL,
|
||||
NAV_DOWN,
|
||||
NAV_INTERRUPT,
|
||||
NAV_NONE,
|
||||
NAV_SELECT,
|
||||
NAV_TOGGLE,
|
||||
NAV_UP,
|
||||
MenuNavigationStart,
|
||||
_NumberedNavigation,
|
||||
curses_radiolist,
|
||||
read_menu_key,
|
||||
reset_menu_navigation_handler,
|
||||
set_menu_navigation_handler,
|
||||
)
|
||||
|
||||
|
||||
@@ -36,6 +45,7 @@ class FakeStdscr:
|
||||
def __init__(self, keys):
|
||||
self.keys = list(keys)
|
||||
self.timeouts = []
|
||||
self.writes = []
|
||||
|
||||
def getch(self):
|
||||
return self.keys.pop(0) if self.keys else -1
|
||||
@@ -43,6 +53,27 @@ class FakeStdscr:
|
||||
def timeout(self, ms):
|
||||
self.timeouts.append(ms)
|
||||
|
||||
def clear(self):
|
||||
pass
|
||||
|
||||
def getmaxyx(self):
|
||||
return (12, 80)
|
||||
|
||||
def addnstr(self, *args):
|
||||
self.writes.append(args)
|
||||
|
||||
def refresh(self):
|
||||
pass
|
||||
|
||||
|
||||
class ExhaustingStdscr(FakeStdscr):
|
||||
"""Fail instead of spinning if a key sequence is not fully handled."""
|
||||
|
||||
def getch(self):
|
||||
if not self.keys:
|
||||
raise AssertionError("menu requested another key after enhanced Enter")
|
||||
return self.keys.pop(0)
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -52,6 +83,271 @@ def test_raw_ss3_arrow_keys_decode():
|
||||
assert read_menu_key(FakeStdscr([27, ord("O"), ord("A")])) == NAV_UP
|
||||
|
||||
|
||||
def test_left_arrow_decodes_to_back():
|
||||
assert read_menu_key(FakeStdscr([curses.KEY_LEFT])) == NAV_BACK
|
||||
assert read_menu_key(FakeStdscr([27, ord("["), ord("D")])) == NAV_BACK
|
||||
assert read_menu_key(FakeStdscr([27, ord("O"), ord("D")])) == NAV_BACK
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("keys", "expected"),
|
||||
[
|
||||
([27, ord("["), ord("1"), ord("3"), ord("u")], NAV_SELECT),
|
||||
(
|
||||
[27, ord("["), ord("1"), ord("3"), ord(";"), ord("1"), ord("u")],
|
||||
NAV_SELECT,
|
||||
),
|
||||
([27, ord("["), ord("2"), ord("7"), ord("u")], NAV_CANCEL),
|
||||
([27, ord("["), ord("3"), ord("2"), ord("u")], NAV_TOGGLE),
|
||||
(
|
||||
[27, ord("["), ord("9"), ord("9"), ord(";"), ord("5"), ord("u")],
|
||||
NAV_INTERRUPT,
|
||||
),
|
||||
([27, ord("["), ord("1"), ord(";"), ord("1"), ord("D")], NAV_BACK),
|
||||
(
|
||||
[27, ord("["), ord("2"), ord("7"), ord(";"), ord("1"), ord(";"), ord("1"), ord("3"), ord("~")],
|
||||
NAV_SELECT,
|
||||
),
|
||||
(
|
||||
[27, ord("["), ord("2"), ord("7"), ord(";"), ord("5"), ord(";"), ord("9"), ord("9"), ord("~")],
|
||||
NAV_INTERRUPT,
|
||||
),
|
||||
(
|
||||
[27, ord("["), ord("1"), ord("3"), ord(";"), ord("1"), ord(":"), ord("3"), ord("u")],
|
||||
NAV_NONE,
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_enhanced_keyboard_sequences_decode(keys, expected):
|
||||
assert read_menu_key(FakeStdscr(keys)) == expected
|
||||
|
||||
|
||||
def test_raw_ctrl_c_cancels():
|
||||
assert read_menu_key(FakeStdscr([3])) == NAV_INTERRUPT
|
||||
|
||||
|
||||
def test_enhanced_enter_selects_filtered_model_while_search_is_active(monkeypatch):
|
||||
fake = ExhaustingStdscr(
|
||||
[
|
||||
ord("/"),
|
||||
ord("g"),
|
||||
27,
|
||||
ord("["),
|
||||
ord("1"),
|
||||
ord("3"),
|
||||
ord("u"),
|
||||
]
|
||||
)
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||||
monkeypatch.setattr(curses, "wrapper", lambda draw: draw(fake))
|
||||
monkeypatch.setattr(curses, "curs_set", lambda _value: None)
|
||||
monkeypatch.setattr(curses, "has_colors", lambda: False)
|
||||
monkeypatch.setattr("hermes_cli.curses_ui.flush_stdin", lambda: None)
|
||||
|
||||
selected = curses_radiolist(
|
||||
"Pick",
|
||||
["alpha", "gpt"],
|
||||
searchable=True,
|
||||
search_labels=["alpha", "gpt"],
|
||||
)
|
||||
|
||||
assert selected == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("enhanced_keys", "expected_event"),
|
||||
[
|
||||
([27, ord("["), ord("9"), ord("9"), ord(";"), ord("5"), ord("u")], "cancel"),
|
||||
([27, ord("["), ord("1"), ord(";"), ord("1"), ord("D")], "back"),
|
||||
],
|
||||
)
|
||||
def test_enhanced_control_keys_dispatch_while_search_is_active(
|
||||
monkeypatch, enhanced_keys, expected_event
|
||||
):
|
||||
class NavigationDispatched(Exception):
|
||||
pass
|
||||
|
||||
fake = FakeStdscr([ord("/"), ord("g"), *enhanced_keys])
|
||||
events = []
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||||
monkeypatch.setattr(curses, "wrapper", lambda draw: draw(fake))
|
||||
monkeypatch.setattr(curses, "curs_set", lambda _value: None)
|
||||
monkeypatch.setattr(curses, "has_colors", lambda: False)
|
||||
|
||||
def handler(event, *_args):
|
||||
events.append(event)
|
||||
if event == expected_event:
|
||||
raise NavigationDispatched()
|
||||
return MenuNavigationStart(allow_back=True) if event == "begin" else None
|
||||
|
||||
token = set_menu_navigation_handler(handler)
|
||||
try:
|
||||
with pytest.raises(NavigationDispatched):
|
||||
curses_radiolist(
|
||||
"Pick",
|
||||
["alpha", "gpt"],
|
||||
searchable=True,
|
||||
search_labels=["alpha", "gpt"],
|
||||
)
|
||||
finally:
|
||||
reset_menu_navigation_handler(token)
|
||||
|
||||
assert events == ["begin", expected_event]
|
||||
|
||||
|
||||
def test_numbered_fallback_ctrl_c_dispatches_scoped_cancel(monkeypatch):
|
||||
class Cancelled(Exception):
|
||||
pass
|
||||
|
||||
fake_error = curses.error("terminal unavailable")
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||||
monkeypatch.setattr(
|
||||
curses,
|
||||
"wrapper",
|
||||
lambda _draw: (_ for _ in ()).throw(fake_error),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.curses_ui._read_numbered_input",
|
||||
lambda _prompt="": _NumberedNavigation.CANCEL,
|
||||
)
|
||||
|
||||
def handler(event, *_args):
|
||||
if event == "cancel":
|
||||
raise Cancelled()
|
||||
return MenuNavigationStart() if event == "begin" else None
|
||||
|
||||
token = set_menu_navigation_handler(handler)
|
||||
try:
|
||||
with pytest.raises(Cancelled):
|
||||
curses_radiolist("Pick", ["a", "b"], cancel_returns=-1)
|
||||
finally:
|
||||
reset_menu_navigation_handler(token)
|
||||
|
||||
|
||||
def test_navigation_handler_programming_error_is_not_hidden_by_fallback(monkeypatch):
|
||||
fake = FakeStdscr([13])
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||||
monkeypatch.setattr(curses, "wrapper", lambda draw: draw(fake))
|
||||
monkeypatch.setattr(curses, "curs_set", lambda _value: None)
|
||||
monkeypatch.setattr(curses, "has_colors", lambda: False)
|
||||
|
||||
def handler(event, *_args):
|
||||
if event == "resolve":
|
||||
raise RuntimeError("navigation contract mismatch")
|
||||
return MenuNavigationStart() if event == "begin" else None
|
||||
|
||||
token = set_menu_navigation_handler(handler)
|
||||
try:
|
||||
with pytest.raises(RuntimeError, match="navigation contract mismatch"):
|
||||
curses_radiolist("Pick", ["a", "b"])
|
||||
finally:
|
||||
reset_menu_navigation_handler(token)
|
||||
|
||||
|
||||
def test_standalone_model_flow_renders_previous_and_reopens_provider(monkeypatch):
|
||||
from hermes_cli import main as main_mod
|
||||
|
||||
provider_screen = FakeStdscr([13])
|
||||
model_back_screen = FakeStdscr([curses.KEY_LEFT])
|
||||
provider_reselect_screen = FakeStdscr([13])
|
||||
model_select_screen = FakeStdscr([13])
|
||||
screens = [
|
||||
provider_screen,
|
||||
model_back_screen,
|
||||
provider_reselect_screen,
|
||||
model_select_screen,
|
||||
]
|
||||
resolved = []
|
||||
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||||
monkeypatch.setattr(curses, "wrapper", lambda draw: draw(screens.pop(0)))
|
||||
monkeypatch.setattr(curses, "curs_set", lambda _value: None)
|
||||
monkeypatch.setattr(curses, "has_colors", lambda: False)
|
||||
monkeypatch.setattr("hermes_cli.curses_ui.flush_stdin", lambda: None)
|
||||
monkeypatch.setattr(main_mod, "_require_tty", lambda _command: None)
|
||||
|
||||
def fake_select_provider_and_model(args=None):
|
||||
resolved.append(
|
||||
curses_radiolist("Select provider:", ["OpenAI"], cancel_returns=-1)
|
||||
)
|
||||
resolved.append(
|
||||
curses_radiolist(
|
||||
"Select default model:",
|
||||
["gpt-5.6-sol"],
|
||||
cancel_returns=-1,
|
||||
searchable=True,
|
||||
search_labels=["gpt-5.6-sol"],
|
||||
)
|
||||
)
|
||||
|
||||
monkeypatch.setattr(
|
||||
main_mod, "select_provider_and_model", fake_select_provider_and_model
|
||||
)
|
||||
|
||||
main_mod.cmd_model(SimpleNamespace(refresh=False))
|
||||
|
||||
assert resolved == [0, 0, 0]
|
||||
assert screens == []
|
||||
assert any(
|
||||
"\u2190 previous" in str(args[2]) for args in model_back_screen.writes
|
||||
)
|
||||
|
||||
|
||||
def test_radiolist_dispatches_contextual_back_navigation(monkeypatch):
|
||||
class WentBack(BaseException):
|
||||
pass
|
||||
|
||||
events = []
|
||||
fake = FakeStdscr([curses.KEY_LEFT])
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||||
monkeypatch.setattr(curses, "wrapper", lambda draw: draw(fake))
|
||||
monkeypatch.setattr(curses, "curs_set", lambda _value: None)
|
||||
monkeypatch.setattr(curses, "has_colors", lambda: False)
|
||||
monkeypatch.setattr("hermes_cli.curses_ui.flush_stdin", lambda: None)
|
||||
|
||||
def handler(event, *_args):
|
||||
events.append(event)
|
||||
if event == "back":
|
||||
raise WentBack()
|
||||
return MenuNavigationStart(allow_back=True) if event == "begin" else None
|
||||
|
||||
token = set_menu_navigation_handler(handler)
|
||||
try:
|
||||
with pytest.raises(WentBack):
|
||||
curses_radiolist("Pick", ["a", "b"])
|
||||
finally:
|
||||
reset_menu_navigation_handler(token)
|
||||
|
||||
assert events == ["begin", "back"]
|
||||
|
||||
|
||||
def test_radiolist_dispatches_contextual_escape_cancellation(monkeypatch):
|
||||
class Cancelled(BaseException):
|
||||
pass
|
||||
|
||||
fake = FakeStdscr([27])
|
||||
monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
|
||||
monkeypatch.setattr(curses, "wrapper", lambda draw: draw(fake))
|
||||
monkeypatch.setattr(curses, "curs_set", lambda _value: None)
|
||||
monkeypatch.setattr(curses, "has_colors", lambda: False)
|
||||
|
||||
def handler(event):
|
||||
if event == "cancel":
|
||||
raise Cancelled()
|
||||
return MenuNavigationStart() if event == "begin" else None
|
||||
|
||||
token = set_menu_navigation_handler(handler)
|
||||
try:
|
||||
with pytest.raises(Cancelled):
|
||||
curses_radiolist("Pick", ["a", "b"])
|
||||
finally:
|
||||
reset_menu_navigation_handler(token)
|
||||
|
||||
|
||||
def test_lone_escape_is_cancel():
|
||||
assert read_menu_key(FakeStdscr([27])) == NAV_CANCEL
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -60,7 +356,3 @@ 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
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -1,6 +1,203 @@
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli import setup as setup_mod
|
||||
|
||||
|
||||
def test_prompt_choice_escape_keeps_default_without_numbered_fallback(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
setup_mod,
|
||||
"_curses_prompt_choice",
|
||||
lambda question, choices, default=0, description=None: -1,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
"builtins.input",
|
||||
lambda *_args, **_kwargs: (_ for _ in ()).throw(
|
||||
AssertionError("Escape must not enter the numbered fallback")
|
||||
),
|
||||
)
|
||||
|
||||
assert setup_mod.prompt_choice("Pick one", ["a", "b"], default=1) == 1
|
||||
|
||||
|
||||
def test_setup_navigation_escape_cancels_and_left_goes_back():
|
||||
state = setup_mod._SetupNavigationState(section_index=1)
|
||||
token = setup_mod._SETUP_NAVIGATION.set(state)
|
||||
try:
|
||||
assert setup_mod._handle_setup_menu_navigation(
|
||||
setup_mod.MenuNavigationEvent.BEGIN
|
||||
).allow_back is True
|
||||
with pytest.raises(setup_mod._SetupCancelled):
|
||||
setup_mod._handle_setup_menu_navigation(
|
||||
setup_mod.MenuNavigationEvent.CANCEL
|
||||
)
|
||||
|
||||
assert setup_mod._handle_setup_menu_navigation(
|
||||
setup_mod.MenuNavigationEvent.BEGIN
|
||||
).allow_back is True
|
||||
with pytest.raises(setup_mod._SetupGoBack) as exc_info:
|
||||
setup_mod._handle_setup_menu_navigation(setup_mod.MenuNavigationEvent.BACK)
|
||||
assert exc_info.value.prompt_index == 1
|
||||
finally:
|
||||
setup_mod._SETUP_NAVIGATION.reset(token)
|
||||
|
||||
|
||||
def test_setup_yes_no_uses_navigable_menu(monkeypatch):
|
||||
calls = []
|
||||
state = setup_mod._SetupNavigationState(section_index=1)
|
||||
token = setup_mod._SETUP_NAVIGATION.set(state)
|
||||
monkeypatch.setattr(
|
||||
setup_mod,
|
||||
"_curses_prompt_choice",
|
||||
lambda question, choices, default=0, description=None: calls.append(
|
||||
(question, choices, default)
|
||||
)
|
||||
or 1,
|
||||
)
|
||||
try:
|
||||
assert setup_mod.prompt_yes_no("Enable it?", default=True) is False
|
||||
finally:
|
||||
setup_mod._SETUP_NAVIGATION.reset(token)
|
||||
|
||||
assert calls == [("Enable it?", ["Yes", "No"], 0)]
|
||||
|
||||
|
||||
def test_setup_steps_move_to_previous_section_or_restart_current_section():
|
||||
calls = []
|
||||
terminal_attempts = 0
|
||||
gateway_attempts = 0
|
||||
|
||||
def model():
|
||||
calls.append("model")
|
||||
|
||||
def terminal():
|
||||
nonlocal terminal_attempts
|
||||
calls.append("terminal")
|
||||
terminal_attempts += 1
|
||||
if terminal_attempts == 1:
|
||||
raise setup_mod._SetupGoBack(prompt_index=0)
|
||||
|
||||
def gateway():
|
||||
nonlocal gateway_attempts
|
||||
calls.append("gateway")
|
||||
gateway_attempts += 1
|
||||
if gateway_attempts == 1:
|
||||
raise setup_mod._SetupGoBack(prompt_index=1)
|
||||
|
||||
state = setup_mod._SetupNavigationState()
|
||||
token = setup_mod._SETUP_NAVIGATION.set(state)
|
||||
try:
|
||||
setup_mod._run_setup_steps(
|
||||
[("Model", model), ("Terminal", terminal), ("Gateway", gateway)]
|
||||
)
|
||||
finally:
|
||||
setup_mod._SETUP_NAVIGATION.reset(token)
|
||||
|
||||
assert calls == [
|
||||
"model",
|
||||
"terminal",
|
||||
"model",
|
||||
"terminal",
|
||||
"gateway",
|
||||
"gateway",
|
||||
]
|
||||
|
||||
|
||||
def test_nested_back_reopens_only_the_immediately_previous_prompt():
|
||||
shown = []
|
||||
attempts = 0
|
||||
|
||||
def model_provider_flow():
|
||||
nonlocal attempts
|
||||
attempts += 1
|
||||
for label in ("provider", "auth method", "existing or reauthenticate"):
|
||||
start = setup_mod._handle_setup_menu_navigation(
|
||||
setup_mod.MenuNavigationEvent.BEGIN
|
||||
)
|
||||
if start.should_replay:
|
||||
setup_mod._handle_setup_menu_navigation(
|
||||
setup_mod.MenuNavigationEvent.RESOLVE, start.replay_value
|
||||
)
|
||||
continue
|
||||
shown.append(label)
|
||||
if label == "existing or reauthenticate" and attempts == 1:
|
||||
setup_mod._handle_setup_menu_navigation(
|
||||
setup_mod.MenuNavigationEvent.BACK
|
||||
)
|
||||
setup_mod._handle_setup_menu_navigation(
|
||||
setup_mod.MenuNavigationEvent.RESOLVE, label
|
||||
)
|
||||
|
||||
state = setup_mod._SetupNavigationState()
|
||||
token = setup_mod._SETUP_NAVIGATION.set(state)
|
||||
try:
|
||||
setup_mod._run_setup_steps([("Model & Provider", model_provider_flow)])
|
||||
finally:
|
||||
setup_mod._SETUP_NAVIGATION.reset(token)
|
||||
|
||||
assert shown[:4] == [
|
||||
"provider",
|
||||
"auth method",
|
||||
"existing or reauthenticate",
|
||||
"auth method",
|
||||
]
|
||||
|
||||
|
||||
def test_section_specific_model_setup_can_go_back_from_model_to_provider(
|
||||
tmp_path, monkeypatch
|
||||
):
|
||||
"""``hermes setup model`` must retain nested setup navigation."""
|
||||
shown = []
|
||||
attempts = 0
|
||||
|
||||
def model_flow(_config):
|
||||
nonlocal attempts
|
||||
attempts += 1
|
||||
for label in ("provider", "model"):
|
||||
start = setup_mod._handle_setup_menu_navigation(
|
||||
setup_mod.MenuNavigationEvent.BEGIN
|
||||
)
|
||||
if start.should_replay:
|
||||
setup_mod._handle_setup_menu_navigation(
|
||||
setup_mod.MenuNavigationEvent.RESOLVE, start.replay_value
|
||||
)
|
||||
continue
|
||||
shown.append(label)
|
||||
if label == "model" and attempts == 1:
|
||||
# This is what the menu does for Left: it only dispatches the
|
||||
# back event when its setup context enables ``← previous``.
|
||||
if start.allow_back:
|
||||
setup_mod._handle_setup_menu_navigation(
|
||||
setup_mod.MenuNavigationEvent.BACK
|
||||
)
|
||||
return
|
||||
setup_mod._handle_setup_menu_navigation(
|
||||
setup_mod.MenuNavigationEvent.RESOLVE, label
|
||||
)
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
monkeypatch.setattr(setup_mod, "is_interactive_stdin", lambda: True)
|
||||
monkeypatch.setattr(
|
||||
setup_mod,
|
||||
"SETUP_SECTIONS",
|
||||
[("model", "Model & Provider", model_flow)],
|
||||
)
|
||||
|
||||
setup_mod.run_setup_wizard(
|
||||
SimpleNamespace(
|
||||
section="model",
|
||||
reset=False,
|
||||
reconfigure=False,
|
||||
quick=False,
|
||||
portal=False,
|
||||
non_interactive=False,
|
||||
)
|
||||
)
|
||||
|
||||
assert shown == ["provider", "model", "provider", "model"]
|
||||
|
||||
|
||||
def test_prompt_strips_bracketed_paste_markers(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
"builtins.input",
|
||||
@@ -20,5 +217,3 @@ def test_prompt_choice_uses_curses_helper(monkeypatch):
|
||||
idx = setup_mod.prompt_choice("Pick one", ["a", "b", "c"], default=0)
|
||||
|
||||
assert idx == 1
|
||||
|
||||
|
||||
|
||||
@@ -148,9 +148,16 @@ class TestQuickFlag:
|
||||
tools="hermes_cli.setup.setup_tools",
|
||||
)
|
||||
from hermes_cli.setup import run_setup_wizard
|
||||
from hermes_cli import setup as setup_mod
|
||||
|
||||
section_indexes = []
|
||||
m["quick"].side_effect = lambda *_args: section_indexes.append(
|
||||
setup_mod._SETUP_NAVIGATION.get().section_index
|
||||
)
|
||||
run_setup_wizard(args)
|
||||
|
||||
m["quick"].assert_called_once()
|
||||
assert section_indexes == [0]
|
||||
# Full reconfigure sections must NOT run.
|
||||
m["model"].assert_not_called()
|
||||
m["terminal"].assert_not_called()
|
||||
@@ -173,10 +180,37 @@ class TestFreshInstall:
|
||||
first="hermes_cli.setup._run_first_time_quick_setup",
|
||||
)
|
||||
from hermes_cli.setup import run_setup_wizard
|
||||
from hermes_cli import setup as setup_mod
|
||||
|
||||
section_indexes = []
|
||||
m["first"].side_effect = lambda *_args: section_indexes.append(
|
||||
setup_mod._SETUP_NAVIGATION.get().section_index
|
||||
)
|
||||
run_setup_wizard(args)
|
||||
|
||||
m["prompt"].assert_called_once()
|
||||
m["first"].assert_called_once()
|
||||
assert section_indexes == [0]
|
||||
|
||||
def test_blank_slate_runs_inside_navigation_step(self, fresh_install):
|
||||
args = _make_setup_args()
|
||||
|
||||
with ExitStack() as stack:
|
||||
m = _enter_fresh_install_patches(
|
||||
stack,
|
||||
prompt=("hermes_cli.setup.prompt_choice", {"return_value": 2}),
|
||||
blank="hermes_cli.setup._run_blank_slate_setup",
|
||||
)
|
||||
from hermes_cli import setup as setup_mod
|
||||
|
||||
section_indexes = []
|
||||
m["blank"].side_effect = lambda *_args: section_indexes.append(
|
||||
setup_mod._SETUP_NAVIGATION.get().section_index
|
||||
)
|
||||
setup_mod.run_setup_wizard(args)
|
||||
|
||||
m["blank"].assert_called_once()
|
||||
assert section_indexes == [0]
|
||||
|
||||
|
||||
class TestArgparse:
|
||||
@@ -198,5 +232,3 @@ class TestArgparse:
|
||||
pass
|
||||
assert captured["args"].reconfigure is True
|
||||
assert captured["args"].quick is False
|
||||
|
||||
|
||||
|
||||
@@ -4,6 +4,8 @@ cannot initialize (e.g. non-TTY, curses unavailable, terminal error)."""
|
||||
import subprocess
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli.config import load_config, save_config
|
||||
|
||||
|
||||
@@ -13,6 +15,46 @@ def _raise_menu(*args, **kwargs):
|
||||
raise subprocess.CalledProcessError(2, ["tput", "clear"])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("sequence", "expected"),
|
||||
[
|
||||
("\x1b[27u", "cancel"),
|
||||
("\x1b[D", "back"),
|
||||
("\x1b[99;5u", "cancel"),
|
||||
],
|
||||
)
|
||||
def test_scoped_numbered_input_handles_navigation_keys(sequence, expected):
|
||||
"""The curses fallback stays escapable on POSIX and native Windows."""
|
||||
from prompt_toolkit.application import create_app_session
|
||||
from prompt_toolkit.input.defaults import create_pipe_input
|
||||
from prompt_toolkit.output import DummyOutput
|
||||
|
||||
from hermes_cli.curses_ui import (
|
||||
MenuNavigationStart,
|
||||
_NUMBERED_BACK_ENABLED,
|
||||
_NumberedNavigation,
|
||||
_read_numbered_input,
|
||||
reset_menu_navigation_handler,
|
||||
set_menu_navigation_handler,
|
||||
)
|
||||
|
||||
def handler(event, *_args):
|
||||
return MenuNavigationStart(allow_back=True) if event == "begin" else None
|
||||
|
||||
token = set_menu_navigation_handler(handler)
|
||||
back_token = _NUMBERED_BACK_ENABLED.set(True)
|
||||
try:
|
||||
with create_pipe_input() as pipe_input:
|
||||
pipe_input.send_text(sequence)
|
||||
with create_app_session(input=pipe_input, output=DummyOutput()):
|
||||
result = _read_numbered_input("Choice: ")
|
||||
finally:
|
||||
_NUMBERED_BACK_ENABLED.reset(back_token)
|
||||
reset_menu_navigation_handler(token)
|
||||
|
||||
assert result is getattr(_NumberedNavigation, expected.upper())
|
||||
|
||||
|
||||
|
||||
|
||||
def test_prompt_model_selection_requires_expensive_confirmation(monkeypatch, capsys):
|
||||
|
||||
Reference in New Issue
Block a user