feat(qqofficial): implement Dify human input button interaction handling and markdown keyboard support

src/langbot/pkg/platform/sources/qqofficial.py

@@ -11,7 +11,7 @@ import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platf
 import langbot_plugin.api.entities.builtin.platform.message as platform_message
 import langbot_plugin.api.entities.builtin.platform.events as platform_events
 import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
-from langbot.libs.qq_official_api.api import QQOfficialClient
+from langbot.libs.qq_official_api.api import QQOfficialClient, build_keyboard_from_form
 from langbot.libs.qq_official_api.qqofficialevent import QQOfficialEvent
 from ...utils import image
 from ..logger import EventLogger
@@ -191,6 +191,7 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
     enable_webhook: bool = False
     message_converter: QQOfficialMessageConverter = QQOfficialMessageConverter()
     event_converter: QQOfficialEventConverter = QQOfficialEventConverter()
+    ap: typing.Any = None
 
     def __init__(self, config: dict, logger: EventLogger):
         enable_webhook = config.get('enable-webhook', False)
@@ -216,6 +217,31 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
         self._stream_ctx_ts: dict[str, float] = {}
         self._fallback_text: dict[str, str] = {}
         self._fallback_text_ts: dict[str, float] = {}
+        # Dify form-action bookkeeping for the human-input button flow.
+        # session_key = "<scene>_<id>" where scene is c2c/group/channel and
+        #   id is user_openid / group_openid / channel_id.
+        # session_key -> {form_data, msg_id, event_id, scene, target_id,
+        #                 sender_id, posted_at}
+        # Set when we send a markdown+keyboard card and consulted when:
+        #   (a) INTERACTION_CREATE fires — we look up the form by
+        #       session_key (button's `data` carries the action_id),
+        #   (b) the resumed-workflow query needs to find a passive-reply
+        #       event_id (INTERACTION_CREATE id, 30-min validity).
+        self._pending_forms: dict[str, dict] = {}
+        # session_key -> most recent ``INTERACTION_CREATE`` event_id, used
+        # as the passive event_id for the resumed query's LLM output.
+        self._session_event_ids: dict[str, dict] = {}
+        # Per-anchor msg_seq counter. QQ accepts up to 5 passive replies
+        # per (msg_id|event_id) within 60 min, but each reuse needs a
+        # fresh ``msg_seq`` — re-sending with msg_seq=1 is silently dedup'd.
+        self._anchor_msg_seq: dict[str, int] = {}
+
+        # Wire button-click handler so webhook mode catches INTERACTION_CREATE.
+        # (ws mode is wired separately via on_event in _run_websocket so the
+        # raw payload bypasses get_message's message-only flattening.)
+        @self.bot.on_interaction()
+        async def _on_interaction(event_data: dict, interaction_id: typing.Optional[str]):
+            await self._handle_interaction_create(event_data, interaction_id)
 
     async def reply_message(
         self,
@@ -227,6 +253,13 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
             message_source,
         )
 
+        # Synthetic event (button-click resume): no inbound platform
+        # object → no msg_id. Route via the cached INTERACTION_CREATE
+        # event_id (valid 30 min, no quota cost).
+        if qq_official_event is None:
+            await self._reply_synthetic(message_source, message)
+            return
+
         content_list = await QQOfficialMessageConverter.yiri2target(message)
 
         # 确定 target_type 和 target_id
@@ -376,6 +409,9 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
             await self.logger.info('QQ Official WebSocket connected and ready')
 
         async def on_event(event_type: str, event_data: dict):
+            # INTERACTION_CREATE is dispatched via bot.on_interaction()
+            # (registered in __init__) so we get the top-level ws_event_id
+            # — needed as the passive-reply event_id. It never reaches here.
             # 只处理消息事件，忽略 READY/RESUMED 等系统事件
             message_event_types = {
                 'C2C_MESSAGE_CREATE',
@@ -439,6 +475,12 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
 
     async def create_message_card(self, message_id: str, event: platform_events.MessageEvent) -> bool:
         source = event.source_platform_object
+        # Synthetic events (button-click resume) have no source object —
+        # they ride a cached INTERACTION_CREATE event_id, not a streamable
+        # msg_id. Skip stream setup; reply_message handles the one-shot
+        # send at is_final.
+        if source is None:
+            return False
         # Streaming API only supports C2C private chat
         if source.t != 'C2C_MESSAGE_CREATE':
             return False
@@ -469,6 +511,29 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
     ):
         # Periodically clean up stale stream contexts
         await self._cleanup_stale_streams()
