2ae50bdddd
Telegram Bot API requires command names to be 1-32 characters. Plugin and skill names that exceed this limit now get truncated. If truncation creates a collision (with core commands, other plugins, or other skills), the name is shortened to 31 chars and a digit 0-9 is appended. Adds _clamp_telegram_names() helper used for both plugin and skill entries in telegram_menu_commands(). Core CommandDef commands are tracked as reserved names so truncated plugin/skill names never shadow them. Addresses the fix from PR #4191 (sroecker) with collision-safe truncation. Tests: 9 new tests covering truncation, digit suffixes, exhaustion, dedup.
856 lines
33 KiB
Python
856 lines
33 KiB
Python
"""Slash command definitions and autocomplete for the Hermes CLI.
|
|
|
|
Central registry for all slash commands. Every consumer -- CLI help, gateway
|
|
dispatch, Telegram BotCommands, Slack subcommand mapping, autocomplete --
|
|
derives its data from ``COMMAND_REGISTRY``.
|
|
|
|
To add a command: add a ``CommandDef`` entry to ``COMMAND_REGISTRY``.
|
|
To add an alias: set ``aliases=("short",)`` on the existing ``CommandDef``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from collections.abc import Callable, Mapping
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from prompt_toolkit.auto_suggest import AutoSuggest, Suggestion
|
|
from prompt_toolkit.completion import Completer, Completion
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CommandDef dataclass
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@dataclass(frozen=True)
|
|
class CommandDef:
|
|
"""Definition of a single slash command."""
|
|
|
|
name: str # canonical name without slash: "background"
|
|
description: str # human-readable description
|
|
category: str # "Session", "Configuration", etc.
|
|
aliases: tuple[str, ...] = () # alternative names: ("bg",)
|
|
args_hint: str = "" # argument placeholder: "<prompt>", "[name]"
|
|
subcommands: tuple[str, ...] = () # tab-completable subcommands
|
|
cli_only: bool = False # only available in CLI
|
|
gateway_only: bool = False # only available in gateway/messaging
|
|
gateway_config_gate: str | None = None # config dotpath; when truthy, overrides cli_only for gateway
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Central registry -- single source of truth
|
|
# ---------------------------------------------------------------------------
|
|
|
|
COMMAND_REGISTRY: list[CommandDef] = [
|
|
# Session
|
|
CommandDef("new", "Start a new session (fresh session ID + history)", "Session",
|
|
aliases=("reset",)),
|
|
CommandDef("clear", "Clear screen and start a new session", "Session",
|
|
cli_only=True),
|
|
CommandDef("history", "Show conversation history", "Session",
|
|
cli_only=True),
|
|
CommandDef("save", "Save the current conversation", "Session",
|
|
cli_only=True),
|
|
CommandDef("retry", "Retry the last message (resend to agent)", "Session"),
|
|
CommandDef("undo", "Remove the last user/assistant exchange", "Session"),
|
|
CommandDef("title", "Set a title for the current session", "Session",
|
|
args_hint="[name]"),
|
|
CommandDef("compress", "Manually compress conversation context", "Session"),
|
|
CommandDef("rollback", "List or restore filesystem checkpoints", "Session",
|
|
args_hint="[number]"),
|
|
CommandDef("stop", "Kill all running background processes", "Session"),
|
|
CommandDef("approve", "Approve a pending dangerous command", "Session",
|
|
gateway_only=True, args_hint="[session|always]"),
|
|
CommandDef("deny", "Deny a pending dangerous command", "Session",
|
|
gateway_only=True),
|
|
CommandDef("background", "Run a prompt in the background", "Session",
|
|
aliases=("bg",), args_hint="<prompt>"),
|
|
CommandDef("btw", "Ephemeral side question using session context (no tools, not persisted)", "Session",
|
|
args_hint="<question>"),
|
|
CommandDef("queue", "Queue a prompt for the next turn (doesn't interrupt)", "Session",
|
|
aliases=("q",), args_hint="<prompt>"),
|
|
CommandDef("status", "Show session info", "Session",
|
|
gateway_only=True),
|
|
CommandDef("profile", "Show active profile name and home directory", "Info"),
|
|
CommandDef("sethome", "Set this chat as the home channel", "Session",
|
|
gateway_only=True, aliases=("set-home",)),
|
|
CommandDef("resume", "Resume a previously-named session", "Session",
|
|
args_hint="[name]"),
|
|
|
|
# Configuration
|
|
CommandDef("config", "Show current configuration", "Configuration",
|
|
cli_only=True),
|
|
CommandDef("provider", "Show available providers and current provider",
|
|
"Configuration"),
|
|
CommandDef("prompt", "View/set custom system prompt", "Configuration",
|
|
cli_only=True, args_hint="[text]", subcommands=("clear",)),
|
|
CommandDef("personality", "Set a predefined personality", "Configuration",
|
|
args_hint="[name]"),
|
|
CommandDef("statusbar", "Toggle the context/model status bar", "Configuration",
|
|
cli_only=True, aliases=("sb",)),
|
|
CommandDef("verbose", "Cycle tool progress display: off -> new -> all -> verbose",
|
|
"Configuration", cli_only=True,
|
|
gateway_config_gate="display.tool_progress_command"),
|
|
CommandDef("yolo", "Toggle YOLO mode (skip all dangerous command approvals)",
|
|
"Configuration"),
|
|
CommandDef("reasoning", "Manage reasoning effort and display", "Configuration",
|
|
args_hint="[level|show|hide]",
|
|
subcommands=("none", "low", "minimal", "medium", "high", "xhigh", "show", "hide", "on", "off")),
|
|
CommandDef("skin", "Show or change the display skin/theme", "Configuration",
|
|
cli_only=True, args_hint="[name]"),
|
|
CommandDef("voice", "Toggle voice mode", "Configuration",
|
|
args_hint="[on|off|tts|status]", subcommands=("on", "off", "tts", "status")),
|
|
|
|
# Tools & Skills
|
|
CommandDef("tools", "Manage tools: /tools [list|disable|enable] [name...]", "Tools & Skills",
|
|
args_hint="[list|disable|enable] [name...]", cli_only=True),
|
|
CommandDef("toolsets", "List available toolsets", "Tools & Skills",
|
|
cli_only=True),
|
|
CommandDef("skills", "Search, install, inspect, or manage skills",
|
|
"Tools & Skills", cli_only=True,
|
|
subcommands=("search", "browse", "inspect", "install")),
|
|
CommandDef("cron", "Manage scheduled tasks", "Tools & Skills",
|
|
cli_only=True, args_hint="[subcommand]",
|
|
subcommands=("list", "add", "create", "edit", "pause", "resume", "run", "remove")),
|
|
CommandDef("reload-mcp", "Reload MCP servers from config", "Tools & Skills",
|
|
aliases=("reload_mcp",)),
|
|
CommandDef("browser", "Connect browser tools to your live Chrome via CDP", "Tools & Skills",
|
|
cli_only=True, args_hint="[connect|disconnect|status]",
|
|
subcommands=("connect", "disconnect", "status")),
|
|
CommandDef("plugins", "List installed plugins and their status",
|
|
"Tools & Skills", cli_only=True),
|
|
|
|
# Info
|
|
CommandDef("commands", "Browse all commands and skills (paginated)", "Info",
|
|
gateway_only=True, args_hint="[page]"),
|
|
CommandDef("help", "Show available commands", "Info"),
|
|
CommandDef("usage", "Show token usage for the current session", "Info"),
|
|
CommandDef("insights", "Show usage insights and analytics", "Info",
|
|
args_hint="[days]"),
|
|
CommandDef("platforms", "Show gateway/messaging platform status", "Info",
|
|
cli_only=True, aliases=("gateway",)),
|
|
CommandDef("paste", "Check clipboard for an image and attach it", "Info",
|
|
cli_only=True),
|
|
CommandDef("update", "Update Hermes Agent to the latest version", "Info",
|
|
gateway_only=True),
|
|
|
|
# Exit
|
|
CommandDef("quit", "Exit the CLI", "Exit",
|
|
cli_only=True, aliases=("exit", "q")),
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Derived lookups -- rebuilt once at import time, refreshed by rebuild_lookups()
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _build_command_lookup() -> dict[str, CommandDef]:
|
|
"""Map every name and alias to its CommandDef."""
|
|
lookup: dict[str, CommandDef] = {}
|
|
for cmd in COMMAND_REGISTRY:
|
|
lookup[cmd.name] = cmd
|
|
for alias in cmd.aliases:
|
|
lookup[alias] = cmd
|
|
return lookup
|
|
|
|
|
|
_COMMAND_LOOKUP: dict[str, CommandDef] = _build_command_lookup()
|
|
|
|
|
|
def resolve_command(name: str) -> CommandDef | None:
|
|
"""Resolve a command name or alias to its CommandDef.
|
|
|
|
Accepts names with or without the leading slash.
|
|
"""
|
|
return _COMMAND_LOOKUP.get(name.lower().lstrip("/"))
|
|
|
|
|
|
def register_plugin_command(cmd: CommandDef) -> None:
|
|
"""Append a plugin-defined command to the registry and refresh lookups."""
|
|
COMMAND_REGISTRY.append(cmd)
|
|
rebuild_lookups()
|
|
|
|
|
|
def rebuild_lookups() -> None:
|
|
"""Rebuild all derived lookup dicts from the current COMMAND_REGISTRY.
|
|
|
|
Called after plugin commands are registered so they appear in help,
|
|
autocomplete, gateway dispatch, Telegram menu, and Slack mapping.
|
|
"""
|
|
global GATEWAY_KNOWN_COMMANDS
|
|
|
|
_COMMAND_LOOKUP.clear()
|
|
_COMMAND_LOOKUP.update(_build_command_lookup())
|
|
|
|
COMMANDS.clear()
|
|
for cmd in COMMAND_REGISTRY:
|
|
if not cmd.gateway_only:
|
|
COMMANDS[f"/{cmd.name}"] = _build_description(cmd)
|
|
for alias in cmd.aliases:
|
|
COMMANDS[f"/{alias}"] = f"{cmd.description} (alias for /{cmd.name})"
|
|
|
|
COMMANDS_BY_CATEGORY.clear()
|
|
for cmd in COMMAND_REGISTRY:
|
|
if not cmd.gateway_only:
|
|
cat = COMMANDS_BY_CATEGORY.setdefault(cmd.category, {})
|
|
cat[f"/{cmd.name}"] = COMMANDS[f"/{cmd.name}"]
|
|
for alias in cmd.aliases:
|
|
cat[f"/{alias}"] = COMMANDS[f"/{alias}"]
|
|
|
|
SUBCOMMANDS.clear()
|
|
for cmd in COMMAND_REGISTRY:
|
|
if cmd.subcommands:
|
|
SUBCOMMANDS[f"/{cmd.name}"] = list(cmd.subcommands)
|
|
for cmd in COMMAND_REGISTRY:
|
|
key = f"/{cmd.name}"
|
|
if key in SUBCOMMANDS or not cmd.args_hint:
|
|
continue
|
|
m = _PIPE_SUBS_RE.search(cmd.args_hint)
|
|
if m:
|
|
SUBCOMMANDS[key] = m.group(0).split("|")
|
|
|
|
GATEWAY_KNOWN_COMMANDS = frozenset(
|
|
name
|
|
for cmd in COMMAND_REGISTRY
|
|
if not cmd.cli_only or cmd.gateway_config_gate
|
|
for name in (cmd.name, *cmd.aliases)
|
|
)
|
|
|
|
|
|
def _build_description(cmd: CommandDef) -> str:
|
|
"""Build a CLI-facing description string including usage hint."""
|
|
if cmd.args_hint:
|
|
return f"{cmd.description} (usage: /{cmd.name} {cmd.args_hint})"
|
|
return cmd.description
|
|
|
|
|
|
# Backwards-compatible flat dict: "/command" -> description
|
|
COMMANDS: dict[str, str] = {}
|
|
for _cmd in COMMAND_REGISTRY:
|
|
if not _cmd.gateway_only:
|
|
COMMANDS[f"/{_cmd.name}"] = _build_description(_cmd)
|
|
for _alias in _cmd.aliases:
|
|
COMMANDS[f"/{_alias}"] = f"{_cmd.description} (alias for /{_cmd.name})"
|
|
|
|
# Backwards-compatible categorized dict
|
|
COMMANDS_BY_CATEGORY: dict[str, dict[str, str]] = {}
|
|
for _cmd in COMMAND_REGISTRY:
|
|
if not _cmd.gateway_only:
|
|
_cat = COMMANDS_BY_CATEGORY.setdefault(_cmd.category, {})
|
|
_cat[f"/{_cmd.name}"] = COMMANDS[f"/{_cmd.name}"]
|
|
for _alias in _cmd.aliases:
|
|
_cat[f"/{_alias}"] = COMMANDS[f"/{_alias}"]
|
|
|
|
|
|
# Subcommands lookup: "/cmd" -> ["sub1", "sub2", ...]
|
|
SUBCOMMANDS: dict[str, list[str]] = {}
|
|
for _cmd in COMMAND_REGISTRY:
|
|
if _cmd.subcommands:
|
|
SUBCOMMANDS[f"/{_cmd.name}"] = list(_cmd.subcommands)
|
|
|
|
# Also extract subcommands hinted in args_hint via pipe-separated patterns
|
|
# e.g. args_hint="[on|off|tts|status]" for commands that don't have explicit subcommands.
|
|
# NOTE: If a command already has explicit subcommands, this fallback is skipped.
|
|
# Use the `subcommands` field on CommandDef for intentional tab-completable args.
|
|
_PIPE_SUBS_RE = re.compile(r"[a-z]+(?:\|[a-z]+)+")
|
|
for _cmd in COMMAND_REGISTRY:
|
|
key = f"/{_cmd.name}"
|
|
if key in SUBCOMMANDS or not _cmd.args_hint:
|
|
continue
|
|
m = _PIPE_SUBS_RE.search(_cmd.args_hint)
|
|
if m:
|
|
SUBCOMMANDS[key] = m.group(0).split("|")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gateway helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Set of all command names + aliases recognized by the gateway.
|
|
# Includes config-gated commands so the gateway can dispatch them
|
|
# (the handler checks the config gate at runtime).
|
|
GATEWAY_KNOWN_COMMANDS: frozenset[str] = frozenset(
|
|
name
|
|
for cmd in COMMAND_REGISTRY
|
|
if not cmd.cli_only or cmd.gateway_config_gate
|
|
for name in (cmd.name, *cmd.aliases)
|
|
)
|
|
|
|
|
|
def _resolve_config_gates() -> set[str]:
|
|
"""Return canonical names of commands whose ``gateway_config_gate`` is truthy.
|
|
|
|
Reads ``config.yaml`` and walks the dot-separated key path for each
|
|
config-gated command. Returns an empty set on any error so callers
|
|
degrade gracefully.
|
|
"""
|
|
gated = [c for c in COMMAND_REGISTRY if c.gateway_config_gate]
|
|
if not gated:
|
|
return set()
|
|
try:
|
|
import yaml
|
|
config_path = os.path.join(
|
|
os.getenv("HERMES_HOME", os.path.expanduser("~/.hermes")),
|
|
"config.yaml",
|
|
)
|
|
if os.path.exists(config_path):
|
|
with open(config_path, encoding="utf-8") as f:
|
|
cfg = yaml.safe_load(f) or {}
|
|
else:
|
|
cfg = {}
|
|
except Exception:
|
|
return set()
|
|
result: set[str] = set()
|
|
for cmd in gated:
|
|
val: Any = cfg
|
|
for key in cmd.gateway_config_gate.split("."):
|
|
if isinstance(val, dict):
|
|
val = val.get(key)
|
|
else:
|
|
val = None
|
|
break
|
|
if val:
|
|
result.add(cmd.name)
|
|
return result
|
|
|
|
|
|
def _is_gateway_available(cmd: CommandDef, config_overrides: set[str] | None = None) -> bool:
|
|
"""Check if *cmd* should appear in gateway surfaces (help, menus, mappings).
|
|
|
|
Unconditionally available when ``cli_only`` is False. When ``cli_only``
|
|
is True but ``gateway_config_gate`` is set, the command is available only
|
|
when the config value is truthy. Pass *config_overrides* (from
|
|
``_resolve_config_gates()``) to avoid re-reading config for every command.
|
|
"""
|
|
if not cmd.cli_only:
|
|
return True
|
|
if cmd.gateway_config_gate:
|
|
overrides = config_overrides if config_overrides is not None else _resolve_config_gates()
|
|
return cmd.name in overrides
|
|
return False
|
|
|
|
|
|
def gateway_help_lines() -> list[str]:
|
|
"""Generate gateway help text lines from the registry."""
|
|
overrides = _resolve_config_gates()
|
|
lines: list[str] = []
|
|
for cmd in COMMAND_REGISTRY:
|
|
if not _is_gateway_available(cmd, overrides):
|
|
continue
|
|
args = f" {cmd.args_hint}" if cmd.args_hint else ""
|
|
alias_parts: list[str] = []
|
|
for a in cmd.aliases:
|
|
# Skip internal aliases like reload_mcp (underscore variant)
|
|
if a.replace("-", "_") == cmd.name.replace("-", "_") and a != cmd.name:
|
|
continue
|
|
alias_parts.append(f"`/{a}`")
|
|
alias_note = f" (alias: {', '.join(alias_parts)})" if alias_parts else ""
|
|
lines.append(f"`/{cmd.name}{args}` -- {cmd.description}{alias_note}")
|
|
return lines
|
|
|
|
|
|
def telegram_bot_commands() -> list[tuple[str, str]]:
|
|
"""Return (command_name, description) pairs for Telegram setMyCommands.
|
|
|
|
Telegram command names cannot contain hyphens, so they are replaced with
|
|
underscores. Aliases are skipped -- Telegram shows one menu entry per
|
|
canonical command.
|
|
"""
|
|
overrides = _resolve_config_gates()
|
|
result: list[tuple[str, str]] = []
|
|
for cmd in COMMAND_REGISTRY:
|
|
if not _is_gateway_available(cmd, overrides):
|
|
continue
|
|
tg_name = cmd.name.replace("-", "_")
|
|
result.append((tg_name, cmd.description))
|
|
return result
|
|
|
|
|
|
_TG_NAME_LIMIT = 32
|
|
|
|
|
|
def _clamp_telegram_names(
|
|
entries: list[tuple[str, str]],
|
|
reserved: set[str],
|
|
) -> list[tuple[str, str]]:
|
|
"""Enforce Telegram's 32-char command name limit with collision avoidance.
|
|
|
|
Names exceeding 32 chars are truncated. If truncation creates a duplicate
|
|
(against *reserved* names or earlier entries in the same batch), the name is
|
|
shortened to 31 chars and a digit ``0``-``9`` is appended to differentiate.
|
|
If all 10 digit slots are taken the entry is silently dropped.
|
|
"""
|
|
used: set[str] = set(reserved)
|
|
result: list[tuple[str, str]] = []
|
|
for name, desc in entries:
|
|
if len(name) > _TG_NAME_LIMIT:
|
|
candidate = name[:_TG_NAME_LIMIT]
|
|
if candidate in used:
|
|
prefix = name[:_TG_NAME_LIMIT - 1]
|
|
for digit in range(10):
|
|
candidate = f"{prefix}{digit}"
|
|
if candidate not in used:
|
|
break
|
|
else:
|
|
# All 10 digit slots exhausted — skip entry
|
|
continue
|
|
name = candidate
|
|
if name in used:
|
|
continue
|
|
used.add(name)
|
|
result.append((name, desc))
|
|
return result
|
|
|
|
|
|
def telegram_menu_commands(max_commands: int = 100) -> tuple[list[tuple[str, str]], int]:
|
|
"""Return Telegram menu commands capped to the Bot API limit.
|
|
|
|
Priority order (higher priority = never bumped by overflow):
|
|
1. Core CommandDef commands (always included)
|
|
2. Plugin slash commands (take precedence over skills)
|
|
3. Built-in skill commands (fill remaining slots, alphabetical)
|
|
|
|
Skills are the only tier that gets trimmed when the cap is hit.
|
|
User-installed hub skills are excluded — accessible via /skills.
|
|
|
|
Returns:
|
|
(menu_commands, hidden_count) where hidden_count is the number of
|
|
skill commands omitted due to the cap.
|
|
"""
|
|
core_commands = list(telegram_bot_commands())
|
|
# Reserve core names so plugin/skill truncation can't collide with them
|
|
reserved_names = {n for n, _ in core_commands}
|
|
all_commands = list(core_commands)
|
|
|
|
# Plugin slash commands get priority over skills
|
|
plugin_entries: list[tuple[str, str]] = []
|
|
try:
|
|
from hermes_cli.plugins import get_plugin_manager
|
|
pm = get_plugin_manager()
|
|
plugin_cmds = getattr(pm, "_plugin_commands", {})
|
|
for cmd_name in sorted(plugin_cmds):
|
|
tg_name = cmd_name.replace("-", "_")
|
|
desc = "Plugin command"
|
|
if len(desc) > 40:
|
|
desc = desc[:37] + "..."
|
|
plugin_entries.append((tg_name, desc))
|
|
except Exception:
|
|
pass
|
|
|
|
# Clamp plugin names to 32 chars with collision avoidance
|
|
plugin_entries = _clamp_telegram_names(plugin_entries, reserved_names)
|
|
reserved_names.update(n for n, _ in plugin_entries)
|
|
all_commands.extend(plugin_entries)
|
|
|
|
# Remaining slots go to built-in skill commands (not hub-installed).
|
|
skill_entries: list[tuple[str, str]] = []
|
|
try:
|
|
from agent.skill_commands import get_skill_commands
|
|
from tools.skills_tool import SKILLS_DIR
|
|
_skills_dir = str(SKILLS_DIR.resolve())
|
|
_hub_dir = str((SKILLS_DIR / ".hub").resolve())
|
|
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.startswith(_skills_dir):
|
|
continue
|
|
if skill_path.startswith(_hub_dir):
|
|
continue
|
|
name = cmd_key.lstrip("/").replace("-", "_")
|
|
desc = info.get("description", "")
|
|
# Keep descriptions short — setMyCommands has an undocumented
|
|
# total payload limit. 40 chars fits 100 commands safely.
|
|
if len(desc) > 40:
|
|
desc = desc[:37] + "..."
|
|
skill_entries.append((name, desc))
|
|
except Exception:
|
|
pass
|
|
|
|
# Clamp skill names to 32 chars with collision avoidance
|
|
skill_entries = _clamp_telegram_names(skill_entries, reserved_names)
|
|
|
|
# Skills fill remaining slots — they're the only tier that gets trimmed
|
|
remaining_slots = max(0, max_commands - len(all_commands))
|
|
hidden_count = max(0, len(skill_entries) - remaining_slots)
|
|
all_commands.extend(skill_entries[:remaining_slots])
|
|
return all_commands[:max_commands], hidden_count
|
|
|
|
|
|
def slack_subcommand_map() -> dict[str, str]:
|
|
"""Return subcommand -> /command mapping for Slack /hermes handler.
|
|
|
|
Maps both canonical names and aliases so /hermes bg do stuff works
|
|
the same as /hermes background do stuff.
|
|
"""
|
|
overrides = _resolve_config_gates()
|
|
mapping: dict[str, str] = {}
|
|
for cmd in COMMAND_REGISTRY:
|
|
if not _is_gateway_available(cmd, overrides):
|
|
continue
|
|
mapping[cmd.name] = f"/{cmd.name}"
|
|
for alias in cmd.aliases:
|
|
mapping[alias] = f"/{alias}"
|
|
return mapping
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Autocomplete
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class SlashCommandCompleter(Completer):
|
|
"""Autocomplete for built-in slash commands, subcommands, and skill commands."""
|
|
|
|
def __init__(
|
|
self,
|
|
skill_commands_provider: Callable[[], Mapping[str, dict[str, Any]]] | None = None,
|
|
) -> None:
|
|
self._skill_commands_provider = skill_commands_provider
|
|
|
|
def _iter_skill_commands(self) -> Mapping[str, dict[str, Any]]:
|
|
if self._skill_commands_provider is None:
|
|
return {}
|
|
try:
|
|
return self._skill_commands_provider() or {}
|
|
except Exception:
|
|
return {}
|
|
|
|
@staticmethod
|
|
def _completion_text(cmd_name: str, word: str) -> str:
|
|
"""Return replacement text for a completion.
|
|
|
|
When the user has already typed the full command exactly (``/help``),
|
|
returning ``help`` would be a no-op and prompt_toolkit suppresses the
|
|
menu. Appending a trailing space keeps the dropdown visible and makes
|
|
backspacing retrigger it naturally.
|
|
"""
|
|
return f"{cmd_name} " if cmd_name == word else cmd_name
|
|
|
|
@staticmethod
|
|
def _extract_path_word(text: str) -> str | None:
|
|
"""Extract the current word if it looks like a file path.
|
|
|
|
Returns the path-like token under the cursor, or None if the
|
|
current word doesn't look like a path. A word is path-like when
|
|
it starts with ``./``, ``../``, ``~/``, ``/``, or contains a
|
|
``/`` separator (e.g. ``src/main.py``).
|
|
"""
|
|
if not text:
|
|
return None
|
|
# Walk backwards to find the start of the current "word".
|
|
# Words are delimited by spaces, but paths can contain almost anything.
|
|
i = len(text) - 1
|
|
while i >= 0 and text[i] != " ":
|
|
i -= 1
|
|
word = text[i + 1:]
|
|
if not word:
|
|
return None
|
|
# Only trigger path completion for path-like tokens
|
|
if word.startswith(("./", "../", "~/", "/")) or "/" in word:
|
|
return word
|
|
return None
|
|
|
|
@staticmethod
|
|
def _path_completions(word: str, limit: int = 30):
|
|
"""Yield Completion objects for file paths matching *word*."""
|
|
expanded = os.path.expanduser(word)
|
|
# Split into directory part and prefix to match inside it
|
|
if expanded.endswith("/"):
|
|
search_dir = expanded
|
|
prefix = ""
|
|
else:
|
|
search_dir = os.path.dirname(expanded) or "."
|
|
prefix = os.path.basename(expanded)
|
|
|
|
try:
|
|
entries = os.listdir(search_dir)
|
|
except OSError:
|
|
return
|
|
|
|
count = 0
|
|
prefix_lower = prefix.lower()
|
|
for entry in sorted(entries):
|
|
if prefix and not entry.lower().startswith(prefix_lower):
|
|
continue
|
|
if count >= limit:
|
|
break
|
|
|
|
full_path = os.path.join(search_dir, entry)
|
|
is_dir = os.path.isdir(full_path)
|
|
|
|
# Build the completion text (what replaces the typed word)
|
|
if word.startswith("~"):
|
|
display_path = "~/" + os.path.relpath(full_path, os.path.expanduser("~"))
|
|
elif os.path.isabs(word):
|
|
display_path = full_path
|
|
else:
|
|
# Keep relative
|
|
display_path = os.path.relpath(full_path)
|
|
|
|
if is_dir:
|
|
display_path += "/"
|
|
|
|
suffix = "/" if is_dir else ""
|
|
meta = "dir" if is_dir else _file_size_label(full_path)
|
|
|
|
yield Completion(
|
|
display_path,
|
|
start_position=-len(word),
|
|
display=entry + suffix,
|
|
display_meta=meta,
|
|
)
|
|
count += 1
|
|
|
|
@staticmethod
|
|
def _extract_context_word(text: str) -> str | None:
|
|
"""Extract a bare ``@`` token for context reference completions."""
|
|
if not text:
|
|
return None
|
|
# Walk backwards to find the start of the current word
|
|
i = len(text) - 1
|
|
while i >= 0 and text[i] != " ":
|
|
i -= 1
|
|
word = text[i + 1:]
|
|
if not word.startswith("@"):
|
|
return None
|
|
return word
|
|
|
|
@staticmethod
|
|
def _context_completions(word: str, limit: int = 30):
|
|
"""Yield Claude Code-style @ context completions.
|
|
|
|
Bare ``@`` or ``@partial`` shows static references and matching
|
|
files/folders. ``@file:path`` and ``@folder:path`` are handled
|
|
by the existing path completion path.
|
|
"""
|
|
lowered = word.lower()
|
|
|
|
# Static context references
|
|
_STATIC_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"),
|
|
)
|
|
for candidate, meta in _STATIC_REFS:
|
|
if candidate.lower().startswith(lowered) and candidate.lower() != lowered:
|
|
yield Completion(
|
|
candidate,
|
|
start_position=-len(word),
|
|
display=candidate,
|
|
display_meta=meta,
|
|
)
|
|
|
|
# If the user typed @file: or @folder:, delegate to path completions
|
|
for prefix in ("@file:", "@folder:"):
|
|
if word.startswith(prefix):
|
|
path_part = word[len(prefix):] or "."
|
|
expanded = os.path.expanduser(path_part)
|
|
if expanded.endswith("/"):
|
|
search_dir, match_prefix = expanded, ""
|
|
else:
|
|
search_dir = os.path.dirname(expanded) or "."
|
|
match_prefix = os.path.basename(expanded)
|
|
|
|
try:
|
|
entries = os.listdir(search_dir)
|
|
except OSError:
|
|
return
|
|
|
|
count = 0
|
|
prefix_lower = match_prefix.lower()
|
|
for entry in sorted(entries):
|
|
if match_prefix and not entry.lower().startswith(prefix_lower):
|
|
continue
|
|
if count >= limit:
|
|
break
|
|
full_path = os.path.join(search_dir, entry)
|
|
is_dir = os.path.isdir(full_path)
|
|
display_path = os.path.relpath(full_path)
|
|
suffix = "/" if is_dir else ""
|
|
kind = "folder" if is_dir else "file"
|
|
meta = "dir" if is_dir else _file_size_label(full_path)
|
|
completion = f"@{kind}:{display_path}{suffix}"
|
|
yield Completion(
|
|
completion,
|
|
start_position=-len(word),
|
|
display=entry + suffix,
|
|
display_meta=meta,
|
|
)
|
|
count += 1
|
|
return
|
|
|
|
# Bare @ or @partial — show matching files/folders from cwd
|
|
query = word[1:] # strip the @
|
|
if not query:
|
|
search_dir, match_prefix = ".", ""
|
|
else:
|
|
expanded = os.path.expanduser(query)
|
|
if expanded.endswith("/"):
|
|
search_dir, match_prefix = expanded, ""
|
|
else:
|
|
search_dir = os.path.dirname(expanded) or "."
|
|
match_prefix = os.path.basename(expanded)
|
|
|
|
try:
|
|
entries = os.listdir(search_dir)
|
|
except OSError:
|
|
return
|
|
|
|
count = 0
|
|
prefix_lower = match_prefix.lower()
|
|
for entry in sorted(entries):
|
|
if match_prefix and not entry.lower().startswith(prefix_lower):
|
|
continue
|
|
if entry.startswith("."):
|
|
continue # skip hidden files in bare @ mode
|
|
if count >= limit:
|
|
break
|
|
full_path = os.path.join(search_dir, entry)
|
|
is_dir = os.path.isdir(full_path)
|
|
display_path = os.path.relpath(full_path)
|
|
suffix = "/" if is_dir else ""
|
|
kind = "folder" if is_dir else "file"
|
|
meta = "dir" if is_dir else _file_size_label(full_path)
|
|
completion = f"@{kind}:{display_path}{suffix}"
|
|
yield Completion(
|
|
completion,
|
|
start_position=-len(word),
|
|
display=entry + suffix,
|
|
display_meta=meta,
|
|
)
|
|
count += 1
|
|
|
|
def get_completions(self, document, complete_event):
|
|
text = document.text_before_cursor
|
|
if not text.startswith("/"):
|
|
# Try @ context completion (Claude Code-style)
|
|
ctx_word = self._extract_context_word(text)
|
|
if ctx_word is not None:
|
|
yield from self._context_completions(ctx_word)
|
|
return
|
|
# Try file path completion for non-slash input
|
|
path_word = self._extract_path_word(text)
|
|
if path_word is not None:
|
|
yield from self._path_completions(path_word)
|
|
return
|
|
|
|
# Check if we're completing a subcommand (base command already typed)
|
|
parts = text.split(maxsplit=1)
|
|
base_cmd = parts[0].lower()
|
|
if len(parts) > 1 or (len(parts) == 1 and text.endswith(" ")):
|
|
sub_text = parts[1] if len(parts) > 1 else ""
|
|
sub_lower = sub_text.lower()
|
|
|
|
# Static subcommand completions
|
|
if " " not in sub_text and base_cmd in SUBCOMMANDS:
|
|
for sub in SUBCOMMANDS[base_cmd]:
|
|
if sub.startswith(sub_lower) and sub != sub_lower:
|
|
yield Completion(
|
|
sub,
|
|
start_position=-len(sub_text),
|
|
display=sub,
|
|
)
|
|
return
|
|
|
|
word = text[1:]
|
|
|
|
for cmd, desc in COMMANDS.items():
|
|
cmd_name = cmd[1:]
|
|
if cmd_name.startswith(word):
|
|
yield Completion(
|
|
self._completion_text(cmd_name, word),
|
|
start_position=-len(word),
|
|
display=cmd,
|
|
display_meta=desc,
|
|
)
|
|
|
|
for cmd, info in self._iter_skill_commands().items():
|
|
cmd_name = cmd[1:]
|
|
if cmd_name.startswith(word):
|
|
description = str(info.get("description", "Skill command"))
|
|
short_desc = description[:50] + ("..." if len(description) > 50 else "")
|
|
yield Completion(
|
|
self._completion_text(cmd_name, word),
|
|
start_position=-len(word),
|
|
display=cmd,
|
|
display_meta=f"⚡ {short_desc}",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Inline auto-suggest (ghost text) for slash commands
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class SlashCommandAutoSuggest(AutoSuggest):
|
|
"""Inline ghost-text suggestions for slash commands and their subcommands.
|
|
|
|
Shows the rest of a command or subcommand in dim text as you type.
|
|
Falls back to history-based suggestions for non-slash 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 get_suggestion(self, buffer, document):
|
|
text = document.text_before_cursor
|
|
|
|
# Only suggest for slash commands
|
|
if not text.startswith("/"):
|
|
# Fall back to history for regular text
|
|
if self._history:
|
|
return self._history.get_suggestion(buffer, document)
|
|
return None
|
|
|
|
parts = text.split(maxsplit=1)
|
|
base_cmd = parts[0].lower()
|
|
|
|
if len(parts) == 1 and not text.endswith(" "):
|
|
# Still typing the command name: /upd → suggest "ate"
|
|
word = text[1:].lower()
|
|
for cmd in COMMANDS:
|
|
cmd_name = cmd[1:] # strip leading /
|
|
if cmd_name.startswith(word) and cmd_name != word:
|
|
return Suggestion(cmd_name[len(word):])
|
|
return None
|
|
|
|
# Command is complete — suggest subcommands or model names
|
|
sub_text = parts[1] if len(parts) > 1 else ""
|
|
sub_lower = sub_text.lower()
|
|
|
|
# Static subcommands
|
|
if base_cmd in SUBCOMMANDS and SUBCOMMANDS[base_cmd]:
|
|
if " " not in sub_text:
|
|
for sub in SUBCOMMANDS[base_cmd]:
|
|
if sub.startswith(sub_lower) and sub != sub_lower:
|
|
return Suggestion(sub[len(sub_text):])
|
|
|
|
# Fall back to history
|
|
if self._history:
|
|
return self._history.get_suggestion(buffer, document)
|
|
return None
|
|
|
|
|
|
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"
|