feat(telegram): add plugin-owned callback actions

cron/jobs.py

@@ -1036,6 +1036,29 @@ def _normalized_inference_axes(job: Dict[str, Any]) -> Tuple[Optional[str], Opti
     )
 
 
+def normalize_inline_buttons(buttons: Optional[List[str]]) -> Optional[List[List[Dict[str, str]]]]:
+    """Normalize repeatable ``LABEL=CALLBACK_DATA`` values for Telegram."""
+    if buttons is None:
+        return None
+    if not isinstance(buttons, list) or not 1 <= len(buttons) <= 8:
+        raise ValueError("buttons must contain between 1 and 8 LABEL=CALLBACK_DATA values")
+    row: List[Dict[str, str]] = []
+    seen = set()
+    for raw in buttons:
+        if not isinstance(raw, str) or "=" not in raw:
+            raise ValueError("each button must use LABEL=CALLBACK_DATA")
+        label, callback_data = (part.strip() for part in raw.split("=", 1))
+        if not label or len(label) > 64:
+            raise ValueError("button labels must contain 1 to 64 characters")
+        if not callback_data or len(callback_data.encode("utf-8")) > 64:
+            raise ValueError("button callback data must contain 1 to 64 bytes")
+        if callback_data in seen:
+            raise ValueError("button callback data must be unique within a job")
+        seen.add(callback_data)
+        row.append({"text": label, "callback_data": callback_data})
+    return [row]
+
+
 def create_job(
     prompt: Optional[str],
     schedule: str,
@@ -1054,6 +1077,7 @@ def create_job(
     workdir: Optional[str] = None,
     no_agent: bool = False,
     attach_to_session: Optional[bool] = None,
+    buttons: Optional[List[str]] = None,
 ) -> Dict[str, Any]:
     """
     Create a new cron job.
@@ -1130,6 +1154,7 @@ def create_job(
     normalized_workdir = _normalize_workdir(workdir)
     normalized_no_agent = bool(no_agent)
     normalized_attach = attach_to_session if isinstance(attach_to_session, bool) else None
+    normalized_inline_keyboard = normalize_inline_buttons(buttons)
 
     # no_agent jobs are meaningless without a script — the script IS the job.
     # Surface this as a clear ValueError at create time so bad configs never
@@ -1220,6 +1245,8 @@ def create_job(
         "enabled_toolsets": normalized_toolsets,
         "workdir": normalized_workdir,
     }
+    if normalized_inline_keyboard is not None:
+        job["inline_keyboard"] = normalized_inline_keyboard
     # Only persist attach_to_session when explicitly set, so existing jobs and
     # the common case stay byte-identical (absent key => fall back to the
     # global cron.mirror_delivery config, default off).

cron/scheduler.py

@@ -1488,6 +1488,7 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
         return msg
 
     delivery_errors = []
+    inline_keyboard = job.get("inline_keyboard")
 
     for target in targets:
         platform_name = target["platform"]
@@ -1677,6 +1678,9 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                     route_metadata["thread_id"] = route_thread_id
                 media_metadata = {"thread_id": thread_id} if thread_id else None
 
+            if platform == Platform.TELEGRAM and inline_keyboard is not None:
+                route_metadata["inline_keyboard"] = inline_keyboard
+
             try:
                 # Send cleaned text (MEDIA tags stripped) — not the raw content.
                 # Route through the gateway's DeliveryRouter so the live send
@@ -1900,7 +1904,15 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                 delivery_errors.extend(target_errors)
                 continue
             # Standalone path: run the async send in a fresh event loop (safe from any thread)
-            coro = _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files)
+            coro = _send_to_platform(
+                platform,
+                pconfig,
+                chat_id,
+                cleaned_delivery_content,
+                thread_id=thread_id,
+                media_files=media_files,
+                inline_keyboard=inline_keyboard,
+            )
             try:
                 result = asyncio.run(coro)
             except RuntimeError as run_err:
@@ -1929,7 +1941,18 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                 try:
                     pool = concurrent.futures.ThreadPoolExecutor(max_workers=1)
                     try:
-                        future = pool.submit(asyncio.run, _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files))
+                        future = pool.submit(
+                            asyncio.run,
+                            _send_to_platform(
+                                platform,
+                                pconfig,
+                                chat_id,
+                                cleaned_delivery_content,
+                                thread_id=thread_id,
+                                media_files=media_files,
+                                inline_keyboard=inline_keyboard,
+                            ),
+                        )
                         result = future.result(timeout=30)
                     finally:
                         pool.shutdown(wait=False)

hermes_cli/cron.py

@@ -310,6 +310,7 @@ def cron_create(args):
         script=getattr(args, "script", None),
         workdir=getattr(args, "workdir", None),
         no_agent=getattr(args, "no_agent", False) or None,
+        buttons=getattr(args, "buttons", None),
     )
     if not result.get("success"):
         print(color(f"Failed to create job: {result.get('error', 'unknown error')}", Colors.RED))
@@ -373,6 +374,7 @@ def cron_edit(args):
         script=getattr(args, "script", None),
         workdir=getattr(args, "workdir", None),
         no_agent=getattr(args, "no_agent", None),
+        buttons=getattr(args, "buttons", None),
     )
     if not result.get("success"):
         print(color(f"Failed to update job: {result.get('error', 'unknown error')}", Colors.RED))

hermes_cli/plugins.py

@@ -171,6 +171,11 @@ def _install_plugin_debug_handler(force: bool = False) -> None:
     #   {"action": "allow"}  /  None             -> normal dispatch
     # Kwargs: event: MessageEvent, gateway: GatewayRunner, session_store.
     "pre_gateway_dispatch",
+    # Platform-native interaction hook. Fired for authorized callback actions
+    # that were not claimed by a built-in platform handler. The first plugin
+    # returning {"handled": True, ...} owns the action. Platform adapters pass
+    # normalized scalar context only; transport objects stay inside the host.
+    "platform_callback",
     # Approval lifecycle hooks. Fired by tools/approval.py when a dangerous
     # command needs an approval decision -- fires for CLI-interactive prompts,
     # gateway/ACP approvals, and smart-mode auxiliary-LLM decisions.

hermes_cli/subcommands/cron.py

@@ -66,6 +66,17 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None:
             "pattern (memory alerts, disk alerts, CI pings)."
         ),
     )
+    cron_create.add_argument(
+        "--button",
+        dest="buttons",
+        action="append",
+        metavar="LABEL=CALLBACK_DATA",
+        help=(
+            "Attach a Telegram inline action to the delivered message. "
+            "Repeat for multiple buttons. Callback data is handled by an "
+            "enabled platform_callback plugin."
+        ),
+    )
     cron_create.add_argument(
         "--workdir",
         help="Absolute path for the job to run from. Injects AGENTS.md / CLAUDE.md / .cursorrules from that directory and uses it as the cwd for terminal/file/code_exec tools. Omit to preserve old behaviour (no project context files).",
@@ -130,6 +141,13 @@ def build_cron_parser(subparsers, *, cmd_cron: Callable) -> None:
         const=False,
         help="Disable no-agent mode on this job (reverts to LLM-driven execution).",
     )
+    cron_edit.add_argument(
+        "--button",
+        dest="buttons",
+        action="append",
+        metavar="LABEL=CALLBACK_DATA",
+        help="Replace the job's Telegram inline actions. Repeat for multiple buttons.",
+    )
     cron_edit.add_argument(
         "--workdir",
         help="Absolute path for the job to run from (injects AGENTS.md etc. and sets terminal cwd). Pass empty string to clear.",

plugins/platforms/telegram/adapter.py

@@ -3881,14 +3881,26 @@ async def send(
         # Skip whitespace-only text to prevent Telegram 400 empty-text errors.
         if not content or not content.strip():
             return SendResult(success=True, message_id=None)
+
+        inline_keyboard = None
+        if metadata and "inline_keyboard" in metadata:
+            inline_keyboard = self._plugin_callback_keyboard(
+                metadata.get("inline_keyboard")
+            )
+            if inline_keyboard is None:
+                return SendResult(
+                    success=False,
+                    error="invalid inline_keyboard metadata",
+                    retryable=False,
+                )
 
         try:
             # Bot API 10.1 rich fast-path: send the raw agent markdown via
             # sendRichMessage so tables/task lists/etc. render natively. Falls
             # through to the legacy MarkdownV2 path on permanent/capability
             # errors or DM-topic routing skips; returns directly on success or
             # on a transient failure (which must NOT be legacy-resent).
-            if self._should_attempt_rich(content, metadata=metadata):
+            if inline_keyboard is None and self._should_attempt_rich(content, metadata=metadata):
                 rich_result = await self._try_send_rich(chat_id, content, reply_to, metadata)
                 if rich_result is not None:
                     if rich_result.success:
@@ -3942,6 +3954,11 @@ async def send(
                 _TimedOut = None  # type: ignore[assignment,misc]
 
             for i, chunk in enumerate(chunks):
+                action_kwargs = (
+                    {"reply_markup": inline_keyboard}
+                    if inline_keyboard is not None and i == len(chunks) - 1
+                    else {}
+                )
                 retried_thread_not_found = False
                 metadata_reply_to = self._metadata_reply_to_message_id(metadata)
                 private_dm_topic_send = self._is_private_dm_topic_send(chat_id, thread_id, metadata)
@@ -3997,6 +4014,7 @@ async def send(
                                 **thread_kwargs,
                                 **self._link_preview_kwargs(),
                                 **self._notification_kwargs(metadata),
+                                **action_kwargs,
                             )
                         except Exception as md_error:
                             # Markdown parsing failed, try plain text
@@ -4011,6 +4029,7 @@ async def send(
                                     **thread_kwargs,
                                     **self._link_preview_kwargs(),
                                     **self._notification_kwargs(metadata),
+                                    **action_kwargs,
                                 )
                             else:
                                 raise
@@ -5940,8 +5959,16 @@ async def _handle_callback_query(
                     )
             return
 
-        # --- Update prompt callbacks ---
+        # --- Plugin-owned platform callbacks ---
         if not data.startswith("update_prompt:"):
+            await self._handle_platform_callback_hook(
+                query,
+                data,
+                query_chat_id=query_chat_id,
+                query_chat_type=query_chat_type,
+                query_thread_id=query_thread_id,
+                query_user_name=query_user_name,
+            )
             return
         answer = data.split(":", 1)[1]  # "y" or "n"
         caller_id = str(getattr(query.from_user, "id", ""))
@@ -5978,6 +6005,117 @@ async def _handle_callback_query(
         except Exception as exc:
             logger.error("Failed to write update response from callback: %s", exc)
 
+    @staticmethod
+    def _plugin_callback_keyboard(rows: object):
+        """Build a bounded Telegram keyboard from normalized plugin output."""
+        if not isinstance(rows, list) or not 1 <= len(rows) <= 8:
+            return None
+        keyboard = []
+        for row in rows:
+            if not isinstance(row, list) or not 1 <= len(row) <= 8:
+                return None
+            built_row = []
+            for button in row:
+                if not isinstance(button, dict):
+                    return None
+                text = button.get("text")
+                callback_data = button.get("callback_data")
+                if not isinstance(text, str) or not text.strip() or len(text) > 64:
+                    return None
+                if (
+                    not isinstance(callback_data, str)
+                    or not callback_data
+                    or len(callback_data.encode("utf-8")) > 64
+                ):
+                    return None
+                built_row.append(
+                    InlineKeyboardButton(text=text, callback_data=callback_data)
+                )
+            keyboard.append(built_row)
+        return InlineKeyboardMarkup(keyboard)
+
+    async def _handle_platform_callback_hook(
+        self,
+        query,
+        data: str,
+        *,
+        query_chat_id,
+        query_chat_type,
+        query_thread_id,
+        query_user_name,
+    ) -> None:
+        """Offer an unclaimed, authorized callback to enabled plugins."""
+        caller_id = str(getattr(query.from_user, "id", ""))
+        chat_type_value = getattr(query_chat_type, "value", query_chat_type)
+        if not self._is_callback_user_authorized(
+            caller_id,
+            chat_id=query_chat_id,
+            chat_type=str(chat_type_value) if chat_type_value is not None else None,
+            thread_id=str(query_thread_id) if query_thread_id is not None else None,
+            user_name=query_user_name,
+        ):
+            await query.answer(text="⛔ You are not authorized to use this action.")
+            return
+
+        message = getattr(query, "message", None)
+        try:
+            from hermes_cli.plugins import invoke_hook
+
+            results = await asyncio.to_thread(
+                invoke_hook,
+                "platform_callback",
+                platform="telegram",
+                callback_data=data,
+                user_id=caller_id,
+                user_name=str(query_user_name or ""),
+                chat_id=str(query_chat_id) if query_chat_id is not None else None,
+                chat_type=str(chat_type_value) if chat_type_value is not None else None,
+                thread_id=str(query_thread_id) if query_thread_id is not None else None,
+                message_id=(
+                    str(getattr(message, "message_id"))
+                    if getattr(message, "message_id", None) is not None
+                    else None
+                ),
+                message_text=str(
+                    getattr(message, "text", None)
+                    or getattr(message, "caption", "")
+                    or ""
+                ),
+            )
+        except Exception:
+            logger.warning("[%s] platform callback hook dispatch failed", self.name, exc_info=True)
+            results = []
+
+        handled = next(
+            (
+                result
+                for result in results
+                if isinstance(result, dict) and result.get("handled") is True
+            ),
+            None,
+        )
+        if handled is None:
+            await query.answer(text="Action unavailable.")
+            return
+
+        answer = handled.get("answer")
+        if not isinstance(answer, str) or not answer.strip():
+            answer = "Done"
+        await query.answer(
+            text=answer.strip()[:200],
+            show_alert=bool(handled.get("show_alert", False)),
+        )
+
+        if "buttons" in handled:
+            reply_markup = self._plugin_callback_keyboard(handled.get("buttons"))
+            if reply_markup is None:
+                logger.warning("[%s] platform callback returned an invalid keyboard", self.name)
+                return
+            try:
+                await query.edit_message_reply_markup(reply_markup=reply_markup)
+            except Exception:
+                logger.debug("[%s] platform callback keyboard update failed", self.name, exc_info=True)
+
     # Maps `gt:<verb>` -> (script-name, extra-args, success-label, is_state).
     # Scripts live in ~/.hermes/scripts/gmail-triage/. `arg` from the callback
     # data is always passed as the first positional arg.

tests/cron/test_scheduler.py

@@ -641,6 +641,34 @@ def test_delivery_skips_wrapping_when_config_disabled(self):
         assert "Cronjob Response" not in sent_content
         assert "The agent cannot see" not in sent_content
 
+    def test_delivery_forwards_job_inline_keyboard(self):
+        """Cron keeps one delivery path while attaching configured actions."""
+        from gateway.config import Platform
+
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
+        keyboard = [[
+            {"text": "Relevant", "callback_data": "zbr:relevant"},
+            {"text": "Zu technisch", "callback_data": "zbr:technical"},
+        ]]
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("tools.send_message_tool._send_to_platform", new=AsyncMock(return_value={"success": True})) as send_mock, \
+             patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}):
+            _deliver_result(
+                {
+                    "id": "radar-job",
+                    "deliver": "origin",
+                    "origin": {"platform": "telegram", "chat_id": "123"},
+                    "inline_keyboard": keyboard,
+                },
+                "Daily Radar",
+            )
+
+        assert send_mock.await_args.kwargs["inline_keyboard"] == keyboard
+
     def test_delivery_extracts_media_tags_before_send(self, tmp_path, monkeypatch):
         """Cron delivery should pass MEDIA attachments separately to the send helper."""
         from gateway.config import Platform