feat(plugins): let plugins register Telegram PTB handlers (ctx.register_telegram_handler)

hermes_cli/plugins.py

@@ -991,6 +991,63 @@ async def _on_approve(ack, body, action):
             action_id,
         )
 
+    # -- telegram handler registration ---------------------------------------
+
+    def register_telegram_handler(self, factory: Callable) -> None:
+        """Register a python-telegram-bot handler factory from a plugin.
+
+        Hermes' Telegram adapter invokes registered factories at ``connect()``
+        time, right after the PTB ``Application`` is built and **before** the
+        core handlers are added. The factory receives
+        ``(application, adapter)`` and wires its own handlers::
+
+            def _wire(application, adapter):
+                from telegram.ext import CallbackQueryHandler
+
+                application.add_handler(
+                    CallbackQueryHandler(_on_button, pattern=r"^myplugin:")
+                )
+
+            ctx.register_telegram_handler(_wire)
+
+        Notes:
+
+        * The factory is called lazily at connect time, so plugins may import
+          ``telegram`` / ``telegram.ext`` inside the factory body — the
+          plugin's ``register()`` still works when PTB is not installed.
+        * PTB dispatches only the *first* matching handler within a group,
+          and the core adapter registers a catch-all ``CallbackQueryHandler``
+          in the default group. Because plugin factories run first,
+          a pattern-scoped ``CallbackQueryHandler`` (e.g. ``pattern=r"^bd:"``)
+          takes precedence for its own callbacks while every other update
+          falls through to the core handlers unchanged. Always scope
+          callback handlers with ``pattern=`` — an unscoped handler would
+          swallow the core button flows (approvals, model picker, clarify).
+        * ``adapter`` is the ``TelegramAdapter`` instance (``adapter.bot``,
+          ``adapter.config`` etc.); treat it as read-only.
+        * Exceptions raised by the factory are caught and logged by the
+          adapter — a broken plugin cannot prevent Telegram from connecting.
+
+        Args:
+            factory: Callable receiving ``(application, adapter)``.
+
+        Raises:
+            ValueError: if ``factory`` is not callable.
+        """
+        if not callable(factory):
+            raise ValueError(
+                f"Plugin '{self.manifest.name}' tried to register a Telegram "
+                f"handler factory with a non-callable factory."
+            )
+        self._manager._telegram_handler_factories.append(
+            (factory, self.manifest.name)
+        )
+        logger.debug(
+            "Plugin %s registered Telegram handler factory: %s",
+            self.manifest.name,
+            getattr(factory, "__name__", repr(factory)),
+        )
+
     # -- hook registration --------------------------------------------------
 
     # -- auxiliary task registration ---------------------------------------
@@ -1222,6 +1279,12 @@ def __init__(self) -> None:
         # ``re.Pattern``, or a constraint dict); ``callback`` is an async
         # function with the slack_bolt signature ``(ack, body, action)``.
         self._slack_action_handlers: List[tuple] = []
+        # Telegram handler factories registered by plugins. Each entry is
+        # (factory, plugin_name); the Telegram adapter invokes factories at
+        # connect() time with (application, adapter) so plugins can wire
+        # their own PTB handlers (pattern-scoped CallbackQueryHandler,
+        # BusinessMessageHandler, etc.) without touching core files.
+        self._telegram_handler_factories: List[tuple] = []
 
     # -----------------------------------------------------------------------
     # Public
@@ -1255,6 +1318,7 @@ def discover_and_load(self, force: bool = False) -> None:
             self._plugin_skills.clear()
             self._aux_tasks.clear()
             self._slack_action_handlers.clear()
+            self._telegram_handler_factories.clear()
             self._context_engine = None
         # Set the flag up front as a re-entrancy guard (a plugin's register()
         # can transitively trigger discovery again), but reset it if the sweep
