diff --git a/hermes_cli/curses_ui.py b/hermes_cli/curses_ui.py index 9a127d936d..577e69d977 100644 --- a/hermes_cli/curses_ui.py +++ b/hermes_cli/curses_ui.py @@ -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() diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 2b12dfdcf8..268752b009 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -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: diff --git a/hermes_cli/setup.py b/hermes_cli/setup.py index 3d43d22eff..716eb6f368 100644 --- a/hermes_cli/setup.py +++ b/hermes_cli/setup.py @@ -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) diff --git a/tests/hermes_cli/test_curses_arrow_keys.py b/tests/hermes_cli/test_curses_arrow_keys.py index 36892213d7..d9e345f64d 100644 --- a/tests/hermes_cli/test_curses_arrow_keys.py +++ b/tests/hermes_cli/test_curses_arrow_keys.py @@ -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 - - - - diff --git a/tests/hermes_cli/test_setup_prompt_menus.py b/tests/hermes_cli/test_setup_prompt_menus.py index 1af86cd9a8..7c12247e75 100644 --- a/tests/hermes_cli/test_setup_prompt_menus.py +++ b/tests/hermes_cli/test_setup_prompt_menus.py @@ -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 - - diff --git a/tests/hermes_cli/test_setup_reconfigure.py b/tests/hermes_cli/test_setup_reconfigure.py index 773b8ed8e9..da2cd9eaf7 100644 --- a/tests/hermes_cli/test_setup_reconfigure.py +++ b/tests/hermes_cli/test_setup_reconfigure.py @@ -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 - - diff --git a/tests/hermes_cli/test_terminal_menu_fallbacks.py b/tests/hermes_cli/test_terminal_menu_fallbacks.py index 8d077a8852..9e69c5190c 100644 --- a/tests/hermes_cli/test_terminal_menu_fallbacks.py +++ b/tests/hermes_cli/test_terminal_menu_fallbacks.py @@ -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):