fix: publish plugin slash commands in Telegram menu

- discover plugin commands before building Telegram command menus
- make plugin command and context engine accessors lazy-load plugins
- add regression coverage for Telegram menu and plugin lookup paths
This commit is contained in:
Stephen Schoettler 2026-04-19 20:56:17 -07:00 committed by Teknium
parent 34ae13e6ed
commit a5e368ebfb
4 changed files with 106 additions and 7 deletions

View File

@ -497,9 +497,8 @@ def _collect_gateway_skill_entries(
# --- Tier 1: Plugin slash commands (never trimmed) ---------------------
plugin_pairs: list[tuple[str, str]] = []
try:
from hermes_cli.plugins import get_plugin_manager
pm = get_plugin_manager()
plugin_cmds = getattr(pm, "_plugin_commands", {})
from hermes_cli.plugins import get_plugin_commands
plugin_cmds = get_plugin_commands()
for cmd_name in sorted(plugin_cmds):
name = sanitize_name(cmd_name) if sanitize_name else cmd_name
if not name:

View File

@ -873,23 +873,31 @@ def get_pre_tool_call_block_message(
return None
def _ensure_plugins_discovered() -> PluginManager:
"""Return the global manager after running idempotent plugin discovery."""
manager = get_plugin_manager()
manager.discover_and_load()
return manager
def get_plugin_context_engine():
"""Return the plugin-registered context engine, or None."""
return get_plugin_manager()._context_engine
return _ensure_plugins_discovered()._context_engine
def get_plugin_command_handler(name: str) -> Optional[Callable]:
"""Return the handler for a plugin-registered slash command, or ``None``."""
entry = get_plugin_manager()._plugin_commands.get(name)
entry = _ensure_plugins_discovered()._plugin_commands.get(name)
return entry["handler"] if entry else None
def get_plugin_commands() -> Dict[str, dict]:
"""Return the full plugin commands dict (name → {handler, description, plugin}).
Safe to call before discovery returns an empty dict if no plugins loaded.
Triggers idempotent plugin discovery so callers can use plugin commands
before any explicit discover_plugins() call.
"""
return get_plugin_manager()._plugin_commands
return _ensure_plugins_discovered()._plugin_commands
def get_plugin_toolsets() -> List[tuple]:

View File

@ -688,6 +688,28 @@ class TestTelegramMenuCommands:
f"Command '{name}' is {len(name)} chars (limit {_TG_NAME_LIMIT})"
)
def test_includes_plugin_commands_via_lazy_discovery(self, tmp_path, monkeypatch):
"""Telegram menu generation should discover plugin slash commands on first access."""
from unittest.mock import patch
import hermes_cli.plugins as plugins_mod
plugin_dir = tmp_path / "plugins" / "cmd-plugin"
plugin_dir.mkdir(parents=True, exist_ok=True)
(plugin_dir / "plugin.yaml").write_text(
"name: cmd-plugin\nversion: 0.1.0\ndescription: Test plugin\n"
)
(plugin_dir / "__init__.py").write_text(
"def register(ctx):\n"
" ctx.register_command('lcm', lambda args: 'ok', description='LCM status and diagnostics')\n"
)
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
with patch.object(plugins_mod, "_plugin_manager", None):
menu, _ = telegram_menu_commands(max_commands=100)
menu_names = {name for name, _ in menu}
assert "lcm" in menu_names
def test_excludes_telegram_disabled_skills(self, tmp_path, monkeypatch):
"""Skills disabled for telegram should not appear in the menu."""
from unittest.mock import patch, MagicMock

View File

@ -795,6 +795,76 @@ class TestPluginCommands:
assert "cmd-b" in cmds
assert cmds["cmd-a"]["description"] == "A"
def test_get_plugin_command_handler_discovers_plugins_lazily(self, tmp_path, monkeypatch):
"""Handler lookup should work before any explicit discover_plugins() call."""
plugins_dir = tmp_path / "hermes_test" / "plugins"
_make_plugin_dir(
plugins_dir,
"cmd-plugin",
register_body='ctx.register_command("lazycmd", lambda a: f"ok:{a}", description="Lazy")',
)
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
import hermes_cli.plugins as plugins_mod
with patch.object(plugins_mod, "_plugin_manager", None):
handler = get_plugin_command_handler("lazycmd")
assert handler is not None
assert handler("x") == "ok:x"
def test_get_plugin_commands_discovers_plugins_lazily(self, tmp_path, monkeypatch):
"""Command listing should trigger plugin discovery on first access."""
plugins_dir = tmp_path / "hermes_test" / "plugins"
_make_plugin_dir(
plugins_dir,
"cmd-plugin",
register_body='ctx.register_command("lazycmd", lambda a: a, description="Lazy")',
)
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
import hermes_cli.plugins as plugins_mod
with patch.object(plugins_mod, "_plugin_manager", None):
cmds = get_plugin_commands()
assert "lazycmd" in cmds
assert cmds["lazycmd"]["description"] == "Lazy"
def test_get_plugin_context_engine_discovers_plugins_lazily(self, tmp_path, monkeypatch):
"""Context engine lookup should work before any explicit discover_plugins() call."""
plugins_dir = tmp_path / "hermes_test" / "plugins"
plugin_dir = plugins_dir / "engine-plugin"
plugin_dir.mkdir(parents=True, exist_ok=True)
(plugin_dir / "plugin.yaml").write_text(
yaml.dump({
"name": "engine-plugin",
"version": "0.1.0",
"description": "Test engine plugin",
})
)
(plugin_dir / "__init__.py").write_text(
"from agent.context_engine import ContextEngine\n\n"
"class StubEngine(ContextEngine):\n"
" @property\n"
" def name(self):\n"
" return 'stub-engine'\n\n"
" def update_from_response(self, usage):\n"
" return None\n\n"
" def should_compress(self, prompt_tokens):\n"
" return False\n\n"
" def compress(self, messages, current_tokens):\n"
" return messages\n\n"
"def register(ctx):\n"
" ctx.register_context_engine(StubEngine())\n"
)
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
import hermes_cli.plugins as plugins_mod
with patch.object(plugins_mod, "_plugin_manager", None):
engine = plugins_mod.get_plugin_context_engine()
assert engine is not None
assert engine.name == "stub-engine"
def test_commands_tracked_on_loaded_plugin(self, tmp_path, monkeypatch):
"""Commands registered during discover_and_load() are tracked on LoadedPlugin."""
plugins_dir = tmp_path / "hermes_test" / "plugins"