From 272f4e4abe618ec74c21cb9325686a9abb4c9aa4 Mon Sep 17 00:00:00 2001
From: teknium1 <127238744+teknium1@users.noreply.github.com>
Date: Sun, 5 Jul 2026 15:17:18 -0700
Subject: [PATCH] feat(plugins): generalize native platform handler
 registration to every gateway platform
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

ctx.register_platform_handler(platform, factory) — the generic surface for
plugins to wire native handlers into any platform adapter at connect()
time. Factories receive (native, adapter): the platform's client/app
object (PTB Application, discord.py Bot, slack_bolt AsyncApp, Teams App,
DingTalkStreamClient, aiohttp web.Application) or None for adapters with
no separate native object.

- BasePlatformAdapter._wire_plugin_handlers(native): shared, isolated
  invocation helper — a raising plugin cannot block a platform connect.
- All 27 connectable adapters call it: telegram/slack/teams/line/
  api_server/msgraph_webhook wire before their dispatch tables freeze;
  the rest hook at connect success.
- register_telegram_handler and get_telegram_handler_factories retained
  as thin back-compat aliases over the telegram bucket.
- Source-invariant test guarantees every adapter with connect() keeps
  calling the hook.
---
 gateway/platforms/api_server.py               |   4 +
 gateway/platforms/base.py                     |  44 +++
 gateway/platforms/bluebubbles.py              |   2 +
 gateway/platforms/msgraph_webhook.py          |   4 +
 gateway/platforms/signal.py                   |   2 +
 gateway/platforms/webhook.py                  |   2 +
 gateway/platforms/weixin.py                   |   2 +
 gateway/platforms/whatsapp_cloud.py           |   2 +
 gateway/platforms/yuanbao.py                  |   6 +-
 hermes_cli/plugins.py                         | 133 +++++++--
 plugins/platforms/dingtalk/adapter.py         |   2 +
 plugins/platforms/discord/adapter.py          |   2 +
 plugins/platforms/email/adapter.py            |   2 +
 plugins/platforms/feishu/adapter.py           |   2 +
 plugins/platforms/google_chat/adapter.py      |   2 +
 plugins/platforms/homeassistant/adapter.py    |   2 +
 plugins/platforms/irc/adapter.py              |   2 +
 plugins/platforms/line/adapter.py             |   4 +
 plugins/platforms/matrix/adapter.py           |   2 +
 plugins/platforms/mattermost/adapter.py       |   2 +
 plugins/platforms/ntfy/adapter.py             |   2 +
 plugins/platforms/photon/adapter.py           |   2 +
 plugins/platforms/raft/adapter.py             |   2 +
 plugins/platforms/simplex/adapter.py          |   2 +
 plugins/platforms/slack/adapter.py            |   8 +
 plugins/platforms/sms/adapter.py              |   2 +
 plugins/platforms/teams/adapter.py            |   5 +
 plugins/platforms/telegram/adapter.py         |  47 +---
 plugins/platforms/wecom/adapter.py            |   2 +
 plugins/platforms/whatsapp/adapter.py         |   4 +
 .../gateway/test_platform_plugin_handlers.py  | 255 ++++++++++++++++++
 .../gateway/test_telegram_plugin_handlers.py  | 165 ------------
 website/docs/developer-guide/plugins/index.md |  73 +++--
 33 files changed, 542 insertions(+), 250 deletions(-)
 create mode 100644 tests/gateway/test_platform_plugin_handlers.py
 delete mode 100644 tests/gateway/test_telegram_plugin_handlers.py

diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py
index 980659c934..c52b2efa88 100644
--- a/gateway/platforms/api_server.py
+++ b/gateway/platforms/api_server.py
@@ -8432,6 +8432,10 @@ class APIServerAdapter(BasePlatformAdapter):
                         self.name, self._host,
                     )
 
+            # Plugin-registered native handlers (aiohttp web.Application —
+            # router routes). Wired before AppRunner.setup() freezes the router.
+            self._wire_plugin_handlers(self._app)
+
             self._runner = web.AppRunner(self._app)
             await self._runner.setup()
             # Bind directly instead of probing 127.0.0.1 first — the old
diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py
index e990255794..94c4f8a3dc 100644
--- a/gateway/platforms/base.py
+++ b/gateway/platforms/base.py
@@ -3703,6 +3703,50 @@ class BasePlatformAdapter(ABC):
         release_scoped_lock(self._platform_lock_scope, identity)
         self._platform_lock_identity = None
 
+    def _wire_plugin_handlers(self, native: Any = None) -> None:
+        """Invoke plugin-registered native handler factories for this platform.
+
+        Plugins call ``ctx.register_platform_handler(<platform>, factory)``
+        at register() time; adapters call this from ``connect()`` once
+        their native client object exists (and, where dispatch order
+        matters, before their own handlers register). Each factory is
+        invoked with ``(native, adapter)``.
+
+        Args:
+            native: The platform's native client/app object to hand to
+                factories (PTB ``Application``, ``commands.Bot``,
+                ``AsyncApp``, aiohttp ``web.Application``, ...). Pass
+                ``None`` for adapters with no separate native object —
+                factories then work against the adapter handle alone.
+
+        Each factory is isolated so a misbehaving plugin can't prevent
+        the platform from connecting.
+        """
+        platform_name = getattr(self.platform, "value", str(self.platform))
+        try:
+            from hermes_cli.plugins import get_plugin_manager
+            factories = get_plugin_manager().get_platform_handler_factories(
+                platform_name
+            )
+        except Exception as e:  # pragma: no cover - defensive
+            logger.warning(
+                "[%s] Could not load plugin handler factories: %s",
+                self.name, e,
+            )
+            return
+        for factory, plugin_name in factories:
+            try:
+                factory(native, self)
+                logger.info(
+                    "[%s] Wired native handlers from plugin '%s'",
+                    self.name, plugin_name,
+                )
+            except Exception as exc:
+                logger.error(
+                    "[%s] Plugin '%s' handler factory raised: %s",
+                    self.name, plugin_name, exc, exc_info=True,
+                )
+
     @property
     def name(self) -> str:
         """Human-readable name for this adapter."""
