refactor(cli/commands): extract completer + platform derivations into commands_completion/commands_platforms with lazy re-exports
- hermes_cli/commands.py keeps CommandDef, COMMAND_REGISTRY, derived lookups, gateway helpers; all moved names resolve via PEP 562 __getattr__ so imports and patch targets keep working - completer: per-command _<x>_completions -> module functions sharing _prefix_completions / _split_args; static refs table; SlashCommandAutoSuggest shares _allowed/_history_suggestion - platforms: telegram/discord/slack derivations, _nested_mapping + _dedupe_sanitized_names folded into _telegram_command_menu_config / _sanitized_rank - COMMAND_REGISTRY + telegram/discord/slack derivations + completer output byte-identical vs base
This commit is contained in:
Teknium
+29
-1321
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,623 @@
|
||||
"""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.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import time
|
||||
from collections.abc import Callable, Iterable, Mapping
|
||||
from itertools import chain
|
||||
from typing import Any, Dict, Optional, Tuple
|
||||
|
||||
from prompt_toolkit.auto_suggest import AutoSuggest, Suggestion
|
||||
from prompt_toolkit.completion import Completer, Completion
|
||||
|
||||
from hermes_cli.commands import COMMANDS, SUBCOMMANDS
|
||||
|
||||
# (config-file signature, personalities) memo for /personality completion.
|
||||
_personalities_memo: Optional[
|
||||
Tuple[Tuple[Optional[str], Optional[int], Optional[int]], Dict[str, Any]]
|
||||
] = None
|
||||
|
||||
|
||||
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.
|
||||
"""
|
||||
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)
|
||||
except Exception:
|
||||
sig = (None, None, None)
|
||||
|
||||
if _personalities_memo is not None and _personalities_memo[0] == sig:
|
||||
return _personalities_memo[1]
|
||||
|
||||
personalities = available_personalities(load_cli_config())
|
||||
_personalities_memo = (sig, personalities)
|
||||
return personalities
|
||||
|
||||
|
||||
def _short_desc(info: Mapping[str, Any], default: str) -> str:
|
||||
"""50-char description preview used in completion menus."""
|
||||
description = str(info.get("description", default))
|
||||
return description[:50] + ("..." if len(description) > 50 else "")
|
||||
|
||||
|
||||
def _file_size_label(path: str) -> str:
|
||||
"""Return a compact human-readable file size, or '' on error."""
|
||||
try:
|
||||
size = os.path.getsize(path)
|
||||
except OSError:
|
||||
return ""
|
||||
if size < 1024:
|
||||
return f"{size}B"
|
||||
if size < 1024 * 1024:
|
||||
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"
|
||||
|
||||
|
||||
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)."""
|
||||
lowered = partial.lower()
|
||||
for name, meta in rows:
|
||||
if name.startswith(lowered) and not (skip_exact and name == lowered):
|
||||
yield Completion(name, start_position=-len(partial), display=name, display_meta=meta)
|
||||
|
||||
|
||||
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."""
|
||||
parts = sub_text.split()
|
||||
if sub_text.endswith(" "):
|
||||
return parts, ""
|
||||
return parts[:-1], (parts[-1] if parts else "")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dynamic argument completers: (sub_text, sub_lower) -> Completion iterator
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
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.
|
||||
"""
|
||||
completed, partial = _split_args(sub_text)
|
||||
if not completed:
|
||||
yield from _prefix_completions(((s, None) for s in ("list", "disable", "enable")), partial)
|
||||
return
|
||||
subcommand = completed[0].lower()
|
||||
if subcommand not in ("enable", "disable"):
|
||||
return
|
||||
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
|
||||
|
||||
# Readonly loader: this runs per keystroke and never mutates config,
|
||||
# so skip the defensive deepcopy of load_config().
|
||||
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]
|
||||
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)
|
||||
except Exception:
|
||||
return
|
||||
|
||||
|
||||
def _handoff_completions(sub_text: str, sub_lower: str):
|
||||
"""/handoff — connected (enabled + configured) 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.
|
||||
"""
|
||||
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:
|
||||
return
|
||||
|
||||
for platform in platforms:
|
||||
name = platform.value
|
||||
if not name.startswith(partial.lower()):
|
||||
continue
|
||||
try:
|
||||
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"
|
||||
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.
|
||||
_DYNAMIC_COMPLETIONS: dict[str, tuple[Callable[..., Any], bool]] = {
|
||||
"/skin": (_skin_completions, True),
|
||||
"/personality": (_personality_completions, True),
|
||||
"/tools": (_tools_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 = text.rpartition(" ")[2]
|
||||
if not word or "://" in word:
|
||||
return None
|
||||
return word if "/" 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*.
|
||||
|
||||
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.
|
||||
"""
|
||||
if expanded.endswith("/"):
|
||||
search_dir, prefix = expanded, ""
|
||||
else:
|
||||
search_dir = os.path.dirname(expanded) or "."
|
||||
prefix = os.path.basename(expanded)
|
||||
try:
|
||||
entries = os.listdir(search_dir)
|
||||
except OSError:
|
||||
return
|
||||
prefix_lower = prefix.lower()
|
||||
count = 0
|
||||
for entry in sorted(entries):
|
||||
if prefix and not entry.lower().startswith(prefix_lower):
|
||||
continue
|
||||
full_path = os.path.join(search_dir, entry)
|
||||
is_dir = os.path.isdir(full_path)
|
||||
if want_dir is not None and want_dir != is_dir:
|
||||
continue
|
||||
if count >= limit:
|
||||
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),
|
||||
)
|
||||
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)."""
|
||||
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
|
||||
yield from _dir_completions(os.path.expanduser(word), word, limit, text_for)
|
||||
|
||||
|
||||
_STATIC_CONTEXT_REFS = (
|
||||
("@diff", "Git working tree diff"),
|
||||
("@staged", "Git staged diff"),
|
||||
("@file:", "Attach a file"),
|
||||
("@folder:", "Attach a folder"),
|
||||
("@git:", "Git log with diffs (e.g. @git:5)"),
|
||||
("@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."""
|
||||
if not query:
|
||||
return 1 # show everything when query is empty
|
||||
lower_file = os.path.basename(filepath).lower()
|
||||
lower_q = query.lower()
|
||||
if lower_file == lower_q:
|
||||
return 100
|
||||
if lower_file.startswith(lower_q):
|
||||
return 80
|
||||
if lower_q in lower_file:
|
||||
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 (_-./).
|
||||
qi = boundary_hits = 0
|
||||
prev = "_" # treat start as boundary
|
||||
for c in lower_file:
|
||||
if qi < len(lower_q) and c == lower_q[qi]:
|
||||
boundary_hits += prev in "_-./"
|
||||
qi += 1
|
||||
prev = c
|
||||
if qi < len(lower_q):
|
||||
return 0
|
||||
return 35 if boundary_hits >= len(lower_q) * 0.5 else 25
|
||||
|
||||
|
||||
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.
|
||||
_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:
|
||||
self._skill_commands_provider = skill_commands_provider
|
||||
self._command_filter = command_filter
|
||||
self._skill_bundles_provider = skill_bundles_provider
|
||||
# Cached project file list for fuzzy @ completions
|
||||
self._file_cache: list[str] = []
|
||||
self._file_cache_time: float = 0.0
|
||||
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))
|
||||
except Exception:
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def _call_provider(provider) -> Mapping[str, dict[str, Any]]:
|
||||
if provider is None:
|
||||
return {}
|
||||
try:
|
||||
return provider() 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)."""
|
||||
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``).
|
||||
|
||||
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.
|
||||
"""
|
||||
try:
|
||||
from agent.skill_commands import _MAX_STACKED_SKILLS as _cap
|
||||
except Exception:
|
||||
_cap = 5
|
||||
|
||||
completed, current_word = _split_args(text)
|
||||
skill_cmds = self._iter_skill_commands()
|
||||
seen: set[str] = set()
|
||||
for token in completed:
|
||||
key = self._normalize_skill_token(token)
|
||||
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
|
||||
|
||||
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).
|
||||
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')}",
|
||||
)
|
||||
|
||||
@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."""
|
||||
if cmd_name != word or cmd_name in SlashCommandCompleter._PICKER_COMMANDS:
|
||||
return cmd_name
|
||||
return f"{cmd_name} "
|
||||
|
||||
@staticmethod
|
||||
def _extract_context_word(text: str) -> str | None:
|
||||
"""Extract a bare ``@`` token for context reference completions."""
|
||||
word = text.rpartition(" ")[2]
|
||||
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."""
|
||||
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.
|
||||
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.
|
||||
yield from _dir_completions(
|
||||
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."""
|
||||
cwd = os.getcwd()
|
||||
now = time.monotonic()
|
||||
if self._file_cache and self._file_cache_cwd == cwd and now - self._file_cache_time < 5.0:
|
||||
return self._file_cache
|
||||
|
||||
files: list[str] = []
|
||||
for cmd in (
|
||||
["rg", "--files", "--sortr=modified", cwd],
|
||||
["rg", "--files", 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",
|
||||
)
|
||||
except (subprocess.TimeoutExpired, OSError):
|
||||
continue
|
||||
if proc.returncode != 0 or not proc.stdout.strip():
|
||||
continue
|
||||
for p in proc.stdout.strip().split("\n")[:5000]:
|
||||
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
|
||||
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)."""
|
||||
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:
|
||||
is_dir = fp.endswith("/")
|
||||
meta = "dir" if is_dir else _file_size_label(os.path.join(os.getcwd(), fp))
|
||||
if query:
|
||||
meta = f"{fp} {meta}" if meta else fp
|
||||
yield Completion(
|
||||
f"@{'folder' if is_dir else 'file'}:{fp}",
|
||||
start_position=-len(word),
|
||||
display=os.path.basename(fp),
|
||||
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)
|
||||
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:
|
||||
yield from _path_completions(path_word)
|
||||
return
|
||||
|
||||
parts = text.split(maxsplit=1)
|
||||
base_cmd = parts[0].lower()
|
||||
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.
|
||||
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)
|
||||
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,
|
||||
)
|
||||
|
||||
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():
|
||||
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)")
|
||||
|
||||
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')}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
class SlashCommandAutoSuggest(AutoSuggest):
|
||||
"""Inline ghost-text for slash commands and their subcommands; history fallback for other input."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
history_suggest: AutoSuggest | 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 get_suggestion(self, buffer, document):
|
||||
text = document.text_before_cursor
|
||||
if not text.startswith("/"):
|
||||
return self._history_suggestion(buffer, document)
|
||||
|
||||
parts = text.split(maxsplit=1)
|
||||
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).
|
||||
word = text[1:].lower()
|
||||
for cmd in sorted(COMMANDS, key=len):
|
||||
cmd_name = cmd[1:]
|
||||
if self._allowed(cmd) and cmd_name.startswith(word) and cmd_name != word:
|
||||
return Suggestion(cmd_name[len(word):])
|
||||
return None
|
||||
|
||||
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.
|
||||
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
|
||||
if remainder.strip():
|
||||
return Suggestion(remainder)
|
||||
|
||||
if not self._allowed(base_cmd):
|
||||
return None
|
||||
if " " not in sub_text:
|
||||
for sub in SUBCOMMANDS.get(base_cmd, ()):
|
||||
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
|
||||
@@ -0,0 +1,495 @@
|
||||
"""Gateway platform command derivations (Telegram / Discord / Slack) from ``COMMAND_REGISTRY``.
|
||||
|
||||
Extracted from :mod:`hermes_cli.commands`, which re-exports every name here so
|
||||
``from hermes_cli.commands import telegram_menu_commands`` keeps working.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from collections.abc import Callable, Mapping, Sequence
|
||||
from typing import Any
|
||||
|
||||
from hermes_cli.commands import (
|
||||
COMMAND_REGISTRY,
|
||||
_is_gateway_available,
|
||||
_iter_plugin_command_entries,
|
||||
_resolve_config_gates,
|
||||
)
|
||||
|
||||
# Logger name parity with the origin module (tests capture "hermes_cli.commands").
|
||||
logger = logging.getLogger("hermes_cli.commands")
|
||||
|
||||
_CMD_NAME_LIMIT = 32
|
||||
"""Max command name length shared by Telegram and Discord."""
|
||||
|
||||
_TG_INVALID_CHARS = re.compile(r"[^a-z0-9_]")
|
||||
_TG_MULTI_UNDERSCORE = re.compile(r"_{2,}")
|
||||
|
||||
|
||||
def _requires_argument(args_hint: str) -> bool:
|
||||
"""True when selecting a command without text would be incomplete."""
|
||||
return args_hint.strip().startswith("<")
|
||||
|
||||
|
||||
def _sanitize_telegram_name(raw: str) -> str:
|
||||
"""Telegram allows only ``[a-z0-9_]``: lowercase → hyphens to underscores → strip the rest → collapse/strip ``_``."""
|
||||
name = _TG_INVALID_CHARS.sub("", raw.lower().replace("-", "_"))
|
||||
return _TG_MULTI_UNDERSCORE.sub("_", name).strip("_")
|
||||
|
||||
|
||||
def _truncate_desc(desc: str, limit: int) -> str:
|
||||
"""Clamp a menu description to *limit* chars with a ``...`` tail."""
|
||||
return desc if len(desc) <= limit else desc[:limit - 3] + "..."
|
||||
|
||||
|
||||
def _clamp_command_names(entries: Sequence[tuple[str, ...]], reserved: set[str]) -> list[tuple[str, ...]]:
|
||||
"""Enforce the 32-char Telegram/Discord name limit with collision avoidance.
|
||||
|
||||
Over-long names are truncated; if that collides with *reserved* or an
|
||||
earlier entry, a 31-char prefix + digit ``0``-``9`` is tried, and the entry
|
||||
is silently dropped when all ten are taken. Duplicate names are dropped.
|
||||
Elements beyond ``(name, desc)`` pass through unchanged.
|
||||
"""
|
||||
used: set[str] = set(reserved)
|
||||
result: list = []
|
||||
for name, desc, *extra in entries:
|
||||
if len(name) > _CMD_NAME_LIMIT:
|
||||
candidate = name[:_CMD_NAME_LIMIT]
|
||||
if candidate in used:
|
||||
prefix = name[:_CMD_NAME_LIMIT - 1]
|
||||
for digit in range(10):
|
||||
candidate = f"{prefix}{digit}"
|
||||
if candidate not in used:
|
||||
break
|
||||
else:
|
||||
continue
|
||||
name = candidate
|
||||
if name in used:
|
||||
continue
|
||||
used.add(name)
|
||||
result.append((name, desc, *extra))
|
||||
return result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Telegram
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def telegram_bot_commands(*, include_plugins: bool = True) -> list[tuple[str, str]]:
|
||||
"""Return (command_name, description) pairs for Telegram setMyCommands.
|
||||
|
||||
Names are Telegram-sanitized; aliases are skipped (one menu entry per
|
||||
canonical command). Built-ins that require arguments are **included** —
|
||||
their handlers show usage text when selected bare. Plugin commands
|
||||
requiring arguments are **excluded** because plugins may lack a no-arg
|
||||
fallback; callers needing source metadata pass ``include_plugins=False``
|
||||
and use :func:`_collect_gateway_skill_entries`.
|
||||
"""
|
||||
overrides = _resolve_config_gates()
|
||||
pairs = [(cmd.name, cmd.description) for cmd in COMMAND_REGISTRY if _is_gateway_available(cmd, overrides)]
|
||||
if include_plugins:
|
||||
pairs += [(n, d) for n, d, hint in _iter_plugin_command_entries() if not _requires_argument(hint)]
|
||||
return [(tg, desc) for name, desc in pairs if (tg := _sanitize_telegram_name(name))]
|
||||
|
||||
|
||||
# Telegram allows 100 BotCommands; the 60-slot default keeps every built-in
|
||||
# plus common skill commands visible while staying under the ~4KB payload
|
||||
# limit. Tunable via platforms.telegram.extra.command_menu.max_commands.
|
||||
_DEFAULT_TELEGRAM_MENU_MAX_COMMANDS = 60
|
||||
_TELEGRAM_BOT_API_MAX_COMMANDS = 100
|
||||
# priority_mode -> rank tables consulted in order ("configured" = user list,
|
||||
# "default" = _TELEGRAM_MENU_PRIORITY); unranked candidates keep stable order after.
|
||||
_TELEGRAM_PRIORITY_TIERS: dict[str, tuple[str, ...]] = {
|
||||
"prepend": ("configured", "default"),
|
||||
"append": ("default", "configured"),
|
||||
"replace": ("configured",),
|
||||
}
|
||||
|
||||
# Built-ins that must survive Telegram's small visible menu cap; everything
|
||||
# else stays dispatchable when typed manually. Order = rank.
|
||||
_TELEGRAM_MENU_PRIORITY = (
|
||||
# Most-typed everyday commands first.
|
||||
"help", "new", "stop", "status", "egress", "resume", "sessions", "model",
|
||||
# Maintenance / diagnostics.
|
||||
"debug", "restart", "update", "verbose", "commands",
|
||||
# Mid-turn session control.
|
||||
"approve", "deny", "queue", "steer", "bg", "btw",
|
||||
# Lower-priority but still useful operational built-ins.
|
||||
"reasoning", "usage", "platforms", "platform", "profile", "whoami",
|
||||
)
|
||||
|
||||
|
||||
def _telegram_command_menu_config() -> dict[str, Any]:
|
||||
"""Normalized ``platforms.telegram.extra.command_menu`` config with safe defaults."""
|
||||
try:
|
||||
from hermes_cli.config import read_raw_config
|
||||
node: Any = read_raw_config() or {}
|
||||
except Exception:
|
||||
node = {}
|
||||
for key in ("platforms", "telegram", "extra", "command_menu"):
|
||||
node = node.get(key) if isinstance(node, Mapping) else None
|
||||
menu_cfg: Mapping[str, Any] = node if isinstance(node, Mapping) else {}
|
||||
|
||||
try:
|
||||
max_commands = int(menu_cfg.get("max_commands", _DEFAULT_TELEGRAM_MENU_MAX_COMMANDS))
|
||||
except (TypeError, ValueError):
|
||||
max_commands = _DEFAULT_TELEGRAM_MENU_MAX_COMMANDS
|
||||
priority_mode = str(menu_cfg.get("priority_mode") or "prepend").strip().lower()
|
||||
raw_priority = menu_cfg.get("priority")
|
||||
if isinstance(raw_priority, list):
|
||||
priority = [str(item) for item in raw_priority if str(item).strip()]
|
||||
else:
|
||||
priority = [raw_priority] if isinstance(raw_priority, str) and raw_priority.strip() else []
|
||||
return {
|
||||
"max_commands": max(1, min(_TELEGRAM_BOT_API_MAX_COMMANDS, max_commands)),
|
||||
"priority_mode": priority_mode if priority_mode in _TELEGRAM_PRIORITY_TIERS else "prepend",
|
||||
"priority": priority,
|
||||
}
|
||||
|
||||
|
||||
def telegram_menu_max_commands() -> int:
|
||||
"""Return configured Telegram BotCommand menu cap with safe bounds."""
|
||||
return int(_telegram_command_menu_config()["max_commands"])
|
||||
|
||||
|
||||
def _sanitized_rank(raw_names: Sequence[str]) -> dict[str, int]:
|
||||
"""name -> rank for the deduped, Telegram-sanitized *raw_names* (first occurrence wins)."""
|
||||
rank: dict[str, int] = {}
|
||||
for raw in raw_names:
|
||||
name = _sanitize_telegram_name(str(raw))
|
||||
if name and name not in rank:
|
||||
rank[name] = len(rank)
|
||||
return rank
|
||||
|
||||
|
||||
def _prioritize_telegram_menu_candidates(
|
||||
candidates: list[tuple[str, str, str, str]],
|
||||
) -> list[tuple[str, str, str, str]]:
|
||||
"""Order ``(final_name, description, source, raw_name)`` candidates; default priority applies to core only.
|
||||
|
||||
``raw_name`` is the pre-clamp name so an explicitly configured long command
|
||||
stays addressable after Telegram name clamping. "replace" mode ignores the
|
||||
built-in defaults entirely.
|
||||
"""
|
||||
# Lazy origin import: tests patch ``hermes_cli.commands._telegram_command_menu_config``.
|
||||
from hermes_cli.commands import _telegram_command_menu_config as menu_config
|
||||
|
||||
menu_cfg = menu_config()
|
||||
configured_rank = _sanitized_rank(menu_cfg["priority"])
|
||||
default_rank = _sanitized_rank(_TELEGRAM_MENU_PRIORITY)
|
||||
tiers = _TELEGRAM_PRIORITY_TIERS[menu_cfg["priority_mode"]]
|
||||
|
||||
def _rank(stable_index: int, candidate: tuple[str, str, str, str]) -> tuple[int, int, int]:
|
||||
final_name, _desc, source, raw_name = candidate
|
||||
indexes = {
|
||||
"configured": configured_rank.get(raw_name, configured_rank.get(final_name)),
|
||||
"default": default_rank.get(final_name) if source == "core" else None,
|
||||
}
|
||||
for tier, table in enumerate(tiers):
|
||||
if indexes[table] is not None:
|
||||
return (tier, indexes[table], stable_index)
|
||||
return (len(tiers), 0, stable_index)
|
||||
|
||||
return [c for _i, c in sorted(enumerate(candidates), key=lambda item: _rank(*item))]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared skill/plugin collection for gateway platforms
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _iter_gateway_skills(platform: str):
|
||||
"""Yield ``(cmd_key, info, rel_parts)`` for skills eligible as gateway slash commands.
|
||||
|
||||
Scan roots are the local ``SKILLS_DIR`` plus every configured
|
||||
``skills.external_dirs`` / trusted project skills dir — a skill anywhere
|
||||
else, or under the hub dir (``SKILLS_DIR/.hub``), is skipped, as are skills
|
||||
disabled for *platform*. Paths are resolved on both sides so symlinked roots
|
||||
(macOS ``/var`` → ``/private/var``) still match, and matching is per path
|
||||
component so ``/my-skills`` never admits ``/my-skills-extra``. ``rel_parts``
|
||||
is the skill dir relative to its root (``("creative", "ascii-art")``) for
|
||||
category derivation. Iterates ``sorted(skill_cmds)`` so first-wins
|
||||
collision handling is alphabetical.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
from agent.skill_commands import get_skill_commands
|
||||
from agent.skill_utils import get_disabled_skill_names, get_external_skills_dirs, get_project_skills_dirs
|
||||
from tools.skills_tool import SKILLS_DIR
|
||||
|
||||
try:
|
||||
disabled = get_disabled_skill_names(platform=platform)
|
||||
except Exception:
|
||||
disabled = set()
|
||||
hub_dir = (SKILLS_DIR / ".hub").resolve()
|
||||
roots = [SKILLS_DIR.resolve()]
|
||||
for getter in (get_external_skills_dirs, get_project_skills_dirs):
|
||||
try:
|
||||
for d in getter():
|
||||
try:
|
||||
roots.append(Path(d).resolve())
|
||||
except Exception:
|
||||
continue
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
skill_cmds = get_skill_commands()
|
||||
for cmd_key in sorted(skill_cmds):
|
||||
info = skill_cmds[cmd_key]
|
||||
skill_path = info.get("skill_md_path", "")
|
||||
if not skill_path:
|
||||
continue
|
||||
sp = Path(skill_path).resolve()
|
||||
if sp.is_relative_to(hub_dir):
|
||||
continue
|
||||
root = next((r for r in roots if sp.is_relative_to(r)), None)
|
||||
if root is None or info.get("name", "") in disabled:
|
||||
continue
|
||||
yield cmd_key, info, sp.parent.relative_to(root).parts
|
||||
|
||||
|
||||
def _collect_gateway_skill_entries(
|
||||
platform: str,
|
||||
max_slots: int | None,
|
||||
reserved_names: set[str],
|
||||
desc_limit: int = 100,
|
||||
sanitize_name: "Callable[[str], str] | None" = None,
|
||||
) -> tuple[list[tuple[str, str, str, str]], int]:
|
||||
"""Collect plugin + skill entries for a gateway platform.
|
||||
|
||||
Plugin slash commands come first and are never trimmed; skill commands
|
||||
(alphabetical, see :func:`_iter_gateway_skills`) fill the remaining
|
||||
*max_slots* — ``None`` returns every candidate for a caller applying its
|
||||
own global cap. *reserved_names* (built-in names) is mutated in place as
|
||||
names are claimed. *sanitize_name* runs before clamping and may return ""
|
||||
to skip an entry. Returns ``(entries, hidden_count)`` with entries of
|
||||
``(name, description, cmd_key, raw_name)`` — ``cmd_key`` is the original
|
||||
skill key ("" for plugins); ``raw_name`` the sanitized pre-clamp name used
|
||||
for configured-priority matching.
|
||||
"""
|
||||
sanitize = sanitize_name or (lambda n: n)
|
||||
# Tier 1: plugin slash commands (never trimmed). No cmd_key — "" placeholder.
|
||||
plugin_entries: list[tuple[str, str, str, str]] = []
|
||||
try:
|
||||
from hermes_cli.plugins import get_plugin_commands
|
||||
plugin_cmds = get_plugin_commands()
|
||||
for cmd_name in sorted(plugin_cmds):
|
||||
if platform == "telegram" and _requires_argument(str(plugin_cmds[cmd_name].get("args_hint") or "")):
|
||||
continue
|
||||
if name := sanitize(cmd_name):
|
||||
desc = _truncate_desc(plugin_cmds[cmd_name].get("description", "Plugin command"), desc_limit)
|
||||
plugin_entries.append((name, desc, "", name))
|
||||
except Exception:
|
||||
pass
|
||||
plugin_entries = _clamp_command_names(plugin_entries, reserved_names)
|
||||
reserved_names.update(n for n, *_rest in plugin_entries)
|
||||
|
||||
# Tier 2: skill commands (the only tier trimmed at the cap). cmd_key and
|
||||
# raw_name survive any clamp-induced rename.
|
||||
skill_entries: list[tuple[str, str, str, str]] = []
|
||||
try:
|
||||
for cmd_key, info, _rel in _iter_gateway_skills(platform):
|
||||
if name := sanitize(cmd_key.lstrip("/")):
|
||||
skill_entries.append((name, _truncate_desc(info.get("description", ""), desc_limit), cmd_key, name))
|
||||
except Exception:
|
||||
pass
|
||||
skill_entries = _clamp_command_names(skill_entries, reserved_names)
|
||||
|
||||
if max_slots is None:
|
||||
return plugin_entries + skill_entries, 0
|
||||
remaining = max(0, max_slots - len(plugin_entries))
|
||||
hidden_count = max(0, len(skill_entries) - remaining)
|
||||
return (plugin_entries + skill_entries[:remaining])[:max_slots], hidden_count
|
||||
|
||||
|
||||
def telegram_menu_commands(max_commands: int = 100) -> tuple[list[tuple[str, str]], int]:
|
||||
"""Return ``(menu_commands, hidden_count)`` for Telegram, capped to the Bot API limit.
|
||||
|
||||
Tier order: core CommandDefs, then plugin slash commands, then skill
|
||||
commands (alphabetical; hub skills and telegram-disabled skills excluded).
|
||||
Tiers keep their relative order unless named in
|
||||
``platforms.telegram.extra.command_menu.priority`` — explicit priority is
|
||||
applied to the combined list *before* the cap, so a prioritized dynamic
|
||||
command can displace an unprioritized core command.
|
||||
"""
|
||||
# Lazy origin import: tests patch ``hermes_cli.commands.telegram_bot_commands``.
|
||||
from hermes_cli.commands import telegram_bot_commands as bot_commands
|
||||
|
||||
core_commands = list(bot_commands(include_plugins=False))
|
||||
entries, hidden_count = _collect_gateway_skill_entries(
|
||||
platform="telegram",
|
||||
max_slots=None,
|
||||
reserved_names={n for n, _ in core_commands},
|
||||
desc_limit=40,
|
||||
sanitize_name=_sanitize_telegram_name,
|
||||
)
|
||||
candidates = [(name, desc, "core", name) for name, desc in core_commands]
|
||||
candidates += [(name, desc, "skill" if cmd_key else "plugin", raw) for name, desc, cmd_key, raw in entries]
|
||||
candidates = _prioritize_telegram_menu_candidates(candidates)
|
||||
overflow_count = max(0, len(candidates) - max_commands)
|
||||
menu = [(name, desc) for name, desc, _source, _raw_name in candidates[:max_commands]]
|
||||
return menu, hidden_count + overflow_count
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Discord
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def discord_skill_commands_by_category(
|
||||
reserved_names: set[str],
|
||||
) -> tuple[dict[str, list[tuple[str, str, str]]], list[tuple[str, str, str]], int]:
|
||||
"""Return ``(categories, uncategorized, hidden_count)`` for Discord ``/skill`` autocomplete.
|
||||
|
||||
Skills nested >= 2 levels under a scan root (``creative/ascii-art/SKILL.md``)
|
||||
are grouped under ``categories[top_level]``; root-level skills are
|
||||
*uncategorized*. Entries are ``(name, description, cmd_key)`` with names
|
||||
clamped to 32 chars and descriptions to 100. Eligibility follows
|
||||
:func:`_iter_gateway_skills`. No per-group cap is applied — the caller
|
||||
flattens everything into one autocomplete callback, which scales to
|
||||
thousands of entries; ``hidden_count`` only reports 32-char clamp
|
||||
collisions against reserved names or earlier skills.
|
||||
"""
|
||||
categories: dict[str, list[tuple[str, str, str]]] = {}
|
||||
uncategorized: list[tuple[str, str, str]] = []
|
||||
# clamped name → origin. Reserved (gateway-builtin) names carry a sentinel
|
||||
# so the warning can distinguish "collided with a reserved command" from
|
||||
# "two skills collided on the 32-char clamp" — the rename-worthy case.
|
||||
names_used: dict[str, str] = dict.fromkeys(reserved_names, "<reserved>")
|
||||
hidden = 0
|
||||
try:
|
||||
for cmd_key, info, rel_parts in _iter_gateway_skills("discord"):
|
||||
# On collision the first (alphabetical) skill wins and the loser is
|
||||
# dropped from the picker; warn loudly, since a silent ``hidden``
|
||||
# count gave skill authors no way to discover the drop.
|
||||
discord_name = cmd_key.lstrip("/")[:32]
|
||||
prior = names_used.get(discord_name)
|
||||
if prior == "<reserved>":
|
||||
logger.warning(
|
||||
"Discord /skill: %r (from %r) collides on its 32-char "
|
||||
"clamp with a reserved gateway command name %r — the "
|
||||
"skill will not appear in the /skill autocomplete. "
|
||||
"Rename the skill's frontmatter ``name:`` to differ "
|
||||
"in its first 32 chars.",
|
||||
discord_name, cmd_key, discord_name,
|
||||
)
|
||||
elif prior is not None:
|
||||
logger.warning(
|
||||
"Discord /skill: %r and %r both clamp to %r on "
|
||||
"Discord's 32-char command-name limit — only %r "
|
||||
"will appear in the /skill autocomplete. Rename "
|
||||
"one skill's frontmatter ``name:`` to differ in "
|
||||
"its first 32 chars.",
|
||||
prior, cmd_key, discord_name, prior,
|
||||
)
|
||||
if prior is not None:
|
||||
hidden += 1
|
||||
continue
|
||||
names_used[discord_name] = cmd_key
|
||||
entry = (discord_name, _truncate_desc(info.get("description", ""), 100), cmd_key)
|
||||
# creative/ascii-art/SKILL.md → category "creative"; root-level skills are uncategorized.
|
||||
(categories.setdefault(rel_parts[0], []) if len(rel_parts) >= 2 else uncategorized).append(entry)
|
||||
except Exception:
|
||||
pass
|
||||
return categories, uncategorized, hidden
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Slack native slash commands
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Slack slash names: lowercase a-z, 0-9, hyphens, underscores, max 32 chars;
|
||||
# an app manifest accepts up to 50 slash commands.
|
||||
_SLACK_MAX_SLASH_COMMANDS = 50
|
||||
_SLACK_NAME_LIMIT = 32
|
||||
_SLACK_INVALID_CHARS = re.compile(r"[^a-z0-9_\-]")
|
||||
_SLACK_RESERVED_COMMANDS = frozenset({
|
||||
# Built-in Slack slash commands that cannot be registered by apps.
|
||||
# https://slack.com/help/articles/201259356-Use-built-in-slash-commands
|
||||
"me", "status", "away", "dnd", "shrug", "remind", "msg", "feed",
|
||||
"who", "collapse", "expand", "leave", "join", "open", "search",
|
||||
"topic", "mute", "pro", "shortcuts",
|
||||
})
|
||||
|
||||
# Canonical commands intentionally NOT given a native Slack slash slot. Slack
|
||||
# caps apps at 50 slash commands and the registry is at that ceiling; rather
|
||||
# than let the clamp silently drop whichever command sorts last (breaking the
|
||||
# Telegram-parity test), low-frequency commands are routed through
|
||||
# ``/hermes <command>`` on Slack only. They stay native on every other surface.
|
||||
# Rule: when a new canonical command tips the registry past the cap, demote a
|
||||
# rarer one-off lookup here (version, whoami, platform, diff, update, ...)
|
||||
# rather than a recurring interactive surface (context, loop, save, approvals).
|
||||
# Keep TIGHT and intentional — the parity test reads this set. (Aliases are
|
||||
# never pinned ahead of canonicals: /bg and /btw became canonical commands
|
||||
# instead, so they win first-pass slots on their own.)
|
||||
_SLACK_VIA_HERMES_ONLY = frozenset({"topup", "moa", "debug", "egress", "init", "version", "diff", "update", "heartbeat", "refine", "review", "pause", "whoami", "platform", "insights"})
|
||||
|
||||
|
||||
def _sanitize_slack_name(raw: str) -> str:
|
||||
"""Lowercase, strip chars outside ``[a-z0-9_-]`` and edge ``-_``, clamp to 32."""
|
||||
return _SLACK_INVALID_CHARS.sub("", raw.lower()).strip("-_")[:_SLACK_NAME_LIMIT]
|
||||
|
||||
|
||||
def slack_native_slashes() -> list[tuple[str, str, str]]:
|
||||
"""Return (slash_name, description, usage_hint) triples for Slack.
|
||||
|
||||
Every gateway-available command (canonical names first, then aliases,
|
||||
then plugin commands) becomes a standalone Slack slash, clamped to the
|
||||
50-command cap with duplicate avoidance. Names colliding with a Slack
|
||||
built-in (``/status``, ``/me``, ...) or listed in _SLACK_VIA_HERMES_ONLY
|
||||
are skipped. ``/hermes`` is always the first entry so the
|
||||
``/hermes <command>`` form keeps working for anything dropped.
|
||||
"""
|
||||
overrides = _resolve_config_gates()
|
||||
available = [cmd for cmd in COMMAND_REGISTRY if _is_gateway_available(cmd, overrides)]
|
||||
# Canonical names first so they win slots at the cap; aliases second; plugins third.
|
||||
wanted = [(cmd.name, cmd.description, cmd.args_hint or "") for cmd in available]
|
||||
wanted += [(alias, f"Alias for /{cmd.name} — {cmd.description}", cmd.args_hint or "")
|
||||
for cmd in available for alias in cmd.aliases]
|
||||
wanted += [(name, desc, hint or "") for name, desc, hint in _iter_plugin_command_entries()]
|
||||
|
||||
entries: list[tuple[str, str, str]] = [("hermes", "Talk to Hermes or run a subcommand", "[subcommand] [args]")]
|
||||
seen = {"hermes"}
|
||||
for name, desc, hint in wanted:
|
||||
slack_name = _sanitize_slack_name(name)
|
||||
if (
|
||||
not slack_name
|
||||
or slack_name in seen
|
||||
or slack_name in _SLACK_RESERVED_COMMANDS
|
||||
or slack_name in _SLACK_VIA_HERMES_ONLY
|
||||
or len(entries) >= _SLACK_MAX_SLASH_COMMANDS
|
||||
):
|
||||
continue
|
||||
# Slack description cap is 2000 chars; keep it short.
|
||||
entries.append((slack_name, desc[:140], hint[:100]))
|
||||
seen.add(slack_name)
|
||||
return entries
|
||||
|
||||
|
||||
def slack_app_manifest(request_url: str = "https://hermes-agent.local/slack/commands") -> dict[str, Any]:
|
||||
"""Return the ``features.slash_commands`` manifest portion for all gateway slashes.
|
||||
|
||||
``request_url`` is schema-required but ignored in Socket Mode (a
|
||||
placeholder is fine). Only this portion is returned so we stay decoupled
|
||||
from the rest of the manifest users configure once in the Slack UI.
|
||||
"""
|
||||
slashes = []
|
||||
for name, desc, usage in slack_native_slashes():
|
||||
entry = {"command": f"/{name}", "description": desc or f"Run /{name}", "should_escape": False, "url": request_url}
|
||||
if usage:
|
||||
entry["usage_hint"] = usage
|
||||
slashes.append(entry)
|
||||
return {"features": {"slash_commands": slashes}}
|
||||
|
||||
|
||||
def slack_subcommand_map() -> dict[str, str]:
|
||||
"""Return name/alias -> "/command" mapping for the Slack ``/hermes`` handler, plugin commands included."""
|
||||
overrides = _resolve_config_gates()
|
||||
mapping: dict[str, str] = {}
|
||||
for cmd in COMMAND_REGISTRY:
|
||||
if _is_gateway_available(cmd, overrides):
|
||||
for name in (cmd.name, *cmd.aliases):
|
||||
mapping[name] = f"/{name}"
|
||||
for name, _description, _args_hint in _iter_plugin_command_entries():
|
||||
mapping.setdefault(name, f"/{name}")
|
||||
return mapping
|
||||
@@ -18,7 +18,7 @@ from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
import hermes_cli.commands as commands_mod
|
||||
import hermes_cli.commands_completion as commands_mod
|
||||
|
||||
|
||||
def _reset_personalities_memo():
|
||||
|
||||
@@ -96,7 +96,7 @@ class TestIntegration:
|
||||
# scheme guard it reached _path_completions and called os.listdir on
|
||||
# every keystroke. Assert no completions AND that the filesystem is
|
||||
# never touched while a URL is under the cursor.
|
||||
import hermes_cli.commands as commands_mod
|
||||
import hermes_cli.commands_completion as commands_mod
|
||||
|
||||
def _fail(*_args, **_kwargs):
|
||||
raise AssertionError("os.listdir must not run for a URL token")
|
||||
|
||||
Reference in New Issue
Block a user