ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

merge: r3-18 group C into simp/r3-18

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:15:46 -07:00
parent aec4f2b745 e1a22b7845
commit b279fd32c7
5 changed files with 937 additions and 1573 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/commands_completion.py
+135 -195
View File
@@ -1,12 +1,12 @@
"""prompt_toolkit completer + inline auto-suggest for slash commands.
Extracted from :mod:`hermes_cli.commands` (which re-exports
``SlashCommandCompleter`` / ``SlashCommandAutoSuggest``). The registry module
stays prompt_toolkit-free so the gateway can import it without the dependency.
Kept out of :mod:`hermes_cli.commands` (which re-exports both classes) so the registry module
stays prompt_toolkit-free for the gateway.
"""
from __future__ import annotations
import functools
import os
import shutil
import subprocess
@@ -29,18 +29,14 @@ _personalities_memo: Optional[
def _personalities_from_cli_config() -> Dict[str, Any]:
"""``available_personalities(load_cli_config())`` memoised on config path+mtime+size.
load_cli_config() does a full YAML parse + deep merge and the completer
runs per keystroke; the result only changes when config.yaml changes on
disk (same pattern as load_env). Falls back to a fresh load when the file
cannot be stat'ed.
load_cli_config() is a full YAML parse + deep merge and the completer runs per keystroke.
Falls back to a fresh load when the file cannot be stat'ed.
"""
global _personalities_memo
from cli import load_cli_config
from hermes_cli.personality import available_personalities
try:
from hermes_cli.config import get_config_path
cfg_path = get_config_path()
st = cfg_path.stat()
sig = (str(cfg_path), st.st_mtime_ns, st.st_size)
@@ -69,15 +65,16 @@ def _file_size_label(path: str) -> str:
return ""
if size < 1024:
return f"{size}B"
if size < 1024 * 1024:
if size < 1024 ** 2:
return f"{size / 1024:.0f}K"
if size < 1024 * 1024 * 1024:
return f"{size / (1024 * 1024):.1f}M"
return f"{size / (1024 * 1024 * 1024):.1f}G"
if size < 1024 ** 3:
return f"{size / 1024 ** 2:.1f}M"
return f"{size / 1024 ** 3:.1f}G"
def _prefix_completions(rows: Iterable[tuple[str, Any]], partial: str, *, skip_exact: bool = True):
"""Yield a Completion per ``(name, meta)`` whose name starts with *partial* (case-folded partial)."""
def _prefix_completions(
rows: Iterable[tuple[str, Any]], partial: str, *, skip_exact: bool = True):
"""A Completion per ``(name, meta)`` whose name starts with the case-folded *partial*."""
lowered = partial.lower()
for name, meta in rows:
if name.startswith(lowered) and not (skip_exact and name == lowered):
@@ -85,46 +82,51 @@ def _prefix_completions(rows: Iterable[tuple[str, Any]], partial: str, *, skip_e
def _split_args(sub_text: str) -> tuple[list[str], str]:
"""``(completed_words, partial)`` for multi-word argument text; a trailing space means a fresh word."""
"""``(completed_words, partial)``; a trailing space means a fresh word."""
parts = sub_text.split()
if sub_text.endswith(" "):
if sub_text.endswith(" ") or not parts:
return parts, ""
return parts[:-1], (parts[-1] if parts else "")
return parts[:-1], parts[-1]
# ---------------------------------------------------------------------------
# Dynamic argument completers: (sub_text, sub_lower) -> Completion iterator
# ---------------------------------------------------------------------------
def _quiet(gen_fn):
"""Generator decorator: any exception while producing completions just ends the stream."""
@functools.wraps(gen_fn)
def wrapper(*args):
try:
yield from gen_fn(*args)
except Exception:
return
return wrapper
@_quiet
def _skin_completions(sub_text: str, sub_lower: str):
"""/skin — available skins."""
try:
from hermes_cli.skin_engine import list_skins
rows = ((s["name"], s.get("description", "") or s.get("source", "")) for s in list_skins())
yield from _prefix_completions(rows, sub_text)
except Exception:
pass
from hermes_cli.skin_engine import list_skins
rows = ((s["name"], s.get("description", "") or s.get("source", "")) for s in list_skins())
yield from _prefix_completions(rows, sub_text)
@_quiet
def _personality_completions(sub_text: str, sub_lower: str):
"""/personality — ``none`` plus configured personalities."""
try:
from hermes_cli.personality import describe_personality
personalities = _personalities_from_cli_config()
rows = chain([("none", "clear personality overlay")],
((name, describe_personality(prompt)) for name, prompt in personalities.items()))
yield from _prefix_completions(rows, sub_text)
except Exception:
pass
from hermes_cli.personality import describe_personality
personalities = _personalities_from_cli_config()
rows = chain(
[("none", "clear personality overlay")],
((name, describe_personality(prompt)) for name, prompt in personalities.items()))
yield from _prefix_completions(rows, sub_text)
def _tools_completions(sub_text: str, sub_lower: str):
"""/tools — subcommand, then toolset / MCP-server names for enable|disable.
Toolsets are offered only when the subcommand would change their state
(enable → currently off, disable → currently on); MCP server prefixes are
always offered.
Toolsets are offered only when the subcommand would change their state; MCP server
prefixes are always offered.
"""
completed, partial = _split_args(sub_text)
if not completed:
@@ -136,35 +138,35 @@ def _tools_completions(sub_text: str, sub_lower: str):
already = set(completed[1:])
try:
from hermes_cli.config import load_config_readonly
from hermes_cli.tools_config import CONFIGURABLE_TOOLSETS, _get_platform_tools, _get_plugin_toolset_keys
from hermes_cli.tools_config import (
CONFIGURABLE_TOOLSETS, _get_platform_tools, _get_plugin_toolset_keys)
# Readonly loader: this runs per keystroke and never mutates config,
# so skip the defensive deepcopy of load_config().
# Readonly loader: per keystroke and never mutates, so skip load_config()'s deepcopy.
config = load_config_readonly()
enabled = _get_platform_tools(config, "cli", include_default_mcp_servers=False)
mcp_servers = config.get("mcp_servers") or {}
want_enabled = subcommand != "enable"
rows = [(k, label) for k, label, _d in CONFIGURABLE_TOOLSETS if (k in enabled) == want_enabled]
rows += [(k, "plugin toolset") for k in sorted(_get_plugin_toolset_keys()) if (k in enabled) == want_enabled]
rows = [(k, label) for k, label, _d in CONFIGURABLE_TOOLSETS]
rows += [(k, "plugin toolset") for k in sorted(_get_plugin_toolset_keys())]
rows = [(k, m) for k, m in rows if (k in enabled) == want_enabled]
if isinstance(mcp_servers, dict):
rows += [(f"{srv}:", f"MCP server '{srv}'") for srv in sorted(mcp_servers)]
yield from _prefix_completions(((k, m) for k, m in rows if k not in already), partial, skip_exact=False)
yield from _prefix_completions(
((k, m) for k, m in rows if k not in already), partial, skip_exact=False)
except Exception:
return
def _handoff_completions(sub_text: str, sub_lower: str):
"""/handoff — connected (enabled + configured) gateway platforms, first arg only.
"""/handoff — connected gateway platforms, first arg only.
A recorded home channel is NOT required to list a platform — it's often
learned at runtime — so the meta hints whether one is set yet.
A home channel is not required (often learned at runtime); the meta hints whether one is set.
"""
completed, partial = _split_args(sub_text)
if completed:
return
try:
from gateway.config import load_gateway_config
gw = load_gateway_config()
platforms = gw.get_connected_platforms()
except Exception:
@@ -178,51 +180,42 @@ def _handoff_completions(sub_text: str, sub_lower: str):
home = gw.get_home_channel(platform)
except Exception:
home = None
meta = f"→ {home.name}" if home and getattr(home, "name", None) else "send this session here"
home_name = getattr(home, "name", None) if home else None
meta = f"→ {home_name}" if home_name else "send this session here"
yield Completion(name, start_position=-len(partial), display=name, display_meta=meta)
# base command -> (handler(sub_text, sub_lower), single_word_only).
# Single-word handlers only run while the first argument is being typed;
# /tools and /handoff parse multi-word input themselves, bypassing the
# static SUBCOMMANDS branch.
# base command -> (handler(sub_text, sub_lower), single_word_only). Single-word handlers only
# run while the first argument is typed; /tools and /handoff parse multi-word input themselves.
_DYNAMIC_COMPLETIONS: dict[str, tuple[Callable[..., Any], bool]] = {
"/skin": (_skin_completions, True),
"/personality": (_personality_completions, True),
"/tools": (_tools_completions, False),
"/handoff": (_handoff_completions, False),
}
"/handoff": (_handoff_completions, False)}
# ---------------------------------------------------------------------------
# Path / @-context completion
# ---------------------------------------------------------------------------
def _extract_path_word(text: str) -> str | None:
"""Path-like word under the cursor (``./``, ``../``, ``~/``, ``/`` or contains ``/``), else None.
Tokens with a ``://`` scheme are excluded — treating a pasted URL as a
path fires os.listdir per keystroke for no useful result.
"""
"""Word under the cursor when it contains ``/`` and no ``://`` scheme (URLs aren't paths)."""
word = text.rpartition(" ")[2]
if not word or "://" in word:
return None
return word if "/" in word else None
return word if word and "://" not in word and "/" in word else None
def _dir_completions(expanded: str, word: str, limit: int, text_for: Callable[[str], str], want_dir: bool | None = None):
"""Yield directory-listing completions for the path *expanded*.
def _dir_completions(
expanded: str, word: str, limit: int, text_for: Callable[[str], str],
want_dir: bool | None = None):
"""Directory-listing completions for *expanded*.
Entries of the parent dir are matched case-insensitively on the typed
basename (all entries after a trailing ``/``), sorted by name, and
limited to *limit*. ``text_for(full_path)`` builds the completion text
(without the trailing ``/``); *want_dir* restricts to dirs / files.
Parent-dir entries are matched case-insensitively on the typed basename (all entries after
a trailing ``/``), sorted, limited to *limit*. ``text_for(full_path)`` builds the completion
text (without trailing ``/``); *want_dir* restricts to dirs / files.
"""
if expanded.endswith("/"):
search_dir, prefix = expanded, ""
else:
search_dir = os.path.dirname(expanded) or "."
prefix = os.path.basename(expanded)
search_dir, prefix = os.path.dirname(expanded) or ".", os.path.basename(expanded)
try:
entries = os.listdir(search_dir)
except OSError:
@@ -230,7 +223,7 @@ def _dir_completions(expanded: str, word: str, limit: int, text_for: Callable[[s
prefix_lower = prefix.lower()
count = 0
for entry in sorted(entries):
if prefix and not entry.lower().startswith(prefix_lower):
if not entry.lower().startswith(prefix_lower):
continue
full_path = os.path.join(search_dir, entry)
is_dir = os.path.isdir(full_path)
@@ -240,22 +233,17 @@ def _dir_completions(expanded: str, word: str, limit: int, text_for: Callable[[s
break
suffix = "/" if is_dir else ""
yield Completion(
text_for(full_path) + suffix,
start_position=-len(word),
display=entry + suffix,
display_meta="dir" if is_dir else _file_size_label(full_path),
)
text_for(full_path) + suffix, start_position=-len(word), display=entry + suffix,
display_meta="dir" if is_dir else _file_size_label(full_path))
count += 1
def _path_completions(word: str, limit: int = 30):
"""Yield Completion objects for file paths matching *word*, keeping the user's path style (~, absolute, relative)."""
"""Path completions for *word*, keeping the user's style (~, absolute, relative)."""
if word.startswith("~"):
text_for = lambda fp: "~/" + os.path.relpath(fp, os.path.expanduser("~")) # noqa: E731
elif os.path.isabs(word):
text_for = lambda fp: fp # noqa: E731
else:
text_for = os.path.relpath
text_for = str if os.path.isabs(word) else os.path.relpath
yield from _dir_completions(os.path.expanduser(word), word, limit, text_for)
@@ -265,14 +253,13 @@ _STATIC_CONTEXT_REFS = (
("@file:", "Attach a file"),
("@folder:", "Attach a folder"),
("@git:", "Git log with diffs (e.g. @git:5)"),
("@url:", "Fetch web content"),
)
("@url:", "Fetch web content"))
def _score_path(filepath: str, query: str) -> int:
"""Score a file path against a fuzzy query. Higher = better match; 0 = no match."""
"""Score a file path against a fuzzy query. Higher = better; 0 = no match; 1 = empty query."""
if not query:
return 1 # show everything when query is empty
return 1
lower_file = os.path.basename(filepath).lower()
lower_q = query.lower()
if lower_file == lower_q:
@@ -283,8 +270,8 @@ def _score_path(filepath: str, query: str) -> int:
return 60
if lower_q in filepath.lower():
return 40
# Abbreviation match: query chars appear in order in the filename ("fo" ~
# "file_operations"); bonus when >= half land on word boundaries (_-./).
# Abbreviation: query chars in order in the filename ("fo" ~ "file_operations"); bonus when
# >= half land on word boundaries (_-./).
qi = boundary_hits = 0
prev = "_" # treat start as boundary
for c in lower_file:
@@ -300,27 +287,20 @@ def _score_path(filepath: str, query: str) -> int:
class SlashCommandCompleter(Completer):
"""Autocomplete for built-in slash commands, subcommands, and skill commands."""
# Commands that open pickers when run bare. No trailing space for these:
# the TUI applies the completion on Enter, and "/model " blocks the picker.
# Bare-run picker commands get no trailing space: "/model " would block the picker.
_PICKER_COMMANDS = frozenset({"model", "skin", "personality"})
# Module-level helpers exposed as staticmethods for existing callers/tests.
_extract_path_word = staticmethod(_extract_path_word)
_dir_completions = staticmethod(_dir_completions)
_path_completions = staticmethod(_path_completions)
_score_path = staticmethod(_score_path)
_skin_completions = staticmethod(_skin_completions)
_personality_completions = staticmethod(_personality_completions)
_tools_completions = staticmethod(_tools_completions)
_handoff_completions = staticmethod(_handoff_completions)
_DYNAMIC_COMPLETIONS = _DYNAMIC_COMPLETIONS
def __init__(
self,
skill_commands_provider: Callable[[], Mapping[str, dict[str, Any]]] | None = None,
command_filter: Callable[[str], bool] | None = None,
skill_bundles_provider: Callable[[], Mapping[str, dict[str, Any]]] | None = None,
) -> None:
skill_bundles_provider: Callable[[], Mapping[str, dict[str, Any]]] | None = None) -> None:
self._skill_commands_provider = skill_commands_provider
self._command_filter = command_filter
self._skill_bundles_provider = skill_bundles_provider
@@ -330,45 +310,35 @@ class SlashCommandCompleter(Completer):
self._file_cache_cwd: str = ""
def _command_allowed(self, slash_command: str) -> bool:
if self._command_filter is None:
return True
try:
return bool(self._command_filter(slash_command))
return self._command_filter is None or bool(self._command_filter(slash_command))
except Exception:
return True
@staticmethod
def _call_provider(provider) -> Mapping[str, dict[str, Any]]:
if provider is None:
return {}
try:
return provider() or {}
return (provider() if provider is not None else None) or {}
except Exception:
return {}
def _iter_skill_commands(self) -> Mapping[str, dict[str, Any]]:
return self._call_provider(self._skill_commands_provider)
def _iter_skill_bundles(self) -> Mapping[str, dict[str, Any]]:
return self._call_provider(self._skill_bundles_provider)
# -- stacked slash-skill completion helpers ---------------------------
@staticmethod
def _normalize_skill_token(token: str) -> str:
"""Canonical hyphenated /slug form; mirrors resolve_skill_command_key() (underscores == hyphens)."""
"""Canonical hyphenated /slug; mirrors resolve_skill_command_key() (``_`` == ``-``)."""
return "/" + token.lstrip("/").replace("_", "-").lower()
def _is_skill_command(self, token: str) -> bool:
return self._normalize_skill_token(token) in self._iter_skill_commands()
def _stacked_skill_completions(self, text: str):
"""Offer skill-command completions for stacked invocations (``/skill-a /skill-b do XYZ``).
"""Skill-command completions for stacked invocations (``/skill-a /skill-b do XYZ``).
Keep suggesting while every completed token is a distinct skill
command, the cap is not reached, and the current word starts with
``/``; once the chain breaks, offer nothing — instruction text must
never be polluted with skill suggestions.
Suggest only while every completed token is a distinct skill command, the cap is not
reached, and the current word starts with ``/`` — instruction text must never be
polluted with skill suggestions.
"""
try:
from agent.skill_commands import _MAX_STACKED_SKILLS as _cap
@@ -383,27 +353,23 @@ class SlashCommandCompleter(Completer):
if key not in skill_cmds or key in seen:
return
seen.add(key)
# A bare space after the chain means they may be starting the instruction.
if len(seen) >= _cap or not current_word.startswith("/"):
return
return # a bare space after the chain may start the instruction
word_key = self._normalize_skill_token(current_word)
for cmd, info in skill_cmds.items():
if cmd in seen or not cmd.startswith(word_key):
continue
# Exact match: trailing space keeps the dropdown visible so the
# next stacked token can be typed immediately (see _completion_text).
# Exact match: trailing space keeps the dropdown open for the next stacked token.
yield Completion(
f"{cmd} " if cmd == word_key else cmd,
start_position=-len(current_word),
display=cmd,
display_meta=f"⚡ {_short_desc(info, 'Skill command')}",
)
display_meta=f"⚡ {_short_desc(info, 'Skill command')}")
@staticmethod
def _completion_text(cmd_name: str, word: str) -> str:
"""Replacement text: on an exact match a no-op replacement makes prompt_toolkit
suppress the menu, so a trailing space is appended — except for _PICKER_COMMANDS."""
"""Replacement text; exact matches get a trailing space (else prompt_toolkit hides the
menu on a no-op replacement) — except _PICKER_COMMANDS."""
if cmd_name != word or cmd_name in SlashCommandCompleter._PICKER_COMMANDS:
return cmd_name
return f"{cmd_name} "
@@ -415,33 +381,29 @@ class SlashCommandCompleter(Completer):
return word if word.startswith("@") else None
def _context_completions(self, word: str, limit: int = 30):
"""Claude Code-style @ completions: static refs, ``@file:``/``@folder:`` paths, else fuzzy project files."""
"""@ completions: static refs, ``@file:``/``@folder:`` paths, else fuzzy project files."""
lowered = word.lower()
for candidate, meta in _STATIC_CONTEXT_REFS:
if candidate.startswith(lowered) and candidate != lowered:
yield Completion(candidate, start_position=-len(word), display=candidate, display_meta=meta)
# Accepting the bare `@file` / `@folder` (no colon yet) lets the picker
# surface entries without first accepting the static hint.
yield Completion(
candidate, start_position=-len(word), display=candidate, display_meta=meta)
# Bare `@file` / `@folder` (no colon yet) already opens the picker.
for prefix in ("@file:", "@folder:"):
bare = prefix[:-1]
if word == bare or word.startswith(prefix):
expanded = os.path.expanduser("" if word == bare else word[len(prefix):])
if not expanded or expanded == ".":
expanded = "./"
# `@folder:` surfaces only directories, `@file:` only regular
# files — otherwise `@folder:` lists every dotfile in cwd.
# `@folder:` = dirs only, `@file:` = files only (else `@folder:` lists dotfiles).
yield from _dir_completions(
expanded, word, limit,
lambda fp: f"{prefix}{os.path.relpath(fp)}",
want_dir=(prefix == "@folder:"),
)
expanded, word, limit, lambda fp: f"{prefix}{os.path.relpath(fp)}",
want_dir=(prefix == "@folder:"))
return
yield from self._fuzzy_file_completions(word, word[1:], limit)
def _get_project_files(self) -> list[str]:
"""Return cached list of project files (refreshed every 5s); rg (gitignore-aware) then fd."""
"""Cached project file list (5s TTL); rg (gitignore-aware) then fd."""
cwd = os.getcwd()
now = time.monotonic()
if self._file_cache and self._file_cache_cwd == cwd and now - self._file_cache_time < 5.0:
@@ -451,15 +413,13 @@ class SlashCommandCompleter(Completer):
for cmd in (
["rg", "--files", "--sortr=modified", cwd],
["rg", "--files", cwd],
["fd", "--type", "f", "--base-directory", cwd],
):
["fd", "--type", "f", "--base-directory", cwd]):
if not shutil.which(cmd[0]):
continue
try:
proc = subprocess.run(
cmd, capture_output=True, text=True, timeout=2,
cwd=cwd, encoding="utf-8", errors="replace",
)
cwd=cwd, encoding="utf-8", errors="replace")
except (subprocess.TimeoutExpired, OSError):
continue
if proc.returncode != 0 or not proc.stdout.strip():
@@ -468,26 +428,21 @@ class SlashCommandCompleter(Completer):
try:
files.append(os.path.relpath(p, cwd) if os.path.isabs(p) else p)
except ValueError:
# Windows: relpath raises for paths on a different mount than
# cwd (\\.\nul, other drive letter). One bad entry must not
# crash the @ autocomplete event loop.
continue
continue # Windows: relpath raises across mounts/drive letters
break
self._file_cache, self._file_cache_time, self._file_cache_cwd = files, now, cwd
return files
def _fuzzy_file_completions(self, word: str, query: str, limit: int = 20):
"""Yield fuzzy file completions for bare @query (no query = recently modified files)."""
"""Fuzzy file completions for bare @query (no query = recently modified files)."""
files = self._get_project_files()
if not query:
ranked = files[:limit]
else:
scored = [(s, fp) for fp in files if (s := _score_path(fp, query)) > 0]
scored.sort(key=lambda x: (-x[0], x[1]))
ranked = [fp for _, fp in scored[:limit]]
for fp in ranked:
if query:
scored = sorted(
((s, fp) for fp in files if (s := _score_path(fp, query)) > 0),
key=lambda x: (-x[0], x[1]))
files = [fp for _, fp in scored]
for fp in files[:limit]:
is_dir = fp.endswith("/")
meta = "dir" if is_dir else _file_size_label(os.path.join(os.getcwd(), fp))
if query:
@@ -496,18 +451,16 @@ class SlashCommandCompleter(Completer):
f"@{'folder' if is_dir else 'file'}:{fp}",
start_position=-len(word),
display=os.path.basename(fp),
display_meta=meta,
)
display_meta=meta)
def get_completions(self, document, complete_event):
text = document.text_before_cursor
if not text.startswith("/"):
ctx_word = self._extract_context_word(text)
path_word = _extract_path_word(text)
if ctx_word is not None:
yield from self._context_completions(ctx_word)
return
path_word = _extract_path_word(text)
if path_word is not None:
elif path_word is not None:
yield from _path_completions(path_word)
return
@@ -516,71 +469,64 @@ class SlashCommandCompleter(Completer):
if len(parts) > 1 or text.endswith(" "):
# Completing arguments: base command already typed.
sub_text = parts[1] if len(parts) > 1 else ""
sub_lower = sub_text.lower()
# Stacked slash-skill chain (`/skill-a /skill-b …`), see
# split_stacked_skill_commands in agent/skill_commands.py.
first_arg = " " not in sub_text
# Stacked slash-skill chain (see split_stacked_skill_commands in agent/skill_commands).
if self._is_skill_command(base_cmd):
yield from self._stacked_skill_completions(text)
return
dynamic = _DYNAMIC_COMPLETIONS.get(base_cmd)
if dynamic is not None:
handler, single_word = dynamic
if not single_word or " " not in sub_text:
yield from handler(sub_text, sub_lower)
return
if " " not in sub_text and base_cmd in SUBCOMMANDS and self._command_allowed(base_cmd):
yield from _prefix_completions(((s, None) for s in SUBCOMMANDS[base_cmd]), sub_text)
handler, single_word = _DYNAMIC_COMPLETIONS.get(base_cmd, (None, False))
if handler is not None and (not single_word or first_arg):
yield from handler(sub_text, sub_text.lower())
elif first_arg and base_cmd in SUBCOMMANDS and self._command_allowed(base_cmd):
yield from _prefix_completions(
((s, None) for s in SUBCOMMANDS[base_cmd]), sub_text)
return
word = text[1:]
def _cmd_completion(cmd_name: str, meta: str):
return Completion(
self._completion_text(cmd_name, word),
start_position=-len(word),
display=f"/{cmd_name}",
display_meta=meta,
)
self._completion_text(cmd_name, word), start_position=-len(word),
display=f"/{cmd_name}", display_meta=meta)
for cmd, desc in COMMANDS.items():
if self._command_allowed(cmd) and cmd[1:].startswith(word):
yield _cmd_completion(cmd[1:], desc)
for cmd, info in self._iter_skill_bundles().items():
for cmd, info in self._call_provider(self._skill_bundles_provider).items():
if cmd[1:].startswith(word):
skill_count = len(info.get("skills", []))
yield _cmd_completion(cmd[1:], f"▣ {_short_desc(info, 'Skill bundle')} ({skill_count} skills)")
yield _cmd_completion(
cmd[1:], f"▣ {_short_desc(info, 'Skill bundle')} ({skill_count} skills)")
for cmd, info in self._iter_skill_commands().items():
if cmd[1:].startswith(word):
yield _cmd_completion(cmd[1:], f"⚡ {_short_desc(info, 'Skill command')}")
try:
from hermes_cli.plugins import get_plugin_commands
for cmd_name, cmd_info in get_plugin_commands().items():
if cmd_name.startswith(word):
yield _cmd_completion(cmd_name, f"🔌 {_short_desc(cmd_info, 'Plugin command')}")
yield _cmd_completion(
cmd_name, f"🔌 {_short_desc(cmd_info, 'Plugin command')}")
except Exception:
pass
class SlashCommandAutoSuggest(AutoSuggest):
"""Inline ghost-text for slash commands and their subcommands; history fallback for other input."""
"""Inline ghost-text for slash commands and subcommands; history fallback for other input."""
def __init__(
self,
history_suggest: AutoSuggest | None = None,
completer: SlashCommandCompleter | None = None,
) -> None:
completer: SlashCommandCompleter | None = None) -> None:
self._history = history_suggest
self._completer = completer # Reuse its model cache
def _allowed(self, cmd: str) -> bool:
return self._completer is None or self._completer._command_allowed(cmd)
def _history_suggestion(self, buffer, document):
return self._history.get_suggestion(buffer, document) if self._history else None
def get_suggestion(self, buffer, document):
text = document.text_before_cursor
if not text.startswith("/"):
@@ -590,8 +536,7 @@ class SlashCommandAutoSuggest(AutoSuggest):
base_cmd = parts[0].lower()
if len(parts) == 1 and not text.endswith(" "):
# Still typing the command name: /upd → "ate". Prefer the SHORTEST
# match so /he ghosts "lp" (/help), not "artbeat" (/heartbeat).
# Still typing the name: prefer the SHORTEST match so /he ghosts "lp", not "artbeat".
word = text[1:].lower()
for cmd in sorted(COMMANDS, key=len):
cmd_name = cmd[1:]
@@ -601,13 +546,11 @@ class SlashCommandAutoSuggest(AutoSuggest):
sub_text = parts[1] if len(parts) > 1 else ""
sub_lower = sub_text.lower()
# Stacked skill chain: ghost-suggest the rest of the next skill name;
# otherwise fall through to the history fallback for instruction text.
# Stacked skill chain: ghost the rest of the next skill name, else history fallback.
if self._completer is not None and self._completer._is_skill_command(base_cmd):
for completion in self._completer._stacked_skill_completions(text):
remainder = completion.text[-completion.start_position:] \
if completion.start_position else completion.text
start = completion.start_position
remainder = completion.text[-start:] if start else completion.text
if remainder.strip():
return Suggestion(remainder)
@@ -618,6 +561,3 @@ class SlashCommandAutoSuggest(AutoSuggest):
if sub.startswith(sub_lower) and sub != sub_lower:
return Suggestion(sub[len(sub_text):])
return self._history_suggestion(buffer, document)
def _history_suggestion(self, buffer, document):
return self._history.get_suggestion(buffer, document) if self._history else None
hermes_cli/completion.py
+39 -84
View File
@@ -1,7 +1,6 @@
"""Shell completion script generation for hermes CLI.
Walks the live argparse parser tree to generate accurate, always-up-to-date completion scripts — no
hardcoded subcommand lists, no extra dependencies.
Walks the live argparse parser tree, so completion scripts never go stale; no extra dependencies.
"""
from __future__ import annotations
@@ -17,23 +16,15 @@ def _walk(parser: argparse.ArgumentParser) -> dict[str, Any]:
for action in parser._actions:
if isinstance(action, argparse._SubParsersAction):
# _choices_actions has one entry per canonical name; aliases are
# omitted, which keeps completion lists clean.
seen: set[str] = set()
# _choices_actions has one entry per canonical name (aliases omitted).
for pseudo in action._choices_actions:
name = pseudo.dest
if name in seen:
continue
seen.add(name)
subparser = action.choices.get(name)
if subparser is None:
if name in subcommands or subparser is None:
continue
info = _walk(subparser)
info["help"] = _clean(pseudo.help or "")
subcommands[name] = info
subcommands[name] = {**_walk(subparser), "help": _clean(pseudo.help or "")}
elif action.option_strings:
flags.extend(o for o in action.option_strings if o.startswith("-"))
return {"flags": flags, "subcommands": subcommands}
@@ -42,22 +33,23 @@ def _clean(text: str, maxlen: int = 60) -> str:
return text.replace("'", "").replace('"', "").replace("\\", "")[:maxlen]
# ---------------------------------------------------------------------------
# Bash
# ---------------------------------------------------------------------------
# Profile actions that take a profile name as their next argument.
_PROFILE_NAME_ACTIONS = ("use", "delete", "show", "alias", "rename", "export")
def _sorted_subcommands(parser: argparse.ArgumentParser) -> list[tuple[str, dict[str, Any]]]:
tree = _walk(parser)
return sorted(tree["subcommands"].items())
def generate_bash(parser: argparse.ArgumentParser) -> str:
tree = _walk(parser)
top_cmds = " ".join(sorted(tree["subcommands"]))
subcommands = _sorted_subcommands(parser)
top_cmds = " ".join(cmd for cmd, _ in subcommands)
cases: list[str] = []
for cmd in sorted(tree["subcommands"]):
info = tree["subcommands"][cmd]
for cmd, info in subcommands:
if cmd == "profile" and info["subcommands"]:
# Profile subcommand: complete actions, then profile names for
# actions that accept a profile argument.
# Complete actions, then profile names for actions that accept a profile argument.
subcmds = " ".join(sorted(info["subcommands"]))
profile_actions = "use delete show alias rename export"
cases.append(
f" profile)\n"
f" case \"$prev\" in\n"
@@ -65,24 +57,21 @@ def generate_bash(parser: argparse.ArgumentParser) -> str:
f" COMPREPLY=($(compgen -W \"{subcmds}\" -- \"$cur\"))\n"
f" return\n"
f" ;;\n"
f" {profile_actions.replace(' ', '|')})\n"
f" {'|'.join(_PROFILE_NAME_ACTIONS)})\n"
f" COMPREPLY=($(compgen -W \"$(_hermes_profiles)\" -- \"$cur\"))\n"
f" return\n"
f" ;;\n"
f" esac\n"
f" ;;"
)
f" ;;")
elif info["subcommands"] or info["flags"]:
words = " ".join(sorted(info["subcommands"]) if info["subcommands"] else info["flags"])
cases.append(
f" {cmd})\n"
f" COMPREPLY=($(compgen -W \"{words}\" -- \"$cur\"))\n"
f" return\n"
f" ;;"
)
f" ;;")
cases_str = "\n".join(cases)
return f"""# Hermes Agent bash completion
# Add to ~/.bashrc:
# eval "$(hermes completion bash)"
@@ -125,35 +114,26 @@ complete -F _hermes_completion hermes
"""
# ---------------------------------------------------------------------------
# Zsh
# ---------------------------------------------------------------------------
def _zsh_describe_lines(subcommands: dict[str, Any], indent: str) -> str:
"""One ``'name:help'`` line per subcommand, sorted, at the given indent."""
return "\n".join(
f"{indent}'{sc}:{_clean(subcommands[sc].get('help', ''))}'"
for sc in sorted(subcommands)
)
f"{indent}'{sc}:{_clean(subcommands[sc].get('help', ''))}'" for sc in sorted(subcommands))
def generate_zsh(parser: argparse.ArgumentParser) -> str:
tree = _walk(parser)
top_cmds_str = _zsh_describe_lines(tree["subcommands"], " " * 16)
subcommands = _sorted_subcommands(parser)
top_cmds_str = _zsh_describe_lines(dict(subcommands), " " * 16)
sub_cases: list[str] = []
for cmd in sorted(tree["subcommands"]):
info = tree["subcommands"][cmd]
for cmd, info in subcommands:
if not info["subcommands"]:
continue
if cmd == "profile":
# Profile subcommand: complete actions, then profile names for
# actions that accept a profile argument.
# Complete actions, then profile names for actions that accept a profile argument.
sub_str = _zsh_describe_lines(info["subcommands"], " " * 24)
sub_cases.append(
f" profile)\n"
f" case ${{line[2]}} in\n"
f" use|delete|show|alias|rename|export)\n"
f" {'|'.join(_PROFILE_NAME_ACTIONS)})\n"
f" _hermes_profiles\n"
f" ;;\n"
f" *)\n"
@@ -164,8 +144,7 @@ def generate_zsh(parser: argparse.ArgumentParser) -> str:
f" _describe 'profile command' profile_cmds\n"
f" ;;\n"
f" esac\n"
f" ;;"
)
f" ;;")
else:
sub_str = _zsh_describe_lines(info["subcommands"], " " * 20)
safe = cmd.replace("-", "_")
@@ -176,10 +155,8 @@ def generate_zsh(parser: argparse.ArgumentParser) -> str:
f"{sub_str}\n"
f" )\n"
f" _describe '{cmd} command' {safe}_cmds\n"
f" ;;"
)
f" ;;")
sub_cases_str = "\n".join(sub_cases)
return f"""#compdef hermes
# Hermes Agent zsh completion
# Add to ~/.zshrc:
@@ -225,15 +202,9 @@ compdef _hermes hermes
"""
# ---------------------------------------------------------------------------
# Fish
# ---------------------------------------------------------------------------
def generate_fish(parser: argparse.ArgumentParser) -> str:
tree = _walk(parser)
top_cmds = sorted(tree["subcommands"])
top_cmds_str = " ".join(top_cmds)
subcommands = _sorted_subcommands(parser)
top_cmds_str = " ".join(cmd for cmd, _ in subcommands)
lines: list[str] = [
"# Hermes Agent fish completion",
"# Add to your config:",
@@ -256,45 +227,29 @@ def generate_fish(parser: argparse.ArgumentParser) -> str:
"complete -c hermes -f -s p -l profile"
" -d 'Profile name' -xa '(__hermes_profiles)'",
"",
"# Top-level subcommands",
]
"# Top-level subcommands"]
for cmd in top_cmds:
info = tree["subcommands"][cmd]
help_text = _clean(info.get("help", ""))
for cmd, info in subcommands:
lines.append(
f"complete -c hermes -f "
f"-n 'not __fish_seen_subcommand_from {top_cmds_str}' "
f"-a {cmd} -d '{help_text}'"
)
lines.append("")
lines.append("# Subcommand completions")
profile_name_actions = {"use", "delete", "show", "alias", "rename", "export"}
for cmd in top_cmds:
info = tree["subcommands"][cmd]
f"-a {cmd} -d '{_clean(info.get('help', ''))}'")
lines += ["", "# Subcommand completions"]
for cmd, info in subcommands:
if not info["subcommands"]:
continue
lines.append(f"# {cmd}")
for sc in sorted(info["subcommands"]):
sinfo = info["subcommands"][sc]
sh = _clean(sinfo.get("help", ""))
for sc, sinfo in sorted(info["subcommands"].items()):
lines.append(
f"complete -c hermes -f "
f"-n '__fish_seen_subcommand_from {cmd}' "
f"-a {sc} -d '{sh}'"
)
# For profile subcommand, complete profile names for relevant actions
if cmd == "profile":
for action in sorted(profile_name_actions):
f"-a {sc} -d '{_clean(sinfo.get('help', ''))}'")
if cmd == "profile": # profile names for the actions that take one
for action in sorted(_PROFILE_NAME_ACTIONS):
lines.append(
f"complete -c hermes -f "
f"-n '__fish_seen_subcommand_from {action}; "
f"and __fish_seen_subcommand_from profile' "
f"-a '(__hermes_profiles)' -d 'Profile name'"
)
f"-a '(__hermes_profiles)' -d 'Profile name'")
lines.append("")
return "\n".join(lines)
hermes_cli/console_engine.py
+321 -560
View File
File diff suppressed because it is too large Load Diff
hermes_cli/curator.py
+241 -439
View File
File diff suppressed because it is too large Load Diff
hermes_cli/curses_ui.py
+201 -295
View File
@@ -1,8 +1,7 @@
"""Shared curses-based UI components for Hermes CLI.
Used by `hermes tools` and `hermes skills` for interactive checklists. Provides a curses multi-
select with keyboard navigation, plus a text-based numbered fallback for terminals without curses
support.
Curses checklist / radiolist / single-select menus with keyboard navigation and fuzzy ``/``
search, plus a numbered text fallback for terminals without curses.
"""
import sys
from contextvars import ContextVar, Token
@@ -12,8 +11,7 @@ from typing import Callable, List, Optional, Protocol, Sequence, Set, Tuple, Uni
from hermes_cli.colors import Colors, color
# Rich radiolist rows: (text, style). style is None | "yellow" | "dim".
# Plain ``str`` items remain fully supported.
# Rich radiolist rows: (text, style). style is None | "yellow" | "dim". Plain ``str`` works too.
RadioItem = Union[str, Sequence[Tuple[str, Optional[str]]]]
@@ -42,16 +40,18 @@ class MenuNavigationEvent(str, Enum):
class MenuNavigationHandler(Protocol):
"""Typed contract between shared menus and a scoped flow controller."""
def __call__(self, event: MenuNavigationEvent, value: object = None) -> MenuNavigationStart | None: ...
def __call__(
self, event: MenuNavigationEvent, value: object = None
) -> MenuNavigationStart | None: ...
_MENU_NAVIGATION_HANDLER: ContextVar[MenuNavigationHandler | None] = ContextVar(
"hermes_menu_navigation_handler", default=None
)
"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]:
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)
@@ -62,7 +62,6 @@ def reset_menu_navigation_handler(token: Token[MenuNavigationHandler | None]) ->
def _notify_scoped_navigation(event: MenuNavigationEvent) -> None:
"""Notify an active menu flow that a text fallback was interrupted (CANCEL) or requested BACK."""
handler = _MENU_NAVIGATION_HANDLER.get()
if handler is not None:
handler(event)
@@ -77,10 +76,10 @@ _NAV_ABORT = object()
def _read_numbered_choice(prompt_text: str) -> int | None | object:
"""Read a numbered fallback choice as a 0-based index.
"""Numbered-fallback choice as a 0-based index.
Returns ``None`` for empty input and ``_NAV_ABORT`` when the prompt was cancelled, backed out
of, interrupted, or given a non-integer (scoped navigation is notified for cancel/back).
``None`` for empty input; ``_NAV_ABORT`` when cancelled, backed out of, interrupted, or
non-integer (scoped navigation is notified for cancel/back).
"""
try:
val = _read_numbered_input(prompt_text)
@@ -97,11 +96,10 @@ def _read_numbered_choice(prompt_text: str) -> int | None | object:
def _read_numbered_input(prompt_text: str) -> str | _NumberedNavigation:
"""Read a numbered fallback choice with setup navigation key bindings.
"""Read a numbered fallback choice.
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.
Plain ``input()`` outside setup flows. Inside a scoped flow, prompt_toolkit supplies
portable Escape / Ctrl+C / Left bindings on POSIX and native Windows.
"""
if _MENU_NAVIGATION_HANDLER.get() is None:
return input(prompt_text)
@@ -111,10 +109,8 @@ def _read_numbered_input(prompt_text: str) -> str | _NumberedNavigation:
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.
# Setup may run without the classic CLI, which normally installs CSI-u aliases at startup.
from hermes_cli.pt_input_extras import install_modify_other_keys_aliases
install_modify_other_keys_aliases()
bindings = KeyBindings()
@@ -134,9 +130,7 @@ def _read_numbered_input(prompt_text: str) -> str | _NumberedNavigation:
def radio_item_plain(item: RadioItem) -> str:
"""Flatten a radiolist item to searchable/plain display text."""
if isinstance(item, str):
return item
return "".join(text for text, _style in item)
return item if isinstance(item, str) else "".join(text for text, _style in item)
def _curses_style_attr(curses, style: Optional[str], *, is_cursor: bool):
@@ -149,9 +143,8 @@ def _curses_style_attr(curses, style: Optional[str], *, is_cursor: bool):
if style == "dim":
attr = curses.A_DIM
if has_colors:
# Pair 3 is the dim-gray status color (extra_color_pairs).
try:
attr |= curses.color_pair(3)
attr |= curses.color_pair(3) # dim-gray status pair (extra_color_pairs)
except curses.error:
pass
return attr
@@ -161,7 +154,6 @@ def _curses_style_attr(curses, style: Optional[str], *, is_cursor: bool):
def _addnstr(stdscr, y: int, x: int, text: str, n: int, attr) -> None:
"""``stdscr.addnstr`` that swallows ``curses.error`` (drawing past the screen edge)."""
import curses
try:
stdscr.addnstr(y, x, text, n, attr)
except curses.error:
@@ -169,23 +161,15 @@ def _addnstr(stdscr, y: int, x: int, text: str, n: int, attr) -> None:
def _draw_title_and_hint(stdscr, title: str, hint: str, max_x: int, *, hint_row: int = 1) -> None:
"""Draw the bold/yellow menu title on row 0 and the dim key hint on ``hint_row``."""
"""Bold/yellow menu title on row 0, dim key hint on ``hint_row``."""
import curses
hattr = curses.A_BOLD | (curses.color_pair(2) if curses.has_colors() else 0)
_addnstr(stdscr, 0, 0, title, max_x - 1, hattr)
_addnstr(stdscr, hint_row, 0, hint, max_x - 1, curses.A_DIM)
def _draw_plain_row(stdscr, y: int, line: str, max_x: int, *, is_cursor: bool) -> None:
"""Draw a plain menu row, bold green when it is the cursor row."""
import curses
_addnstr(stdscr, y, 0, line, max_x - 1, _curses_style_attr(curses, None, is_cursor=is_cursor))
def _draw_segments(stdscr, y: int, x: int, segments, max_x: int) -> None:
"""Draw ``(text, attr)`` segments left to right from column ``x``, clipped at the screen edge."""
"""Draw ``(text, attr)`` segments left to right from column ``x``, clipped at the edge."""
col = x
for text, attr in segments:
remaining = max_x - 1 - col
@@ -199,7 +183,6 @@ def _draw_segments(stdscr, y: int, x: int, segments, max_x: int) -> None:
def _draw_description_line(stdscr, y: int, text: str, max_x: int) -> None:
"""Draw a description line, highlighting ★ in yellow when colors exist."""
import curses
star_attr = curses.color_pair(2) if curses.has_colors() else curses.A_NORMAL
segments = []
for i, part in enumerate(text.split("★")):
@@ -210,54 +193,50 @@ def _draw_description_line(stdscr, y: int, text: str, max_x: int) -> None:
_draw_segments(stdscr, y, 0, segments, max_x)
def _draw_radio_item(stdscr, y: int, x: int, item: RadioItem, max_x: int, *, is_cursor: bool) -> None:
def _draw_radio_item(
stdscr, y: int, x: int, item: RadioItem, max_x: int, *, is_cursor: bool) -> None:
"""Draw a plain or segmented radiolist item starting at column ``x``."""
import curses
if isinstance(item, str):
attr = _curses_style_attr(curses, None, is_cursor=is_cursor)
_addnstr(stdscr, y, x, item, max(0, max_x - 1 - x), attr)
return
_draw_segments(
stdscr, y, x,
((text, _curses_style_attr(curses, style, is_cursor=is_cursor)) for text, style in item),
max_x,
)
max_x)
def _draw_plain_row(stdscr, y: int, line: str, max_x: int, *, is_cursor: bool) -> None:
"""Draw a plain menu row, bold green when it is the cursor row."""
import curses
_addnstr(stdscr, y, 0, line, max_x - 1, _curses_style_attr(curses, None, is_cursor=is_cursor))
_WORD_BOUNDARY = frozenset("-_/. ")
def _is_boundary(target: str, index: int) -> bool:
"""True if position ``index`` in ``target`` starts a word.
Mirrors ``isBoundary`` in the TS scorer: start-of-string, after a separator char, or a
lower->upper camelCase transition.
"""
"""Mirrors ``isBoundary`` in the TS scorer: start, after a separator, or lower->Upper."""
if index == 0:
return True
prev = target[index - 1]
if prev in _WORD_BOUNDARY:
return True
# camelCase / lower->upper transition (e.g. the `O` in `gptO`).
cur = target[index]
return prev == prev.lower() and cur != cur.lower() and cur == cur.upper()
def _token_score(orig: str, lower: str, token: str) -> float | None:
"""Score one token against a target. None if the token isn't a subsequence.
"""Score one token against a target; None if not a subsequence.
Faithful port of ``fuzzyScore`` in ui-tui and web ``fuzzy.ts`` so all three surfaces rank
model ids identically: contiguous runs, word-boundary/first-char starts, prefixes and exact
matches outrank scattered hits. Matching runs against ``lower`` while boundary detection
uses ``orig`` so the camelCase rule works, exactly as in the TS scorer.
Faithful port of ``fuzzyScore`` in ui-tui / web ``fuzzy.ts`` so all surfaces rank model ids
identically. Matches run against ``lower``; boundary detection uses ``orig`` for camelCase.
"""
score = 0.0
prev = -1
search_from = 0
positions: list[int] = []
for ch in token:
idx = lower.find(ch, search_from)
if idx < 0:
@@ -274,27 +253,15 @@ def _token_score(orig: str, lower: str, token: str) -> float | None:
score += 5
prev = idx
search_from = idx + 1
# Prefix bonus: the token matched a contiguous prefix of the target.
if positions and positions[0] == 0 and positions[-1] == len(positions) - 1:
score += 8
# Exact full match dominates everything else.
score += 8 # contiguous prefix
if lower == token:
score += 20
# Slightly prefer shorter targets when scores are otherwise close.
score -= len(lower) * 0.01
return score
score += 20 # exact match dominates
return score - len(lower) * 0.01 # slightly prefer shorter targets
def _fuzzy_score(label: str, query: str) -> float | None:
"""Aggregate score for a multi-token query (AND). None if any token fails.
Mirrors ``fuzzyScoreMulti`` in the TS scorer: every whitespace-separated token must match; per-
token scores are summed.
"""
"""Multi-token AND score (``fuzzyScoreMulti``): sum of per-token scores, None if any fails."""
lower = label.lower()
total = 0.0
for token in query.lower().split():
@@ -306,16 +273,11 @@ def _fuzzy_score(label: str, query: str) -> float | None:
def _filter_indices(items: List[str], query: str) -> List[int]:
"""Return item indices matching *query*, ranked best-first.
An empty query keeps every item in original order. Otherwise items are filtered to fuzzy matches
and sorted by score descending, ties broken by original index so equal-scoring rows keep their
catalog order.
"""
"""Item indices matching *query*, best-first; ties keep catalog order. Empty query = all."""
q = query.strip()
if not q:
return list(range(len(items)))
scored = [(i, score) for i, label in enumerate(items) if (score := _fuzzy_score(label, q)) is not None]
scored = [(i, s) for i, label in enumerate(items) if (s := _fuzzy_score(label, q)) is not None]
scored.sort(key=lambda pair: (-pair[1], pair[0]))
return [i for i, _ in scored]
@@ -342,7 +304,8 @@ def _move_filtered_cursor(filtered: List[int], cursor: int, cursor_pos: int, del
return filtered[(cursor_pos + delta) % len(filtered)] if filtered else cursor
def _scroll_for_cursor(scroll_offset: int, cursor_pos: int, visible_rows: int, total_rows: int) -> int:
def _scroll_for_cursor(
scroll_offset: int, cursor_pos: int, visible_rows: int, total_rows: int) -> int:
"""Clamp scroll offset so the cursor remains visible."""
visible_rows = max(1, visible_rows)
if cursor_pos < scroll_offset:
@@ -352,23 +315,20 @@ def _scroll_for_cursor(scroll_offset: int, cursor_pos: int, visible_rows: int, t
return max(0, min(scroll_offset, max(0, total_rows - visible_rows)))
def _handle_active_search_key(curses_mod, key: int, search: _SearchState) -> tuple[bool, bool, bool]:
"""Handle a key while the search prompt is active."""
def _handle_active_search_key(
curses_mod, key: int, search: _SearchState) -> tuple[bool, bool, bool]:
"""Handle a key while the search prompt is active -> ``(handled, confirm, changed)``."""
if not search.active:
return False, False, False
if key == 27:
# Esc stops search AND clears the query, restoring the full list (so a
# no-match filter can't strand the user on an empty list). Signals
# `changed` when there was a query so the driver resets scroll/cursor.
# Esc stops search AND clears the query so a no-match filter can't strand the user on
# an empty list; `changed` when there was a query so the driver resets scroll/cursor.
had_query = bool(search.query)
search.active = False
search.query = ""
return True, False, had_query
if key in (curses_mod.KEY_ENTER, 10, 13):
return True, True, False
if key in (curses_mod.KEY_BACKSPACE, 127, 8):
search.query = search.query[:-1]
elif key == 21: # Ctrl+U
@@ -381,11 +341,9 @@ def _handle_active_search_key(curses_mod, key: int, search: _SearchState) -> tup
def flush_stdin() -> None:
"""Flush any stray bytes from the stdin input buffer.
"""Drain stray stdin bytes after ``curses.wrapper()`` and before the next ``input()``.
Must be called after ``curses.wrapper()`` returns, and before the next ``input()`` /
``getpass.getpass()`` call. ``curses.endwin()`` restores the terminal but does NOT drain the OS
input buffer.
``curses.endwin()`` restores the terminal but does NOT drain the OS input buffer.
"""
try:
if sys.stdin.isatty():
@@ -395,8 +353,7 @@ def flush_stdin() -> None:
pass
# Normalized menu actions returned by ``read_menu_key``. Using sentinels keeps
# every menu's key-handling branch identical and free of raw escape-byte logic.
# Normalized menu actions returned by ``read_menu_key``.
NAV_UP = "up"
NAV_DOWN = "down"
NAV_BACK = "back"
@@ -408,11 +365,10 @@ NAV_NONE = "none"
def read_menu_key(stdscr) -> str:
"""Read one keypress and normalize it to a menu action.
"""Read one keypress and normalize it to a ``NAV_*`` action.
Returns one of the ``NAV_*`` constants. A lone ESC (no continuation byte within a short window)
is the only thing that maps to ``NAV_CANCEL`` via the escape path; ``q`` also cancels. Unknown
sequences map to ``NAV_NONE`` so the caller simply ignores them rather than misfiring.
A lone ESC (no continuation byte within a short window) and ``q`` cancel; unknown sequences
map to ``NAV_NONE`` so the caller ignores them rather than misfiring.
"""
return _decode_menu_key(stdscr, stdscr.getch())
@@ -425,7 +381,7 @@ def _parse_int(value: str, default=0):
def _parse_csi_u_key(raw: str) -> tuple[int, int, int] | None:
"""Parse a Kitty/CSI-u key into ``(codepoint, modifier, event_type)``; None without a codepoint."""
"""Kitty/CSI-u key -> ``(codepoint, modifier, event_type)``; None without a codepoint."""
parts = raw.split(";")
codepoint = _parse_int(parts[0].split(":", 1)[0])
if not codepoint:
@@ -437,22 +393,19 @@ def _parse_csi_u_key(raw: str) -> tuple[int, int, int] | None:
def _parse_csi_numbers(raw: str) -> list[int]:
"""Parse semicolon-delimited CSI numbers for modifyOtherKeys."""
"""Semicolon-delimited CSI numbers for modifyOtherKeys."""
return [_parse_int(part.split(":", 1)[0]) for part in raw.split(";")]
_ENHANCED_NAV = {10: NAV_SELECT, 13: NAV_SELECT, 27: NAV_CANCEL, 32: NAV_TOGGLE}
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.
if codepoint in _ENHANCED_NAV:
return _ENHANCED_NAV[codepoint]
# CSI-u encodes Ctrl+C as `c` plus the Ctrl modifier bit; lock-state bits may be added, so
# test the Ctrl bit rather than 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
@@ -472,70 +425,101 @@ def _read_csi_tail(stdscr) -> tuple[str, int | None]:
return "".join(raw), None
_CSI_FINAL_NAV = {ord("A"): NAV_UP, ord("k"): NAV_UP, ord("B"): NAV_DOWN, ord("j"): NAV_DOWN, ord("D"): NAV_BACK}
_CSI_FINAL_NAV = {
ord("A"): NAV_UP, ord("k"): NAV_UP, ord("B"): NAV_DOWN, ord("j"): NAV_DOWN, ord("D"): NAV_BACK}
def _decode_menu_key(stdscr, key: int) -> str:
"""Normalize an already-read keypress to a menu action.
Split out from ``read_menu_key`` so search-aware loops can peek the raw key (e.g. to catch
``/``) before falling back to nav decoding.
"""
import curses
if key in (curses.KEY_UP, ord("k")):
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(" "):
return NAV_TOGGLE
if key == ord("q"):
return NAV_CANCEL
if key == 27: # ESC — could be a lone ESC (cancel) or an escape sequence.
# Wait briefly for a continuation byte. On slow PTYs (SSH/tmux) the
# bytes of an arrow key can arrive across separate reads, so a tiny
# timeout avoids misreading a split sequence as a bare ESC.
try:
stdscr.timeout(60)
nxt = stdscr.getch()
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 _CSI_FINAL_NAV:
return _CSI_FINAL_NAV[final]
if final == ord("u"):
enhanced = _parse_csi_u_key(raw_params)
if enhanced is not None:
codepoint, modifier, event_type = enhanced
if event_type == 3: # key release
return NAV_NONE
return _enhanced_key_action(codepoint, 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])
def _decode_escape_sequence(stdscr) -> str:
"""Decode the bytes after an ESC: lone ESC cancels, CSI/SS3 sequences map to actions."""
nxt = stdscr.getch()
if nxt == -1:
return NAV_CANCEL # genuine lone ESC
if nxt not in (ord("["), ord("O")): # not a CSI / SS3 introducer — swallow
return NAV_NONE
raw_params, final = _read_csi_tail(stdscr)
if final in _CSI_FINAL_NAV:
return _CSI_FINAL_NAV[final]
if final == ord("u"):
enhanced = _parse_csi_u_key(raw_params)
if enhanced is not None:
codepoint, modifier, event_type = enhanced
if event_type == 3: # key release
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
return _enhanced_key_action(codepoint, 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
# Sentinel: an on_action reducer returns this to mean "keep looping" (the
# keypress changed cursor/selection state but didn't resolve the menu).
def _decode_menu_key(stdscr, key: int) -> str:
"""Normalize an already-read keypress to a menu action (lets loops peek the raw key first)."""
import curses
plain = {
curses.KEY_UP: NAV_UP, ord("k"): NAV_UP, curses.KEY_DOWN: NAV_DOWN, ord("j"): NAV_DOWN,
curses.KEY_LEFT: NAV_BACK, 3: NAV_INTERRUPT, # 3 = Ctrl+C in raw/cbreak mode
curses.KEY_ENTER: NAV_SELECT, 10: NAV_SELECT, 13: NAV_SELECT,
ord(" "): NAV_TOGGLE, ord("q"): NAV_CANCEL}
if key in plain:
return plain[key]
if key != 27:
return NAV_NONE
# ESC: wait briefly for a continuation byte. On slow PTYs (SSH/tmux) an arrow key's bytes
# can arrive across separate reads, so a tiny timeout avoids misreading it as a bare ESC.
try:
stdscr.timeout(60)
return _decode_escape_sequence(stdscr)
finally:
stdscr.timeout(-1) # restore blocking mode
# An on_action reducer returns this to keep looping (state changed, menu not resolved).
_KEEP = object()
# Sentinel from ``_route_key``: the key was consumed by the search prompt; redraw, no action.
_CONSUMED = object()
def _init_colors(curses, extra_color_pairs: bool) -> None:
curses.curs_set(0)
if curses.has_colors():
curses.start_color()
curses.use_default_colors()
curses.init_pair(1, curses.COLOR_GREEN, -1)
curses.init_pair(2, curses.COLOR_YELLOW, -1)
if extra_color_pairs:
curses.init_pair(3, 8 if curses.COLORS > 8 else curses.COLOR_WHITE, -1)
def _route_key(curses, stdscr, key: int, search: _SearchState, use_search: bool):
"""Turn a raw key into a ``NAV_*`` action, honoring an active ``/`` search prompt.
Returns ``(action, changed)``; ``action`` may be ``_CONSUMED`` (redraw only) or
``NAV_SELECT`` for Enter inside the prompt; ``changed`` means the query changed.
"""
if use_search and search.active and key == 27:
# Enhanced keys (Ghostty/Kitty) also start with ESC: decode the full sequence before
# treating a genuine Escape as "stop search".
action = _decode_menu_key(stdscr, key)
if action == NAV_CANCEL:
search.active = False
search.query = ""
return _CONSUMED, True
return (_CONSUMED if action == NAV_NONE else action), False
if use_search and search.active:
# Active search consumes query-editing keys; nav keys fall through.
handled, confirm, changed = _handle_active_search_key(curses, key, search)
if confirm:
return NAV_SELECT, changed
if handled:
return _CONSUMED, changed
return _decode_menu_key(stdscr, key), changed
if use_search and key == ord("/"):
search.active = True
return _CONSUMED, False
return _decode_menu_key(stdscr, key), False
def _run_curses_menu(
*,
@@ -550,18 +534,15 @@ def _run_curses_menu(
fallback,
cancel_value,
searchable=False,
search_labels=None,
):
search_labels=None):
"""Shared curses single-/multi-select event loop.
Owns the non-TTY guard, ``curses.wrapper`` setup, the per-frame clear/refresh cycle, scroll
math, key dispatch with cursor wrap, and the KeyboardInterrupt / curses-unavailable
fallback; per-menu behavior comes in as callbacks so rendering stays byte-identical to the
old hand-rolled loops. ``draw_row`` always receives the ORIGINAL item index (filtering
doesn't change rendering); ``on_action`` returns ``_KEEP`` to continue or any other value to
resolve the menu; a ``draw_footer`` row budget must be included in ``reserve_bottom``; with
``searchable``, ``/`` filters over ``search_labels`` (length must equal ``item_count``) and
results are original indices.
Owns the non-TTY guard, ``curses.wrapper`` setup, clear/refresh cycle, scroll math, key
dispatch with cursor wrap, and the KeyboardInterrupt / curses-unavailable fallback.
``draw_row`` always receives the ORIGINAL item index; ``on_action`` returns ``_KEEP`` to
continue or any other value to resolve; a ``draw_footer`` row budget must be included in
``reserve_bottom``; with ``searchable``, ``/`` filters over ``search_labels`` (length ==
``item_count``) and results are original indices.
"""
navigation_handler = _MENU_NAVIGATION_HANDLER.get()
@@ -577,9 +558,8 @@ def _run_curses_menu(
_notify(MenuNavigationEvent.RESOLVE, navigation_start.replay_value)
return navigation_start.replay_value
# 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).
# Non-TTY stdin: curses and input() both hang or spin, so return the cancel value directly
# (the numbered fallback is only for curses errors on a real TTY).
if not sys.stdin.isatty():
return cancel_value
@@ -611,14 +591,7 @@ def _run_curses_menu(
return True
def _draw(stdscr):
curses.curs_set(0)
if curses.has_colors():
curses.start_color()
curses.use_default_colors()
curses.init_pair(1, curses.COLOR_GREEN, -1)
curses.init_pair(2, curses.COLOR_YELLOW, -1)
if extra_color_pairs:
curses.init_pair(3, 8 if curses.COLORS > 8 else curses.COLOR_WHITE, -1)
_init_colors(curses, extra_color_pairs)
cursor = initial_cursor
scroll_offset = 0
search = _SearchState()
@@ -626,72 +599,40 @@ def _run_curses_menu(
while True:
stdscr.clear()
max_y, max_x = stdscr.getmaxyx()
filtered = _filter_indices(search_labels, search.query) if use_search else list(range(item_count))
filtered = (
_filter_indices(search_labels, search.query) if use_search
else list(range(item_count)))
cursor, cursor_pos = _reconcile_cursor(filtered, cursor)
items_start = draw_header(stdscr, max_y, max_x, search=search, back_enabled=allow_back)
items_start = draw_header(
stdscr, max_y, max_x, search=search, back_enabled=allow_back)
visible_rows = max(1, max_y - items_start - reserve_bottom)
scroll_offset = _scroll_for_cursor(scroll_offset, cursor_pos, visible_rows, len(filtered))
scroll_offset = _scroll_for_cursor(
scroll_offset, cursor_pos, visible_rows, len(filtered))
if use_search and search.query and not filtered:
_addnstr(stdscr, items_start, 0, " No matches", max_x - 1, curses.A_DIM)
for draw_i, i in enumerate(filtered[scroll_offset : scroll_offset + visible_rows]):
y = draw_i + items_start
if y >= max_y - reserve_bottom:
break
draw_row(stdscr, y, i, i == cursor, max_x)
if draw_footer is not None:
draw_footer(stdscr, max_y, max_x)
stdscr.refresh()
key = stdscr.getch()
if use_search and 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 use_search and search.active:
# Active search consumes query-editing keys; nav keys
# fall through to be decoded below.
handled, confirm, changed = _handle_active_search_key(curses, key, search)
if changed:
scroll_offset = 0
action, changed = _route_key(curses, stdscr, stdscr.getch(), search, use_search)
if changed:
scroll_offset = 0
if search.active: # Esc-clear resets scroll only; the loop re-reconciles
cursor, cursor_pos = _reconcile_cursor(
_filter_indices(search_labels, search.query), cursor
)
if confirm:
if filtered and _resolve(on_action(NAV_SELECT, cursor)):
return
continue
if handled:
continue
action = _decode_menu_key(stdscr, key)
elif use_search and key == ord("/"):
search.active = True
_filter_indices(search_labels, search.query), cursor)
if action is _CONSUMED:
continue
else:
action = _decode_menu_key(stdscr, key)
if action == NAV_UP:
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, NAV_INTERRUPT) or (
action == NAV_BACK and allow_back
):
action == NAV_BACK and allow_back):
if action == NAV_SELECT and use_search and not filtered:
continue
if action in (NAV_CANCEL, NAV_INTERRUPT):
@@ -704,7 +645,6 @@ def _run_curses_menu(
curses.wrapper(_draw)
flush_stdin()
return result_holder[0] if result_holder[0] is not _KEEP else cancel_value
except KeyboardInterrupt:
_notify(MenuNavigationEvent.CANCEL)
return cancel_value
@@ -718,17 +658,14 @@ def curses_checklist(
selected: Set[int],
*,
cancel_returns: Set[int] | None = None,
status_fn: Optional[Callable[[Set[int]], str]] = None,
) -> Set[int]:
status_fn: Optional[Callable[[Set[int]], str]] = None) -> Set[int]:
"""Curses multi-select checklist. Returns set of selected indices.
``cancel_returns`` (default: the original *selected*) is returned on ESC/q.
``status_fn(chosen)`` renders on the bottom row for live aggregate info such as token
estimates.
``status_fn(chosen)`` renders on the bottom row for live aggregate info such as token estimates.
"""
if cancel_returns is None:
cancel_returns = set(selected)
chosen = set(selected)
def _draw_row(stdscr, y, i, is_cursor, max_x):
@@ -739,8 +676,7 @@ def curses_checklist(
def _draw_footer(stdscr, max_y, max_x):
import curses
status_text = status_fn(chosen)
if status_text:
# Right-align on the bottom row
if status_text: # right-aligned on the bottom row
sx = max(0, max_x - len(status_text) - 1)
sattr = curses.A_DIM | (curses.color_pair(3) if curses.has_colors() else 0)
_addnstr(stdscr, max_y - 1, sx, status_text, max_x - sx - 1, sattr)
@@ -763,8 +699,7 @@ def curses_checklist(
draw_footer=_draw_footer if status_fn else None,
extra_color_pairs=bool(status_fn),
fallback=lambda: _numbered_fallback(title, items, selected, cancel_returns, status_fn),
cancel_value=cancel_returns,
)
cancel_value=cancel_returns)
def _search_hint(search, searchable: bool, confirm: str, cancel: str, back_enabled: bool) -> str:
@@ -780,7 +715,6 @@ def _search_hint(search, searchable: bool, confirm: str, cancel: str, back_enabl
def _simple_header(title: str, confirm: str, cancel: str, searchable: bool):
"""``draw_header`` callback: title on row 0, key hint on row 1, items start on row 3."""
def _draw_header(stdscr, max_y, max_x, search=None, back_enabled=False):
hint = _search_hint(search, searchable, confirm, cancel, back_enabled)
_draw_title_and_hint(stdscr, title, hint, max_x)
@@ -797,49 +731,37 @@ def curses_radiolist(
cancel_returns: int | None = None,
description: str | None = None,
searchable: bool = False,
search_labels: List[str] | None = None,
) -> int:
search_labels: List[str] | None = None) -> int:
"""Curses single-select radio list. Returns the selected index.
Items are plain strings or ``(text, style)`` segment sequences
(``None``/``"yellow"``/``"dim"``); the cursor row is forced green, unselected rows honor
segment styles. ``description`` is shown between title and list so context survives the
curses screen clear. With ``searchable``, ``/`` filters over ``search_labels`` (default:
display labels) and the returned value is always the ORIGINAL item index, never a filtered
row position.
Items are plain strings or ``(text, style)`` segment sequences (``None``/``"yellow"``/
``"dim"``); the cursor row is forced green. ``description`` is shown between title and list
so context survives the curses screen clear. With ``searchable``, ``/`` filters over
``search_labels`` (default: display labels); the return value is always the ORIGINAL index.
"""
if cancel_returns is None:
cancel_returns = selected
desc_lines = description.splitlines() if description else []
plain_labels = [radio_item_plain(item) for item in items] if searchable else None
def _draw_header(stdscr, max_y, max_x, search=None, back_enabled=False):
# Description lines — paint ★ yellow so the sale legend matches rows.
row = 1
for dline in desc_lines[: max(0, max_y - 2)]:
for dline in desc_lines[: max(0, max_y - 2)]: # ★ painted yellow to match rows
_draw_description_line(stdscr, row, dline, max_x)
row += 1
hint = _search_hint(search, searchable, "ENTER/SPACE select", "ESC cancel", back_enabled)
_draw_title_and_hint(stdscr, title, hint, max_x, hint_row=row)
# One blank row between the hint and the item list.
return row + 2
return row + 2 # one blank row between the hint and the item list
def _draw_row(stdscr, y, i, is_cursor, max_x):
radio = "\u25cf" if i == selected else "\u25cb"
arrow = "\u2192" if is_cursor else " "
prefix = f" {arrow} ({radio}) "
_draw_plain_row(stdscr, y, prefix, max_x, is_cursor=is_cursor)
_draw_radio_item(
stdscr, y, len(prefix), items[i], max_x, is_cursor=is_cursor
)
_draw_radio_item(stdscr, y, len(prefix), items[i], max_x, is_cursor=is_cursor)
def _on_action(action, cursor):
if action in (NAV_SELECT, NAV_TOGGLE):
return cursor
return cancel_returns # NAV_CANCEL
return cursor if action in (NAV_SELECT, NAV_TOGGLE) else cancel_returns # NAV_CANCEL
return _run_curses_menu(
initial_cursor=selected,
@@ -848,13 +770,16 @@ def curses_radiolist(
draw_row=_draw_row,
on_action=_on_action,
reserve_bottom=1,
# Dim gray (pair 3) for unselected "was …" sale chrome.
extra_color_pairs=True,
extra_color_pairs=True, # dim gray (pair 3) for unselected "was …" sale chrome
fallback=lambda: _radio_numbered_fallback(title, items, selected, cancel_returns),
cancel_value=cancel_returns,
searchable=searchable,
search_labels=(list(search_labels) if search_labels is not None else plain_labels) if searchable else None,
)
search_labels=(
(list(search_labels) if search_labels is not None else plain_labels)
if searchable else None))
_ANSI_STYLE = {"yellow": Colors.YELLOW, "dim": Colors.DIM}
def format_radio_item_ansi(item: RadioItem) -> str:
@@ -862,23 +787,14 @@ def format_radio_item_ansi(item: RadioItem) -> str:
if isinstance(item, str):
return item
return "".join(
color(text, _ANSI_STYLE[style]) if style in _ANSI_STYLE else text for text, style in item
)
_ANSI_STYLE = {"yellow": Colors.YELLOW, "dim": Colors.DIM}
color(text, _ANSI_STYLE[style]) if style in _ANSI_STYLE else text for text, style in item)
def _radio_numbered_fallback(
title: str,
items: List[RadioItem],
selected: int,
cancel_returns: int,
) -> int:
title: str, items: List[RadioItem], selected: int, cancel_returns: int) -> int:
"""Text-based numbered fallback for radio selection."""
print(color(f"\n {title}", Colors.YELLOW))
print(color(" Select by number, Enter to confirm.\n", Colors.DIM))
for i, label in enumerate(items):
marker = color("(\u25cf)", Colors.GREEN) if i == selected else "(\u25cb)"
print(f" {marker} {i + 1:>2}. {format_radio_item_ansi(label)}")
@@ -895,12 +811,11 @@ def curses_single_select(
default_index: int = 0,
*,
cancel_label: str = "Cancel",
searchable: bool = False,
) -> int | None:
searchable: bool = False) -> int | None:
"""Curses single-select menu. Returns selected index or None on cancel.
When ``searchable`` is true, ``/`` opens a type-to-filter prompt; the returned value is always
the original item index (or None for cancel).
With ``searchable``, ``/`` opens a type-to-filter prompt; the return value is always the
original item index (or None for cancel).
"""
all_items = list(items) + [cancel_label]
cancel_idx = len(items)
@@ -910,9 +825,7 @@ def curses_single_select(
_draw_plain_row(stdscr, y, f" {arrow} {all_items[i]}", max_x, is_cursor=is_cursor)
def _on_action(action, cursor):
if action == NAV_SELECT:
# Selecting the synthetic cancel row resolves to None, mirroring
# the old post-loop ``>= cancel_idx`` guard.
if action == NAV_SELECT: # the synthetic cancel row resolves to None
return None if cursor >= cancel_idx else cursor
if action in (NAV_CANCEL, NAV_INTERRUPT):
return None
@@ -928,15 +841,10 @@ def curses_single_select(
fallback=lambda: _numbered_single_fallback(title, all_items, cancel_idx),
cancel_value=None,
searchable=searchable,
search_labels=list(all_items) if searchable else None,
)
search_labels=list(all_items) if searchable else None)
def _numbered_single_fallback(
title: str,
items: List[str],
cancel_idx: int,
) -> int | None:
def _numbered_single_fallback(title: str, items: List[str], cancel_idx: int) -> int | None:
"""Text-based numbered fallback for single-select."""
print(f"\n {title}\n")
for i, label in enumerate(items, 1):
@@ -951,13 +859,11 @@ def _numbered_fallback(
items: List[str],
selected: Set[int],
cancel_returns: Set[int],
status_fn: Optional[Callable[[Set[int]], str]] = None,
) -> Set[int]:
status_fn: Optional[Callable[[Set[int]], str]] = None) -> Set[int]:
"""Text-based toggle fallback for terminals without curses."""
chosen = set(selected)
print(color(f"\n {title}", Colors.YELLOW))
print(color(" Toggle by number, Enter to confirm.\n", Colors.DIM))
while True:
for i, label in enumerate(items):
marker = color("[✓]", Colors.GREEN) if i in chosen else "[ ]"