diff --git a/gateway/platforms/bluebubbles.py b/gateway/platforms/bluebubbles.py
index 23e1621f3b..6306c92b9e 100644
--- a/gateway/platforms/bluebubbles.py
+++ b/gateway/platforms/bluebubbles.py
@@ -320,6 +320,8 @@ class BlueBubblesAdapter(BasePlatformAdapter):
         # This is required for the server to know where to send events
         await self._register_webhook()
 
+        # Plugin-registered native handlers (ctx.register_platform_handler).
+        self._wire_plugin_handlers(None)
         return True
 
     async def disconnect(self) -> None:
diff --git a/gateway/platforms/msgraph_webhook.py b/gateway/platforms/msgraph_webhook.py
index 3fea2ca13a..6675fadf9c 100644
--- a/gateway/platforms/msgraph_webhook.py
+++ b/gateway/platforms/msgraph_webhook.py
@@ -171,6 +171,10 @@ class MSGraphWebhookAdapter(BasePlatformAdapter):
         app.router.add_get(self._webhook_path, self._handle_validation)
         app.router.add_post(self._webhook_path, self._handle_notification)
 
+        # Plugin-registered native handlers (aiohttp web.Application —
+        # router routes). Wired before AppRunner.setup() freezes the router.
+        self._wire_plugin_handlers(app)
+
         self._runner = web.AppRunner(app)
         await self._runner.setup()
         site = web.TCPSite(self._runner, self._host, self._port)
diff --git a/gateway/platforms/signal.py b/gateway/platforms/signal.py
index 6a8363730c..4e46f2b2b2 100644
--- a/gateway/platforms/signal.py
+++ b/gateway/platforms/signal.py
@@ -405,6 +405,8 @@ class SignalAdapter(BasePlatformAdapter):
             self._health_monitor_task = asyncio.create_task(self._health_monitor())
 
             logger.info("Signal: connected to %s", self.http_url)
+            # Plugin-registered native handlers (ctx.register_platform_handler).
+            self._wire_plugin_handlers(None)
             return True
         finally:
             if not self._running:
diff --git a/gateway/platforms/webhook.py b/gateway/platforms/webhook.py
index 99eefcd634..f13dddd1c2 100644
--- a/gateway/platforms/webhook.py
+++ b/gateway/platforms/webhook.py
@@ -341,6 +341,8 @@ class WebhookAdapter(BasePlatformAdapter):
             self._port,
             route_names,
         )
+        # Plugin-registered native handlers (ctx.register_platform_handler).
+        self._wire_plugin_handlers(None)
         return True
 
     async def disconnect(self) -> None:
diff --git a/gateway/platforms/weixin.py b/gateway/platforms/weixin.py
index 5cefbffeb2..8c2b18b765 100644
--- a/gateway/platforms/weixin.py
+++ b/gateway/platforms/weixin.py
@@ -1341,6 +1341,8 @@ class WeixinAdapter(BasePlatformAdapter):
                 self.name,
                 self._group_policy,
             )
+        # Plugin-registered native handlers (ctx.register_platform_handler).
+        self._wire_plugin_handlers(None)
         return True
 
     async def disconnect(self) -> None:
diff --git a/gateway/platforms/whatsapp_cloud.py b/gateway/platforms/whatsapp_cloud.py
index bf371f0495..d8cdd192b4 100644
--- a/gateway/platforms/whatsapp_cloud.py
+++ b/gateway/platforms/whatsapp_cloud.py
@@ -492,6 +492,8 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
                 "incoming webhook POSTs will be refused with 503. Set "
                 "the app secret to enable inbound message delivery."
             )
+        # Plugin-registered native handlers (ctx.register_platform_handler).
+        self._wire_plugin_handlers(None)
         return True
 
     async def disconnect(self) -> None:
diff --git a/gateway/platforms/yuanbao.py b/gateway/platforms/yuanbao.py
index 46254bf012..4e9949e402 100644
--- a/gateway/platforms/yuanbao.py
+++ b/gateway/platforms/yuanbao.py
@@ -5068,7 +5068,11 @@ class YuanbaoAdapter(BasePlatformAdapter):
 
         Delegates to ConnectionManager.open().
         """
-        return await self._connection.open()
+        ok = await self._connection.open()
+        if ok:
+            # Plugin-registered native handlers (ctx.register_platform_handler).
+            self._wire_plugin_handlers(None)
+        return ok
 
     async def disconnect(self) -> None:
         """Cancel background tasks and close the WebSocket connection."""
diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py
index 29e7558bd7..781c639b48 100644
--- a/hermes_cli/plugins.py
+++ b/hermes_cli/plugins.py
@@ -3071,6 +3071,83 @@ class PluginContext:
         )
         return handle
 
+    # -- platform handler registration ----------------------------------------
+
+    def register_platform_handler(self, platform: str, factory: Callable) -> None:
+        """Register a native-client handler factory for a gateway platform.
+
+        The generic surface for plugins that need to receive platform
+        events the core adapter doesn't route (extra update types, native
+        button callbacks, reaction/member events, webhook routes, ...).
+
+        The adapter for ``platform`` invokes registered factories at
+        ``connect()`` time, after its native client object is built and
+        before (or as) its own handlers register. The factory receives
+        ``(native, adapter)``::
+
+            def _wire(native, adapter):
+                # native: the platform's client/app object (see table)
+                # adapter: the platform adapter instance (treat read-only)
+                ...
+
+            ctx.register_platform_handler("discord", _wire)
+
+        What ``native`` is per platform (None when the adapter has no
+        separate native client — the adapter itself is then the only
+        useful handle):
+
+        =============  ======================================================
+        telegram       python-telegram-bot ``Application`` (add_handler)
+        discord        ``discord.ext.commands.Bot`` (add_listener / events)
+        slack          ``slack_bolt.async_app.AsyncApp`` (event/action)
+        matrix         the Matrix client (event callbacks)
+        teams          Microsoft Teams ``App`` (on_message / on_card_action)
+        dingtalk       ``DingTalkStreamClient`` (register_callback_handler)
+        line           aiohttp ``web.Application`` (router)
+        others         ``None`` — connect-time hook with the adapter handle
+        =============  ======================================================
+
+        Notes:
+
+        * Factories are invoked lazily at connect time, so platform SDK
+          imports belong inside the factory body — ``register()`` keeps
+          working when the SDK isn't installed.
+        * Factories are isolated: an exception is logged and the platform
+          still connects.
+        * When hooking dispatch tables that stop at the first match
+          (e.g. PTB callback handlers), always scope your handler
+          (pattern prefixes, specific event types) so core flows keep
+          working.
+
+        Args:
+            platform: Gateway platform name, lowercase (``"telegram"``,
+                ``"discord"``, ``"slack"``, ...).
+            factory: Callable receiving ``(native, adapter)``.
+
+        Raises:
+            ValueError: if ``factory`` is not callable or ``platform`` is
+                empty.
+        """
+        if not callable(factory):
+            raise ValueError(
+                f"Plugin '{self.manifest.name}' tried to register a platform "
+                f"handler factory with a non-callable factory."
+            )
+        key = (platform or "").strip().lower()
+        if not key:
+            raise ValueError(
+                f"Plugin '{self.manifest.name}' tried to register a platform "
+                f"handler factory with an empty platform name."
+            )
+        self._manager._platform_handler_factories.setdefault(key, []).append(
+            (factory, self.manifest.name)
+        )
+        logger.debug(
+            "Plugin %s registered %s handler factory: %s",
+            self.manifest.name, key,
+            getattr(factory, "__name__", repr(factory)),
+        )
+
     # -- telegram handler registration ---------------------------------------
 
     def register_telegram_handler(self, factory: Callable) -> None:
@@ -3114,19 +3191,9 @@ class PluginContext:
         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)),
-        )
+        # Thin alias over the generic surface — kept for back-compat and
+        # for the Telegram-specific docs above.
+        self.register_platform_handler("telegram", factory)
 
     # -- hook registration --------------------------------------------------
 
@@ -3749,12 +3816,15 @@ class PluginManager:
         # full plugin loads.
         self._predeclared_modules: Dict[str, types.ModuleType] = {}
         self._predeclared_tools: Dict[str, List[str]] = {}
-        # 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] = []
+        # Native platform handler factories registered by plugins, keyed by
+        # lowercase platform name. Each entry is (factory, plugin_name);
+        # the platform's adapter invokes factories at connect() time with
+        # (native_client, adapter) so plugins can wire their own handlers
+        # (PTB handlers, discord.py listeners, slack_bolt events, webhook
+        # routes, ...) without touching core files.
+        # ``register_telegram_handler`` is a thin alias writing into the
+        # "telegram" bucket.
+        self._platform_handler_factories: Dict[str, List[tuple]] = {}
 
     # -----------------------------------------------------------------------
     # Registration ledger internals
@@ -4101,7 +4171,7 @@ class PluginManager:
             self._slack_action_handlers.clear()
             self._predeclared_modules.clear()
             self._predeclared_tools.clear()
-            self._telegram_handler_factories.clear()
+            self._platform_handler_factories.clear()
             self._context_engine = None
             with self._hook_timeout_lock:
                 self._hook_running_callbacks.clear()
@@ -5957,21 +6027,28 @@ class PluginManager:
         return list(self._slack_action_handlers)
 
     # -----------------------------------------------------------------------
-    # Telegram handler factory accessor
+    # Platform handler factory accessors
     # -----------------------------------------------------------------------
 
-    def get_telegram_handler_factories(self) -> List[tuple]:
-        """Return the list of plugin-registered Telegram handler factories.
+    def get_platform_handler_factories(self, platform: str) -> List[tuple]:
+        """Return plugin-registered handler factories for one platform.
 
         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.
+        platform's adapter at connect time; each factory is invoked with
+        ``(native_client, adapter)`` so plugins can wire their own native
+        handlers before/alongside the core ones.
 
         Plugins register factories via
-        :meth:`PluginContext.register_telegram_handler`.
+        :meth:`PluginContext.register_platform_handler` (or the
+        Telegram-specific alias
+        :meth:`PluginContext.register_telegram_handler`).
         """
-        return list(self._telegram_handler_factories)
+        key = (platform or "").strip().lower()
+        return list(self._platform_handler_factories.get(key, []))
+
+    def get_telegram_handler_factories(self) -> List[tuple]:
+        """Back-compat alias for ``get_platform_handler_factories("telegram")``."""
+        return self.get_platform_handler_factories("telegram")
 
     # -----------------------------------------------------------------------
     # Introspection