+
+        # Dify human-input pause: when the runner attaches `_form_data` to
+        # the final chunk, finalize any in-flight stream session and send
+        # a markdown + keyboard message instead. Plain-text content from
+        # earlier chunks is already on the stream; we close it cleanly
+        # and the buttons land as a separate reply.
+        form_data = getattr(bot_message, '_form_data', None) if not isinstance(bot_message, dict) else None
+        if is_final:
+            _resume = getattr(bot_message, '_resume_from_form', None) if not isinstance(bot_message, dict) else None
+            _open_new = getattr(bot_message, '_open_new_card', None) if not isinstance(bot_message, dict) else None
+            if self.ap is not None:
+                self.ap.logger.info(
+                    f'QQ Official reply_message_chunk final: '
+                    f'type={type(bot_message).__name__} '
+                    f'is_final={is_final} '
+                    f'form_data_present={form_data is not None} '
+                    f'resume_from_form={_resume} open_new_card={_open_new} '
+                    f'content_len={len(getattr(bot_message, "content", "") or "")}'
+                )
+        if form_data and is_final:
+            await self._handle_form_chunk(message_source, message, form_data)
+            return
+
         # 提取纯文本内容（当前 chunk 的文本）
         text_parts = []
         for msg in message:
@@ -557,3 +622,462 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
         ],
     ):
         return super().unregister_listener(event_type, callback)
