feat: add feishu interactive callback adapter

connector/__main__.py

@@ -262,7 +262,7 @@ async def main():
             logger.warning("Feishu adapter disabled due to invalid mode config.")
         else:
             platforms["feishu"] = feishu_server or feishu_long_clients[0]
-            if config.feishu_mode == "webhook" and feishu_server is not None:
+            if feishu_server is not None:
                 await feishu_server.start()
             for feishu_long_client in feishu_long_clients:
                 await feishu_long_client.start()

connector/config.py

@@ -135,6 +135,7 @@ class ConnectorConfig:
     feishu_bind_host: str = "127.0.0.1"
     feishu_bind_port: int = 8094
     feishu_webhook_path: str = "/feishu/events"
+    feishu_callback_path: str = "/feishu/callback"
     feishu_domain: str = "feishu"  # feishu | lark
     feishu_mode: str = "websocket"  # websocket | webhook
     feishu_require_mention: bool = True
@@ -367,6 +368,9 @@ def load_config() -> ConnectorConfig:
     cfg.feishu_webhook_path = os.environ.get(
         "OPENCLAW_CONNECTOR_FEISHU_PATH", "/feishu/events"
     )
+    cfg.feishu_callback_path = os.environ.get(
+        "OPENCLAW_CONNECTOR_FEISHU_CALLBACK_PATH", "/feishu/callback"
+    )
     cfg.feishu_domain = (
         os.environ.get("OPENCLAW_CONNECTOR_FEISHU_DOMAIN", "feishu").strip() or "feishu"
     )

connector/platforms/feishu_installation_manager.py

@@ -158,6 +158,10 @@ class FeishuInstallationManager:
     def binding_count(self) -> int:
         return len(self._bindings)
 
+    @property
+    def registry(self):
+        return self._registry
+
     def bindings(self) -> List[FeishuBinding]:
         return list(self._bindings.values())
 

connector/platforms/feishu_webhook.py

@@ -42,6 +42,19 @@ except ImportError:  # pragma: no cover
         safe_request_json,
     )
 
+try:
+    from services.connector_callback_contract import (
+        CallbackActorContext,
+        CallbackDecisionCode,
+        ConnectorCallbackContract,
+    )
+except ImportError:  # pragma: no cover
+    from services.connector_callback_contract import (  # type: ignore
+        CallbackActorContext,
+        CallbackDecisionCode,
+        ConnectorCallbackContract,
+    )
+
 logger = logging.getLogger(__name__)
 
 FEISHU_WEBHOOK_MAX_BODY_BYTES = 256 * 1024
@@ -58,6 +71,12 @@ _PLACEHOLDER_TYPES = {
     "media": "<media>",
     "sticker": "<sticker>",
 }
+_FEISHU_CALLBACK_POLICY_MAP = {
+    "approval.approve": "admin",
+    "approval.reject": "admin",
+    "command.status": "public",
+    "command.run": "run",
+}
 
 
 def _import_aiohttp_web():
@@ -213,6 +232,31 @@ def parse_feishu_message_text(message: Dict[str, Any]) -> str:
     return str(parsed.get("text", "") or "").strip()
 
 
+def _infer_callback_action_type(command_text: str, button: Dict[str, Any]) -> str:
+    explicit = str(button.get("action_type", "") or "").strip()
+    if explicit:
+        return explicit
+    normalized = str(command_text or "").strip().lower()
+    if normalized.startswith("/approve"):
+        return "approval.approve"
+    if normalized.startswith("/reject"):
+        return "approval.reject"
+    if normalized.startswith("/run"):
+        return "command.run"
+    if normalized.startswith("/status"):
+        return "command.status"
+    return "command.unknown"
+
+
+def _force_approval_command(command_text: str) -> str:
+    normalized = str(command_text or "").strip()
+    if not normalized:
+        return normalized
+    if normalized.startswith("/run") and "--approval" not in normalized:
+        return f"{normalized} --approval"
+    return normalized
+
+
 @dataclass
 class FeishuDeliveryTarget:
     channel_id: str