diff --git a/plugins/platforms/dingtalk/adapter.py b/plugins/platforms/dingtalk/adapter.py
index d59db86a20..fcca09e358 100644
--- a/plugins/platforms/dingtalk/adapter.py
+++ b/plugins/platforms/dingtalk/adapter.py
@@ -377,6 +377,8 @@ class DingTalkAdapter(BasePlatformAdapter):
             self._stream_task = asyncio.create_task(self._run_stream())
             self._mark_connected()
             logger.info("[%s] Connected via Stream Mode", self.name)
+            # Plugin-registered native handlers (DingTalkStreamClient — register_callback_handler()).
+            self._wire_plugin_handlers(self._stream_client)
             return True
         except Exception as e:
             logger.error("[%s] Failed to connect: %s", self.name, e)
diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py
index 2ea115b1fb..c01031cf33 100644
--- a/plugins/platforms/discord/adapter.py
+++ b/plugins/platforms/discord/adapter.py
@@ -1476,6 +1476,8 @@ class DiscordAdapter(BasePlatformAdapter):
 
             self._running = True
             self._start_liveness_probe()
+            # Plugin-registered native handlers (discord.py Bot — add_listener()/event hooks).
+            self._wire_plugin_handlers(self._client)
             return True
 
         except asyncio.TimeoutError:
diff --git a/plugins/platforms/email/adapter.py b/plugins/platforms/email/adapter.py
index 704524e4ef..89ead8a82a 100644
--- a/plugins/platforms/email/adapter.py
+++ b/plugins/platforms/email/adapter.py
@@ -796,6 +796,8 @@ class EmailAdapter(BasePlatformAdapter):
         self._running = True
         self._poll_task = asyncio.create_task(self._poll_loop())
         print(f"[Email] Connected as {self._address}")
+        # Plugin-registered native handlers (ctx.register_platform_handler).
+        self._wire_plugin_handlers(None)
         return True
 
     async def disconnect(self) -> None:
diff --git a/plugins/platforms/feishu/adapter.py b/plugins/platforms/feishu/adapter.py
index d2f3352657..42a70e79fb 100644
--- a/plugins/platforms/feishu/adapter.py
+++ b/plugins/platforms/feishu/adapter.py
@@ -1816,6 +1816,8 @@ class FeishuAdapter(BasePlatformAdapter):
             await self._connect_with_retry()
             self._mark_connected()
             logger.info("[Feishu] Connected in %s mode (%s)", self._connection_mode, self._domain_name)
+            # Plugin-registered native handlers (lark_oapi client).
+            self._wire_plugin_handlers(self._client)
             return True
         except Exception as exc:
             await self._release_app_lock()
diff --git a/plugins/platforms/google_chat/adapter.py b/plugins/platforms/google_chat/adapter.py
index ba11710897..41c9b65500 100644
--- a/plugins/platforms/google_chat/adapter.py
+++ b/plugins/platforms/google_chat/adapter.py
@@ -1147,6 +1147,8 @@ class GoogleChatAdapter(BasePlatformAdapter):
             self._max_messages,
             self._max_bytes,
         )
+        # Plugin-registered native handlers (ctx.register_platform_handler).
+        self._wire_plugin_handlers(None)
         return True
 
     async def disconnect(self) -> None:
diff --git a/plugins/platforms/homeassistant/adapter.py b/plugins/platforms/homeassistant/adapter.py
index 6564460d47..37a7397d4b 100644
--- a/plugins/platforms/homeassistant/adapter.py
+++ b/plugins/platforms/homeassistant/adapter.py
@@ -157,6 +157,8 @@ class HomeAssistantAdapter(BasePlatformAdapter):
             self._listen_task = asyncio.create_task(self._listen_loop())
             self._running = True
             logger.info("[%s] Connected to %s", self.name, self._hass_url)
+            # Plugin-registered native handlers (ctx.register_platform_handler).
+            self._wire_plugin_handlers(None)
             return True
 
         except Exception as e:
diff --git a/plugins/platforms/irc/adapter.py b/plugins/platforms/irc/adapter.py
index 030300b797..ce3ec4ed59 100644
--- a/plugins/platforms/irc/adapter.py
+++ b/plugins/platforms/irc/adapter.py
@@ -241,6 +241,8 @@ class IRCAdapter(BasePlatformAdapter):
 
         self._mark_connected()
         logger.info("IRC: connected to %s:%s as %s, joined %s", self.server, self.port, self._current_nick, self.channel)
+        # Plugin-registered native handlers (ctx.register_platform_handler).
+        self._wire_plugin_handlers(None)
         return True
 
     async def disconnect(self) -> None:
diff --git a/plugins/platforms/line/adapter.py b/plugins/platforms/line/adapter.py
index b31c1a209d..b8d3ae10cd 100644
--- a/plugins/platforms/line/adapter.py
+++ b/plugins/platforms/line/adapter.py
@@ -855,6 +855,10 @@ class LineAdapter(BasePlatformAdapter):
             self._handle_media,
         )
 
+        # Plugin-registered native handlers (aiohttp web.Application —
+        # router routes). Wired before AppRunner.setup() freezes the router.
+        self._wire_plugin_handlers(self._app)
+
         self._runner = web.AppRunner(self._app)
         try:
             await self._runner.setup()
diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py
index 0da6f37c96..3268fd9d19 100644
--- a/plugins/platforms/matrix/adapter.py
+++ b/plugins/platforms/matrix/adapter.py
@@ -2142,6 +2142,8 @@ class MatrixAdapter(BasePlatformAdapter):
         # Start the sync loop.
         self._sync_task = asyncio.create_task(self._sync_loop())
         self._mark_connected()
+        # Plugin-registered native handlers (Matrix client — event callbacks).
+        self._wire_plugin_handlers(self._client)
         return True
 
     async def disconnect(self) -> None:
diff --git a/plugins/platforms/mattermost/adapter.py b/plugins/platforms/mattermost/adapter.py
index 4e80958b35..6962fbf615 100644
--- a/plugins/platforms/mattermost/adapter.py
+++ b/plugins/platforms/mattermost/adapter.py
@@ -339,6 +339,8 @@ class MattermostAdapter(BasePlatformAdapter):
         # Start WebSocket in background.
         self._ws_task = asyncio.create_task(self._ws_loop())
         self._mark_connected()
+        # Plugin-registered native handlers (ctx.register_platform_handler).
+        self._wire_plugin_handlers(None)
         return True
 
     async def disconnect(self) -> None:
diff --git a/plugins/platforms/ntfy/adapter.py b/plugins/platforms/ntfy/adapter.py
index 935610c320..b9fb08c7ef 100644
--- a/plugins/platforms/ntfy/adapter.py
+++ b/plugins/platforms/ntfy/adapter.py
@@ -221,6 +221,8 @@ class NtfyAdapter(BasePlatformAdapter):
             self._stream_task = asyncio.create_task(self._run_stream())
             self._mark_connected()
             logger.info("[%s] Connected — subscribing to %s/%s", self.name, self._server, self._topic)
+            # Plugin-registered native handlers (ctx.register_platform_handler).
+            self._wire_plugin_handlers(None)
             return True
         except Exception as e:
             logger.error("[%s] Failed to connect: %s", self.name, e)
diff --git a/plugins/platforms/photon/adapter.py b/plugins/platforms/photon/adapter.py
index b0bc3e388c..306e279802 100644
--- a/plugins/platforms/photon/adapter.py
+++ b/plugins/platforms/photon/adapter.py
@@ -968,6 +968,8 @@ class PhotonAdapter(BasePlatformAdapter):
             "[photon] connected — sidecar on %s:%d, streaming inbound over gRPC",
             self._sidecar_bind, self._sidecar_port,
         )
+        # Plugin-registered native handlers (ctx.register_platform_handler).
+        self._wire_plugin_handlers(None)
         return True
 
     async def disconnect(self) -> None:
diff --git a/plugins/platforms/raft/adapter.py b/plugins/platforms/raft/adapter.py
index 7ad661f910..d31ee4601a 100644
--- a/plugins/platforms/raft/adapter.py
+++ b/plugins/platforms/raft/adapter.py
@@ -513,6 +513,8 @@ class RaftAdapter(BasePlatformAdapter):
         logger.info("[raft] Raft channel listening on %s:%d%s", self._host, bound_port, self._path)
 
         self._spawn_bridge(bound_port)
+        # Plugin-registered native handlers (ctx.register_platform_handler).
+        self._wire_plugin_handlers(None)
         return True
 
     async def disconnect(self) -> None:
diff --git a/plugins/platforms/simplex/adapter.py b/plugins/platforms/simplex/adapter.py
index 9eaa4d856e..b4f493e456 100644
--- a/plugins/platforms/simplex/adapter.py
+++ b/plugins/platforms/simplex/adapter.py
@@ -240,6 +240,8 @@ class SimplexAdapter(BasePlatformAdapter):
         if hasattr(self, "_mark_connected"):
             self._mark_connected()
         logger.info("SimpleX: connected to %s", self.ws_url)
+        # Plugin-registered native handlers (ctx.register_platform_handler).
+        self._wire_plugin_handlers(None)
         return True
 
     async def disconnect(self) -> None:
diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py
index b83b70d61d..cd8e237d3b 100644
--- a/plugins/platforms/slack/adapter.py
+++ b/plugins/platforms/slack/adapter.py
@@ -2425,6 +2425,14 @@ class SlackAdapter(BasePlatformAdapter):
                     len(_plugin_handlers),
                 )
 
+            # Generic plugin-registered native handlers
+            # (ctx.register_platform_handler("slack", ...)). Factories get
+            # the slack_bolt AsyncApp — the full app.event()/app.action()/
+            # app.command() surface, not just Block Kit actions. Wired
+            # before Socket Mode starts so bolt's matcher knows about them
+            # before events dispatch.
+            self._wire_plugin_handlers(self._app)
+
             # Bring up the handler and watchdog atomically. ``_running`` only
             # flips to True after the handler is alive so the watchdog loop
             # observes the live task immediately; on any failure here we tear
diff --git a/plugins/platforms/sms/adapter.py b/plugins/platforms/sms/adapter.py
index 0c081242d9..37db336e7a 100644
--- a/plugins/platforms/sms/adapter.py
+++ b/plugins/platforms/sms/adapter.py
@@ -166,6 +166,8 @@ class SmsAdapter(BasePlatformAdapter):
             self._webhook_port,
             redact_phone(self._from_number),
         )
+        # Plugin-registered native handlers (ctx.register_platform_handler).
+        self._wire_plugin_handlers(None)
         return True
 
     async def disconnect(self) -> None:
