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.
This commit is contained in:
teknium1 2026-07-29 09:49:01 -07:00 committed by Teknium
parent 4fe5410d57
commit 1f45ff9e8a
5 changed files with 341 additions and 36 deletions

View file

@ -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 `` (NM 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,

View file

@ -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",

View file

@ -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)

View file

@ -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

View file

@ -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": " (1120 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"] == " (2125 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"] == " (1111 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