feat(eko): add quick replies for exec approvals

plugins/platforms/eko/README.md

@@ -335,6 +335,9 @@ targets. Without the explicit format, standalone delivery falls back to DM push.
 - Webhook signature verification via `X-Eko-Signature` (HMAC-SHA256-Base64)
 - Image receiving: download inbound pictures, cache locally, vision tool integration
 - Image sending: native multipart upload with reply token + push fallback
+- Selectable prompts: clarify choices, slash confirmations, and dangerous
+  command approvals use Eko quick replies via `/bot/v1/message/quickreply`
+  when a reply token is available, with text fallback otherwise
 - File sending: push files to users via multipart upload
 - Sticker webhook events: surface `[sticker]` placeholder
 
@@ -358,7 +361,6 @@ None currently.
 
 | Feature | Description | Notes |
 |---------|-------------|-------|
-| Quick reply buttons | Tap-to-respond options for users | Eko supports it via `/bot/v1/message/quickreply` |
 | Compact tool progress | One-shot progress message on no-edit platforms | Issue #32 (pended — core gateway change) |
 
 ### Low priority

plugins/platforms/eko/adapter.py

@@ -671,6 +671,154 @@ def _has_mention_trigger(self, text: str) -> bool:
     # Outbound send (text)
     # ------------------------------------------------------------------
 