diff --git a/plugins/platforms/teams/adapter.py b/plugins/platforms/teams/adapter.py
index 2a66ed087e..88bbf5a184 100644
--- a/plugins/platforms/teams/adapter.py
+++ b/plugins/platforms/teams/adapter.py
@@ -856,6 +856,11 @@ class TeamsAdapter(BasePlatformAdapter):
             ) -> InvokeResponse[AdaptiveCardActionMessageResponse]:
                 return await self._on_card_action(ctx)
 
+            # Plugin-registered native handlers (Teams App — on_message /
+            # on_card_action / on_* decorators). Wired before initialize()
+            # so plugin routes register alongside ours.
+            self._wire_plugin_handlers(self._app)
+
             # initialize() calls register_route() on the bridge, which adds
             # POST /api/messages to aiohttp_app automatically
             await self._app.initialize()
diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py
index 501912bd2b..a7cd3b5a85 100644
--- a/plugins/platforms/telegram/adapter.py
+++ b/plugins/platforms/telegram/adapter.py
@@ -4363,44 +4363,6 @@ class TelegramAdapter(BasePlatformAdapter):
         # it observes alongside, never displaces, the core handlers.
         app.add_handler(TypeHandler(Update, self._on_platform_update), group=99)
 
-    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.
 
@@ -4644,8 +4606,13 @@ class TelegramAdapter(BasePlatformAdapter):
             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()
+            # Plugins register via ctx.register_telegram_handler (alias of
+            # ctx.register_platform_handler("telegram", ...)); factories
+            # receive (application, adapter). PTB dispatches the first
+            # matching handler per group, so pattern-scoped plugin handlers
+            # take precedence for their own updates while everything else
+            # falls through to the core handlers below.
+            self._wire_plugin_handlers(self._app)
 
             # Register handlers via the single registration site (#64176).
             self._register_handlers(self._app)
diff --git a/plugins/platforms/wecom/adapter.py b/plugins/platforms/wecom/adapter.py
index 929178df53..c26d4a8350 100644
--- a/plugins/platforms/wecom/adapter.py
+++ b/plugins/platforms/wecom/adapter.py
@@ -644,6 +644,8 @@ class WeComAdapter(BasePlatformAdapter):
             self._listen_task = asyncio.create_task(self._listen_loop())
             self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())
             logger.info("[%s] Connected to %s", self.name, self._ws_url)
+            # Plugin-registered native handlers (ctx.register_platform_handler).
+            self._wire_plugin_handlers(None)
             return True
         except Exception as exc:
             message = f"WeCom startup failed: {exc}"
diff --git a/plugins/platforms/whatsapp/adapter.py b/plugins/platforms/whatsapp/adapter.py
index fd21387244..282cb7c0cb 100644
--- a/plugins/platforms/whatsapp/adapter.py
+++ b/plugins/platforms/whatsapp/adapter.py
@@ -660,6 +660,8 @@ class WhatsAppAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
                                     self._bridge_process = None  # Not managed by us
                                     self._http_session = aiohttp.ClientSession()
                                     self._poll_task = asyncio.create_task(self._poll_messages())
