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

Mirrors the Slack precedent (register_slack_action_handler): plugins queue
a factory at register() time; the Telegram adapter invokes each factory
with (application, adapter) at connect() time, before the core handlers
register, so pattern-scoped plugin handlers take precedence for their own
updates while everything else falls through unchanged. Factories are
isolated — a raising plugin cannot prevent Telegram from connecting.

Unblocks standalone plugins that need PTB update types the core adapter
doesn't route (Telegram Business API secretary bots, custom callback
prefixes, chat-member events) without touching core files.

hermes_cli/plugins.py

@@ -991,6 +991,63 @@ class PluginContext:
             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 @@ class PluginManager:
         # ``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 @@ class PluginManager:
             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 @@ class PluginManager:
         """
         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 @@ class TelegramAdapter(BasePlatformAdapter):
             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 @@ class TelegramAdapter(BasePlatformAdapter):
             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.
 :::