+    async def send_exec_approval(
+        self,
+        chat_id: str,
+        command: str,
+        session_key: str,
+        description: str = "dangerous command",
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Render dangerous-command approval as Eko quick replies.
+
+        Eko quick replies are reply-token only.  If no fresh token is
+        available, return unsupported so the gateway sends its text fallback.
+        Button ``value`` fields are set to slash commands (e.g. ``/approve``)
+        so the tap arrives as a real command that bypasses the agent-active
+        queue in base.py.
+        """
+        if not self._client:
+            return SendResult(success=False, error="Eko adapter not connected")
+
+        token, used_reply = self._consume_reply_token(chat_id)
+        if not used_reply:
+            return SendResult(success=False, error="No Eko reply token available")
+
+        cmd_preview = command[:3800] + "..." if len(command) > 3800 else command
+        prompt = (
+            "⚠️ Command Approval Required\n\n"
+            f"```{cmd_preview}```\n\n"
+            f"Reason: {description}"
+        )
+
+        try:
+            await self._client.reply_quick_reply(
+                token,
+                prompt,
+                ["Approve Once", "Approve Session", "Approve Always", "Deny"],
+                values=["/approve", "/approve session", "/approve always", "/deny"],
+            )
+        except Exception as exc:
+            logger.debug(
+                "Eko: exec-approval quick reply failed, falling back to text prompt: %s",
+                exc,
+            )
+            return SendResult(success=False, error=str(exc), retryable=True)
+
+        return SendResult(success=True, message_id=token)
+
+    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:
+        """Render slash confirmations as Eko quick replies when possible.
+
+        Eko quick replies are reply-token only.  If no fresh token is
+        available, return unsupported so the gateway sends its text fallback.
+        Button ``value`` fields are set to slash commands (e.g. ``/approve``)
+        so the tap arrives as a real command that bypasses the agent-active
+        queue in base.py.
+        """
+        if not self._client:
+            return SendResult(success=False, error="Eko adapter not connected")
+
+        token, used_reply = self._consume_reply_token(chat_id)
+        if not used_reply:
+            return SendResult(success=False, error="No Eko reply token available")
+
+        try:
+            await self._client.reply_quick_reply(
+                token,
+                message,
+                ["Approve Once", "Always Approve", "Cancel"],
+                values=["/approve", "/always", "/cancel"],
+            )
+        except Exception as exc:
+            logger.debug(
+                "Eko: slash-confirm quick reply failed, falling back to text prompt: %s",
+                exc,
+            )
+            return SendResult(success=False, error=str(exc), retryable=True)
+
+        return SendResult(success=True, message_id=token)
+
+    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:
+        """Render clarify choices as Eko quick replies when possible.
+
+        Eko quick replies are reply-token only.  If no fresh token is
+        available, fall back to the base numbered-text prompt.
+        """
+        if not choices or not self._client:
+            return await super().send_clarify(
+                chat_id=chat_id,
+                question=question,
+                choices=choices,
+                clarify_id=clarify_id,
+                session_key=session_key,
+                metadata=metadata,
+            )
+
+        token, used_reply = self._consume_reply_token(chat_id)
+        if not used_reply:
+            return await super().send_clarify(
+                chat_id=chat_id,
+                question=question,
+                choices=choices,
+                clarify_id=clarify_id,
+                session_key=session_key,
+                metadata=metadata,
+            )
+
+        try:
+            await self._client.reply_quick_reply(
+                token,
+                question,
+                [str(c) for c in choices],
+            )
+        except Exception as exc:
+            logger.debug(
+                "Eko: quick reply failed, falling back to text prompt: %s",
+                exc,
+            )
+            return await super().send_clarify(
+                chat_id=chat_id,
+                question=question,
+                choices=choices,
+                clarify_id=clarify_id,
+                session_key=session_key,
+                metadata=metadata,
+            )
+
+        # Eko quick-reply taps arrive back as ordinary text messages with a
+        # fresh reply token.  Mark this clarify as text-capturing so the
+        # gateway resolves it instead of starting a new agent turn.
+        from tools.clarify_gateway import mark_awaiting_text
+        mark_awaiting_text(clarify_id)
+        return SendResult(success=True, message_id=token)
+
     async def send(
         self,
         chat_id: str,

plugins/platforms/eko/client.py

@@ -254,6 +254,44 @@ async def reply_text(self, reply_token: str, message: str) -> None:
         data.add_field("replyToken", reply_token)
         await self._request_form("/bot/v1/message/text", data=data)
 
+    async def reply_quick_reply(
+        self,
+        reply_token: str,
+        message: str,
+        choices: List[str],
+        values: Optional[List[str]] = None,
+    ) -> None:
+        """Send a quick-reply prompt using a reply token.
+
+        ``choices`` sets the display label (``data.text``).  ``values`` sets
+        the reply payload (``value``).  When ``values`` is omitted, each
+        choice is used as its own value (backward-compatible).
+        """
+        items = [
+            {
+                "data": {"text": choice},
+                "type": "label",
+                "value": (values[i] if values else choice),
+            }
+            for i, choice in enumerate(choices)
+        ]
+        await self._request_json_post(
+            "/bot/v1/message/quickreply",
+            json={
+                "replyToken": reply_token,
+                "message": {
+                    "data": message,
+                    "meta": {
+                        "quickreply": {
+                            "template": "default",
+                            "items": items,
+                        }
+                    },
+                },
+            },
+            expect_json=False,
+        )
+
     async def push_text(self, uid: str, message: str) -> None:
         """Push a text message to a user by uid."""
         await self._request_json_post(

tests/gateway/test_eko_plugin.py

@@ -642,6 +642,138 @@ async def test_not_connected_returns_error(self):
         assert "not connected" in result.error
 
 
+# ---------------------------------------------------------------------------
+# 5a. Selectable quick replies
+# ---------------------------------------------------------------------------
+
+class TestExecApprovalQuickReplies:
+
+    @pytest.mark.asyncio
+    async def test_send_exec_approval_uses_quick_reply_with_reply_token(self):
+        adapter = EkoAdapter.__new__(EkoAdapter)
+        adapter._reply_tokens = {"chat1": ("tok_abc", time.time() + 50)}
+        adapter._client = MagicMock(reply_quick_reply=AsyncMock())
+
+        result = await adapter.send_exec_approval(
+            chat_id="chat1",
+            command="rm -rf /tmp/example",
+            session_key="sk-eko",
+            description="test approval",
+        )
+
+        assert result.success
+        adapter._client.reply_quick_reply.assert_called_once()
+        args, kwargs = adapter._client.reply_quick_reply.call_args
+        assert args[0] == "tok_abc"
+        assert "rm -rf /tmp/example" in args[1]
+        assert "test approval" in args[1]
+        assert args[2] == [
+            "Approve Once",
+            "Approve Session",
+            "Approve Always",
+            "Deny",
+        ]
+        assert kwargs["values"] == [
+            "/approve",
+            "/approve session",
+            "/approve always",
+            "/deny",
+        ]
+        assert "chat1" not in adapter._reply_tokens
+
+    @pytest.mark.asyncio
+    async def test_send_exec_approval_without_reply_token_uses_text_fallback(self):
+        adapter = EkoAdapter.__new__(EkoAdapter)
+        adapter._reply_tokens = {}
+        adapter._client = MagicMock(reply_quick_reply=AsyncMock())
+
+        result = await adapter.send_exec_approval(
+            chat_id="chat1",
+            command="rm -rf /tmp/example",
+            session_key="sk-eko",
+        )
+
+        assert not result.success
+        adapter._client.reply_quick_reply.assert_not_called()
+
+
+class TestSlashConfirmQuickReplies:
+
+    @pytest.mark.asyncio
+    async def test_send_slash_confirm_uses_quick_reply_with_reply_token(self):
+        adapter = EkoAdapter.__new__(EkoAdapter)
+        adapter._reply_tokens = {"chat1": ("tok_abc", time.time() + 50)}
+        adapter._client = MagicMock(reply_quick_reply=AsyncMock())
+
+        result = await adapter.send_slash_confirm(
+            chat_id="chat1",
+            title="/new",
+            message="Confirm /new?",
+            session_key="sk-eko",
+            confirm_id="confirm-1",
+        )
+
+        assert result.success
+        adapter._client.reply_quick_reply.assert_called_once_with(
+            "tok_abc",
+            "Confirm /new?",
+            ["Approve Once", "Always Approve", "Cancel"],
+            values=["/approve", "/always", "/cancel"],
+        )
+        assert "chat1" not in adapter._reply_tokens
+
+    @pytest.mark.asyncio
+    async def test_send_slash_confirm_without_reply_token_uses_text_fallback(self):
+        adapter = EkoAdapter.__new__(EkoAdapter)
+        adapter._reply_tokens = {}
+        adapter._client = MagicMock(reply_quick_reply=AsyncMock())
+
+        result = await adapter.send_slash_confirm(
+            chat_id="chat1",
+            title="/new",
+            message="Confirm /new?",
+            session_key="sk-eko",
+            confirm_id="confirm-1",
+        )
+
+        assert not result.success
+        adapter._client.reply_quick_reply.assert_not_called()
+
+
+class TestClarifyQuickReplies:
+
+    @pytest.mark.asyncio
+    async def test_send_clarify_uses_quick_reply_with_reply_token(self):
+        from tools import clarify_gateway as cm
+
+        cm.clear_session("sk-eko")
+        cm.register("cid-eko", "sk-eko", "Pick one?", ["A", "B"])
+
+        adapter = EkoAdapter.__new__(EkoAdapter)
+        adapter._reply_tokens = {"chat1": ("tok_abc", time.time() + 50)}
+        adapter._client = MagicMock(reply_quick_reply=AsyncMock())
+
+        try:
+            result = await adapter.send_clarify(
+                chat_id="chat1",
+                question="Pick one?",
+                choices=["A", "B"],
+                clarify_id="cid-eko",
+                session_key="sk-eko",
+            )
+
+            assert result.success
+            adapter._client.reply_quick_reply.assert_called_once_with(
+                "tok_abc", "Pick one?", ["A", "B"]
+            )
+            assert "chat1" not in adapter._reply_tokens
+            pending = cm.get_pending_for_session("sk-eko")
+            assert pending is not None
+            assert pending.awaiting_text is True
+        finally:
+            cm.clear_session("sk-eko")
+
+
 # ---------------------------------------------------------------------------
 # 5b. Outbound chunking
 # ---------------------------------------------------------------------------
@@ -1051,6 +1183,29 @@ async def test_reply_picture_sends_multipart(self):
         call_args = mock_session.post.call_args
         assert call_args[0][0].endswith("/bot/v1/message/picture")
 
+    @pytest.mark.asyncio
+    async def test_reply_quick_reply_sends_json(self):
+        mock_aiohttp = _mock_aiohttp_for_post(200)
+        client = _make_eko_client()
+
+        with patch.dict("sys.modules", {"aiohttp": mock_aiohttp}):
+            await client.reply_quick_reply(
+                "reply_tok", "Pick one?", ["A", "B"], values=["/a", "/b"],
+            )
+
+        mock_session = mock_aiohttp.ClientSession.return_value
+        mock_session.post.assert_called_once()
+        call_args = mock_session.post.call_args
+        assert call_args[0][0].endswith("/bot/v1/message/quickreply")
+        payload = call_args.kwargs["json"]
+        assert payload["replyToken"] == "reply_tok"
+        assert payload["message"]["data"] == "Pick one?"
+        items = payload["message"]["meta"]["quickreply"]["items"]
+        assert items == [
+            {"data": {"text": "A"}, "type": "label", "value": "/a"},
+            {"data": {"text": "B"}, "type": "label", "value": "/b"},
+        ]
+
     @pytest.mark.asyncio
     async def test_push_file_sends_multipart(self):
         mock_aiohttp = _mock_aiohttp_for_post(200)