diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index f224aab323fcd..475d8f8386d2f 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -3582,6 +3582,24 @@ class BasePlatformAdapter(ABC): def is_connected(self) -> bool: """Check if adapter is currently connected.""" return self._running + + def _fire_gateway_hook(self, name: str, **kwargs: Any) -> None: + """Fire a ``gateway_*`` platform-boundary observer hook (see VALID_HOOKS). + + Observer-only: ``invoke_hook`` isolates each callback and this wrapper + swallows any plugin-layer error so a misbehaving plugin can't break the + adapter. A ``has_hook`` guard skips all dispatch when nothing subscribes + (the common case). Callers pass the hook's documented kwargs (e.g. the + normalized ``gateway_platform_event`` envelope) — never raw SDK objects. + """ + try: + from hermes_cli.plugins import get_plugin_manager + mgr = get_plugin_manager() + if not mgr.has_hook(name): + return + mgr.invoke_hook(name, **kwargs) + except Exception as exc: + logger.debug("[%s] %s hook fire error: %s", self.name, name, exc) def set_message_handler(self, handler: MessageHandler) -> None: """ diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 09f19f2d338b4..ce04883cd38ba 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -220,6 +220,23 @@ VALID_HOOKS: Set[str] = { "kanban_task_claimed", "kanban_task_completed", "kanban_task_blocked", + # Gateway platform-boundary observer hooks (#64176). Observer-only; each + # callback isolated by invoke_hook. Normalized envelopes only — NO raw + # platform SDK objects in the payload (per #64176 / #64182 ground rule); + # raw access is a separate capability-gated action API, not a hook. + # + # gateway_platform_event — inbound platform event (reactions, forwards, + # edits, chat-member) as a normalized envelope. Kwargs: platform, + # event_type, payload (event_type-specific dict). Fired for Telegram + # reactions today; other event types + fire-sites land with #64176's + # taxonomy (#64231). + # gateway_session_titled / gateway_message_delivered / gateway_thread_created + # — reserved (fire-sites pending #64176); names registered so plugins + # can subscribe ahead of implementation. + "gateway_session_titled", + "gateway_message_delivered", + "gateway_thread_created", + "gateway_platform_event", } ENTRY_POINTS_GROUP = "hermes_agent.plugins" diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index bda9b8c35d15a..91cc773134a93 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -245,6 +245,7 @@ try: CallbackQueryHandler, MessageHandler as TelegramMessageHandler, ContextTypes, + TypeHandler, filters, ) from telegram.constants import ParseMode, ChatType @@ -261,6 +262,7 @@ except ImportError: Application = Any CommandHandler = Any CallbackQueryHandler = Any + TypeHandler = Any TelegramMessageHandler = Any HTTPXRequest = Any filters = None @@ -3679,6 +3681,64 @@ class TelegramAdapter(BasePlatformAdapter): if self._post_connect_task is asyncio.current_task(): self._post_connect_task = None + async def _on_platform_update(self, update, context) -> None: + """Catch-all PTB handler firing ``gateway_platform_event`` per inbound update. + + Normalizes the update into a stable envelope (no raw SDK objects — see + #64176) and fires the observer hook. Registered in a dedicated high + group so it observes alongside — never displaces — the core handlers. + Normalization is wrapped so a malformed update can't raise into PTB + dispatch — the observer can't break the adapter. + """ + try: + event = self._normalize_platform_event(update) + except Exception as exc: + logger.debug("[%s] gateway_platform_event normalize error: %s", self.name, exc) + return + if event is None: + return + self._fire_gateway_hook("gateway_platform_event", **event) + + def _normalize_platform_event(self, update) -> Optional[Dict[str, Any]]: + """Map an inbound PTB update to a normalized ``gateway_platform_event`` + envelope ``{platform, event_type, payload}``, or ``None`` if unsupported. + + Reaction (the motivating use case: a plugin that re-renders or reacts to + a message when the user reacts to it) is normalized to the fields a + plugin consumes: ``emojis`` (standard unicode), ``custom_emoji_ids`` + (custom reaction emojis — PTB exposes ``custom_emoji_id`` with no + ``.emoji``), ``chat_id``, ``message_id``, ``thread_id``. Other update + types (forward, edit, chat-member) return ``None`` for now; their + payload contracts land with #64176's taxonomy (#64231). + """ + mr = getattr(update, "message_reaction", None) + if mr is None: + return None + chat = getattr(mr, "chat", None) + new_reaction = getattr(mr, "new_reaction", None) or [] + emojis: List[str] = [] + custom_emoji_ids: List[str] = [] + for r in new_reaction: + emoji = getattr(r, "emoji", None) + if emoji is not None: + emojis.append(emoji) + custom_id = getattr(r, "custom_emoji_id", None) + if custom_id is not None: + custom_emoji_ids.append(str(custom_id)) + return { + "platform": "telegram", + "event_type": "reaction", + "payload": { + "emojis": emojis, + "custom_emoji_ids": custom_emoji_ids, + "chat_id": str(getattr(chat, "id", "")) if chat is not None else None, + "message_id": str(getattr(mr, "message_id", "")), + # Reactions don't carry thread_id; plugins route the reply via + # their own send/edit cache or the future action API (#64176). + "thread_id": None, + }, + } + async def connect(self, *, is_reconnect: bool = False) -> bool: """Connect to Telegram via polling or webhook. @@ -3910,6 +3970,9 @@ class TelegramAdapter(BasePlatformAdapter): )) # Handle inline keyboard button callbacks (update prompts) self._app.add_handler(CallbackQueryHandler(self._handle_callback_query)) + # gateway_platform_event observer (see _on_platform_update); group 99 + # so it observes alongside — never displaces — the core handlers. + self._app.add_handler(TypeHandler(Update, self._on_platform_update), group=99) # Start polling — retry initialize() for transient TLS resets. # Each attempt is capped by _init_timeout so a single unreachable diff --git a/tests/gateway/test_gateway_platform_event_hook.py b/tests/gateway/test_gateway_platform_event_hook.py new file mode 100644 index 0000000000000..6f46cc0f03f08 --- /dev/null +++ b/tests/gateway/test_gateway_platform_event_hook.py @@ -0,0 +1,251 @@ +"""Tests for the ``gateway_platform_event`` observer hook (#64176's observer half). + +Covers the normalized-envelope pattern that replaces raw-SDK handler args: +* the four ``gateway_*`` hooks are registered in ``VALID_HOOKS`` +* ``BasePlatformAdapter._fire_gateway_hook`` routes to ``invoke_hook`` with a + ``has_hook`` no-subscriber fast-path and per-call error isolation +* ``TelegramAdapter._normalize_platform_event`` maps an inbound PTB update to a + stable ``{platform, event_type, payload}`` envelope (no raw SDK objects), + including custom-emoji reactions +* ``_on_platform_update`` fires ``gateway_platform_event`` with that envelope + and swallows normalization errors so the observer can't break the adapter +""" + +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + + +_repo = str(Path(__file__).resolve().parents[2]) +if _repo not in sys.path: + sys.path.insert(0, _repo) + + +# --------------------------------------------------------------------------- +# python-telegram-bot is an optional dep; mock it so the adapter imports +# (same shim as test_telegram_network_reconnect / test_telegram_plugin_handlers). +# --------------------------------------------------------------------------- +def _ensure_telegram_mock() -> None: + if "telegram" in sys.modules and hasattr(sys.modules["telegram"], "__file__"): + return + telegram_mod = MagicMock() + telegram_mod.ext.ContextTypes.DEFAULT_TYPE = type(None) + telegram_mod.constants.ParseMode.MARKDOWN_V2 = "MarkdownV2" + telegram_mod.constants.ChatType.GROUP = "group" + telegram_mod.constants.ChatType.SUPERGROUP = "supergroup" + telegram_mod.constants.ChatType.CHANNEL = "channel" + telegram_mod.constants.ChatType.PRIVATE = "private" + for name in ("telegram", "telegram.ext", "telegram.constants", "telegram.request"): + sys.modules.setdefault(name, telegram_mod) + + +_ensure_telegram_mock() + +from plugins.platforms.telegram.adapter import TelegramAdapter # noqa: E402 +from hermes_cli.plugins import VALID_HOOKS # noqa: E402 + + +def _adapter() -> TelegramAdapter: + """Build a TelegramAdapter without the heavy __init__. + + _fire_gateway_hook / _normalize_platform_event only need self.name (a + read-only property over self.platform), so set a stand-in platform. + """ + a = object.__new__(TelegramAdapter) + a.platform = SimpleNamespace(value="telegram") # name -> "Telegram" + return a + + +def _reaction(*, emoji=None, custom_emoji_id=None): + """A PTB ReactionType stand-in. + + PTB exposes ``.emoji`` for standard-emoji reactions and + ``.custom_emoji_id`` for custom-emoji reactions (one or the other). Set + both explicitly so the MagicMock doesn't auto-supply a truthy attribute. + """ + r = MagicMock() + r.emoji = emoji + r.custom_emoji_id = custom_emoji_id + return r + + +def _reaction_update(reactions, chat_id=123, message_id=456): + """A PTB Update stand-in carrying a message_reaction with ``reactions``.""" + update = MagicMock() + update.message_reaction = MagicMock() + update.message_reaction.chat.id = chat_id + update.message_reaction.message_id = message_id + update.message_reaction.new_reaction = list(reactions) + return update + + +# --------------------------------------------------------------------------- +# Hook registration +# --------------------------------------------------------------------------- + +class TestHookRegistration: + def test_gateway_hooks_are_valid(self): + """register_hook rejects names not in VALID_HOOKS, so the four new + platform-boundary hooks must be present there.""" + assert "gateway_platform_event" in VALID_HOOKS + assert "gateway_session_titled" in VALID_HOOKS + assert "gateway_message_delivered" in VALID_HOOKS + assert "gateway_thread_created" in VALID_HOOKS + + +# --------------------------------------------------------------------------- +# BasePlatformAdapter._fire_gateway_hook — routing + isolation +# --------------------------------------------------------------------------- + +class TestFireGatewayHook: + def test_routes_to_invoke_hook_with_kwargs(self): + a = _adapter() + captured: dict = {} + + def fake_invoke(name, **kwargs): + captured["name"] = name + captured["kwargs"] = kwargs + + mgr = MagicMock() + mgr.has_hook.return_value = True + mgr.invoke_hook.side_effect = fake_invoke + + with patch("hermes_cli.plugins.get_plugin_manager", return_value=mgr): + a._fire_gateway_hook( + "gateway_platform_event", + platform="telegram", event_type="reaction", payload={"emojis": ["x"]}, + ) + + assert captured["name"] == "gateway_platform_event" + assert captured["kwargs"] == { + "platform": "telegram", "event_type": "reaction", "payload": {"emojis": ["x"]}, + } + + def test_skips_dispatch_when_no_subscriber(self): + """has_hook False -> invoke_hook never called.""" + a = _adapter() + mgr = MagicMock() + mgr.has_hook.return_value = False + + with patch("hermes_cli.plugins.get_plugin_manager", return_value=mgr): + a._fire_gateway_hook("gateway_platform_event", platform="telegram") + + mgr.has_hook.assert_called_once_with("gateway_platform_event") + mgr.invoke_hook.assert_not_called() + + def test_plugin_layer_error_is_isolated(self): + """A raising invoke_hook OR get_plugin_manager must not propagate.""" + a = _adapter() + mgr = MagicMock() + mgr.has_hook.return_value = True + mgr.invoke_hook.side_effect = RuntimeError("plugin boom") + + with patch("hermes_cli.plugins.get_plugin_manager", return_value=mgr): + a._fire_gateway_hook("gateway_platform_event", platform="telegram") # no raise + + +# --------------------------------------------------------------------------- +# TelegramAdapter._normalize_platform_event — envelope normalization +# --------------------------------------------------------------------------- + +class TestNormalizePlatformEvent: + def test_standard_emoji_reaction_normalized(self): + """A message_reaction update becomes {platform, event_type, payload} with + exactly the fields a real plugin consumes — no raw SDK objects.""" + a = _adapter() + update = _reaction_update([_reaction(emoji="\U0001F44E")], chat_id=123, message_id=456) + + assert a._normalize_platform_event(update) == { + "platform": "telegram", + "event_type": "reaction", + "payload": { + "emojis": ["\U0001F44E"], + "custom_emoji_ids": [], + "chat_id": "123", + "message_id": "456", + "thread_id": None, + }, + } + + def test_custom_emoji_reaction_normalized(self): + """Custom-emoji reactions expose custom_emoji_id (no .emoji) — captured + separately so a string-joining consumer never sees None.""" + a = _adapter() + update = _reaction_update([_reaction(custom_emoji_id="555123")]) + + event = a._normalize_platform_event(update) + assert event["payload"]["emojis"] == [] + assert event["payload"]["custom_emoji_ids"] == ["555123"] + + def test_mixed_reactions_split_correctly(self): + """A reaction set with standard + custom emojis splits into both lists.""" + a = _adapter() + update = _reaction_update([ + _reaction(emoji="\U0001F44D"), + _reaction(custom_emoji_id="555"), + _reaction(emoji="\U0001F525"), + ]) + + event = a._normalize_platform_event(update) + assert event["payload"]["emojis"] == ["\U0001F44D", "\U0001F525"] + assert event["payload"]["custom_emoji_ids"] == ["555"] + + def test_non_reaction_update_returns_none(self): + """Unsupported update types return None (payload contracts pending #64231).""" + a = _adapter() + update = MagicMock() + update.message_reaction = None # e.g. an edited_message or chat_member update + + assert a._normalize_platform_event(update) is None + + +# --------------------------------------------------------------------------- +# TelegramAdapter._on_platform_update — fire-site +# --------------------------------------------------------------------------- + +class TestOnPlatformUpdate: + def test_fires_gateway_platform_event_with_envelope(self): + a = _adapter() + seen: list = [] + a._fire_gateway_hook = lambda name, **kw: seen.append((name, kw)) # type: ignore[assignment] + + asyncio.run(a._on_platform_update( + _reaction_update([_reaction(emoji="\U0001F44E")], 123, 456), context=MagicMock(), + )) + + assert len(seen) == 1 + name, kwargs = seen[0] + assert name == "gateway_platform_event" + assert kwargs["platform"] == "telegram" + assert kwargs["event_type"] == "reaction" + assert kwargs["payload"]["emojis"] == ["\U0001F44E"] + assert kwargs["payload"]["chat_id"] == "123" + + def test_unsupported_update_does_not_fire(self): + a = _adapter() + seen: list = [] + a._fire_gateway_hook = lambda name, **kw: seen.append((name, kw)) # type: ignore[assignment] + + update = MagicMock() + update.message_reaction = None + asyncio.run(a._on_platform_update(update, context=MagicMock())) + + assert seen == [] + + def test_normalize_error_does_not_propagate(self): + """A malformed update that makes normalize raise must be swallowed — the + observer can't break the adapter (regression guard for the try/except).""" + a = _adapter() + a._fire_gateway_hook = lambda *a_, **kw: pytest.fail("must not fire on normalize error") # type: ignore[assignment] + + def boom(update): + raise RuntimeError("malformed update") + + a._normalize_platform_event = boom # type: ignore[assignment] + asyncio.run(a._on_platform_update(MagicMock(), context=MagicMock())) # must not raise