From 1f45ff9e8ae98c32f2318612d48143f651289a3c Mon Sep 17 00:00:00 2001
From: teknium1 <127238744+teknium1@users.noreply.github.com>
Date: Wed, 29 Jul 2026 09:49:01 -0700
Subject: [PATCH] refactor(gateway): shared exec-approval/picker formatting
 cores in base adapter

- base._format_exec_approval(command, description, smart_denied): shared
  header/fence/reason/smart-deny assembly driven by _EA_* template attrs and
  an _ea_escape() hook; base._format_choice_page(options, page, per_page):
  shared pagination core returning (page_options, meta) incl. the
  ' (N-M of T)' page_info suffix; base._truncate_preview: the shared
  truncate-with-ellipsis idiom.
- telegram (HTML attrs + _html.escape hook), feishu (card markdown attrs),
  matrix (head-only; local reaction-legend tail) rewired; telegram's
  provider/model keyboard pagination and slash-confirm preview use the
  shared cores. All user-visible strings byte-identical (parity-tested).
- slack/discord/teams left untouched: their formatting interleaves
  platform-specific budget arithmetic (Slack 3000-char section budget
  subtraction, Discord mention-prefix + dual content/embed budgets, Teams
  adaptive-card blocks) beyond template params.
- tests/gateway/test_interactive_prompt_base.py covers the cores + parity.
---
 gateway/platforms/base.py                     |  85 +++++++
 plugins/platforms/feishu/adapter.py           |  11 +-
 plugins/platforms/matrix/adapter.py           |  11 +-
 plugins/platforms/telegram/adapter.py         |  54 ++---
 tests/gateway/test_interactive_prompt_base.py | 216 ++++++++++++++++++
 5 files changed, 341 insertions(+), 36 deletions(-)
 create mode 100644 tests/gateway/test_interactive_prompt_base.py

diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py
index 3963cef866d..a2e00343d3f 100644
--- a/gateway/platforms/base.py
+++ b/gateway/platforms/base.py
@@ -3622,6 +3622,91 @@ class BasePlatformAdapter(ABC):
             # about it never being awaited, then drop silently.
             coro.close()
 
