"""M5: approval / clarify / choice-picker frames (interactive banners). Mixin for ``adapter.IrisAdapter``. Hermes detects these methods on the adapter type; each emits a high-priority ``notification`` (pushed even when a device is live) and, with a live device, an interactive ``picker.choice`` card whose selection runs the stored callback (``pickers.py``). """ import time from typing import Any from gateway.platforms.base import SendResult from . import protocol from .classify import ( _mint_message_id, _mint_picker_id, _push_preview, _thread_id_from_metadata, ) from .mixin_base import IrisAdapterBase from .pickers import ( _approval_picker_callback, _clarify_is_multi, _clarify_picker_callback, ) class PickerHandlers(IrisAdapterBase): """Interactive pickers + approvals (see module docstring).""" async def send_slash_confirm( # noqa: PLR0913 self, chat_id: str, title: str, message: str, session_key: str, confirm_id: str, metadata: dict[str, Any] | None = None, ) -> SendResult: """Banner + push for a slash-command approval prompt. The gateway's text fallback still renders the actionable prompt (the app has no inline buttons yet); the notification is the push-visible signal (high priority: pushed even when a device is live). """ thread_id = _thread_id_from_metadata(metadata) await self._broadcast_or_log( chat_id, protocol.notification( chat_id, protocol.NOTIF_APPROVAL, title or "Approval needed", _push_preview(message), thread_id=thread_id, ), ) return await super().send_slash_confirm( chat_id, title, message, session_key, confirm_id, metadata=metadata ) async def send_exec_approval( # noqa: PLR0913 self, chat_id: str, command: str, session_key: str, description: str = "dangerous command", metadata: dict[str, Any] | None = None, allow_permanent: bool = True, allow_session: bool = True, smart_denied: bool = False, ) -> SendResult: """Interactive exec-approval picker (buttons) for a dangerous command. Hermes calls this (detected on the adapter type) when the agent wants to run a command that needs approval; the agent thread blocks until the user decides. With a live device we render the same choice set as the native adapters (Allow Once / Session / Always / Deny, gated by the same flags) as a ``picker.choice`` card, reusing the clarify/slash picker mechanism. A tap resolves via ``resolve_gateway_approval`` (the same primitive the text ``/approve`` / ``/deny`` handlers use), unblocking the agent, and a short confirmation is delivered as a normal message. A high-priority ``approval`` notification is also emitted so a backgrounded device is woken (pushed even when live). With no live device the picker could never be answered, so report failure and let hermes fall back to the text ``/approve`` prompt. """ if not self._http_server.has_devices(): return SendResult(success=False, error="no live devices for approval picker") thread_id = _thread_id_from_metadata(metadata) # High-priority banner + push (wakes a backgrounded device). await self._broadcast_or_log( chat_id, protocol.notification( chat_id, protocol.NOTIF_APPROVAL, "Approval needed", _push_preview(description or command), thread_id=thread_id, ), ) # Choice set mirrors the native adapters (telegram/relay). frame_choices = [{"value": "once", "label": "\u2705 Allow Once", "is_current": False}] if not smart_denied and allow_session: frame_choices.append( {"value": "session", "label": "\u2705 Allow Session", "is_current": False} ) if allow_permanent: frame_choices.append( {"value": "always", "label": "\u2705 Always Allow", "is_current": False} ) frame_choices.append({"value": "deny", "label": "\u274c Deny", "is_current": False}) cmd_preview = command if len(command) <= 1500 else command[:1500] + "\u2026" title = ( "\u26a0\ufe0f **Command approval required**\n\n" f"```\n{cmd_preview}\n```\n\n" f"Reason: {description}" ) if smart_denied: title += "\n\n**Smart DENY:** owner override applies to this one operation only." picker_id = _mint_picker_id() self._pending_pickers[picker_id] = { "chat_id": chat_id, "thread_id": thread_id, "on_choice_selected": _approval_picker_callback(session_key), } await self._broadcast_or_log( chat_id, protocol.picker_choice(picker_id, title, frame_choices, chat_id, thread_id=thread_id), ) return SendResult(success=True, message_id=picker_id) async def send_choice_picker( # noqa: PLR0913 self, chat_id: str, title: str, choices: list, session_key: str, on_choice_selected, metadata: dict[str, Any] | None = None, ) -> SendResult: """Send an interactive choice picker (one tap → one value). The generic companion to Telegram's inline-keyboard pickers, used by ``/reasoning``, ``/fast``, and any future finite-choice slash command (hermes detects this method on the adapter type). Emits a ``picker.choice`` frame; the app answers with ``picker.select``, which runs ``on_choice_selected(chat_id, value)`` and delivers the returned text as a normal message. Outboxed, so a reconnecting device re-renders a still-pending picker. With no live device the picker could never be answered, so report failure and let hermes fall back to the text status card. """ if not self._http_server.has_devices(): return SendResult(success=False, error="no live devices for picker") thread_id = _thread_id_from_metadata(metadata) picker_id = _mint_picker_id() self._pending_pickers[picker_id] = { "chat_id": chat_id, "thread_id": thread_id, "on_choice_selected": on_choice_selected, } await self._broadcast_or_log( chat_id, protocol.picker_choice(picker_id, title, choices, chat_id, thread_id=thread_id), ) return SendResult(success=True, message_id=picker_id) async def send_clarify( # noqa: PLR0913 self, chat_id: str, question: str, choices: list | None, clarify_id: str, session_key: str, metadata: dict[str, Any] | None = None, ) -> SendResult: """Banner + push for a clarify prompt. Single-select clarifies with a live device render as an interactive ``picker.choice`` card (one tap per option + an "Other" free-text button), reusing the slash-command picker mechanism. A real pick resolves via ``resolve_gateway_clarify`` (the agent then continues and replies); "Other" flips the entry to text-capture. Multi-select, open-ended, and no-live-device clarifies fall back to a numbered text list whose reply the gateway's text-intercept captures via ``mark_awaiting_text``. """ thread_id = _thread_id_from_metadata(metadata) await self._broadcast_or_log( chat_id, protocol.notification( chat_id, protocol.NOTIF_CLARIFY, "Question", _push_preview(question), thread_id=thread_id, ), ) # Single-select + live device → interactive picker card. if choices and not _clarify_is_multi(clarify_id) and self._http_server.has_devices(): picker_id = _mint_picker_id() self._pending_pickers[picker_id] = { "chat_id": chat_id, "thread_id": thread_id, "on_choice_selected": _clarify_picker_callback( clarify_id, [str(c) for c in choices] ), } frame_choices = [ {"value": f"c{i}", "label": str(c)[:75], "is_current": False} for i, c in enumerate(choices) ] frame_choices.append( {"value": "other", "label": "✏️ Other (type your answer)", "is_current": False} ) await self._broadcast_or_log( chat_id, protocol.picker_choice( picker_id, f"❓ {question}", frame_choices, chat_id, thread_id=thread_id ), ) return SendResult(success=True, message_id=picker_id) # Text fallback (multi-select / open-ended / no live device). if choices: lines = [f"❓ {question}", ""] for i, choice in enumerate(choices, start=1): lines.append(f" {i}. {choice}") lines.append("") if _clarify_is_multi(clarify_id): lines.append( "Multiple selections allowed — reply with the numbers " 'separated by commas or spaces (e.g. "1, 3"), the option ' "text, or your own answer." ) else: lines.append("Reply with the number, the option text, or your own answer.") text = "\n".join(lines) # Text fallback: enable text-capture so the gateway intercept # picks up the user's typed reply (e.g. "2" or choice text). from tools.clarify_gateway import mark_awaiting_text mark_awaiting_text(clarify_id) else: text = f"❓ {question}" message_id = _mint_message_id() await self._broadcast_or_log( chat_id, protocol.message( chat_id=chat_id, message_id=message_id, role=protocol.ROLE_ASSISTANT, text=text, thread_id=thread_id, ts=int(time.time() * 1000), ), ) return SendResult(success=True, message_id=message_id)