@@ -1928,6 +1992,23 @@ def get_slack_action_handlers(self) -> List[tuple]:
         """
         return list(self._slack_action_handlers)
 
+    # -----------------------------------------------------------------------
+    # Telegram handler factory accessor
+    # -----------------------------------------------------------------------
+
+    def get_telegram_handler_factories(self) -> List[tuple]:
+        """Return the list of plugin-registered Telegram handler factories.
+
+        Each entry is a ``(factory, plugin_name)`` tuple. Consumed by the
+        Telegram adapter at connect time; each factory is invoked with
+        ``(application, adapter)`` so plugins can wire their own PTB
+        handlers before the core handlers are added.
+
+        Plugins register factories via
+        :meth:`PluginContext.register_telegram_handler`.
+        """
+        return list(self._telegram_handler_factories)
+
     # -----------------------------------------------------------------------
     # Introspection
     # -----------------------------------------------------------------------

plugins/platforms/telegram/adapter.py

@@ -2915,6 +2915,44 @@ async def _run_post_connect_housekeeping(self) -> None:
             if self._post_connect_task is asyncio.current_task():
                 self._post_connect_task = None
 
+    def _wire_plugin_handlers(self) -> None:
+        """Invoke plugin-registered Telegram handler factories.
+
+        Plugins call ``ctx.register_telegram_handler(factory)`` at
+        register() time; the manager queues the factories and this method
+        invokes them with ``(application, adapter)`` right after the PTB
+        Application is built and BEFORE the core handlers are added. PTB
+        dispatches only the first matching handler per group, so plugin
+        handlers registered first take precedence for the updates they
+        scope to (e.g. a ``CallbackQueryHandler`` with a ``pattern=``
+        prefix, business_message updates) while everything else falls
+        through to the core handlers.
+
+        Each factory is isolated so a misbehaving plugin can't prevent
+        Telegram from connecting.
+        """
+        try:
+            from hermes_cli.plugins import get_plugin_manager
+            factories = get_plugin_manager().get_telegram_handler_factories()
+        except Exception as e:  # pragma: no cover - defensive
+            logger.warning(
+                "[%s] Could not load plugin Telegram handler factories: %s",
+                self.name, e,
+            )
+            return
+        for factory, plugin_name in factories:
+            try:
+                factory(self._app, self)
+                logger.info(
+                    "[%s] Wired Telegram handlers from plugin '%s'",
+                    self.name, plugin_name,
+                )
+            except Exception as exc:
+                logger.error(
+                    "[%s] Plugin '%s' Telegram handler factory raised: %s",
+                    self.name, plugin_name, exc, exc_info=True,
+                )
+
     async def connect(self, *, is_reconnect: bool = False) -> bool:
         """Connect to Telegram via polling or webhook.
 
@@ -3102,7 +3140,11 @@ def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict:
             builder = builder.request(request).get_updates_request(get_updates_request)
             self._app = builder.build()
             self._bot = self._app.bot