+
+    # ------------------------------------------------------------------
+    # Dify human-input button-interaction support
+    # ------------------------------------------------------------------
+
+    _PENDING_FORM_TTL = 1800  # 30 min — matches QQ passive-reply window.
+    _MAX_REPLIES_PER_ANCHOR = 5  # QQ hard limit per msg_id / event_id.
+
+    def _next_msg_seq(self, anchor: str) -> typing.Optional[int]:
+        """Return the next msg_seq for an anchor, or ``None`` if the
+        anchor has already been used 5 times (further sends would be
+        silently dropped by QQ)."""
+        if not anchor:
+            return 1
+        used = self._anchor_msg_seq.get(anchor, 0)
+        if used >= self._MAX_REPLIES_PER_ANCHOR:
+            return None
+        self._anchor_msg_seq[anchor] = used + 1
+        return used + 1
+
+    async def _reply_synthetic(
+        self,
+        message_source: platform_events.MessageEvent,
+        message: platform_message.MessageChain,
+    ) -> None:
+        """Deliver a reply for a synthetic (button-click-resume) event.
+
+        Synthetic events have ``source_platform_object=None`` and no
+        fresh inbound msg_id. The previous INTERACTION_CREATE id we
+        cached in :attr:`_session_event_ids` is a valid passive-reply
+        anchor (``event_id``) for up to 30 minutes — use it.
+        """
+        if isinstance(message_source, platform_events.GroupMessage):
+            target_type = 'group'
+            group = getattr(message_source, 'group', None) or (
+                message_source.sender.group if hasattr(message_source.sender, 'group') else None
+            )
+            target_id = str(group.id) if group else None
+        else:
+            target_type = 'c2c'
+            target_id = str(message_source.sender.id) if message_source.sender else None
+
+        if not target_id:
+            await self.logger.warning('QQ Official: synthetic reply has no target_id; dropping')
+            return
+
+        session_key = f'{target_type}_{target_id}'
+        cached = self._session_event_ids.get(session_key)
+        event_id = cached.get('event_id') if cached else None
+        if cached and (time.time() - cached.get('posted_at', 0)) > self._PENDING_FORM_TTL:
+            event_id = None
+
+        if not event_id:
+            await self.logger.warning(
+                f'QQ Official: no cached event_id for {session_key}; '
+                f'cannot deliver synthetic reply within passive-reply window'
+            )
+            return
+
+        content_list = await QQOfficialMessageConverter.yiri2target(message)
+        text_parts = [c['content'] for c in content_list if c.get('type') == 'text' and c.get('content')]
+        if not text_parts:
+            await self.logger.info('QQ Official: synthetic reply has no text content; skipping')
+            return
+        text = '\n\n'.join(text_parts)
+
+        msg_seq = self._next_msg_seq(event_id)
+        if msg_seq is None:
+            await self.logger.warning(
+                f'QQ Official: anchor {event_id!r} exhausted (>5 passive replies); '
+                f'cannot deliver synthetic reply for {session_key}'
+            )
+            return
+
+        try:
+            if target_type == 'c2c':
+                await self.bot.send_private_text_msg(
+                    user_openid=target_id,
+                    content=text,
+                    event_id=event_id,
+                    msg_seq=msg_seq,
+                )
+            elif target_type == 'group':
+                await self.bot.send_group_text_msg(
+                    group_openid=target_id,
+                    content=text,
+                    event_id=event_id,
+                    msg_seq=msg_seq,
+                )
+        except Exception:
+            await self.logger.error(f'QQ Official: synthetic reply delivery failed: {traceback.format_exc()}')
+
+    def _resolve_target_from_source(self, source: QQOfficialEvent) -> typing.Optional[tuple[str, str]]:
+        """Return ``(target_type, target_id)`` for sending a reply, or
+        ``None`` if the scene cannot host a markdown+keyboard message."""
+        if source is None:
+            return None
+        if source.t == 'C2C_MESSAGE_CREATE':
+            return 'c2c', source.user_openid
+        if source.t == 'GROUP_AT_MESSAGE_CREATE':
+            return 'group', source.group_openid
+        if source.t == 'AT_MESSAGE_CREATE':
+            return 'channel', source.channel_id
+        # DIRECT_MESSAGE_CREATE uses the guild DM API which does not accept
+        # markdown+keyboard at the time of writing — caller falls back to text.
+        return None
+
+    def _resolve_target_from_event(
+        self, message_source: platform_events.MessageEvent
+    ) -> typing.Optional[tuple[str, str]]:
+        """Resolve ``(target_type, target_id)`` from the public event.
+
+        Prefers the platform-native source when present; falls back to
+        the synthesized event's sender/group fields so button-click
+        resume queries can still find a destination.
+        """
+        source = message_source.source_platform_object
+        if source is not None:
+            return self._resolve_target_from_source(source)
+        if isinstance(message_source, platform_events.GroupMessage):
+            group = getattr(message_source, 'group', None) or (
+                message_source.sender.group
+                if message_source.sender and hasattr(message_source.sender, 'group')
+                else None
+            )
+            if group and getattr(group, 'id', None):
+                return 'group', str(group.id)
+        if isinstance(message_source, platform_events.FriendMessage):
+            if message_source.sender and getattr(message_source.sender, 'id', None):
+                return 'c2c', str(message_source.sender.id)
+        return None
+
+    def _prune_pending_forms(self) -> None:
+        now = time.time()
+        stale = [k for k, v in self._pending_forms.items() if now - v.get('posted_at', 0) > self._PENDING_FORM_TTL]
+        for k in stale:
+            self._pending_forms.pop(k, None)
+        stale_e = [
+            k for k, v in self._session_event_ids.items() if now - v.get('posted_at', 0) > self._PENDING_FORM_TTL
+        ]
+        for k in stale_e:
+            self._session_event_ids.pop(k, None)
+
+    async def _handle_form_chunk(
+        self,
+        message_source: platform_events.MessageEvent,
+        message: platform_message.MessageChain,
+        form_data: dict,
+    ) -> None:
+        """Send the markdown + keyboard form prompt for a Dify pause.
+
+        Called from ``reply_message_chunk`` when the runner attaches
+        ``_form_data`` to the final chunk. Replaces what would otherwise
+        be a plain-text numbered-list fallback.
+        """
+        if self.ap is not None:
+            self.ap.logger.info(
+                f'QQ Official _handle_form_chunk entered; '
+                f'source_present={message_source.source_platform_object is not None} '
+                f'form_actions={len(form_data.get("actions") or [])}'
+            )
+        self._prune_pending_forms()
+
+        source = message_source.source_platform_object
+        scene_target = self._resolve_target_from_event(message_source)
+        if scene_target is None:
+            # No rich-UI fit — fall through to existing text path.
+            await self.logger.info('QQ Official: form chunk on unsupported scene; falling back to text')
+            text_parts = [m.text for m in message if type(m) is platform_message.Plain]
+            fallback_msg = platform_message.MessageChain([platform_message.Plain(text='\n\n'.join(text_parts))])
+            try:
+                await self.reply_message(message_source, fallback_msg)
+            except Exception:
+                await self.logger.error(f'QQ Official: form fallback text send failed: {traceback.format_exc()}')
+            return
+
+        target_type, target_id = scene_target
+        session_key = f'{target_type}_{target_id}'
+
+        # Cancel any in-flight stream / fallback ctx so plain-text prefix
+        # doesn't continue alongside the keyboard message.
+        msg_id = getattr(source, 'd_id', '') or '' if source is not None else ''
+        if msg_id:
+            self._stream_ctx.pop(msg_id, None)
+            self._stream_ctx_ts.pop(msg_id, None)
+            self._fallback_text.pop(msg_id, None)
+            self._fallback_text_ts.pop(msg_id, None)
+
+        node_title = form_data.get('node_title') or 'Confirmation needed'
+        form_content = form_data.get('form_content') or ''
+        parts = [f'### {node_title}']
+        if form_content.strip():
+            parts.append(form_content.strip())
+        parts.append('请点击下方按钮选择：')
+        markdown_content = '\n\n'.join(parts)
+
+        keyboard = build_keyboard_from_form(form_data, buttons_per_row=2)
+        if not keyboard.get('content', {}).get('rows'):
+            # No actions to render — fall back to plain text.
+            text_msg = platform_message.MessageChain([platform_message.Plain(text=markdown_content)])
+            try:
+                await self.reply_message(message_source, text_msg)
+            except Exception:
+                await self.logger.error(f'QQ Official: empty-keyboard fallback send failed: {traceback.format_exc()}')
+            return
+
+        # Prefer the inbound msg_id (no quota cost). If the source is a
+        # synthetic event from a prior click, the cached interaction id
+        # serves as event_id for up to 30 min.
+        event_id = None
+        if not msg_id:
+            cached = self._session_event_ids.get(session_key)
+            if cached and (time.time() - cached.get('posted_at', 0)) < self._PENDING_FORM_TTL:
+                event_id = cached.get('event_id')
+
+        anchor = msg_id or event_id or ''
+        msg_seq = self._next_msg_seq(anchor)
+        if msg_seq is None:
+            await self.logger.warning(
+                f'QQ Official: anchor {anchor!r} exhausted (>5 passive replies); '
+                f'cannot deliver form card for session={session_key}'
+            )
+            return
+
+        try:
+            await self.bot.send_markdown_keyboard(
+                target_type=target_type,
+                target_id=target_id,
+                markdown_content=markdown_content,
+                keyboard=keyboard,
+                msg_id=msg_id if (msg_id and not event_id) else None,
+                event_id=event_id,
+                msg_seq=msg_seq,
+            )
+            if self.ap is not None:
+                self.ap.logger.info(
+                    f'QQ Official: form card sent '
+                    f'target={target_type}/{target_id} '
+                    f'msg_id={msg_id!r} event_id={event_id!r} msg_seq={msg_seq}'
+                )
+        except Exception:
+            if self.ap is not None:
+                self.ap.logger.error(
+                    f'QQ Official: send_markdown_keyboard failed, falling back to text: {traceback.format_exc()}'
+                )
+            await self.logger.error(
+                f'QQ Official: send_markdown_keyboard failed, falling back to text: {traceback.format_exc()}'
+            )
+            text_msg = platform_message.MessageChain([platform_message.Plain(text=markdown_content)])
+            try:
+                await self.reply_message(message_source, text_msg)
+            except Exception:
+                pass
+            return
+
+        sender_id = ''
+        if source is not None:
+            sender_id = (
+                getattr(source, 'user_openid', None)
+                or getattr(source, 'member_openid', None)
+                or getattr(source, 'd_author_id', None)
+                or ''
+            )
+        if not sender_id and message_source.sender is not None:
+            sender_id = str(getattr(message_source.sender, 'id', '') or '')
+        self._pending_forms[session_key] = {
+            'form_data': form_data,
+            'msg_id': msg_id,
+            'sender_id': sender_id,
+            'target_type': target_type,
+            'target_id': target_id,
+            'source_event_t': source.t if source is not None else None,
+            'posted_at': time.time(),
+        }
+        await self.logger.info(
+            f'QQ Official: form posted session={session_key} actions={len(form_data.get("actions") or [])}'
+        )
+
+    async def _handle_interaction_create(
+        self,
+        event_data: dict,
+        ws_event_id: typing.Optional[str] = None,
+    ) -> None:
+        """Handle a button-click INTERACTION_CREATE event.
+
+        Two IDs at play (QQ keeps them separate):
+            ws_event_id   top-level payload ``id`` (or webhook ``X-Bot-
+                          Event-Id``). The ONLY value accepted as
+                          ``event_id`` for subsequent passive replies.
+            d['id']       the interaction id — used for PUT
+                          /interactions/{id} ack. Cannot be reused as
+                          event_id (QQ returns 40034025 if you try).
+
+        Layout (https://bot.q.qq.com/.../msg-btn.html):
+            chat_type     0 channel / 1 group / 2 c2c
+            data.resolved.button_data  what we set as ``action.data``
+            data.resolved.button_id    ``id`` field on the button row
+        """
+        import langbot_plugin.api.entities.builtin.provider.session as provider_session
+
+        if self.ap is not None:
+            self.ap.logger.info(
+                f'QQ Official _handle_interaction_create entered; '
+                f'ws_event_id={ws_event_id!r} '
+                f'interaction_id={(event_data.get("id") if isinstance(event_data, dict) else None)!r} '
+                f'chat_type={event_data.get("chat_type") if isinstance(event_data, dict) else None}'
+            )
+
+        if not isinstance(event_data, dict):
+            await self.logger.warning(f'QQ Official: INTERACTION_CREATE event_data is not dict: {type(event_data)}')
+            return
+
+        # ACK uses the interaction id, NOT the ws event id.
+        interaction_id = event_data.get('id') or ''
+        if interaction_id:
+            asyncio.create_task(self.bot.ack_interaction(interaction_id, code=0))
+
+        resolved = (event_data.get('data') or {}).get('resolved') or {}
+        action_id = str(resolved.get('button_data') or resolved.get('button_id') or '').strip()
+        if not action_id:
+            await self.logger.warning('QQ Official: INTERACTION_CREATE missing button_data/button_id; ignoring')
+            return
+
+        chat_type = event_data.get('chat_type')
+        scene_target: typing.Optional[tuple[str, str]] = None
+        if chat_type == 2 or event_data.get('user_openid'):
+            scene_target = ('c2c', event_data.get('user_openid') or '')
+        elif chat_type == 1 or event_data.get('group_openid'):
+            scene_target = ('group', event_data.get('group_openid') or '')
+        elif chat_type == 0 or event_data.get('channel_id'):
+            scene_target = ('channel', event_data.get('channel_id') or '')
+
+        if not scene_target or not scene_target[1]:
+            await self.logger.warning(f'QQ Official: INTERACTION_CREATE missing scene/target; raw={event_data}')
+            return
+
+        target_type, target_id = scene_target
+        session_key = f'{target_type}_{target_id}'
+
+        self._prune_pending_forms()
+        pending = self._pending_forms.pop(session_key, None)
+        if not pending:
+            await self.logger.warning(
+                f'QQ Official: no pending form for session {session_key}; click ignored (action_id={action_id!r})'
+            )
+            return
+
+        # Cache ws_event_id so a follow-up pause / text reply can use it
+        # as event_id for passive delivery (30-min window). Falls back to
+        # the interaction_id only if no ws_event_id was provided (e.g.
+        # tests / older payload shape) — QQ will reject that value but
+        # we log so the mismatch is debuggable.
+        cached_event_id = ws_event_id or interaction_id
+        if cached_event_id:
+            self._session_event_ids[session_key] = {
+                'event_id': cached_event_id,
+                'posted_at': time.time(),
+            }
+            # New anchor → fresh 5-reply budget.
+            self._anchor_msg_seq[cached_event_id] = 0
+            if self.ap is not None and not ws_event_id:
+                self.ap.logger.warning(
+                    'QQ Official: INTERACTION_CREATE lacked ws_event_id; '
+                    'falling back to interaction_id (passive reply may be rejected)'
+                )
+
+        form_data: dict = pending.get('form_data') or {}
+        actions = form_data.get('actions') or []
+        matched = next(
+            (a for a in actions if str(a.get('id', '')) == action_id),
+            None,
+        )
+        action_title = (matched or {}).get('title') or action_id
+
+        sender_id = pending.get('sender_id') or event_data.get('user_openid') or event_data.get('member_openid') or ''
+
+        # Build resume payload matching the shape every other adapter uses
+        # (DingTalk / Lark / Telegram / WeCom). The runner's
+        # _merge_pending_form_action consumes this verbatim.
+        if target_type == 'group' or target_type == 'channel':
+            launcher_type = provider_session.LauncherTypes.GROUP
+            launcher_id = target_id
+        else:
+            launcher_type = provider_session.LauncherTypes.PERSON
+            launcher_id = sender_id or target_id
+
+        form_action_data = {
+            'form_token': form_data.get('form_token', ''),
+            'workflow_run_id': form_data.get('workflow_run_id', ''),
+            'action_id': action_id,
+            'action_title': action_title,
+            'node_title': form_data.get('node_title', ''),
+            'user': f'{launcher_type.value}_{launcher_id}',
+            'inputs': {},
+        }
+
+        message_chain = platform_message.MessageChain([platform_message.Plain(text=f'[Form Action: {action_title}]')])
+
+        if launcher_type == provider_session.LauncherTypes.GROUP:
+            synthetic_event: platform_events.MessageEvent = platform_events.GroupMessage(
+                sender=platform_entities.GroupMember(
+                    id=sender_id or launcher_id,
+                    member_name='',
+                    permission='MEMBER',
+                    group=platform_entities.Group(
+                        id=launcher_id,
+                        name='',
+                        permission=platform_entities.Permission.Member,
+                    ),
+                    special_title='',
+                ),
+                message_chain=message_chain,
+                time=int(time.time()),
+                source_platform_object=None,
+            )
+        else:
+            synthetic_event = platform_events.FriendMessage(
+                sender=platform_entities.Friend(
+                    id=sender_id or launcher_id,
+                    nickname='',
+                    remark='',
+                ),
+                message_chain=message_chain,
+                time=int(time.time()),
+                source_platform_object=None,
+            )
+
+        if self.ap is None:
+            await self.logger.error('QQ Official: ap not injected; cannot enqueue button-click query')
+            return
+
+        bot_uuid = ''
+        pipeline_uuid = None
+        for bot in self.ap.platform_mgr.bots:
+            if bot.adapter is self:
+                bot_uuid = bot.bot_entity.uuid
+                pipeline_uuid = bot.bot_entity.use_pipeline_uuid
+                break
+
+        try:
+            await self.ap.query_pool.add_query(
+                bot_uuid=bot_uuid,
+                launcher_type=launcher_type,
+                launcher_id=launcher_id,
+                sender_id=sender_id or launcher_id,
+                message_event=synthetic_event,
+                message_chain=message_chain,
+                adapter=self,
+                pipeline_uuid=pipeline_uuid,
+                variables={
+                    '_dify_form_action': form_action_data,
+                    '_routed_by_rule': True,
+                },
+            )
+            await self.logger.info(
+                f'QQ Official: button-click query enqueued action_id={action_id!r} session={session_key}'
+            )
+        except Exception:
+            await self.logger.error(f'QQ Official: enqueue button-click query failed: {traceback.format_exc()}')