+                                    # Plugin-registered native handlers.
+                                    self._wire_plugin_handlers(None)
                                     return True
                                 stale_reason = (
                                     f"running={running_hash or 'unversioned'}, disk={disk_hash}"
@@ -820,6 +822,8 @@ class WhatsAppAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
             
             self._mark_connected()
             print(f"[{self.name}] Bridge started on port {self._bridge_port}")
+            # Plugin-registered native handlers.
+            self._wire_plugin_handlers(None)
             return True
             
         except Exception as e:
diff --git a/tests/gateway/test_platform_plugin_handlers.py b/tests/gateway/test_platform_plugin_handlers.py
new file mode 100644
index 0000000000..663033ba70
--- /dev/null
+++ b/tests/gateway/test_platform_plugin_handlers.py
@@ -0,0 +1,255 @@
+"""Tests for plugin-registered native platform handler factories.
+
+Covers:
+* ``PluginContext.register_platform_handler`` validation + queuing
+* ``PluginContext.register_telegram_handler`` back-compat alias
+* ``PluginManager.get_platform_handler_factories`` accessor (+ telegram alias)
+* ``BasePlatformAdapter._wire_plugin_handlers`` invoking factories with
+  ``(native, adapter)`` — exercised through the Telegram adapter
+* Defensive isolation: a factory that raises does NOT prevent other
+  factories from wiring or the platform from connecting.
+* Platform scoping: factories for platform A never fire for platform B.
+* ``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_platform_handler — validation + queuing
+# ===========================================================================
+
+class TestRegisterPlatformHandlerAPI:
+    def test_factory_is_queued_with_plugin_name(self):
+        mgr, ctx = _make_ctx()
+
+        def factory(native, adapter):  # pragma: no cover - never called
+            pass
+
+        ctx.register_platform_handler("discord", factory)
+
+        factories = mgr.get_platform_handler_factories("discord")
+        assert len(factories) == 1
+        fn, plugin_name = factories[0]
+        assert fn is factory
+        assert plugin_name == "test_plugin"
+
+    def test_platform_key_is_normalized(self):
+        mgr, ctx = _make_ctx()
+        ctx.register_platform_handler("  Slack ", lambda n, a: None)
+        assert len(mgr.get_platform_handler_factories("slack")) == 1
+
+    def test_non_callable_factory_raises(self):
+        _, ctx = _make_ctx()
+        with pytest.raises(ValueError, match="non-callable"):
+            ctx.register_platform_handler("discord", "nope")  # type: ignore[arg-type]
+
+    def test_empty_platform_raises(self):
+        _, ctx = _make_ctx()
+        with pytest.raises(ValueError, match="empty platform"):
+            ctx.register_platform_handler("  ", lambda n, a: None)
+
+    def test_platform_scoping(self):
+        """Factories for platform A never appear in platform B's list."""
+        mgr, ctx = _make_ctx()
+        ctx.register_platform_handler("discord", lambda n, a: None)
+        ctx.register_platform_handler("matrix", lambda n, a: None)
+        assert len(mgr.get_platform_handler_factories("discord")) == 1
+        assert len(mgr.get_platform_handler_factories("matrix")) == 1
+        assert mgr.get_platform_handler_factories("slack") == []
+
+    def test_accessor_returns_copy(self):
+        mgr, ctx = _make_ctx()
+        ctx.register_platform_handler("telegram", lambda n, a: None)
+
+        got = mgr.get_platform_handler_factories("telegram")
+        got.append(("junk", "junk"))
+        assert len(mgr.get_platform_handler_factories("telegram")) == 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_platform_handler("telegram", lambda n, a: None)
+
+        names = [n for _, n in mgr.get_platform_handler_factories("telegram")]
+        assert names == ["plugin_a", "plugin_b"]
+
+    def test_force_rediscovery_clears_factories(self):
+        mgr, ctx = _make_ctx()
+        ctx.register_platform_handler("telegram", lambda n, a: None)
+        assert len(mgr.get_platform_handler_factories("telegram")) == 1
+
+        mgr.discover_and_load(force=True)
+        assert mgr.get_platform_handler_factories("telegram") == []
+
+
+# ===========================================================================
+# Telegram back-compat alias
+# ===========================================================================
+
+class TestTelegramAlias:
+    def test_register_telegram_handler_routes_to_telegram_bucket(self):
+        mgr, ctx = _make_ctx()
+
+        def factory(application, adapter):  # pragma: no cover
+            pass
+
+        ctx.register_telegram_handler(factory)
+
+        assert mgr.get_platform_handler_factories("telegram") == [
+            (factory, "test_plugin")
+        ]
+        # Legacy accessor still works.
+        assert mgr.get_telegram_handler_factories() == [(factory, "test_plugin")]
+
+    def test_alias_non_callable_raises(self):
+        _, ctx = _make_ctx()
+        with pytest.raises(ValueError, match="non-callable"):
+            ctx.register_telegram_handler("not-a-callable")  # type: ignore[arg-type]
+
+
+# ===========================================================================
+# BasePlatformAdapter._wire_plugin_handlers (via TelegramAdapter)
+# ===========================================================================
+
+class TestAdapterPluginWiring:
+    def test_factory_invoked_with_native_and_adapter(self):
+        adapter = _make_adapter()
+        calls = []
+
+        def factory(native, adp):
+            calls.append((native, adp))
+            native.add_handler(MagicMock())
+
+        mgr = MagicMock()
+        mgr.get_platform_handler_factories.return_value = [(factory, "biz_plugin")]
+
+        with patch("hermes_cli.plugins.get_plugin_manager", return_value=mgr):
+            adapter._wire_plugin_handlers(adapter._app)
+
+        assert calls == [(adapter._app, adapter)]
+        adapter._app.add_handler.assert_called_once()
+        # Adapter asked for its own platform's factories.
+        mgr.get_platform_handler_factories.assert_called_once_with("telegram")
+
+    def test_no_factories_is_a_noop(self):
+        adapter = _make_adapter()
+        mgr = MagicMock()
+        mgr.get_platform_handler_factories.return_value = []
+
+        with patch("hermes_cli.plugins.get_plugin_manager", return_value=mgr):
+            adapter._wire_plugin_handlers(adapter._app)
+
+        adapter._app.add_handler.assert_not_called()
+
+    def test_raising_factory_does_not_block_others(self):
+        adapter = _make_adapter()
+        wired = []
+
+        def bad_factory(native, adp):
+            raise RuntimeError("boom")
+
+        def good_factory(native, adp):
+            wired.append("good")
+
+        mgr = MagicMock()
+        mgr.get_platform_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(adapter._app)  # 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(adapter._app)  # must not raise
+
+    def test_native_none_supported(self):
+        """Adapters without a separate native client pass None."""
+        adapter = _make_adapter()
+        seen = []
+
+        mgr = MagicMock()
+        mgr.get_platform_handler_factories.return_value = [
+            (lambda native, adp: seen.append(native), "p"),
+        ]
+        with patch("hermes_cli.plugins.get_plugin_manager", return_value=mgr):
+            adapter._wire_plugin_handlers(None)
+        assert seen == [None]
+
+
+# ===========================================================================
+# Every adapter calls _wire_plugin_handlers in connect() — source invariant
+# ===========================================================================
+
+def test_all_connectable_adapters_wire_plugin_handlers():
+    """Invariant: every platform adapter with a connect() implementation
+    calls ``_wire_plugin_handlers`` somewhere in its source, so plugins can
+    rely on the hook existing on every platform (native may be None)."""
+    import glob
+
+    repo = Path(_repo)
+    adapter_files = sorted(
+        glob.glob(str(repo / "plugins" / "platforms" / "*" / "adapter.py"))
+    ) + [
+        str(repo / "gateway" / "platforms" / name)
+        for name in (
+            "api_server.py", "bluebubbles.py", "msgraph_webhook.py",
+            "signal.py", "webhook.py", "weixin.py",
+            "whatsapp_cloud.py", "yuanbao.py",
+        )
+    ]
+    missing = []
+    for f in adapter_files:
+        src = Path(f).read_text()
+        if "async def connect(" not in src:
+            continue
+        if "_wire_plugin_handlers" not in src:
+            missing.append(f)
+    assert not missing, f"adapters missing plugin-handler wiring: {missing}"
diff --git a/tests/gateway/test_telegram_plugin_handlers.py b/tests/gateway/test_telegram_plugin_handlers.py
deleted file mode 100644
index c66b443455..0000000000
--- a/tests/gateway/test_telegram_plugin_handlers.py
+++ /dev/null
@@ -1,165 +0,0 @@
-"""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
diff --git a/website/docs/developer-guide/plugins/index.md b/website/docs/developer-guide/plugins/index.md
index 43075a0c76..0a27acadef 100644
--- a/website/docs/developer-guide/plugins/index.md
+++ b/website/docs/developer-guide/plugins/index.md
@@ -1279,18 +1279,59 @@ def register(ctx):
 - Standard slack_bolt rules apply — `await ack()` within 3 seconds, then do longer work.
 - For multi-workspace deployments the handler fires for clicks from any connected workspace; use `body["team"]["id"]` if you need to scope behaviour.
 
-This is the public way for plugins to participate in Slack interactivity. Older plugins may patch `SlackAdapter.connect`; prefer this API instead.
+This is the public way for plugins to participate in Slack interactivity. Older plugins may patch `SlackAdapter.connect`; prefer this API instead. For the full slack_bolt surface (events, shortcuts, commands — not just Block Kit actions), use the generic `register_platform_handler("slack", ...)` below.
 
-### Register Telegram (PTB) handlers
+### Register native platform handlers (any platform)
 
-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.
+Plugins that need to receive platform events the core adapter doesn't route — extra update types, native button callbacks, reaction/member events, webhook routes — can register a handler factory that the platform's adapter invokes at connect time. This works on **every** gateway platform.
+
+```python
+def register(ctx):
+    def _wire(native, adapter):
+        # native: the platform's client/app object (see table below)
+        # adapter: the platform adapter instance (treat as read-only)
+        # Import platform SDKs HERE so register() works without them.
+        ...
+
+    ctx.register_platform_handler("discord", _wire)
+```
+
+**Signature:** `ctx.register_platform_handler(platform, factory) -> None`
+
+| Parameter | Type | Description |
+|-----------|------|-------------|
+| `platform` | `str` | Gateway platform name, lowercase (`"telegram"`, `"discord"`, `"slack"`, `"matrix"`, ...) |
+| `factory` | callable | Receives `(native, adapter)` at connect time |
+
+**What `native` is, per platform:**
+
+| Platform | `native` object | Typical hooks |
+|----------|-----------------|---------------|
+| `telegram` | PTB `Application` | `add_handler` — any update type, pattern-scoped callbacks |
+| `discord` | `discord.ext.commands.Bot` | `add_listener` — reactions, member events, threads, voice |
+| `slack` | `slack_bolt.AsyncApp` | `app.event()` / `app.action()` / `app.command()` |
+| `matrix` | Matrix client | event callbacks |
+| `teams` | Teams `App` | `on_message` / `on_card_action` decorators |
+| `dingtalk` | `DingTalkStreamClient` | `register_callback_handler` for other stream topics |
+| `feishu` | lark_oapi client | API calls; event routing |
+| `line`, `api_server`, `msgraph_webhook` | aiohttp `web.Application` | `router.add_get/post` — custom routes (wired before the router freezes) |
+| everything else (whatsapp, signal, irc, email, sms, ntfy, wecom, weixin, bluebubbles, yuanbao, ...) | `None` | connect-time hook; work through the `adapter` handle |
+
+**Runtime behavior:**
+
+- Factories are queued at plugin-load time and invoked when the platform connects — for platforms where dispatch order matters (Telegram, Slack, Teams, aiohttp routers) they run **before** the core handlers register, so scoped plugin handlers take precedence and everything else falls through.
+- **Always scope handlers you add to first-match dispatch tables.** On Telegram, use `CallbackQueryHandler(..., pattern=r"^myplugin:")` — an unscoped handler would swallow the core button flows (exec approvals, model picker, clarify prompts).
+- Each factory is isolated: if it raises, the error is logged and the platform still connects.
+- Import platform SDKs inside the factory body, not at module level — `register()` must work when the SDK isn't installed.
+- One plugin can register factories for several platforms; each fires only when its platform connects.
+
+**Telegram alias:** `ctx.register_telegram_handler(factory)` is a back-compat alias for `ctx.register_platform_handler("telegram", factory)`.
+
+Example — Telegram, pattern-scoped inline buttons:
 
 ```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):
@@ -1302,21 +1343,21 @@ def register(ctx):
             CallbackQueryHandler(_on_button, pattern=r"^myplugin:")
         )
 
-    ctx.register_telegram_handler(_wire)
+    ctx.register_platform_handler("telegram", _wire)
 ```
 
-**Signature:** `ctx.register_telegram_handler(factory) -> None`
+Example — Discord, reaction events:
 
-| Parameter | Type | Description |
-|-----------|------|-------------|
-| `factory` | callable | Receives `(application, adapter)` — the PTB `Application` and the `TelegramAdapter` instance (`adapter.bot`, `adapter.config`; treat it as read-only) |
+```python
+def register(ctx):
+    def _wire(bot, adapter):
+        async def on_raw_reaction_add(payload):
+            ...  # e.g. reaction-based voting / moderation
 
-**Runtime behavior:**
+        bot.add_listener(on_raw_reaction_add, "on_raw_reaction_add")
 
-- 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.
+    ctx.register_platform_handler("discord", _wire)
+```
 
 :::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.