diff --git a/docs/relay-connector-contract.md b/docs/relay-connector-contract.md index 5e5344288bfd..499e23756b3b 100644 --- a/docs/relay-connector-contract.md +++ b/docs/relay-connector-contract.md @@ -395,6 +395,8 @@ The gateway calls the transport with action dicts. Source of truth: | `typing` | `chat_id`, `content?`, `metadata?` | `{success: bool}` | | `follow_up` | `session_key`, `kind`, `content`, `metadata?` | `{success: bool, message_id?, error?}` | | `send_media` | `chat_id`, `media_kind`, `source_url`, `content?` (caption), `filename?`, `reply_to?`, `metadata?` | `{success: bool, message_id?, error?}` | +| `prompt` | `chat_id`, `prompt_kind`, `prompt_id`, `content` (the question), `options[]{id,label,style?}`, `timeout_s?`, `reply_to?`, `metadata?` | `{success: bool, message_id?, error?}` | +| `react` | `chat_id`, `message_id`, `emoji`, `remove?`, `metadata?` | `{success: bool, error?}` | `get_chat_info(chat_id)` is a separate proxied call returning at least `{name, type}`. @@ -430,6 +432,52 @@ receipt, not lazily. A parallel `media` array (same order) adds `kind`, `mime`, `size`, `filename`, `caption` metadata; `message_type` reflects the first attachment's kind (`image`/`audio`/`document`). +**`prompt` (Phase 3 interactive).** One platform-abstract op renders the +gateway's highest-stakes interactions (exec approvals, slash confirms, +clarify pickers) with NATIVE controls: Discord button components, Telegram +inline keyboards, Slack Block Kit actions, WhatsApp button messages (≤3 +options) / list messages (4–10; >10 degrades to the numbered-text fallback). +`prompt_kind` (`approval`/`clarify`/`choice`) is a styling hint only. +`prompt_id` is gateway-minted (8 hex) and opaque to the connector; each +option's callback payload carries the token `hp1::` +(≤64 bytes — Telegram's `callback_data` cap binds every lane; option ids are +`[A-Za-z0-9_.-]`, ≤32 chars). `style` maps per-platform +(primary/success/danger/secondary). `timeout_s` is advisory on the wire — +expiry is enforced GATEWAY-side (the pending-prompt registry drops expired +entries; a stale press falls through as typed text, mirroring the native +adapters' "approval expired" edit). + +**`prompt_response` (Phase 3 inbound).** The user's press crosses back as a +normal inbound MessageEvent carrying +`prompt_response: {prompt_id, option_id, label?, prompt_message_id?}` — never +a bare platform `custom_id`. The event's `text` mirrors `/{option_id}` with +`message_type: "command"` so a gateway predating the field routes the press +as a typed reply instead of dropping it. The SOURCE is the authentic +CLICKING user (connector-observed: Telegram `callback_query.from`, Slack +`block_actions.user`, WhatsApp `messages[].from`, Discord interaction +member/user), so gateway-side authorization gates apply to a button press +exactly as to a typed `/approve`. Ingest lanes: Telegram `callback_query` +(polled, `allowed_updates` widened; best-effort `answerCallbackQuery` +spinner-stop), Slack `POST /slack/interactions` (raw-bytes HMAC + replay +window, same posture as `/slack/commands`), WhatsApp interactive +`button_reply`/`list_reply` (webhook normalize arm), Discord type-3 +component interactions (passthrough §5.1 sanitized forward; the type-3 edge +ack is `DEFERRED_UPDATE` so no visible "thinking…" reply). Foreign +callback payloads (another integration's buttons) never become prompt +events: Telegram/Slack/WhatsApp drop them at the connector; Discord type-3 +forwards keep the legacy custom_id-as-text shape. + +**`react` (Phase 3 ack lifecycle).** Adds/removes the bot's own `emoji` +reaction on `message_id` — restoring the native adapters' 👀→✅/❌ +processing-lifecycle acks over the relay. Unicode emoji on the wire; the +Slack sender maps to Slack's name vocabulary (`eyes`, `white_check_mark`, …) +and treats `already_reacted`/`no_reaction` as success (idempotent). Telegram +uses `setMessageReaction` (empty set = remove; Telegram's curated-emoji +restriction can reject glyphs — the failure is structured and the gateway +treats reactions as cosmetic). WhatsApp sends a reaction message (empty +emoji = remove). Reactions are best-effort by contract: a `react` failure +must never fail a turn. + **`typing` `content?` (Slack status clear).** A `typing` frame normally omits `content` — the connector renders its platform's active indicator ("is typing…" Assistant status on Slack, one-shot typing elsewhere). An **empty diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index e5a5da577672..4f2c0e98fa4f 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -1807,6 +1807,16 @@ class MessageEvent: reply_to_author_id: Optional[str] = None reply_to_author_name: Optional[str] = None reply_to_is_own_message: bool = False # True when the user replied to this bot/assistant's message + + # Structured interactive-prompt reply (relay Phase 3). Present when this + # event is the user answering a native interactive prompt rendered by the + # relay connector (Discord component / Telegram inline keyboard / Slack + # Block Kit / WhatsApp button-list). Shape mirrors the wire contract: + # {prompt_id, option_id, label?, prompt_message_id?}. The RelayAdapter + # consumes it in _on_inbound (routing to the approval/slash-confirm/ + # clarify resolvers) BEFORE normal dispatch; native adapters never set it + # (their button callbacks resolve in-process). + prompt_response: Optional[Dict[str, Any]] = None # Auto-loaded skill(s) for topic/channel bindings (e.g., Telegram DM Topics, # Discord channel_skill_bindings). A single name or ordered list. diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py index 97f98da13727..31ff41d67d93 100644 --- a/gateway/relay/adapter.py +++ b/gateway/relay/adapter.py @@ -95,6 +95,14 @@ class RelayAdapter(BasePlatformAdapter): # None when either is absent (media lanes then degrade to the # pre-media text fallbacks). self._media_client: Optional["RelayMediaClient"] = None + # Phase 3 interactive: prompt_id -> pending-prompt state. A `prompt` + # op renders native options; the user's pick comes back inbound as a + # prompt_response naming this id. The registry maps it back to the + # waiting primitive (approval / slash-confirm / clarify) so the click + # resolves EXACTLY like the native adapters' button callbacks. + # Entries expire lazily (see _pop_prompt) so an unanswered prompt + # never leaks. Keyed by our own minted 8-hex ids. + self._pending_prompts: Dict[str, Dict[str, Any]] = {} # ── capability surface (from descriptor) ───────────────────────────── @property @@ -235,6 +243,12 @@ class RelayAdapter(BasePlatformAdapter): async def _on_inbound(self, event) -> None: """Bridge a connector-delivered MessageEvent into the normal adapter path.""" self._capture_scope(event) + # Phase 3: a structured prompt answer resolves its waiting primitive + # (approval/confirm/clarify) and is CONSUMED — it must not also + # dispatch as a chat message. Unknown/expired prompt ids fall through + # (the command-shaped text then behaves like a typed reply). + if await self._consume_prompt_response(event): + return await self._localize_inbound_media(event) await self.handle_message(event) @@ -432,6 +446,11 @@ class RelayAdapter(BasePlatformAdapter): event = self._discord_interaction_to_event(forward) if event is not None: self._capture_scope(event) + # Phase 3: a component press carrying a Hermes prompt token + # resolves its waiting primitive and is consumed (same + # gate as _on_inbound's prompt_response arm). + if await self._consume_prompt_response(event): + return await self.handle_message(event) return logger.info( @@ -499,7 +518,49 @@ class RelayAdapter(BasePlatformAdapter): scope_id=str(guild_id) if guild_id else None, # Discord guild → generic scope slot message_id=str(payload.get("id")) if payload.get("id") else None, ) - return MessageEvent(text=text, message_type=message_type, source=source) + event = MessageEvent(text=text, message_type=message_type, source=source) + if itype == 3: + # Phase 3: a component press whose custom_id is a Hermes prompt + # token (hp1::) becomes a STRUCTURED prompt + # answer — _on_inbound's _consume_prompt_response then resolves + # the waiting approval/confirm/clarify, replacing the bare- + # custom_id-as-text stub. Foreign custom_ids keep the legacy + # best-effort TEXT shape. + decoded = self._decode_prompt_token(text) + if decoded: + prompt_id, option_id = decoded + msg = payload.get("message") or {} + prompt_message_id = ( + str(msg.get("id")) if isinstance(msg, dict) and msg.get("id") else None + ) + event.prompt_response = { + "prompt_id": prompt_id, + "option_id": option_id, + "prompt_message_id": prompt_message_id, + } + event.text = f"/{option_id}" + event.message_type = MessageType.COMMAND + return event + + @staticmethod + def _decode_prompt_token(token: str): + """Decode an hp1:: callback token, or None. + + Mirrors the connector's promptCodec.decodePromptCallback (the token + alphabet is [A-Za-z0-9_.-], ≤32 per id) so both ends agree on what is + — and is not — a Hermes prompt answer. + """ + import re + + if not token: + return None + parts = token.split(":") + if len(parts) != 3 or parts[0] != "hp1": + return None + id_re = re.compile(r"^[A-Za-z0-9_.\-]{1,32}$") + if not id_re.match(parts[1]) or not id_re.match(parts[2]): + return None + return parts[1], parts[2] @staticmethod def _render_interaction_options(options) -> list: @@ -1051,3 +1112,413 @@ class RelayAdapter(BasePlatformAdapter): chat_id, file_path, caption=caption, file_name=file_name, reply_to=reply_to, metadata=metadata, **kwargs, ) + + # ── Phase 3 interactive: prompt + react ────────────────────────────── + + def _mint_prompt(self, kind: str, state: Dict[str, Any], timeout_s: float = 3600.0) -> str: + """Register a pending prompt and return its 8-hex id. + + ``state`` carries what the resolver needs when the answer comes back + (session_key, resolver kind, per-kind extras). Expiry is enforced + gateway-side on consumption (_pop_prompt) — the wire's timeout_s is + advisory only. + """ + import secrets + import time + + prompt_id = secrets.token_hex(4) + self._pending_prompts[prompt_id] = { + **state, + "kind": kind, + "expires_at": time.time() + timeout_s, + } + # Opportunistic sweep so abandoned prompts can't accumulate: drop + # anything already expired (cheap — dict is small by construction). + now = time.time() + for stale in [k for k, v in self._pending_prompts.items() if v.get("expires_at", 0) < now]: + self._pending_prompts.pop(stale, None) + return prompt_id + + def _pop_prompt(self, prompt_id: str) -> Optional[Dict[str, Any]]: + """Consume a pending prompt: one answer wins, expired entries miss.""" + import time + + state = self._pending_prompts.pop(str(prompt_id), None) + if not state: + return None + if state.get("expires_at", 0) < time.time(): + return None + return state + + async def _send_prompt( + self, + chat_id: str, + *, + prompt_kind: str, + text: str, + prompt_id: str, + options: list, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + timeout_s: Optional[int] = None, + ) -> Optional[SendResult]: + """Egress one `prompt` op; None when the lane is unavailable. + + Mirrors _send_media's progressive-enhancement posture: op-gated on the + descriptor advertising `prompt`, and every failure returns None so the + caller falls back to its numbered-text base behaviour (which is the + pre-Phase-3 UX, still fully functional via typed replies). + """ + if self._transport is None or not self.descriptor.supports_op("prompt"): + return None + action: Dict[str, Any] = { + "op": "prompt", + "chat_id": chat_id, + "content": text, + "prompt_kind": prompt_kind, + "prompt_id": prompt_id, + "options": options, + "reply_to": reply_to, + "metadata": self._with_scope(chat_id, metadata), + } + if timeout_s is not None: + action["timeout_s"] = int(timeout_s) + try: + result = await self._transport.send_outbound( + action, + platform=self._platform_by_chat.get(str(chat_id)), + ) + except Exception: # noqa: BLE001 - transport failure degrades to fallback + logger.debug("relay prompt transport failure", exc_info=True) + return None + if not result.get("success"): + logger.warning( + "relay prompt declined for %s: %s", chat_id, result.get("error") + ) + return None + return SendResult( + success=True, + message_id=result.get("message_id"), + raw_response=result, + ) + + async def send_exec_approval( + self, + chat_id: str, + command: str, + session_key: str, + description: str = "dangerous command", + metadata: Optional[Dict[str, Any]] = None, + allow_permanent: bool = True, + allow_session: bool = True, + smart_denied: bool = False, + ) -> SendResult: + """Native-button exec approval over the relay (Phase 3). + + Renders the same choice set as the native adapters (Allow Once / + Session / Always / Deny, gated by the same flags) through the + connector's `prompt` op. The user's press comes back as a + prompt_response and resolves via tools.approval.resolve_gateway_approval + — the exact mechanism the native button handlers use. When the lane is + unavailable the send FAILS (success=False) so gateway/run.py's existing + button→text fallback takes over (same contract as a native adapter's + failed button send). + """ + options: list = [{"id": "once", "label": "✅ Allow Once", "style": "success"}] + if not smart_denied and allow_session: + options.append({"id": "session", "label": "✅ Session", "style": "primary"}) + if allow_permanent: + options.append({"id": "always", "label": "✅ Always", "style": "primary"}) + options.append({"id": "deny", "label": "❌ Deny", "style": "danger"}) + + cmd_preview = command if len(command) <= 1500 else command[:1500] + "..." + text = ( + "⚠️ **Command Approval Required**\n\n" + f"```\n{cmd_preview}\n```\n" + f"Reason: {description}" + ) + if smart_denied: + text += "\n\n**Smart DENY:** owner override applies to this one operation only." + + prompt_id = self._mint_prompt( + "exec_approval", + {"session_key": session_key, "chat_id": str(chat_id)}, + ) + result = await self._send_prompt( + chat_id, + prompt_kind="approval", + text=text, + prompt_id=prompt_id, + options=options, + metadata=metadata, + ) + if result is not None: + return result + # Lane unavailable: unregister and let run.py's text fallback run. + self._pending_prompts.pop(prompt_id, None) + return SendResult(success=False, error="relay prompt op unavailable") + + async def send_slash_confirm( + self, + chat_id: str, + title: str, + message: str, + session_key: str, + confirm_id: str, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + """Three-button slash-command confirmation over the relay (Phase 3). + + Resolves via tools.slash_confirm.resolve() — the same primitive the + native button handlers call. Falls back (success=False) to the + gateway's text-intercept flow when the prompt lane is unavailable. + """ + options = [ + {"id": "once", "label": "✅ Approve Once", "style": "success"}, + {"id": "always", "label": "🔒 Always Approve", "style": "primary"}, + {"id": "cancel", "label": "❌ Cancel", "style": "danger"}, + ] + text = f"**{title}**\n\n{message}" if title else message + prompt_id = self._mint_prompt( + "slash_confirm", + { + "session_key": session_key, + "confirm_id": confirm_id, + "chat_id": str(chat_id), + }, + ) + result = await self._send_prompt( + chat_id, + prompt_kind="approval", + text=text, + prompt_id=prompt_id, + options=options, + metadata=metadata, + ) + if result is not None: + return result + self._pending_prompts.pop(prompt_id, None) + return SendResult(success=False, error="relay prompt op unavailable") + + async def send_clarify( + self, + chat_id: str, + question: str, + choices: Optional[list], + clarify_id: str, + session_key: str, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + """Native-button clarify over the relay (Phase 3). + + Multiple-choice clarifies render as a `prompt` op (one button per + choice + Other). A press resolves via + tools.clarify_gateway.resolve_gateway_clarify with the CHOICE TEXT + (never the option id); "Other" flips the clarify to text-capture + exactly like the native adapters. Open-ended clarifies and + unavailable lanes fall back to the base numbered-text behaviour. + + Option ids are positional (c0..cN / other) — choice text can be + arbitrary UTF-8 and would blow the 64-byte callback budget; the + registry maps ids back to the real strings on the answer. + """ + if choices and self.descriptor.supports_op("prompt"): + options = [ + {"id": f"c{i}", "label": str(choice)[:75]} + for i, choice in enumerate(choices) + ] + options.append({"id": "other", "label": "✏️ Other (type your answer)"}) + prompt_id = self._mint_prompt( + "clarify", + { + "session_key": session_key, + "clarify_id": clarify_id, + "choices": [str(c) for c in choices], + "chat_id": str(chat_id), + }, + ) + result = await self._send_prompt( + chat_id, + prompt_kind="clarify", + text=f"❓ {question}", + prompt_id=prompt_id, + options=options, + metadata=metadata, + ) + if result is not None: + return result + self._pending_prompts.pop(prompt_id, None) + return await super().send_clarify( + chat_id, question, choices, clarify_id, session_key, metadata=metadata + ) + + async def _consume_prompt_response(self, event) -> bool: + """Route an inbound prompt_response to its waiting primitive. + + Returns True when the event was a prompt answer (consumed — do NOT + dispatch it as a chat message), False otherwise. Unknown/expired + prompt ids fall through to normal dispatch: the command-shaped text + ("/once", "/deny", …) then behaves like a typed reply, which the + approval/confirm text lanes already understand — the same stale-tap + degradation the native adapters implement with an "expired" edit. + """ + pr = getattr(event, "prompt_response", None) + if not isinstance(pr, dict): + return False + prompt_id = str(pr.get("prompt_id") or "") + option_id = str(pr.get("option_id") or "") + if not prompt_id or not option_id: + return False + state = self._pop_prompt(prompt_id) + if state is None: + logger.info( + "relay prompt_response for unknown/expired prompt %s (option=%s) — " + "falling through to text dispatch", + prompt_id, + option_id, + ) + return False + + kind = state.get("kind") + chat_id = str(state.get("chat_id") or getattr(event.source, "chat_id", "")) + session_key = str(state.get("session_key") or "") + try: + if kind == "exec_approval": + from tools.approval import resolve_gateway_approval + + choice = option_id if option_id in {"once", "session", "always", "deny"} else "deny" + count = resolve_gateway_approval(session_key, choice) + label = { + "once": "✅ Approved once", + "session": "✅ Approved for session", + "always": "✅ Approved permanently", + "deny": "❌ Denied", + }.get(choice, "Resolved") + if not count: + label = "⌛ Approval expired — no command was waiting." + # Acknowledge in-channel (the connector's prompt message can't + # be edited cross-platform yet — edit support varies; a short + # confirmation preserves the audit trail the native edit gives). + await self.send(chat_id, label, metadata=self._prompt_reply_metadata(event)) + if count: + self.resume_typing_for_chat(chat_id) + elif kind == "slash_confirm": + from tools import slash_confirm as slash_confirm_mod + + choice = option_id if option_id in {"once", "always", "cancel"} else "cancel" + result_text = await slash_confirm_mod.resolve( + session_key, str(state.get("confirm_id") or ""), choice + ) + label = { + "once": "✅ Approved once", + "always": "🔒 Always approve", + "cancel": "❌ Cancelled", + }.get(choice, "Resolved") + await self.send(chat_id, label, metadata=self._prompt_reply_metadata(event)) + if result_text: + await self.send( + chat_id, str(result_text), metadata=self._prompt_reply_metadata(event) + ) + elif kind == "clarify": + from tools.clarify_gateway import mark_awaiting_text, resolve_gateway_clarify + + clarify_id = str(state.get("clarify_id") or "") + if option_id == "other": + mark_awaiting_text(clarify_id) + await self.send( + chat_id, + "✏️ Type your answer:", + metadata=self._prompt_reply_metadata(event), + ) + else: + choices = state.get("choices") or [] + try: + idx = int(option_id[1:]) if option_id.startswith("c") else -1 + except ValueError: + idx = -1 + if 0 <= idx < len(choices): + resolve_gateway_clarify(clarify_id, str(choices[idx])) + await self.send( + chat_id, + f"✅ {choices[idx]}", + metadata=self._prompt_reply_metadata(event), + ) + else: + # Unmappable option: flip to text capture so the user + # can answer by typing (never dead-end a clarify). + mark_awaiting_text(clarify_id) + else: + logger.warning("relay prompt_response with unknown kind %r", kind) + except Exception: # noqa: BLE001 - a resolver failure must not kill the reader + logger.warning("relay prompt_response resolution failed", exc_info=True) + return True + + def _prompt_reply_metadata(self, event) -> Dict[str, Any]: + """Thread/topic metadata so prompt acks land where the prompt lives.""" + meta: Dict[str, Any] = {} + thread_id = getattr(event.source, "thread_id", None) + if thread_id: + meta["thread_id"] = str(thread_id) + return meta + + # ── Phase 3 ack lifecycle (👀 → ✅/❌) ──────────────────────────────── + + async def _react( + self, + chat_id: str, + message_id: str, + emoji: str, + *, + remove: bool = False, + ) -> bool: + """Egress one `react` op; best-effort (False on any failure). + + Reactions are cosmetic by contract: a failure is logged at debug and + never surfaces to the caller's flow (mirrors the native Discord + adapter's _add_reaction posture). + """ + if self._transport is None or not self.descriptor.supports_op("react"): + return False + if not chat_id or not message_id: + return False + try: + result = await self._transport.send_outbound( + { + "op": "react", + "chat_id": chat_id, + "message_id": message_id, + "emoji": emoji, + "remove": remove, + "metadata": self._with_scope(chat_id, None), + }, + platform=self._platform_by_chat.get(str(chat_id)), + ) + return bool(result.get("success")) + except Exception: # noqa: BLE001 - reactions are cosmetic + logger.debug("relay react failed", exc_info=True) + return False + + async def on_processing_start(self, event) -> None: + """Add the 👀 in-progress reaction (op-gated; silent no-op otherwise).""" + message_id = getattr(event, "message_id", None) or getattr( + event.source, "message_id", None + ) + chat_id = getattr(event.source, "chat_id", None) + if message_id and chat_id: + await self._react(str(chat_id), str(message_id), "👀") + + async def on_processing_complete(self, event, outcome) -> None: + """Swap 👀 for ✅/❌ per outcome (op-gated; silent no-op otherwise).""" + from gateway.platforms.base import ProcessingOutcome + + message_id = getattr(event, "message_id", None) or getattr( + event.source, "message_id", None + ) + chat_id = getattr(event.source, "chat_id", None) + if not (message_id and chat_id): + return + await self._react(str(chat_id), str(message_id), "👀", remove=True) + if outcome == ProcessingOutcome.SUCCESS: + await self._react(str(chat_id), str(message_id), "✅") + elif outcome == ProcessingOutcome.FAILURE: + await self._react(str(chat_id), str(message_id), "❌") diff --git a/gateway/relay/ws_transport.py b/gateway/relay/ws_transport.py index 0e951941211a..9e97dfecfdb8 100644 --- a/gateway/relay/ws_transport.py +++ b/gateway/relay/ws_transport.py @@ -251,6 +251,16 @@ def _event_from_wire(raw: Dict[str, Any]) -> MessageEvent: # connector that doesn't send it, a dm, or a no-context platform, so # this is purely additive and byte-identical to today when unset. channel_context=_render_relay_context(raw.get("context")), + # Structured interactive-prompt reply (Phase 3): carried verbatim off + # the wire when present ({prompt_id, option_id, label?, + # prompt_message_id?}). The RelayAdapter's inbound bridge consumes it + # to resolve pending approvals/confirms/clarifies; a gateway that + # predates the resolvers just sees the command-shaped text. + prompt_response=( + dict(raw["prompt_response"]) + if isinstance(raw.get("prompt_response"), dict) + else None + ), ) diff --git a/tests/gateway/relay/stub_connector.py b/tests/gateway/relay/stub_connector.py index 25867ca33eff..0a9bee357fda 100644 --- a/tests/gateway/relay/stub_connector.py +++ b/tests/gateway/relay/stub_connector.py @@ -44,6 +44,9 @@ class StubConnector: self.next_send_result: Dict[str, Any] = {"success": True, "message_id": "m1"} # Canned result for the next send_media op (Phase 2; override per-test). self.next_media_result: Dict[str, Any] = {"success": True, "message_id": "md1"} + # Canned results for the Phase 3 interactive ops (override per-test). + self.next_prompt_result: Dict[str, Any] = {"success": True, "message_id": "pm1"} + self.next_react_result: Dict[str, Any] = {"success": True} # Canned result for the next send_follow_up (override per-test). Default # mimics a resolved capability egress; set success=False to simulate an # absent/expired capability or a tenant mismatch on the connector side. @@ -84,6 +87,10 @@ class StubConnector: return dict(self.next_send_result) if action.get("op") == "send_media": return dict(self.next_media_result) + if action.get("op") == "prompt": + return dict(self.next_prompt_result) + if action.get("op") == "react": + return dict(self.next_react_result) return {"success": True} async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: diff --git a/tests/gateway/relay/test_relay_interactive.py b/tests/gateway/relay/test_relay_interactive.py new file mode 100644 index 000000000000..849bba47d0d6 --- /dev/null +++ b/tests/gateway/relay/test_relay_interactive.py @@ -0,0 +1,416 @@ +"""Relay Phase 3 interactive tests — prompt op egress, prompt_response +consumption, and the react ack lifecycle. + +Covers: + - send_exec_approval / send_slash_confirm / send_clarify render through ONE + `prompt` op with the right option sets, honoring op gating (legacy + connectors get the base/text behaviour or a structured failure that + triggers run.py's text fallback); + - the pending-prompt registry: mint → consume-once → expiry; + - _consume_prompt_response routes answers to the approval / slash-confirm / + clarify resolvers and CONSUMES the event; unknown/expired ids fall + through to normal dispatch; + - the Discord type-3 hp1 decode (structured prompt_response replacing the + bare-custom_id stub; foreign custom_ids keep the legacy text shape); + - on_processing_start/complete drive react ops (👀 → ✅/❌), op-gated and + best-effort. +""" + +from __future__ import annotations + +import time +from typing import Any, Dict, Optional + +import pytest + +from gateway.config import PlatformConfig +from gateway.platforms.base import MessageEvent, MessageType, ProcessingOutcome +from gateway.relay.adapter import RelayAdapter +from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor +from gateway.session import SessionSource + +from tests.gateway.relay.stub_connector import StubConnector + +FULL_OPS = ( + "send", + "edit", + "typing", + "get_chat_info", + "send_media", + "prompt", + "react", +) + + +def make_desc(**kw) -> CapabilityDescriptor: + base = dict( + contract_version=CONTRACT_VERSION, + platform="telegram", + label="Telegram", + max_message_length=4096, + supports_draft_streaming=False, + supports_edit=True, + supports_threads=True, + markdown_dialect="markdown_v2", + len_unit="utf16", + supported_ops=FULL_OPS, + ) + base.update(kw) + return CapabilityDescriptor(**base) + + +def _adapter(**desc_kw) -> tuple[RelayAdapter, StubConnector]: + stub = StubConnector(make_desc(**desc_kw)) + adapter = RelayAdapter(PlatformConfig(), make_desc(**desc_kw), transport=stub) + return adapter, stub + + +def _event( + prompt_response: Optional[Dict[str, Any]] = None, + text: str = "/once", + chat_id: str = "c1", +) -> MessageEvent: + return MessageEvent( + text=text, + message_type=MessageType.COMMAND, + source=SessionSource( + platform="telegram", chat_id=chat_id, chat_type="dm", user_id="u1" + ), + prompt_response=prompt_response, + ) + + +# ── egress: the three prompt surfaces ──────────────────────────────────── + + +@pytest.mark.asyncio +async def test_exec_approval_renders_full_option_set(): + adapter, stub = _adapter() + result = await adapter.send_exec_approval( + "c1", "rm -rf /tmp/x", "sess:1", description="deletes files" + ) + assert result.success is True + assert result.message_id == "pm1" + action = stub.sent[-1] + assert action["op"] == "prompt" + assert action["prompt_kind"] == "approval" + ids = [o["id"] for o in action["options"]] + assert ids == ["once", "session", "always", "deny"] + assert "rm -rf /tmp/x" in action["content"] + assert "deletes files" in action["content"] + # The registry holds the pending prompt keyed by the wire's prompt_id. + assert action["prompt_id"] in adapter._pending_prompts + state = adapter._pending_prompts[action["prompt_id"]] + assert state["kind"] == "exec_approval" + assert state["session_key"] == "sess:1" + + +@pytest.mark.asyncio +async def test_exec_approval_smart_denied_and_flag_gating(): + adapter, stub = _adapter() + await adapter.send_exec_approval( + "c1", "cmd", "s", smart_denied=True, allow_permanent=True, allow_session=True + ) + ids = [o["id"] for o in stub.sent[-1]["options"]] + assert ids == ["once", "deny"] # smart-deny: no session/always + await adapter.send_exec_approval( + "c1", "cmd", "s", allow_session=True, allow_permanent=False + ) + ids = [o["id"] for o in stub.sent[-1]["options"]] + assert ids == ["once", "session", "deny"] + + +@pytest.mark.asyncio +async def test_exec_approval_without_prompt_op_fails_for_text_fallback(): + adapter, stub = _adapter(supported_ops=("send", "edit", "typing")) + result = await adapter.send_exec_approval("c1", "cmd", "s") + # success=False → gateway/run.py falls back to the text approval prompt + # (same contract as a failed native button send). + assert result.success is False + assert all(a["op"] != "prompt" for a in stub.sent) + assert adapter._pending_prompts == {} # nothing left pending + + +@pytest.mark.asyncio +async def test_slash_confirm_renders_three_options(): + adapter, stub = _adapter() + result = await adapter.send_slash_confirm( + "c1", "Reload MCP", "This invalidates the prompt cache.", "sess:1", "cf-9" + ) + assert result.success is True + action = stub.sent[-1] + ids = [o["id"] for o in action["options"]] + assert ids == ["once", "always", "cancel"] + assert "Reload MCP" in action["content"] + state = adapter._pending_prompts[action["prompt_id"]] + assert state == { + **state, + "kind": "slash_confirm", + "confirm_id": "cf-9", + "session_key": "sess:1", + } + + +@pytest.mark.asyncio +async def test_clarify_renders_choices_plus_other_with_positional_ids(): + adapter, stub = _adapter() + result = await adapter.send_clarify( + "c1", + "Which environment?", + ["staging — the safe one", "production"], + "cl-1", + "sess:1", + ) + assert result.success is True + action = stub.sent[-1] + assert action["prompt_kind"] == "clarify" + ids = [o["id"] for o in action["options"]] + # Positional ids (choice text is arbitrary UTF-8; ids must be callback-safe). + assert ids == ["c0", "c1", "other"] + labels = [o["label"] for o in action["options"]] + assert labels[0].startswith("staging") + state = adapter._pending_prompts[action["prompt_id"]] + assert state["choices"] == ["staging — the safe one", "production"] + + +@pytest.mark.asyncio +async def test_clarify_open_ended_uses_base_text_path(monkeypatch): + adapter, stub = _adapter() + # No choices → base class question-only text send (no prompt op). + result = await adapter.send_clarify("c1", "What now?", None, "cl-2", "sess:1") + assert result.success is True + assert all(a["op"] != "prompt" for a in stub.sent) + + +@pytest.mark.asyncio +async def test_prompt_decline_degrades_clarify_to_numbered_text(monkeypatch): + adapter, stub = _adapter() + stub.next_prompt_result = {"success": False, "error": "nope"} + marked: list[str] = [] + monkeypatch.setattr( + "tools.clarify_gateway.mark_awaiting_text", lambda cid: marked.append(cid) + ) + result = await adapter.send_clarify( + "c1", "Which?", ["a", "b"], "cl-3", "sess:1" + ) + # Falls back to the base numbered-text clarify (a plain send). + assert result.success is True + assert stub.sent[-1]["op"] == "send" + assert "1. a" in stub.sent[-1]["content"] + assert marked == ["cl-3"] + assert adapter._pending_prompts == {} + + +# ── the pending-prompt registry ────────────────────────────────────────── + + +def test_registry_mint_consume_once_and_expiry(): + adapter, _stub = _adapter() + pid = adapter._mint_prompt("exec_approval", {"session_key": "s"}, timeout_s=3600) + assert adapter._pop_prompt(pid) is not None + assert adapter._pop_prompt(pid) is None # one answer wins + stale = adapter._mint_prompt("exec_approval", {"session_key": "s"}, timeout_s=-1) + assert adapter._pop_prompt(stale) is None # expired misses + + +# ── inbound consumption ────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_prompt_response_resolves_exec_approval(monkeypatch): + adapter, stub = _adapter() + await adapter.send_exec_approval("c1", "cmd", "sess:9") + prompt_id = stub.sent[-1]["prompt_id"] + + calls: list[tuple[str, str]] = [] + monkeypatch.setattr( + "tools.approval.resolve_gateway_approval", + lambda sk, choice, **kw: calls.append((sk, choice)) or 1, + ) + event = _event({"prompt_id": prompt_id, "option_id": "session"}) + consumed = await adapter._consume_prompt_response(event) + assert consumed is True + assert calls == [("sess:9", "session")] + # Consumed prompts leave the registry; the ack landed as a plain send. + assert prompt_id not in adapter._pending_prompts + assert stub.sent[-1]["op"] == "send" + assert "session" in stub.sent[-1]["content"].lower() + + +@pytest.mark.asyncio +async def test_prompt_response_resolves_slash_confirm(monkeypatch): + adapter, stub = _adapter() + await adapter.send_slash_confirm("c1", "T", "msg", "sess:9", "cf-1") + prompt_id = stub.sent[-1]["prompt_id"] + + resolved: list[tuple] = [] + + async def fake_resolve(session_key, confirm_id, choice, **kw): + resolved.append((session_key, confirm_id, choice)) + return "done!" + + monkeypatch.setattr("tools.slash_confirm.resolve", fake_resolve) + event = _event({"prompt_id": prompt_id, "option_id": "always"}) + assert await adapter._consume_prompt_response(event) is True + assert resolved == [("sess:9", "cf-1", "always")] + # The handler's result text went out as a follow-up send. + sends = [a for a in stub.sent if a["op"] == "send"] + assert any("done!" in a["content"] for a in sends) + + +@pytest.mark.asyncio +async def test_prompt_response_resolves_clarify_choice_and_other(monkeypatch): + adapter, stub = _adapter() + await adapter.send_clarify("c1", "Which?", ["alpha", "beta"], "cl-9", "s") + prompt_id = stub.sent[-1]["prompt_id"] + + resolved: list[tuple] = [] + marked: list[str] = [] + monkeypatch.setattr( + "tools.clarify_gateway.resolve_gateway_clarify", + lambda cid, resp: resolved.append((cid, resp)) or True, + ) + monkeypatch.setattr( + "tools.clarify_gateway.mark_awaiting_text", lambda cid: marked.append(cid) + ) + # Positional id maps back to the REAL choice text. + event = _event({"prompt_id": prompt_id, "option_id": "c1"}) + assert await adapter._consume_prompt_response(event) is True + assert resolved == [("cl-9", "beta")] + + # "Other" flips to text capture. + await adapter.send_clarify("c1", "Which?", ["a"], "cl-10", "s") + prompt_id2 = stub.sent[-1]["prompt_id"] + event2 = _event({"prompt_id": prompt_id2, "option_id": "other"}) + assert await adapter._consume_prompt_response(event2) is True + assert marked == ["cl-10"] + + +@pytest.mark.asyncio +async def test_unknown_or_expired_prompt_falls_through(): + adapter, _stub = _adapter() + event = _event({"prompt_id": "deadbeef", "option_id": "once"}) + assert await adapter._consume_prompt_response(event) is False + assert await adapter._consume_prompt_response(_event(None)) is False + # Malformed shapes never consume. + assert await adapter._consume_prompt_response(_event({"prompt_id": ""})) is False + + +# ── Discord type-3 hp1 decode ──────────────────────────────────────────── + + +def test_discord_component_interaction_decodes_prompt_token(): + adapter, _stub = _adapter() + + class Forward: + platform = "discord" + method = "POST" + path = "/interactions/bot1" + body = ( + b'{"type": 3, "id": "i1", "channel_id": "ch1", "guild_id": "g1",' + b' "message": {"id": "pm55"},' + b' "member": {"user": {"id": "u1", "username": "ben"}},' + b' "data": {"custom_id": "hp1:a1b2c3d4:deny"}}' + ) + + event = adapter._discord_interaction_to_event(Forward()) + assert event is not None + assert event.prompt_response == { + "prompt_id": "a1b2c3d4", + "option_id": "deny", + "prompt_message_id": "pm55", + } + assert event.text == "/deny" + assert event.message_type == MessageType.COMMAND + + +def test_discord_foreign_custom_id_keeps_legacy_text_shape(): + adapter, _stub = _adapter() + + class Forward: + platform = "discord" + method = "POST" + path = "/interactions/bot1" + body = ( + b'{"type": 3, "id": "i1", "channel_id": "ch1", "guild_id": "g1",' + b' "data": {"custom_id": "someones_button"}}' + ) + + event = adapter._discord_interaction_to_event(Forward()) + assert event is not None + assert event.prompt_response is None + assert event.text == "someones_button" + assert event.message_type == MessageType.TEXT + + +def test_decode_prompt_token_matches_connector_codec(): + adapter, _stub = _adapter() + assert adapter._decode_prompt_token("hp1:p1:deny") == ("p1", "deny") + assert adapter._decode_prompt_token("ea:once:3") is None + assert adapter._decode_prompt_token("hp1:p1") is None + assert adapter._decode_prompt_token("hp1:bad id:x") is None + assert adapter._decode_prompt_token("") is None + + +# ── react ack lifecycle ────────────────────────────────────────────────── + + +def _reactable_event() -> MessageEvent: + return MessageEvent( + text="do something", + message_type=MessageType.TEXT, + source=SessionSource( + platform="discord", + chat_id="ch1", + chat_type="channel", + user_id="u1", + message_id="m42", + ), + message_id="m42", + ) + + +@pytest.mark.asyncio +async def test_processing_lifecycle_reacts_eyes_then_check(): + adapter, stub = _adapter() + event = _reactable_event() + await adapter.on_processing_start(event) + await adapter.on_processing_complete(event, ProcessingOutcome.SUCCESS) + reacts = [a for a in stub.sent if a["op"] == "react"] + assert [(r["emoji"], r.get("remove", False)) for r in reacts] == [ + ("👀", False), + ("👀", True), + ("✅", False), + ] + assert all(r["message_id"] == "m42" and r["chat_id"] == "ch1" for r in reacts) + + +@pytest.mark.asyncio +async def test_processing_failure_reacts_cross(): + adapter, stub = _adapter() + event = _reactable_event() + await adapter.on_processing_complete(event, ProcessingOutcome.FAILURE) + emojis = [a["emoji"] for a in stub.sent if a["op"] == "react"] + assert emojis[-1] == "❌" + + +@pytest.mark.asyncio +async def test_react_is_op_gated_and_best_effort(): + adapter, stub = _adapter(supported_ops=("send", "edit", "typing")) + event = _reactable_event() + await adapter.on_processing_start(event) + await adapter.on_processing_complete(event, ProcessingOutcome.SUCCESS) + assert all(a["op"] != "react" for a in stub.sent) # never hit the wire + # And a connector decline never raises. + adapter2, stub2 = _adapter() + stub2.next_react_result = {"success": False, "error": "nope"} + await adapter2.on_processing_start(event) # must not raise + + +@pytest.mark.asyncio +async def test_cancelled_outcome_removes_eyes_without_verdict(): + adapter, stub = _adapter() + event = _reactable_event() + await adapter.on_processing_complete(event, ProcessingOutcome.CANCELLED) + reacts = [(a["emoji"], a.get("remove", False)) for a in stub.sent if a["op"] == "react"] + assert reacts == [("👀", True)] # eyes removed, no ✅/❌