feat(relay): Phase 3 interactive — native prompt UX (approvals/confirms/clarify) + react ack lifecycle (#71404)

- RelayAdapter.send_exec_approval / send_slash_confirm / send_clarify:
  override the base text fallbacks with ONE platform-abstract `prompt` op
  (connector renders Discord components / Telegram inline keyboards /
  Slack Block Kit / WhatsApp buttons+lists). Option sets mirror the native
  adapters exactly (once/session/always/deny with the same
  allow_session/allow_permanent/smart_denied gating; once/always/cancel;
  choices + Other). Clarify option ids are positional (c0..cN/other) —
  choice text is arbitrary UTF-8, callback budgets are 64 bytes.
- Pending-prompt registry: gateway-minted 8-hex prompt ids →
  {kind, session_key, extras}; one answer wins, lazy expiry, unanswered
  prompts swept opportunistically. Wire timeout_s stays advisory.
- _consume_prompt_response (wired into _on_inbound AND the Discord
  passthrough lane): routes answers to the SAME primitives the native
  button handlers call — tools.approval.resolve_gateway_approval,
  tools.slash_confirm.resolve, tools.clarify_gateway
  resolve/mark_awaiting_text — then acks in-channel. Unknown/expired ids
  fall through as command-shaped text (typed-reply degradation, the
  relay's analog of the native 'approval expired' edit).
- Discord type-3 stub replaced: an hp1:<prompt>:<option> custom_id decodes
  to a structured prompt_response (codec mirrored from the connector's
  promptCodec); foreign custom_ids keep the legacy best-effort text shape.
- MessageEvent.prompt_response field + ws_transport wire parsing (additive).
- react ack lifecycle: on_processing_start/complete → `react` ops
  (👀 → ✅/❌, remove-then-add), op-gated on supported_ops, best-effort by
  contract (a react failure never touches the turn).
- Op gating throughout: a connector not advertising `prompt` gets
  success=False from send_exec_approval/send_slash_confirm (run.py's text
  fallback takes over — same contract as a failed native button send) and
  the base numbered-text clarify; `react` silently no-ops.
- docs/relay-connector-contract.md §4: prompt / prompt_response / react
  semantics (callback token, budgets, authorization-parity, foreign-id
  behavior, per-platform react mappings).
- tests: tests/gateway/relay/test_relay_interactive.py (19) — option-set
  rendering + gating matrices, registry consume-once/expiry, resolver
  routing for all three kinds (monkeypatched primitives), fall-through
  cases, Discord hp1 decode + foreign-id shape, react lifecycle
  (success/failure/cancelled), op-gated/best-effort react.

Cross-repo pair: gateway-gateway 'Phase 3 interactive' PR (prompt/react
senders on all four lanes + interaction ingest).

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:<prompt_id>:<option_id>`
+(≤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

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.

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:<prompt_id>:<option_id>) 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:<prompt_id>:<option_id> 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), "❌")

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
+        ),
     )
 
 

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]:

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 ✅/❌