-
+
+            # Wire plugin-provided PTB handlers BEFORE the core handlers.
+            # See _wire_plugin_handlers for precedence + isolation notes.
+            self._wire_plugin_handlers()
+
             # Register handlers
             self._app.add_handler(TelegramMessageHandler(
                 filters.TEXT & ~filters.COMMAND,

tests/gateway/test_telegram_plugin_handlers.py

@@ -0,0 +1,165 @@
+"""Tests for plugin-registered Telegram PTB handler factories.
+
+Covers:
+* ``PluginContext.register_telegram_handler`` validation + queuing
+* ``PluginManager.get_telegram_handler_factories`` accessor
+* ``TelegramAdapter._wire_plugin_handlers`` invoking factories with
+  ``(application, adapter)``
+* Defensive isolation: a factory that raises does NOT prevent the
+  adapter from wiring other factories or continuing to connect.
+* ``discover_and_load(force=True)`` clears queued factories.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# Ensure the repo root is importable when this test runs directly
+# ---------------------------------------------------------------------------
+_repo = str(Path(__file__).resolve().parents[2])
+if _repo not in sys.path:
+    sys.path.insert(0, _repo)
+
+from plugins.platforms.telegram.adapter import TelegramAdapter  # noqa: E402
+from gateway.config import PlatformConfig  # noqa: E402
+
+from hermes_cli.plugins import (  # noqa: E402
+    PluginContext,
+    PluginManager,
+    PluginManifest,
+)
+
+
+def _make_ctx(name: str = "test_plugin") -> tuple[PluginManager, PluginContext]:
+    mgr = PluginManager()
+    manifest = PluginManifest(name=name, version="0.1.0", description="test")
+    ctx = PluginContext(manifest=manifest, manager=mgr)
+    return mgr, ctx
+
+
+def _make_adapter() -> TelegramAdapter:
+    config = PlatformConfig(enabled=True, token="test-token", extra={})
+    adapter = TelegramAdapter(config)
+    adapter._app = MagicMock()
+    adapter._bot = MagicMock()
+    return adapter
+
+
+# ===========================================================================
+# PluginContext.register_telegram_handler — validation + queuing
+# ===========================================================================
+
+class TestRegisterTelegramHandlerAPI:
+    def test_factory_is_queued_with_plugin_name(self):
+        mgr, ctx = _make_ctx()
+
+        def factory(application, adapter):  # pragma: no cover - never called
+            pass
+
+        ctx.register_telegram_handler(factory)
+
+        factories = mgr.get_telegram_handler_factories()
+        assert len(factories) == 1
+        fn, plugin_name = factories[0]
+        assert fn is factory
+        assert plugin_name == "test_plugin"
+
+    def test_non_callable_factory_raises(self):
+        _, ctx = _make_ctx()
+        with pytest.raises(ValueError, match="non-callable"):
+            ctx.register_telegram_handler("not-a-callable")  # type: ignore[arg-type]
+
+    def test_accessor_returns_copy(self):
+        mgr, ctx = _make_ctx()
+        ctx.register_telegram_handler(lambda app, adapter: None)
+
+        got = mgr.get_telegram_handler_factories()
+        got.append(("junk", "junk"))
+        assert len(mgr.get_telegram_handler_factories()) == 1
+
+    def test_multiple_plugins_each_recorded(self):
+        mgr = PluginManager()
+        for name in ("plugin_a", "plugin_b"):
+            manifest = PluginManifest(name=name, version="0.1.0", description="t")
+            ctx = PluginContext(manifest=manifest, manager=mgr)
+            ctx.register_telegram_handler(lambda app, adapter: None)
+
+        names = [n for _, n in mgr.get_telegram_handler_factories()]
+        assert names == ["plugin_a", "plugin_b"]
+
+    def test_force_rediscovery_clears_factories(self):
+        mgr, ctx = _make_ctx()
+        ctx.register_telegram_handler(lambda app, adapter: None)
+        assert len(mgr.get_telegram_handler_factories()) == 1
+
+        # force=True clears queued registrations before the re-scan; the
+        # scan itself finds nothing in an isolated HERMES_HOME.
+        mgr.discover_and_load(force=True)
+        assert mgr.get_telegram_handler_factories() == []
+
+
+# ===========================================================================
+# TelegramAdapter._wire_plugin_handlers
+# ===========================================================================
+
+class TestTelegramAdapterPluginWiring:
+    def test_factory_invoked_with_application_and_adapter(self):
+        adapter = _make_adapter()
+        calls = []
+
+        def factory(application, adp):
+            calls.append((application, adp))
+            application.add_handler(MagicMock())
+
+        mgr = MagicMock()
+        mgr.get_telegram_handler_factories.return_value = [(factory, "biz_plugin")]
+
+        with patch("hermes_cli.plugins.get_plugin_manager", return_value=mgr):
+            adapter._wire_plugin_handlers()
+
+        assert calls == [(adapter._app, adapter)]
+        adapter._app.add_handler.assert_called_once()
+
+    def test_no_factories_is_a_noop(self):
+        adapter = _make_adapter()
+        mgr = MagicMock()
+        mgr.get_telegram_handler_factories.return_value = []
+
+        with patch("hermes_cli.plugins.get_plugin_manager", return_value=mgr):
+            adapter._wire_plugin_handlers()
+
+        adapter._app.add_handler.assert_not_called()
+
+    def test_raising_factory_does_not_block_others(self):
+        adapter = _make_adapter()
+        wired = []
+
+        def bad_factory(application, adp):
+            raise RuntimeError("boom")
+
+        def good_factory(application, adp):
+            wired.append("good")
+
+        mgr = MagicMock()
+        mgr.get_telegram_handler_factories.return_value = [
+            (bad_factory, "bad_plugin"),
+            (good_factory, "good_plugin"),
+        ]
+
+        with patch("hermes_cli.plugins.get_plugin_manager", return_value=mgr):
+            adapter._wire_plugin_handlers()  # must not raise
+
+        assert wired == ["good"]
+
+    def test_manager_load_failure_does_not_raise(self):
+        adapter = _make_adapter()
+        with patch(
+            "hermes_cli.plugins.get_plugin_manager",
+            side_effect=RuntimeError("plugin system down"),
+        ):
+            adapter._wire_plugin_handlers()  # must not raise

website/docs/guides/build-a-hermes-plugin.md

@@ -905,6 +905,43 @@ def register(ctx):
 
 This is the public way for plugins to participate in Slack interactivity. Older plugins may patch `SlackAdapter.connect`; prefer this API instead.
 
+### Register Telegram (PTB) handlers
+
+Plugins that need to receive Telegram updates the core adapter doesn't route — inline button callbacks with their own prefix, Business API updates, chat-member events, etc. — can register a handler factory that the Telegram adapter invokes at connect time.
+
+```python
+def register(ctx):
+    def _wire(application, adapter):
+        # Called with the PTB Application right after it is built,
+        # BEFORE the core handlers are added. Import telegram here so
+        # register() works even when PTB isn't installed.
+        from telegram.ext import CallbackQueryHandler
+
+        async def _on_button(update, context):
+            query = update.callback_query
+            await query.answer()
+            # ...handle "myplugin:*" callbacks
+
+        application.add_handler(
+            CallbackQueryHandler(_on_button, pattern=r"^myplugin:")
+        )
+
+    ctx.register_telegram_handler(_wire)
+```
+
+**Signature:** `ctx.register_telegram_handler(factory) -> None`
+
+| Parameter | Type | Description |
+|-----------|------|-------------|
+| `factory` | callable | Receives `(application, adapter)` — the PTB `Application` and the `TelegramAdapter` instance (`adapter.bot`, `adapter.config`; treat it as read-only) |
+
+**Runtime behavior:**
+
+- The factory is queued at plugin-load time and invoked when the Telegram platform connects, before the core handlers register. PTB dispatches only the first matching handler per group, so plugin handlers take precedence for the updates they scope to; everything else falls through to core.
+- **Always scope `CallbackQueryHandler` with a `pattern=` prefix** (e.g. `r"^myplugin:"`). An unscoped handler would swallow the core button flows (exec approvals, model picker, clarify prompts).
+- The factory is isolated: if it raises, the error is logged and Telegram still connects.
+- The adapter polls with `allowed_updates=Update.ALL_TYPES`, so non-message update types (e.g. `business_connection`, `chat_member`) already arrive without extra configuration.
+
 :::tip
 This guide covers **general plugins** (tools, hooks, slash commands, CLI commands). The sections below sketch the authoring pattern for each specialized plugin type; each links to its full guide for field reference and examples.
 :::