@@ -256,6 +300,8 @@ class FeishuWebhookServer:
         self._tenant_access_token_expires_at: Dict[str, float] = {}
         self._bot_open_ids: Dict[str, str] = {}
         self._bot_open_id: str = ""
+        self._callback_contracts: Dict[str, ConnectorCallbackContract] = {}
+        self._callback_contract_secrets: Dict[str, str] = {}
 
     async def start(self):
         aiohttp, web = _import_aiohttp_web()
@@ -268,13 +314,15 @@ class FeishuWebhookServer:
                 "(OPENCLAW_CONNECTOR_FEISHU_APP_ID / APP_SECRET missing)"
             )
             return
-        if not any(
+        has_event_ingress = any(
             binding.verification_token
             for binding in self._installation_manager.bindings()
-        ):
+        )
+        has_callback_ingress = bool(str(self.config.feishu_callback_path or "").strip())
+        if not has_event_ingress and not has_callback_ingress:
             logger.info(
                 "Feishu webhook adapter disabled "
-                "(OPENCLAW_CONNECTOR_FEISHU_VERIFICATION_TOKEN missing)"
+                "(verification token and callback path missing)"
             )
             return
         logger.info(
@@ -285,7 +333,13 @@ class FeishuWebhookServer:
             self.config.feishu_domain,
         )
         self.app = web.Application(client_max_size=FEISHU_WEBHOOK_MAX_BODY_BYTES)