+    # ── Shared interactive-prompt formatting cores ─────────────────────────
+    # Template attrs for ``_format_exec_approval``. Adapters override these to
+    # keep their historical, platform-specific wording byte-identical while
+    # sharing the assembly logic (header → fenced command preview → reason →
+    # optional smart-deny note).
+    _EA_HEADER: str = "⚠️ Command Approval Required\n\n"
+    _EA_CODE_OPEN: str = "```\n"
+    _EA_CODE_CLOSE: str = "\n```\n"
+    _EA_REASON_LABEL: str = "Reason: "
+    _EA_SMART_DENY_LINE: str = (
+        "\n\nSmart DENY: owner override applies to this one operation only."
+    )
+    _EA_CMD_BUDGET: int = 3000
+
+    @staticmethod
+    def _truncate_preview(text: str, budget: int, suffix: str = "...") -> str:
+        """Truncate ``text`` to ``budget`` chars, appending ``suffix`` when cut.
+
+        The shared ``x[:budget] + "..." if len(x) > budget else x`` idiom used
+        by every adapter's approval/confirm preview construction.
+        """
+        text = str(text or "")
+        return text[:budget] + suffix if len(text) > budget else text
+
+    def _ea_escape(self, text: str) -> str:
+        """Escape hook applied to the command preview and reason text.
+
+        Default is pass-through; HTML-mode platforms (Telegram) override.
+        """
+        return text
+
+    def _format_exec_approval(
+        self,
+        command: str,
+        description: str = "dangerous command",
+        smart_denied: bool = False,
+    ) -> str:
+        """Shared formatting core for exec-approval prompt text.
+
+        Assembles ``_EA_HEADER`` + fenced command preview (truncated to
+        ``_EA_CMD_BUDGET``) + ``_EA_REASON_LABEL`` + description, plus
+        ``_EA_SMART_DENY_LINE`` when ``smart_denied``. Button construction
+        stays platform-local; adapters with additional trailing instructions
+        (e.g. reaction legends) append them to this core.
+        """
+        cmd_preview = self._truncate_preview(str(command or ""), self._EA_CMD_BUDGET)
+        text = (
+            f"{self._EA_HEADER}"
+            f"{self._EA_CODE_OPEN}{self._ea_escape(cmd_preview)}{self._EA_CODE_CLOSE}"
+            f"{self._EA_REASON_LABEL}{self._ea_escape(description)}"
+        )
+        if smart_denied:
+            text += self._EA_SMART_DENY_LINE
+        return text
+
+    @staticmethod
+    def _format_choice_page(
+        options: list,
+        page: int,
+        per_page: int,
+    ) -> "tuple[list, Dict[str, Any]]":
+        """Shared pagination core for picker keyboards/menus.
+
+        Clamps ``page`` into range, slices ``options`` for that page and
+        returns ``(page_options, meta)`` where ``meta`` carries ``page``,
+        ``total_pages``, ``start``, ``end``, ``total`` and ``page_info`` —
+        the `` (N–M of T)`` suffix text (empty when everything fits on one
+        page). Option/button rendering stays platform-local.
+        """
+        total = len(options)
+        total_pages = max(1, (total + per_page - 1) // per_page)
+        page = max(0, min(page, total_pages - 1))
+        start = page * per_page
+        end = min(start + per_page, total)
+        page_info = f" ({start + 1}–{end} of {total})" if total_pages > 1 else ""
+        meta: Dict[str, Any] = {
+            "page": page,
+            "total_pages": total_pages,
+            "start": start,
+            "end": end,
+            "total": total,
+            "page_info": page_info,
+        }
+        return options[start:end], meta
+
     async def send_slash_confirm(
         self,
         chat_id: str,
diff --git a/plugins/platforms/feishu/adapter.py b/plugins/platforms/feishu/adapter.py
index f3756a8589b..d86dc597a81 100644
--- a/plugins/platforms/feishu/adapter.py
+++ b/plugins/platforms/feishu/adapter.py
@@ -1999,6 +1999,13 @@ class FeishuAdapter(BasePlatformAdapter):
             logger.error("[Feishu] Failed to edit message %s: %s", message_id, exc, exc_info=True)
             return SendResult(success=False, error=str(exc))
 
+    # Template attrs for the shared _format_exec_approval core. The card
+    # header carries the title, so the text core starts at the code fence.
+    _EA_HEADER = ""
+    _EA_REASON_LABEL = "**Reason:** "
+    _EA_SMART_DENY_LINE = "\n\n**Smart DENY:** owner override applies to this one operation only."
+    _EA_CMD_BUDGET = 3000
+
     async def send_exec_approval(
         self, chat_id: str, command: str, session_key: str,
         description: str = "dangerous command",
@@ -2018,7 +2025,6 @@ class FeishuAdapter(BasePlatformAdapter):
 
         try:
             approval_id = next(self._approval_counter)
-            cmd_preview = command[:3000] + "..." if len(command) > 3000 else command
 
             def _btn(label: str, action_name: str, btn_type: str = "default") -> dict:
                 return {
@@ -2034,7 +2040,6 @@ class FeishuAdapter(BasePlatformAdapter):
                 if allow_permanent:
                     actions.append(_btn("✅ Always", "approve_always"))
             actions.append(_btn("❌ Deny", "deny", "danger"))
-            scope_note = "\n\n**Smart DENY:** owner override applies to this one operation only." if smart_denied else ""
             card = {
                 "config": {"wide_screen_mode": True},
                 "header": {
@@ -2044,7 +2049,7 @@ class FeishuAdapter(BasePlatformAdapter):
                 "elements": [
                     {
                         "tag": "markdown",
-                        "content": f"```\n{cmd_preview}\n```\n**Reason:** {description}{scope_note}",
+                        "content": self._format_exec_approval(command, description, smart_denied),
                     },
                     {
                         "tag": "action",
diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py
index c6b2dcdd615..8ed47897e0f 100644
--- a/plugins/platforms/matrix/adapter.py
+++ b/plugins/platforms/matrix/adapter.py
@@ -2225,6 +2225,12 @@ class MatrixAdapter(BasePlatformAdapter):
             chat_id, video_path, "m.video", caption, reply_to, metadata=metadata
         )
 
+    # Template attrs for the shared _format_exec_approval core. Matrix keeps
+    # the smart-deny/scope wording in its local tail (reaction legend), so the
+    # core is used for the header + fence + reason head only.
+    _EA_HEADER = "⚠️ **Dangerous command requires approval**\n"
+    _EA_CMD_BUDGET = 2000
+
     async def send_exec_approval(
         self,
         chat_id: str,
@@ -2241,7 +2247,6 @@ class MatrixAdapter(BasePlatformAdapter):
             return SendResult(success=False, error="Not connected")
 
         requester_user_id = str((metadata or {}).get("requester_user_id") or "") or None
-        cmd_preview = command[:2000] + "..." if len(command) > 2000 else command
         scope_choices = ""
         if smart_denied:
             scope_choices = "Smart DENY: owner override applies to this one operation only.\n"
@@ -2258,9 +2263,7 @@ class MatrixAdapter(BasePlatformAdapter):
                 reaction_legend_parts.append("♾️ = approve always")
         reaction_legend_parts.append("❎ = deny")
         text = (
-            "⚠️ **Dangerous command requires approval**\n"
-            f"```\n{cmd_preview}\n```\n"
-            f"Reason: {description}\n\n"
+            f"{self._format_exec_approval(command, description)}\n\n"
             f"{scope_choices}Reply `!approve` to execute once, or `!deny` to cancel.\n\n"
             "You can also click the reaction to approve:\n"
             + "\n".join(reaction_legend_parts)
diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py
index 3f8538d8832..686519c45da 100644
--- a/plugins/platforms/telegram/adapter.py
+++ b/plugins/platforms/telegram/adapter.py
@@ -5318,6 +5318,16 @@ class TelegramAdapter(BasePlatformAdapter):
             logger.warning("[%s] send_update_prompt failed: %s", self.name, _redact_telegram_error_text(e))
             return SendResult(success=False, error=_redact_telegram_error_text(e))
 
+    # Template attrs for the shared _format_exec_approval core (HTML mode).
+    _EA_HEADER = "⚠️ <b>Command Approval Required</b>\n\n"
+    _EA_CODE_OPEN = "<pre>"
+    _EA_CODE_CLOSE = "</pre>\n\n"
+    _EA_SMART_DENY_LINE = "\n\n<b>Smart DENY:</b> owner override applies to this one operation only."
+    _EA_CMD_BUDGET = 3800
+
+    def _ea_escape(self, text: str) -> str:
+        return _html.escape(text)
+
     async def send_exec_approval(
         self, chat_id: str, command: str, session_key: str,
         description: str = "dangerous command",
@@ -5335,14 +5345,7 @@ class TelegramAdapter(BasePlatformAdapter):
             return SendResult(success=False, error="Not connected")
 
         try:
-            cmd_preview = command[:3800] + "..." if len(command) > 3800 else command
-            text = (
-                f"⚠️ <b>Command Approval Required</b>\n\n"
-                f"<pre>{_html.escape(cmd_preview)}</pre>\n\n"
-                f"Reason: {_html.escape(description)}"
-            )
-            if smart_denied:
-                text += "\n\n<b>Smart DENY:</b> owner override applies to this one operation only."
+            text = self._format_exec_approval(command, description, smart_denied)
 
             # Resolve thread context for thread replies
             thread_id = self._metadata_thread_id(metadata)
@@ -5410,7 +5413,7 @@ class TelegramAdapter(BasePlatformAdapter):
             return SendResult(success=False, error="Not connected")
 
         try:
-            preview = self.format_message(message if len(message) <= 3800 else message[:3800] + "...")
+            preview = self.format_message(self._truncate_preview(message, 3800))
 
             keyboard = InlineKeyboardMarkup([
                 [
@@ -5774,14 +5777,11 @@ class TelegramAdapter(BasePlatformAdapter):
             for p in providers:
                 buttons.append(_provider_button(p))
 
-        page_size = self._PROVIDER_PAGE_SIZE
-        total = len(buttons)
-        total_pages = max(1, (total + page_size - 1) // page_size)
-        page = max(0, min(page, total_pages - 1))
-
-        start = page * page_size
-        end = min(start + page_size, total)
-        page_buttons = buttons[start:end]
+        page_buttons, page_meta = self._format_choice_page(
+            buttons, page, self._PROVIDER_PAGE_SIZE
+        )
+        page = page_meta["page"]
+        total_pages = page_meta["total_pages"]
 
         rows = [page_buttons[i : i + 2] for i in range(0, len(page_buttons), 2)]
 
@@ -5796,19 +5796,16 @@ class TelegramAdapter(BasePlatformAdapter):
 
         rows.append([InlineKeyboardButton("✗ Cancel", callback_data="mx")])
 
-        page_info = f" ({start + 1}–{end} of {total})" if total_pages > 1 else ""
-        return InlineKeyboardMarkup(rows), page_info
+        return InlineKeyboardMarkup(rows), page_meta["page_info"]
 
     def _build_model_keyboard(self, models: list, page: int) -> tuple:
         """Build paginated model buttons. Returns (keyboard, page_info_text)."""
-        page_size = self._MODEL_PAGE_SIZE
-        total = len(models)
-        total_pages = max(1, (total + page_size - 1) // page_size)
-        page = max(0, min(page, total_pages - 1))
-
-        start = page * page_size
-        end = min(start + page_size, total)
-        page_models = models[start:end]
+        page_models, page_meta = self._format_choice_page(
+            models, page, self._MODEL_PAGE_SIZE
+        )
+        page = page_meta["page"]
+        total_pages = page_meta["total_pages"]
+        start = page_meta["start"]
 
         buttons: list = []
         for i, model_id in enumerate(page_models):
@@ -5837,8 +5834,7 @@ class TelegramAdapter(BasePlatformAdapter):
             InlineKeyboardButton("✗ Cancel", callback_data="mx"),
         ])
 
-        page_info = f" ({start + 1}–{end} of {total})" if total_pages > 1 else ""
-        return InlineKeyboardMarkup(rows), page_info
+        return InlineKeyboardMarkup(rows), page_meta["page_info"]
 
     async def _handle_model_picker_callback(
         self, query, data: str, chat_id: str
diff --git a/tests/gateway/test_interactive_prompt_base.py b/tests/gateway/test_interactive_prompt_base.py
new file mode 100644
index 00000000000..86f0726d7a4
--- /dev/null
+++ b/tests/gateway/test_interactive_prompt_base.py
@@ -0,0 +1,216 @@
+"""Tests for the shared interactive-prompt formatting cores in BasePlatformAdapter.
+
+Covers ``_format_exec_approval`` (template-attr driven exec-approval text),
+``_format_choice_page`` (picker pagination core), ``_truncate_preview``, and
+byte-parity of the rewired adapters (telegram/feishu/matrix) against their
+historical inline formatting.
+"""
+
+import html as _html
+
+from gateway.platforms.base import BasePlatformAdapter
+
+
+def _bare(cls):
+    """Bare instance without running __init__ (documented test pattern)."""
+    return object.__new__(cls)
+
+
+class _DefaultAdapter(BasePlatformAdapter):
+    """Concrete subclass using only base-class template attrs."""
+
+    async def connect(self):  # pragma: no cover - not used
+        pass
+
+    async def disconnect(self):  # pragma: no cover - not used
+        pass
+
+    async def get_chat_info(self, chat_id):  # pragma: no cover - not used
+        return {}
+
+    async def send(self, *a, **k):  # pragma: no cover - not used
+        raise NotImplementedError
+
+
+class TestTruncatePreview:
+    def test_short_text_unchanged(self):
+        assert BasePlatformAdapter._truncate_preview("abc", 10) == "abc"
+
+    def test_exact_budget_unchanged(self):
+        assert BasePlatformAdapter._truncate_preview("x" * 10, 10) == "x" * 10
+
+    def test_over_budget_truncates_with_suffix(self):
+        out = BasePlatformAdapter._truncate_preview("x" * 11, 10)
+        assert out == "x" * 10 + "..."
+
+    def test_none_coerced_to_empty(self):
+        assert BasePlatformAdapter._truncate_preview(None, 10) == ""
+
+    def test_custom_suffix(self):
+        out = BasePlatformAdapter._truncate_preview("abcdef", 3, suffix="!")
+        assert out == "abc!"
+
+
+class TestFormatExecApproval:
+    def test_default_template(self):
+        ad = _bare(_DefaultAdapter)
+        text = ad._format_exec_approval("rm -rf /", "scary")
+        assert text == (
+            "⚠️ Command Approval Required\n\n"
+            "```\nrm -rf /\n```\n"
+            "Reason: scary"
+        )
+
+    def test_smart_denied_appends_line(self):
+        ad = _bare(_DefaultAdapter)
+        text = ad._format_exec_approval("ls", "d", smart_denied=True)
+        assert text.endswith(
+            "\n\nSmart DENY: owner override applies to this one operation only."
+        )
+
+    def test_command_truncated_to_budget(self):
+        ad = _bare(_DefaultAdapter)
+        text = ad._format_exec_approval("x" * 5000, "d")
+        assert "x" * 3000 + "..." in text
+        assert "x" * 3001 not in text
+
+    def test_escape_hook_applied_to_command_and_reason(self):
+        class Escaping(_DefaultAdapter):
+            def _ea_escape(self, text: str) -> str:
+                return _html.escape(text)
+
+        ad = _bare(Escaping)
+        text = ad._format_exec_approval("echo <hi>", "a & b")
+        assert "echo &lt;hi&gt;" in text
+        assert "a &amp; b" in text
+
+    def test_empty_command(self):
+        ad = _bare(_DefaultAdapter)
+        text = ad._format_exec_approval("", "d")
+        assert "```\n\n```" in text
+
+
+class TestFormatChoicePage:
+    def test_single_page_no_page_info(self):
+        opts, meta = BasePlatformAdapter._format_choice_page([1, 2, 3], 0, 10)
+        assert opts == [1, 2, 3]
+        assert meta["page_info"] == ""
+        assert meta["total_pages"] == 1
+        assert meta["page"] == 0
+
+    def test_multi_page_slicing_and_info(self):
+        options = list(range(25))
+        opts, meta = BasePlatformAdapter._format_choice_page(options, 1, 10)
+        assert opts == list(range(10, 20))
+        assert meta == {
+            "page": 1,
+            "total_pages": 3,
+            "start": 10,
+            "end": 20,
+            "total": 25,
+            "page_info": " (11–20 of 25)",
+        }
+
+    def test_page_clamped_high(self):
+        opts, meta = BasePlatformAdapter._format_choice_page(list(range(25)), 99, 10)
+        assert meta["page"] == 2
+        assert opts == list(range(20, 25))
+        assert meta["page_info"] == " (21–25 of 25)"
+
+    def test_page_clamped_negative(self):
+        opts, meta = BasePlatformAdapter._format_choice_page(list(range(25)), -5, 10)
+        assert meta["page"] == 0
+        assert opts == list(range(10))
+
+    def test_empty_options(self):
+        opts, meta = BasePlatformAdapter._format_choice_page([], 0, 10)
+        assert opts == []
+        assert meta["total_pages"] == 1
+        assert meta["page_info"] == ""
+
+    def test_last_partial_page(self):
+        opts, meta = BasePlatformAdapter._format_choice_page(list(range(11)), 1, 10)
+        assert opts == [10]
+        assert meta["page_info"] == " (11–11 of 11)"
+
+
+class TestAdapterParity:
+    """Rewired adapters produce byte-identical text vs their historical inline code."""
+
+    def test_telegram_parity(self):
+        from plugins.platforms.telegram.adapter import TelegramAdapter
+
+        def old(command, description, smart_denied):
+            cmd_preview = command[:3800] + "..." if len(command) > 3800 else command
+            text = (
+                f"⚠️ <b>Command Approval Required</b>\n\n"
+                f"<pre>{_html.escape(cmd_preview)}</pre>\n\n"
+                f"Reason: {_html.escape(description)}"
+            )
+            if smart_denied:
+                text += "\n\n<b>Smart DENY:</b> owner override applies to this one operation only."
+            return text
+
+        ad = _bare(TelegramAdapter)
+        for cmd in ["rm -rf /", "x" * 5000, "echo <hi> & 'stuff'", ""]:
+            for sd in (False, True):
+                assert ad._format_exec_approval(cmd, "why <b>&</b>", sd) == old(
+                    cmd, "why <b>&</b>", sd
+                )
+
+    def test_feishu_parity(self):
+        from plugins.platforms.feishu.adapter import FeishuAdapter
+
+        def old(command, description, smart_denied):
+            cmd_preview = command[:3000] + "..." if len(command) > 3000 else command
+            scope_note = (
+                "\n\n**Smart DENY:** owner override applies to this one operation only."
+                if smart_denied
+                else ""
+            )
+            return f"```\n{cmd_preview}\n```\n**Reason:** {description}{scope_note}"
+
+        ad = _bare(FeishuAdapter)
+        for cmd in ["rm -rf /", "x" * 5000, ""]:
+            for sd in (False, True):
+                assert ad._format_exec_approval(cmd, "reason", sd) == old(cmd, "reason", sd)
+
+    def test_matrix_parity(self):
+        from plugins.platforms.matrix.adapter import MatrixAdapter
+
+        def old_head(command, description):
+            cmd_preview = command[:2000] + "..." if len(command) > 2000 else command
+            return (
+                "⚠️ **Dangerous command requires approval**\n"
+                f"```\n{cmd_preview}\n```\n"
+                f"Reason: {description}"
+            )
+
+        ad = _bare(MatrixAdapter)
+        for cmd in ["rm -rf /", "x" * 5000, ""]:
+            assert ad._format_exec_approval(cmd, "reason") == old_head(cmd, "reason")
+
+    def test_telegram_pagination_parity(self):
+        """_format_choice_page matches the old _build_*_keyboard arithmetic."""
+
+        def old(options, page, page_size):
+            total = len(options)
+            total_pages = max(1, (total + page_size - 1) // page_size)
+            page = max(0, min(page, total_pages - 1))
+            start = page * page_size
+            end = min(start + page_size, total)
+            page_info = f" ({start + 1}–{end} of {total})" if total_pages > 1 else ""
+            return options[start:end], page, total_pages, page_info
+
+        for n in (0, 1, 8, 9, 10, 25):
+            options = list(range(n))
+            for page in (-3, 0, 1, 2, 99):
+                for per in (8, 10):
+                    o_opts, o_page, o_tp, o_info = old(options, page, per)
+                    n_opts, meta = BasePlatformAdapter._format_choice_page(
+                        options, page, per
+                    )
+                    assert n_opts == o_opts
+                    assert meta["page"] == o_page
+                    assert meta["total_pages"] == o_tp
+                    assert meta["page_info"] == o_info