-        self.app.router.add_post(self.config.feishu_webhook_path, self.handle_event)
+        if has_event_ingress:
+            self.app.router.add_post(self.config.feishu_webhook_path, self.handle_event)
+        if has_callback_ingress:
+            self.app.router.add_post(
+                self.config.feishu_callback_path,
+                self.handle_callback,
+            )
         self.runner = web.AppRunner(self.app)
         await self.runner.setup()
         self.site = web.TCPSite(
@@ -330,6 +384,29 @@ class FeishuWebhookServer:
             return _make_response(web, status=400, text=str(exc))
         return _make_response(web, status=200, text="OK")
 
+    async def handle_callback(self, request):
+        _, web = _import_aiohttp_web()
+        try:
+            body = await request.read()
+        except Exception:
+            return _make_response(web, status=400, text="Bad request")
+        if len(body) > FEISHU_WEBHOOK_MAX_BODY_BYTES:
+            return _make_response(web, status=413, text="Payload too large")
+        try:
+            payload = json.loads(body or b"{}")
+        except json.JSONDecodeError:
+            return _make_response(web, status=400, text="Bad JSON")
+        try:
+            response = await self.process_callback_payload(payload)
+        except ValueError as exc:
+            logger.warning("Feishu callback rejected: %s", exc)
+            return _make_json_response(
+                web,
+                {"ok": False, "error": str(exc)},
+                status=403,
+            )
+        return _make_json_response(web, response)
+
     def _is_challenge(self, payload: Dict[str, Any]) -> bool:
         return bool(
             payload.get("challenge")
@@ -343,6 +420,120 @@ class FeishuWebhookServer:
         except ValueError:
             return False
 
+    def _extract_callback_action(self, payload: Dict[str, Any]) -> Tuple[
+        Dict[str, Any],
+        Dict[str, Any],
+        Dict[str, Any],
+        Dict[str, Any],
+        str,
+        str,
+    ]:
+        header = payload.get("header") or {}
+        event = payload.get("event") or {}
+        action = payload.get("action") or event.get("action") or {}
+        if not action and isinstance(event.get("actions"), list):
+            first_action = event.get("actions")[0] if event.get("actions") else {}
+            if isinstance(first_action, dict):
+                action = first_action
+        if not isinstance(action, dict):
+            raise ValueError("invalid_callback_action")
+        raw_value = action.get("value") or {}
+        if isinstance(raw_value, str):
+            raw_value = _json_loads_safe(raw_value)
+        if not isinstance(raw_value, dict):
+            raise ValueError("invalid_callback_value")
+        envelope = raw_value.get("callback_envelope") or {}
+        callback_payload = raw_value.get("payload") or {}
+        if not isinstance(envelope, dict) or not isinstance(callback_payload, dict):
+            raise ValueError("invalid_callback_envelope")
+        workspace_id = str(
+            header.get("tenant_key")
+            or event.get("tenant_key")
+            or callback_payload.get("workspace_id")
+            or ""
+        ).strip()
+        account_id = str(callback_payload.get("account_id", "") or "").strip()
+        return header, event, envelope, callback_payload, workspace_id, account_id
+
+    def _callback_contract_for_binding(
+        self,
+        *,
+        binding: FeishuBinding,
+        signing_secret: str,
+    ) -> ConnectorCallbackContract:
+        cache_key = self._cache_key_for_binding(binding)
+        if (
+            self._callback_contracts.get(cache_key) is not None
+            and self._callback_contract_secrets.get(cache_key) == signing_secret
+        ):
+            return self._callback_contracts[cache_key]
+        contract = ConnectorCallbackContract(
+            signing_secret=signing_secret,
+            installation_registry=self._installation_manager.registry,
+            action_policy_map=_FEISHU_CALLBACK_POLICY_MAP,
+        )
+        self._callback_contracts[cache_key] = contract
+        self._callback_contract_secrets[cache_key] = signing_secret
+        return contract
+
+    def _actor_context_for_callback(
+        self,
+        *,
+        actor_id: str,
+        actor_open_id: str,
+        channel_id: str,
+        message_id: str,
+        workspace_id: str,
+        account_id: str,
+        command_text: str,
+    ) -> Tuple[CallbackActorContext, CommandRequest]:
+        request = CommandRequest(
+            platform="feishu",
+            sender_id=actor_id or actor_open_id,
+            channel_id=channel_id or actor_id or actor_open_id,
+            username=actor_id or actor_open_id,
+            message_id=message_id or f"cb-{secrets.token_hex(4)}",
+            text=command_text,
+            timestamp=time.time(),
+            workspace_id=workspace_id,
+            thread_id=message_id,
+            metadata={
+                "account_id": account_id,
+                "sender_open_id": actor_open_id,
+                "interactive_callback": True,
+            },
+        )
+        actor = CallbackActorContext(
+            is_admin=self.router._is_admin(request.sender_id),
+            is_trusted=self.router._is_trusted(request),
+            user_id=request.sender_id,
+            tenant_id=workspace_id or request.workspace_id or "",
+        )
+        return actor, request
+
+    def _build_callback_response(
+        self,
+        *,
+        ok: bool,
+        text: str,
+        response_type: str = "info",
+        card: Optional[Dict[str, Any]] = None,
+        duplicate: bool = False,
+        decision_code: str = "",
+    ) -> Dict[str, Any]:
+        response = {
+            "ok": ok,
+            "duplicate": duplicate,
+            "decision_code": decision_code,
+            "toast": {
+                "type": response_type,
+                "content": text[:500] if text else "",
+            },
+        }
+        if card is not None:
+            response["card"] = card
+        return response
+
     def _resolve_inbound_binding(self, payload: Dict[str, Any]) -> FeishuBinding:
         header = payload.get("header") or {}
         verification_token = (
@@ -474,16 +665,17 @@ class FeishuWebhookServer:
                 return
         response = await self.router.handle(request)
         resp_text = str(getattr(response, "text", "") or "").strip()
-        if resp_text:
-            await self._send_reply(
-                FeishuDeliveryTarget(
-                    channel_id=request.channel_id,
-                    reply_to_message_id=request.thread_id,
-                    workspace_id=request.workspace_id,
-                    account_id=str(request.metadata.get("account_id", "") or ""),
-                ),
-                resp_text,
-            )
+        buttons = getattr(response, "buttons", []) or []
+        target = FeishuDeliveryTarget(
+            channel_id=request.channel_id,
+            reply_to_message_id=request.thread_id,
+            workspace_id=request.workspace_id,
+            account_id=str(request.metadata.get("account_id", "") or ""),
+        )
+        if buttons:
+            await self._send_interactive_reply(target, resp_text, buttons)
+        elif resp_text:
+            await self._send_reply(target, resp_text)
 
     def _resolve_delivery_binding(
         self, *, workspace_id: str = "", account_id: str = ""
@@ -493,6 +685,110 @@ class FeishuWebhookServer:
             account_id=account_id or self._bound_account_id,
         )
 
+    async def process_callback_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        _, _, envelope_dict, callback_payload, workspace_id, account_id = (
+            self._extract_callback_action(payload)
+        )
+        resolution, binding, secrets = self._resolve_delivery_binding(
+            workspace_id=workspace_id,
+            account_id=account_id,
+        )
+        if binding is None or not resolution.ok:
+            raise ValueError(resolution.reject_reason or "missing_binding")
+        signing_secret = str(
+            secrets.get("app_secret", "") or binding.app_secret or ""
+        ).strip()
+        if not signing_secret:
+            raise ValueError("missing_callback_signing_secret")
+        contract = self._callback_contract_for_binding(
+            binding=binding,
+            signing_secret=signing_secret,
+        )
+        event = payload.get("event") or {}
+        operator = payload.get("operator") or event.get("operator") or {}
+        operator_id = operator.get("operator_id") or operator.get("sender_id") or {}
+        actor_id = str(
+            operator.get("user_id")
+            or operator_id.get("user_id")
+            or callback_payload.get("actor_user_id")
+            or ""
+        ).strip()
+        actor_open_id = str(
+            operator.get("open_id")
+            or operator_id.get("open_id")
+            or callback_payload.get("actor_open_id")
+            or ""
+        ).strip()
+        command_text = str(callback_payload.get("command", "") or "").strip()
+        actor, request = self._actor_context_for_callback(
+            actor_id=actor_id,
+            actor_open_id=actor_open_id,
+            channel_id=str(
+                payload.get("open_chat_id")
+                or event.get("open_chat_id")
+                or callback_payload.get("channel_id")
+                or ""
+            ).strip(),
+            message_id=str(
+                payload.get("open_message_id")
+                or event.get("open_message_id")
+                or callback_payload.get("message_id")
+                or ""
+            ).strip(),
+            workspace_id=workspace_id or binding.workspace_id,
+            account_id=binding.account_id,
+            command_text=command_text,
+        )
+        decision = contract.evaluate(
+            platform="feishu",
+            envelope_dict=envelope_dict,
+            payload=callback_payload,
+            actor=actor,
+        )
+        if decision.decision_code == CallbackDecisionCode.REJECT_REPLAY.value:
+            return self._build_callback_response(
+                ok=True,
+                text="Action already processed.",
+                response_type="info",
+                duplicate=True,
+                decision_code=decision.decision_code,
+            )
+        if not decision.ok and not decision.requires_approval:
+            raise ValueError(decision.message or decision.decision_code)
+        request.text = (
+            _force_approval_command(request.text)
+            if decision.requires_approval
+            else request.text
+        )
+        contract.acknowledge_request(envelope_dict.get("request_id", ""))
+        response = await self.router.handle(request)
+        contract.complete_request(envelope_dict.get("request_id", ""))
+        response_text = str(getattr(response, "text", "") or "").strip() or (
+            "Action processed."
+        )
+        response_buttons = getattr(response, "buttons", []) or []
+        card = None
+        if response_buttons:
+            card = self._build_interactive_card(
+                FeishuDeliveryTarget(
+                    channel_id=request.channel_id,
+                    reply_to_message_id=request.thread_id,
+                    workspace_id=request.workspace_id,
+                    account_id=binding.account_id,
+                ),
+                response_text,
+                response_buttons,
+                binding=binding,
+                secrets=secrets,
+            )
+        return self._build_callback_response(
+            ok=True,
+            text=response_text,
+            response_type="success",
+            card=card,
+            decision_code=decision.decision_code,
+        )
+
     async def _get_tenant_access_token(
         self,
         *,
@@ -628,6 +924,167 @@ class FeishuWebhookServer:
             self._bot_open_id = bot_open_id
         return bot_open_id
 
+    def _build_card_button_value(
+        self,
+        button: Dict[str, Any],
+        *,
+        target: FeishuDeliveryTarget,
+        binding: FeishuBinding,
+        signing_secret: str,
+    ) -> Dict[str, Any]:
+        contract = self._callback_contract_for_binding(
+            binding=binding,
+            signing_secret=signing_secret,
+        )
+        command_text = str(button.get("value", "") or "").strip()
+        callback_payload = {
+            "label": str(button.get("label", "") or "").strip(),
+            "command": command_text,
+            "approval_id": str(button.get("approval_id", "") or "").strip(),
+            "workspace_id": target.workspace_id or binding.workspace_id,
+            "account_id": target.account_id or binding.account_id,
+            "channel_id": target.channel_id,
+            "message_id": target.reply_to_message_id,
+        }
+        envelope = contract.build_envelope(
+            request_id=secrets.token_hex(12),
+            workspace_id=callback_payload["workspace_id"],
+            action_type=_infer_callback_action_type(command_text, button),
+            payload=callback_payload,
+        )
+        return {
+            "callback_envelope": dict(envelope.__dict__),
+            "payload": callback_payload,
+        }
+
+    def _build_interactive_card(
+        self,
+        target: FeishuDeliveryTarget,
+        text: str,
+        buttons: list[dict],
+        *,
+        binding: FeishuBinding,
+        secrets: Dict[str, str],
+    ) -> Dict[str, Any]:
+        signing_secret = str(
+            secrets.get("app_secret", "") or binding.app_secret or ""
+        ).strip()
+        if not signing_secret:
+            raise RuntimeError("feishu_callback_signing_secret_missing")
+        actions = []
+        for button in buttons[:6]:
+            command_text = str(button.get("value", "") or "").strip()
+            if not command_text:
+                continue
+            actions.append(
+                {
+                    "tag": "button",
+                    "type": str(button.get("style", "") or "default"),
+                    "text": {
+                        "tag": "plain_text",
+                        "content": str(button.get("label", "") or "OpenClaw"),
+                    },
+                    "value": self._build_card_button_value(
+                        button,
+                        target=target,
+                        binding=binding,
+                        signing_secret=signing_secret,
+                    ),
+                }
+            )
+        return {
+            "config": {"wide_screen_mode": True},
+            "header": {
+                "template": "blue",
+                "title": {"tag": "plain_text", "content": "OpenClaw"},
+            },
+            "elements": [
+                {"tag": "markdown", "content": text or "OpenClaw"},
+                {"tag": "action", "actions": actions},
+            ],
+        }
+
+    async def _send_interactive_reply(
+        self,
+        target: FeishuDeliveryTarget,
+        text: str,
+        buttons: list[dict],
+    ) -> None:
+        resolution, binding, secrets = self._resolve_delivery_binding(
+            workspace_id=target.workspace_id,
+            account_id=target.account_id,
+        )
+        if binding is None or not resolution.ok:
+            logger.warning(
+                "Feishu interactive reply dropped: no workspace binding available (%s / %s)",
+                target.workspace_id or "no-workspace",
+                target.account_id or "no-account",
+            )
+            return
+        token = await self._get_tenant_access_token(
+            binding=binding,
+            workspace_id=target.workspace_id,
+            account_id=target.account_id,
+        )
+        api_base = _resolve_domain_base(binding.domain)
+        card = self._build_interactive_card(
+            target,
+            text,
+            buttons,
+            binding=binding,
+            secrets=secrets,
+        )
+        payload = {
+            "content": json.dumps(card, ensure_ascii=False),
+            "msg_type": "interactive",
+        }
+        headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json; charset=utf-8",
+        }
+        if target.reply_to_message_id:
+            url = (
+                f"{api_base}/open-apis/im/v1/messages/"
+                f"{target.reply_to_message_id}/reply"
+            )
+        else:
+            url = f"{api_base}/open-apis/im/v1/messages?receive_id_type=chat_id"
+            payload["receive_id"] = target.channel_id
+        try:
+            data = safe_request_json(
+                method="POST",
+                url=url,
+                json_body=payload,
+                headers=headers,
+                content_type="application/json; charset=utf-8",
+                timeout_sec=15,
+                allow_hosts=_allowed_api_hosts(binding.domain),
+                policy=STANDARD_OUTBOUND_POLICY,
+            )
+        except SafeIOHTTPError as exc:
+            if resolution.installation is not None:
+                self._installation_manager.mark_api_error(
+                    resolution.installation.installation_id,
+                    error_code=exc.reason,
+                    status_code=exc.status_code,
+                    details={"phase": "interactive_reply"},
+                )
+            logger.warning(
+                "Feishu interactive reply failed: status=%s", exc.status_code
+            )
+            return
+        if data.get("code", 0) != 0:
+            if resolution.installation is not None:
+                self._installation_manager.mark_api_error(
+                    resolution.installation.installation_id,
+                    error_code=str(data.get("msg", "unknown") or "unknown"),
+                    status_code=200,
+                    details={"phase": "interactive_reply"},
+                )
+            logger.warning(
+                "Feishu interactive reply failed: %s", data.get("msg", "unknown")
+            )
+
     async def _send_reply(self, target: FeishuDeliveryTarget, text: str) -> None:
         resolution, binding, _ = self._resolve_delivery_binding(
             workspace_id=target.workspace_id,

connector/router.py

@@ -524,6 +524,7 @@ class CommandRouter:
 
         pending_count = res.get("pending_count")
         lines = []
+        buttons = []
         for i in items:
             # IMPORTANT (stability): the backend approval schema uses:
             # `approval_id`, `template_id`, `status`, `requested_by`, `source`.
@@ -539,11 +540,35 @@ class CommandRouter:
             lines.append(
                 f"- {approval_id} [{status}] template={template_id} by={requested_by} source={source}"
             )
+        for i in items[:3]:
+            approval_id = i.get("approval_id") or i.get("id") or "unknown"
+            short_id = str(approval_id)[:8]
+            buttons.append(
+                {
+                    "label": f"Approve {short_id}",
+                    "value": f"/approve {approval_id}",
+                    "action_type": "approval.approve",
+                    "approval_id": approval_id,
+                    "style": "primary",
+                }
+            )
+            buttons.append(
+                {
+                    "label": f"Reject {short_id}",
+                    "value": f"/reject {approval_id}",
+                    "action_type": "approval.reject",
+                    "approval_id": approval_id,
+                    "style": "danger",
+                }
+            )
 
         header = "Pending Approvals"
         if isinstance(pending_count, int):
             header += f" ({pending_count})"
-        return CommandResponse(text=header + ":\n" + "\n".join(lines))
+        return CommandResponse(
+            text=header + ":\n" + "\n".join(lines),
+            buttons=buttons,
+        )
 
     async def _handle_approve(
         self, req: CommandRequest, args: List[str]

tests/chat_connector/test_config.py

@@ -53,6 +53,7 @@ class TestConnectorConfig(unittest.TestCase):
                 "OPENCLAW_CONNECTOR_FEISHU_WORKSPACE_ID": "tenant-alpha",
                 "OPENCLAW_CONNECTOR_FEISHU_WORKSPACE_NAME": "Alpha Workspace",
                 "OPENCLAW_CONNECTOR_FEISHU_BINDINGS_JSON": '[{"account_id":"acct-extra","workspace_id":"tenant-beta","app_id":"cli_extra","app_secret":"sec_extra"}]',
+                "OPENCLAW_CONNECTOR_FEISHU_CALLBACK_PATH": "/feishu/cards",
                 "OPENCLAW_CONNECTOR_FEISHU_ALLOWED_USERS": "u1,u2",
                 "OPENCLAW_CONNECTOR_FEISHU_ALLOWED_CHATS": "oc_a,oc_b",
                 "OPENCLAW_CONNECTOR_FEISHU_DOMAIN": "lark",
@@ -71,6 +72,7 @@ class TestConnectorConfig(unittest.TestCase):
             self.assertEqual(cfg.feishu_workspace_id, "tenant-alpha")
             self.assertEqual(cfg.feishu_workspace_name, "Alpha Workspace")
             self.assertIn("acct-extra", cfg.feishu_bindings_json)
+            self.assertEqual(cfg.feishu_callback_path, "/feishu/cards")
             self.assertEqual(cfg.feishu_allowed_users, ["u1", "u2"])
             self.assertEqual(cfg.feishu_allowed_chats, ["oc_a", "oc_b"])
             self.assertEqual(cfg.feishu_domain, "lark")

tests/test_f69_feishu_callbacks.py

@@ -0,0 +1,235 @@
+import json
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import AsyncMock, MagicMock, patch
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from connector.config import ConnectorConfig
+from connector.contract import CommandResponse
+from connector.platforms.feishu_installation_manager import FeishuInstallationManager
+from connector.platforms.feishu_webhook import FeishuDeliveryTarget, FeishuWebhookServer
+from services.connector_installation_registry import ConnectorInstallationRegistry
+from services.secret_store import SecretStore
+
+
+def _event_payload():
+    return {
+        "schema": "2.0",
+        "header": {
+            "event_id": "fe-evt-f69-1",
+            "event_type": "im.message.receive_v1",
+            "tenant_key": "tenant-1",
+            "token": "verify-token",
+        },
+        "event": {
+            "sender": {
+                "sender_id": {
+                    "user_id": "u_sender",
+                    "open_id": "ou_sender",
+                }
+            },
+            "message": {
+                "message_id": "om_1",
+                "chat_id": "oc_dm_1",
+                "chat_type": "p2p",
+                "message_type": "text",
+                "content": json.dumps({"text": "/approvals"}),
+                "mentions": [],
+                "root_id": "om_root_1",
+            },
+        },
+    }
+
+
+class TestF69FeishuCallbacks(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.tmpdir = tempfile.TemporaryDirectory()
+        self.state_dir = self.tmpdir.name
+        self.secret_store = SecretStore(state_dir=self.state_dir)
+        self.registry = ConnectorInstallationRegistry(
+            state_dir=self.state_dir,
+            secret_store=self.secret_store,
+        )
+        self.config = ConnectorConfig()
+        self.config.feishu_app_id = "cli_test"
+        self.config.feishu_app_secret = "sec_test"
+        self.config.feishu_account_id = "acct-default"
+        self.config.feishu_default_account_id = "acct-default"
+        self.config.feishu_verification_token = "verify-token"
+        self.config.feishu_workspace_id = "tenant-1"
+        self.config.feishu_mode = "webhook"
+        self.config.feishu_callback_path = "/feishu/callback"
+        self.router = MagicMock()
+        self.router.handle = AsyncMock(return_value=CommandResponse(text="OK"))
+        self.router._is_admin = MagicMock(return_value=False)
+        self.router._is_trusted = MagicMock(return_value=False)
+        self.manager = FeishuInstallationManager(
+            self.config,
+            registry=self.registry,
+            secret_store=self.secret_store,
+            state_dir=self.state_dir,
+        )
+        self.server = FeishuWebhookServer(
+            self.config,
+            self.router,
+            installation_manager=self.manager,
+        )
+        self.binding = self.manager.get_binding("acct-default")
+
+    async def asyncTearDown(self):
+        self.tmpdir.cleanup()
+
+    def _callback_body(
+        self,
+        *,
+        button,
+        actor_user_id="admin-user",
+        actor_open_id="ou_admin",
+    ):
+        value = self.server._build_card_button_value(
+            button,
+            target=FeishuDeliveryTarget(
+                channel_id="oc_dm_1",
+                reply_to_message_id="om_card_1",
+                workspace_id="tenant-1",
+                account_id="acct-default",
+            ),
+            binding=self.binding,
+            signing_secret=self.binding.app_secret,
+        )
+        return {
+            "header": {"tenant_key": "tenant-1"},
+            "event": {
+                "open_chat_id": "oc_dm_1",
+                "open_message_id": "om_card_1",
+                "operator": {
+                    "operator_id": {
+                        "user_id": actor_user_id,
+                        "open_id": actor_open_id,
+                    }
+                },
+                "action": {"value": value},
+            },
+        }
+
+    async def test_approvals_response_uses_interactive_card(self):
+        self.router.handle = AsyncMock(
+            return_value=CommandResponse(
+                text="Pending Approvals (1):\n- apr_1 [pending]",
+                buttons=[
+                    {
+                        "label": "Approve apr_1",
+                        "value": "/approve apr_1",
+                        "action_type": "approval.approve",
+                        "approval_id": "apr_1",
+                        "style": "primary",
+                    }
+                ],
+            )
+        )
+        with (
+            patch.object(
+                self.server, "_get_tenant_access_token", AsyncMock(return_value="tok_1")
+            ),
+            patch(
+                "connector.platforms.feishu_webhook.safe_request_json",
+                return_value={"code": 0, "data": {}},
+            ) as mock_safe,
+        ):
+            await self.server.process_event_payload(_event_payload())
+
+        kwargs = mock_safe.call_args.kwargs
+        self.assertEqual(kwargs["json_body"]["msg_type"], "interactive")
+        card = json.loads(kwargs["json_body"]["content"])
+        action = card["elements"][1]["actions"][0]
+        self.assertEqual(action["text"]["content"], "Approve apr_1")
+        self.assertEqual(
+            action["value"]["callback_envelope"]["action_type"], "approval.approve"
+        )
+        self.assertEqual(action["value"]["payload"]["command"], "/approve apr_1")
+
+    async def test_admin_callback_routes_and_duplicate_is_deduped(self):
+        self.router._is_admin = MagicMock(return_value=True)
+        self.router._is_trusted = MagicMock(return_value=True)
+        self.router.handle = AsyncMock(
+            return_value=CommandResponse(text="[Approved] apr_1")
+        )
+        body = self._callback_body(
+            button={
+                "label": "Approve apr_1",
+                "value": "/approve apr_1",
+                "action_type": "approval.approve",
+                "approval_id": "apr_1",
+                "style": "primary",
+            }
+        )
+
+        first = await self.server.process_callback_payload(body)
+        second = await self.server.process_callback_payload(body)
+
+        self.assertTrue(first["ok"])
+        self.assertEqual(first["decision_code"], "cb_accept_admin")
+        req = self.router.handle.call_args.args[0]
+        self.assertEqual(req.text, "/approve apr_1")
+        self.assertTrue(second["ok"])
+        self.assertTrue(second["duplicate"])
+        self.assertEqual(self.router.handle.await_count, 1)
+
+    async def test_run_callback_degrades_to_approval_for_untrusted_actor(self):
+        self.router._is_admin = MagicMock(return_value=False)
+        self.router._is_trusted = MagicMock(return_value=False)
+        self.router.handle = AsyncMock(
+            return_value=CommandResponse(text="[Approval Requested]")
+        )
+        body = self._callback_body(
+            button={
+                "label": "Run template",
+                "value": "/run template_x prompt=city",
+                "action_type": "command.run",
+            },
+            actor_user_id="u_untrusted",
+            actor_open_id="ou_untrusted",
+        )
+
+        response = await self.server.process_callback_payload(body)
+
+        self.assertTrue(response["ok"])
+        self.assertEqual(response["decision_code"], "cb_require_approval")
+        req = self.router.handle.call_args.args[0]
+        self.assertIn("--approval", req.text)
+
+    async def test_stale_callback_is_rejected(self):
+        body = self._callback_body(
+            button={
+                "label": "Approve apr_1",
+                "value": "/approve apr_1",
+                "action_type": "approval.approve",
+                "approval_id": "apr_1",
+            }
+        )
+        envelope = body["event"]["action"]["value"]["callback_envelope"]
+        envelope["timestamp"] = envelope["timestamp"] - 1000
+
+        with self.assertRaisesRegex(ValueError, "timestamp_out_of_window"):
+            await self.server.process_callback_payload(body)
+
+    async def test_invalid_signature_is_rejected(self):
+        body = self._callback_body(
+            button={
+                "label": "Approve apr_1",
+                "value": "/approve apr_1",
+                "action_type": "approval.approve",
+                "approval_id": "apr_1",
+            }
+        )
+        body["event"]["action"]["value"]["callback_envelope"]["signature"] = "bad"
+
+        with self.assertRaisesRegex(ValueError, "signature_mismatch"):
+            await self.server.process_callback_payload(body)
+
+
+if __name__ == "__main__":
+    unittest.main()