diff --git a/README.md b/README.md index bc1bd07ebc..92cdc8ff59 100644 --- a/README.md +++ b/README.md @@ -1080,6 +1080,7 @@ DINGTALK_CLIENT_SECRET=your_client_secret 3. Under **Events**, subscribe to `im.message.receive_v1` and select **Long Connection** mode. 4. Copy the App ID and App Secret. Set `FEISHU_APP_ID` and `FEISHU_APP_SECRET` in `.env` and enable the channel in `config.yaml`. 5. The bot supports inbound text, image, and file messages. Inbound attachment downloads are limited to 20 MB per attachment. +6. _(Optional)_ Interactive clarification cards: set `card_interaction_enabled: true` under `channels.feishu` to let the agent's option-based clarifications render as clickable button cards, preserving any background context before the question. Then, in the developer console, open **Callback Configuration** (回调配置 — a separate tab from **Event Subscriptions**), subscribe to `card.action.trigger` in **Long Connection** mode, and publish a new app version. A click resolves the clarification exactly like typing the choice; only the user who triggered the clarification can operate its card. Button answers are saved in the conversation history, and subsequent clarifications continue in the same thread. **WeChat Setup** diff --git a/README_zh.md b/README_zh.md index 4dde4f6a64..437351e457 100644 --- a/README_zh.md +++ b/README_zh.md @@ -541,6 +541,7 @@ DINGTALK_CLIENT_SECRET=your_client_secret 2. 添加权限:`im:message`、`im:message.p2p_msg:readonly`、`im:resource`。 3. 在 **事件订阅** 中订阅 `im.message.receive_v1`,连接方式选择 **长连接**。 4. 复制 App ID 和 App Secret,在 `.env` 中设置 `FEISHU_APP_ID` 和 `FEISHU_APP_SECRET`,并在 `config.yaml` 中启用该渠道。 +5. _(可选)_ 交互式澄清卡片:在 `channels.feishu` 下设置 `card_interaction_enabled: true`,让 agent 的选项式提问渲染为可点击的按钮卡片,并在问题前保留背景说明。随后在开发者后台打开 **回调配置**(与「事件订阅」是不同的标签页),订阅 `card.action.trigger` 并选择 **长连接**,最后发布新版本。点击按钮等同于输入该选项作答;只有触发该次提问的用户才能操作对应卡片。按钮答案会保存到对话历史中,后续澄清继续在同一会话内进行。 **WeChat 配置** diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 58cfde97c1..86043decaa 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -175,7 +175,7 @@ More specific `AGENTS.md` files in backend code directories contain the subsyste The backend is split into two layers with a strict dependency direction: - **Harness** (`packages/harness/deerflow/`): Publishable agent framework package (`deerflow-harness`). Import prefix: `deerflow.*`. Contains agent orchestration, tools, sandbox, models, MCP, skills, config — everything needed to build and run agents. -- **App** (`app/`): Unpublished application code. Import prefix: `app.*`. Contains the FastAPI Gateway API and IM channel integrations (Feishu, Slack, Telegram, DingTalk). +- **App** (`app/`): Unpublished application code. Import prefix: `app.*`. Contains the FastAPI Gateway API and IM channel integrations (Feishu, Slack, Telegram, DingTalk). Feishu option clarifications can render as interactive button cards behind `channels.feishu.card_interaction_enabled` (default off); clicks are originator-validated and one-time, and fail closed across workers — ownership lives in `app/channels/feishu_card_interaction.py`. **Dependency rule**: App imports deerflow, but deerflow never imports app. This boundary is enforced by `tests/test_harness_boundary.py` which runs in CI. diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 06cd79233c..79e46cb8e9 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -36,7 +36,7 @@ Cursor writes drain before cancellation; token expiry drains cursor/auth clearin 5. Feishu/Telegram chat: `runs.stream()` → accumulate AI text → publish multiple outbound updates (`is_final=False`) → publish final outbound (`is_final=True`) 6. Slack/Discord chat: `runs.wait()` → extract final response → publish outbound 6b. GitHub chat (`ChannelRunPolicy.fire_and_forget=True`): `runs.create()` returns once the run is `pending`; the manager does not wait for the final state and does not publish an outbound; the agent posts its own reply via sandbox `gh`. `ConflictError` on a busy thread still trips the standard `THREAD_BUSY_MESSAGE` path (log-only on GitHub); with `buffer_followups_on_busy=True` (GitHub's default) the triggering message is captured into a per-thread buffer so a concurrent comment is not silently dropped (see "Follow-up buffering while busy"). -7. Feishu channel sends one running reply card up front, then patches the same card for each outbound update (card JSON sets `config.update_multi=true` for Feishu's patch API requirement). Messages already sent inside an existing Feishu topic carry a compact source-message preview in that card, and queued same-thread follow-ups patch their own source message's card from queued → running → final without falling back to the generic busy reply. +7. Feishu patches one card per source (`config.update_multi=true`), with topic previews and queued → running → final states. Opt-in cards show context/question; bounded tokens preserve request ID/source. Originator/option checks consume once. Card replies keep bound aliases in fields/metadata; ordinary replies preserve `/new`. Accepted intake retires only that card's buttons/text hint before terminal PATCH; rejected intake keeps text fallback. 8. Telegram streaming: the "Working on it..." placeholder message is registered as the stream target; non-final updates `editMessageText` it in place (channel-side throttle: 1s in private chats, 3s in groups due to Telegram's 20 msg/min group cap; 4096-char truncation; rate-limited updates dropped); the final update performs the last edit and splits >4096 texts into follow-up messages 9. DingTalk AI Card mode (when `card_template_id` configured): `runs.stream()` → create card with initial text → stream updates via `PUT /v1.0/card/streaming` → finalize on `is_final=True`. Falls back to `sampleMarkdown` if card creation or streaming fails 10. For commands (`/new`, `/status`, `/models`, `/model`, `/memory`, `/goal`, `/agent`, `/help`): handle locally or query Gateway API. `/agent list` reads the effective owner's Custom Agents; `/agent use ` validates in the same owner bucket, creates a new thread, and persists `channel_agent_name` in its metadata. A Custom Agent writes canonical `metadata.agent_name` so thread-search routes Web continuation to `/workspace/agents//chats/`; `lead_agent` omits it and stays on the ordinary chat route. diff --git a/backend/app/channels/feishu.py b/backend/app/channels/feishu.py index e677c46387..dad37f3cfc 100644 --- a/backend/app/channels/feishu.py +++ b/backend/app/channels/feishu.py @@ -18,8 +18,15 @@ from typing import Any, Literal, TypeVar from app.channels.base import Channel, ChannelStopTimeout from app.channels.commands import is_known_channel_command, strip_leading_mentions -from app.channels.connection_identity import attach_connection_identity +from app.channels.connection_identity import attach_connection_identity, lookup_thread_id +from app.channels.feishu_card_interaction import ( + ClarificationCardRequest, + PendingCardEntry, + PendingCardRegistry, + build_button_card, +) from app.channels.message_bus import ( + HUMAN_INPUT_RESPONSE_METADATA_KEY, PENDING_CLARIFICATION_METADATA_KEY, RESOLVED_FROM_PENDING_CLARIFICATION_METADATA_KEY, InboundMessage, @@ -42,6 +49,7 @@ from deerflow.uploads.manager import ( logger = logging.getLogger(__name__) PENDING_CLARIFICATION_TTL_SECONDS = 30 * 60 +_PENDING_CARD_METADATA_KEY = "_feishu_pending_card_id" # Channel-store deadlines. ``_on_message`` runs on lark-oapi's private event loop, # which also carries the SDK's websocket reads, pongs and its 120s ping loop: # every second the bridge blocks that loop is a second the connection cannot @@ -104,6 +112,7 @@ class _Routing: topic_id: str | None from_stored_mapping: bool from_pending: bool + clarification_card_id: str | None = None @dataclass(slots=True) @@ -168,6 +177,12 @@ class FeishuChannel(Channel): self._running_card_ids: dict[str, str] = {} self._running_card_tasks: dict[str, asyncio.Task] = {} self._pending_clarifications: dict[tuple[str, str], list[dict[str, Any]]] = {} + # Interactive clarification cards (button answers). Off by default: the + # feature needs a one-time console subscription to card.action.trigger. + self._card_interaction_enabled = bool(config.get("card_interaction_enabled", False)) + self._card_registry = PendingCardRegistry() + self._CallBackToast = None + self._P2CardActionTriggerResponse = None self._pending_inbound_batches: dict[tuple[str, str], dict[str, Any]] = {} self._CreateFileRequest = None self._CreateFileRequestBody = None @@ -240,15 +255,20 @@ class FeishuChannel(Channel): return self._thread is not None and self._thread.is_alive() def _build_event_handler(self, lark): - return ( + builder = ( lark.EventDispatcherHandler.builder("", "") .register_p2_im_message_receive_v1(self._on_message) .register_p2_im_message_message_read_v1(self._on_ignored_message_event) .register_p2_im_message_reaction_created_v1(self._on_ignored_message_event) .register_p2_im_message_reaction_deleted_v1(self._on_ignored_message_event) .register_p2_im_message_recalled_v1(self._on_ignored_message_event) - .build() ) + if self._card_interaction_enabled: + # Card clicks arrive as WS EVENT frames routed through the SDK's + # callback processor map (verified against lark-oapi source); the + # synchronous return value becomes the click's toast ack. + builder = builder.register_p2_card_action_trigger(self._on_card_action) + return builder.build() async def start(self) -> None: if self._running: @@ -277,6 +297,13 @@ class FeishuChannel(Channel): return self._lark = lark + from lark_oapi.event.callback.model.p2_card_action_trigger import ( + CallBackToast, + P2CardActionTriggerResponse, + ) + + self._CallBackToast = CallBackToast + self._P2CardActionTriggerResponse = P2CardActionTriggerResponse self._CreateMessageRequest = CreateMessageRequest self._CreateMessageRequestBody = CreateMessageRequestBody self._ReplyMessageRequest = ReplyMessageRequest @@ -667,12 +694,12 @@ class FeishuChannel(Channel): except Exception: logger.exception("[Feishu] failed to add reaction '%s' to message %s", emoji_type, message_id) - async def _reply_card(self, message_id: str, text: str) -> str | None: - """Reply with an interactive card and return the created card message ID.""" + async def _reply_card(self, message_id: str, text: str, *, raw_card_content: bool = False) -> str | None: + """Reply with text or serialized card JSON and return the card message ID.""" if not self._api_client: return None - content = self._build_card_content(text) + content = text if raw_card_content else self._build_card_content(text) request = self._ReplyMessageRequest.builder().message_id(message_id).request_body(self._ReplyMessageRequestBody.builder().msg_type("interactive").content(content).build()).build() response = await asyncio.to_thread(self._api_client.im.v1.message.reply, request) if not response.success(): @@ -691,12 +718,15 @@ class FeishuChannel(Channel): if not response.success(): raise RuntimeError(f"Feishu card creation failed: code={response.code}, msg={response.msg}, log_id={response.get_log_id()}") - async def _update_card(self, message_id: str, text: str) -> None: - """Patch an existing card message in place.""" + async def _update_card(self, message_id: str, text: str, *, raw_card_content: bool = False) -> None: + """Patch an existing card message in place. + + ``raw_card_content=True`` means *text* is already a serialized card JSON. + """ if not self._api_client or not self._PatchMessageRequest: return - content = self._build_card_content(text) + content = text if raw_card_content else self._build_card_content(text) request = self._PatchMessageRequest.builder().message_id(message_id).request_body(self._PatchMessageRequestBody.builder().content(content).build()).build() response = await asyncio.to_thread(self._api_client.im.v1.message.patch, request) if not response.success(): @@ -781,9 +811,71 @@ class FeishuChannel(Channel): except Exception: logger.exception("[Feishu] failed to send running reply for message %s", message_id) + async def _bind_card_thread(self, msg: OutboundMessage) -> str | None: + """Bind answers to their thread without changing the chat's /new route.""" + user_id = self._non_empty_str(msg.metadata.get("user_id")) + if not (user_id and msg.thread_id and msg.thread_ts): + return None + # Provider message IDs cannot address this namespace, and the alias + # is reused across a thread's cards without exposing callback tokens. + topic_id = f"feishu-card:{msg.thread_id}" + try: + if msg.connection_id: + if self._connection_repo is None or not msg.owner_user_id: + return None + await self._connection_repo.set_thread_id(connection_id=msg.connection_id, owner_user_id=msg.owner_user_id, provider=self.name, external_conversation_id=msg.chat_id, thread_id=msg.thread_id, external_topic_id=topic_id) + else: + store = self.config.get("channel_store") + if store is None: + return None + await store.set_thread_id(self.name, msg.chat_id, msg.thread_id, topic_id=topic_id, user_id=user_id) + except Exception: + logger.exception("[Feishu] cannot bind clarification card to thread %s; using text fallback", msg.thread_id) + return None + return topic_id + + def _register_pending_card(self, msg: OutboundMessage, request: ClarificationCardRequest, card_message_id: str | None, *, topic_id: str) -> None: + """Register a delivered button card so its clicks can be validated later.""" + user_id = self._non_empty_str(msg.metadata.get("user_id")) + if not (user_id and msg.thread_id and card_message_id): + logger.warning("[Feishu] button card delivered without full binding context, clicks will fail closed: request_id=%s", request.request_id) + return + self._card_registry.register( + callback_token=request.callback_token, + request_id=request.request_id, + source=request.source, + thread_id=msg.thread_id, + chat_id=msg.chat_id, + originator_open_id=user_id, + options=request.options, + option_ids=request.option_ids, + card_message_id=card_message_id, + topic_id=topic_id, + connection_id=msg.connection_id, + owner_user_id=msg.owner_user_id, + ) + + async def _remember_card_delivery(self, msg: OutboundMessage, request: ClarificationCardRequest | None, card_message_id: str | None, *, topic_id: str | None) -> None: + # Admit callbacks immediately after the SDK returns. Auxiliary message + # aliases may involve slow storage, but the thread alias is already bound. + if request and topic_id: + self._register_pending_card(msg, request, card_message_id, topic_id=topic_id) + self._remember_pending_clarification(msg, card_message_id, topic_id=topic_id) + await self._remember_thread_mapping(msg, msg.thread_ts, card_message_id) + async def _send_card_message(self, msg: OutboundMessage) -> None: """Send or update the Feishu card tied to the current request.""" source_message_id = msg.thread_ts + # Option clarifications render as an interactive button card on the + # final outbound; everything else keeps the plain markdown card. + card_request = ClarificationCardRequest.from_metadata(msg.metadata) if (self._card_interaction_enabled and msg.is_final) else None + card_topic_id = await self._bind_card_thread(msg) if card_request else None + if card_topic_id is None: + card_request = None + card_text = build_button_card(card_request) if card_request else self._compose_card_text(msg.text, msg.metadata) + # Button cards arrive as serialized card JSON and bypass the markdown + # wrapper; the plain path keeps its exact historical call signature. + raw_card_kw: dict[str, bool] = {"raw_card_content": True} if card_request else {} if source_message_id: running_card_id = self._running_card_ids.get(source_message_id) awaited_running_card_task = False @@ -795,9 +887,8 @@ class FeishuChannel(Channel): running_card_id = await running_card_task if running_card_id: - card_text = self._compose_card_text(msg.text, msg.metadata) try: - await self._update_card(running_card_id, card_text) + await self._update_card(running_card_id, card_text, **raw_card_kw) except Exception: if not msg.is_final: raise @@ -805,20 +896,18 @@ class FeishuChannel(Channel): "[Feishu] failed to patch running card %s, falling back to final reply", running_card_id, ) - fallback_card_id = await self._reply_card(source_message_id, card_text) - await self._remember_thread_mapping(msg, source_message_id, fallback_card_id) - self._remember_pending_clarification(msg, fallback_card_id) + fallback_card_id = await self._reply_card(source_message_id, card_text, **raw_card_kw) + await self._remember_card_delivery(msg, card_request, fallback_card_id, topic_id=card_topic_id) else: - await self._remember_thread_mapping(msg, source_message_id, running_card_id) - self._remember_pending_clarification(msg, running_card_id) + await self._remember_card_delivery(msg, card_request, running_card_id, topic_id=card_topic_id) logger.info("[Feishu] running card updated: source=%s card=%s", source_message_id, running_card_id) elif msg.is_final: final_card_id = await self._reply_card( source_message_id, - self._compose_card_text(msg.text, msg.metadata), + card_text, + **raw_card_kw, ) - await self._remember_thread_mapping(msg, source_message_id, final_card_id) - self._remember_pending_clarification(msg, final_card_id) + await self._remember_card_delivery(msg, card_request, final_card_id, topic_id=card_topic_id) elif awaited_running_card_task: logger.warning( "[Feishu] running card task finished without message_id for source=%s, skipping duplicate non-final creation", @@ -846,13 +935,12 @@ class FeishuChannel(Channel): if store is None or not msg.thread_id: return - metadata_topic_ids = [ - msg.metadata.get("message_id"), - msg.metadata.get("root_id"), - msg.metadata.get("parent_id"), - msg.metadata.get("thread_id"), - msg.metadata.get("topic_id"), - ] + topic_id = self._non_empty_str(msg.metadata.get("topic_id")) + metadata_topic_ids = [msg.metadata.get("message_id"), topic_id] + if not (topic_id and topic_id.startswith("feishu-card:")): + metadata_topic_ids.extend([msg.metadata.get("root_id"), msg.metadata.get("parent_id"), msg.metadata.get("thread_id")]) + # A reply to an older card stays in that card's thread. Its inherited + # root/parent must not overwrite the conversation switched by /new. user_id = "" raw_user_id = msg.metadata.get("user_id") if isinstance(raw_user_id, str): @@ -927,12 +1015,124 @@ class FeishuChannel(Channel): logger.info("[Feishu] %s cancelled by channel stop", name) raise StoreLookupUnresolved("cancelled by channel stop") from None - def _remember_pending_clarification(self, msg: OutboundMessage, card_message_id: str | None) -> None: + def _card_action_toast(self, content: str, *, toast_type: str = "info"): + """Build the synchronous toast ack for a card click.""" + toast = self._CallBackToast() + toast.type = toast_type + toast.content = content + resp = self._P2CardActionTriggerResponse() + resp.toast = toast + return resp + + def _on_card_action(self, data): + """Sync card-click callback on the lark SDK loop: validate, ack, dispatch. + + Only validation and the toast ack happen on this thread; the state + change (terminal card update + hidden synthetic inbound) is dispatched + to the Gateway loop. Registry consumption is one-time, so a validated + click can never be replayed, and a non-originator click leaves the + entry intact for the real originator. + """ + try: + value = getattr(data.event.action, "value", None) or {} + callback_token = value.get("callback_token") + option_id = value.get("option_id") + option_value = value.get("option_value") + operator_open_id = getattr(data.event.operator, "open_id", None) + except AttributeError: + return self._card_action_toast("无法处理的卡片操作", toast_type="warning") + if not all(isinstance(item, str) and item for item in (callback_token, option_id, option_value, operator_open_id)): + return self._card_action_toast("无法处理的卡片操作", toast_type="warning") + + entry, reason = self._card_registry.consume_if_originator(callback_token, operator_open_id, option_id=option_id, option_value=option_value) + if reason == "not_originator": + return self._card_action_toast("仅提问发起人可操作", toast_type="warning") + if entry is None: + return self._card_action_toast("此卡片已失效或已处理", toast_type="warning") + + scheduled = False + if self._main_loop and self._main_loop.is_running(): + scheduled = bool( + self._submit_threadsafe_coroutine( + self._resolve_card_click(entry, option_id=option_id, option_value=option_value, operator_open_id=operator_open_id), + self._main_loop, + name="resolve_card_click", + msg_id=entry.get("card_message_id") or entry["request_id"], + ) + ) + if not scheduled: + # Never ack success for a click we could not dispatch: the entry is + # already consumed, so an honest warning beats a fake "已选择". + logger.warning("[Feishu] card click could not be dispatched (no running loop or loop closing): request_id=%s", entry["request_id"]) + return self._card_action_toast("处理失败,请改用文字回复", toast_type="warning") + return self._card_action_toast(f"已选择:{option_value}", toast_type="success") + + async def _resolve_card_click(self, entry: PendingCardEntry, *, option_id: str, option_value: str, operator_open_id: str) -> None: + """Apply a validated click: inject the hidden human-input response first + (carrying the routing identity the ChannelManager keys on), then flip the + card to its terminal state only after the answer is admitted to the bus — + a dropped inbound must never leave a card claiming a choice was recorded.""" + card_message_id = entry.get("card_message_id") + response = { + "version": 1, + "kind": "human_input_response", + "source": entry["source"], + "request_id": entry["request_id"], + "response_kind": "option", + "option_id": option_id, + "value": option_value, + } + inbound = self._make_inbound( + chat_id=entry["chat_id"], + user_id=operator_open_id, + text=option_value, + thread_ts=card_message_id, + metadata={ + "message_id": card_message_id, + "user_id": operator_open_id, + "topic_id": entry.get("topic_id"), + "hide_from_ui": True, + HUMAN_INPUT_RESPONSE_METADATA_KEY: response, + }, + ) + inbound.topic_id = entry.get("topic_id") + inbound.connection_id = entry.get("connection_id") + inbound.owner_user_id = entry.get("owner_user_id") + # Bound-identity admission re-reads the connection with the message's + # workspace_id, and Feishu connections are bound with workspace_id = + # chat_id (see _bind_connection_from_connect_code). Without this the + # re-read misses and every click is rejected as unbound. + inbound.workspace_id = entry.get("chat_id") + + published = await self._publish_inbound_or_drop(inbound) + if not published: + logger.error("[Feishu] card answer dropped by intake queue; card left clickable but the entry is consumed (request_id=%s)", entry["request_id"]) + return + + self._forget_pending_card_clarification(entry) + if card_message_id: + try: + await self._update_card(card_message_id, f"✅ 已选择:{option_value}") + except Exception: + logger.exception("[Feishu] failed to mark card resolved: %s", card_message_id) + + def _forget_pending_card_clarification(self, entry: PendingCardEntry) -> None: + """Retire only this accepted card's plain-text continuation hint.""" + key = self._pending_key(entry["chat_id"], entry["originator_open_id"]) + with self._thread_lock: + pending_items = self._pending_clarifications.get(key, []) + remaining = [pending for pending in pending_items if not (pending["card_message_id"] == entry.get("card_message_id") and pending["thread_id"] == entry["thread_id"])] + if remaining: + self._pending_clarifications[key] = remaining + else: + self._pending_clarifications.pop(key, None) + + def _remember_pending_clarification(self, msg: OutboundMessage, card_message_id: str | None, *, topic_id: str | None = None) -> None: if not msg.is_final or msg.metadata.get(PENDING_CLARIFICATION_METADATA_KEY) is not True: return user_id = self._non_empty_str(msg.metadata.get("user_id")) - topic_id = self._non_empty_str(msg.metadata.get("topic_id")) + topic_id = topic_id or self._non_empty_str(msg.metadata.get("topic_id")) source_message_id = self._non_empty_str(msg.thread_ts) or self._non_empty_str(msg.metadata.get("message_id")) if not (user_id and topic_id and msg.thread_id and source_message_id and card_message_id): return @@ -958,7 +1158,7 @@ class FeishuChannel(Channel): msg.thread_id, ) - def _consume_pending_clarification(self, chat_id: str, user_id: str) -> dict[str, Any] | None: + def _consume_pending_clarification(self, chat_id: str, user_id: str, *, reply_to_ids: list[str] | None = None) -> dict[str, Any] | None: key = self._pending_key(chat_id, user_id) with self._thread_lock: pending_items = self._pending_clarifications.get(key) @@ -966,19 +1166,23 @@ class FeishuChannel(Channel): return None now = time.time() - while pending_items: - pending = pending_items.pop(0) - created_at = pending.get("created_at") - if isinstance(created_at, (int, float)) and now - created_at <= PENDING_CLARIFICATION_TTL_SECONDS: - if pending_items: - self._pending_clarifications[key] = pending_items - else: - self._pending_clarifications.pop(key, None) - return pending - logger.info("[Feishu] pending clarification expired: chat_id=%s user_id=%s", chat_id, user_id) - - self._pending_clarifications.pop(key, None) - return None + pending_items = [pending for pending in pending_items if isinstance(pending.get("created_at"), (int, float)) and now - pending["created_at"] <= PENDING_CLARIFICATION_TTL_SECONDS] + # Explicit replies target a card/source, while accepted ordinary + # replies use their current thread alias. Shared roots cannot select + # a card belonging to a conversation replaced by /new. + if reply_to_ids is None: + matched = 0 if pending_items else None + else: + matched = next( + (index for field in ("card_message_id", "source_message_id", "topic_id") for candidate in reply_to_ids for index, pending in enumerate(pending_items) if pending.get(field) == candidate), + None, + ) + pending = pending_items.pop(matched) if matched is not None else None + if pending_items: + self._pending_clarifications[key] = pending_items + else: + self._pending_clarifications.pop(key, None) + return pending def _route_without_store(self, parsed: _ParsedMessage) -> _Routing: """Routing for a channel constructed without a ``channel_store`` (no lookups possible).""" @@ -987,11 +1191,14 @@ class FeishuChannel(Channel): topic_id = None if parsed.msg_type != InboundMessageType.CHAT: return _Routing(topic_id, False, False) - pending = self._consume_pending_clarification(parsed.chat_id, parsed.sender_id) + if self._card_interaction_enabled and parsed.chat_type == "p2p" and not (parsed.root_id or parsed.parent_id or parsed.feishu_thread_id): + return _Routing(None, False, False) + reply_to_ids = [parsed.parent_id] if self._card_interaction_enabled and parsed.parent_id else None + pending = self._consume_pending_clarification(parsed.chat_id, parsed.sender_id, reply_to_ids=reply_to_ids) pending_topic_id = self._non_empty_str(pending.get("topic_id")) if pending else None if not pending_topic_id: return _Routing(topic_id, False, False) - return _Routing(pending_topic_id, False, True) + return _Routing(pending_topic_id, False, True, self._non_empty_str(pending.get("card_message_id"))) async def _resolve_routing(self, attempt: _RoutingAttempt) -> _Routing: """Decide a message's topic on the Gateway loop, consulting and updating the channel store. @@ -1010,13 +1217,23 @@ class FeishuChannel(Channel): if candidates: stored = await self._first_stored_topic(store, parsed.chat_id, candidates) if stored: + pending = ( + self._consume_pending_clarification(parsed.chat_id, parsed.sender_id, reply_to_ids=[parsed.parent_id]) if self._card_interaction_enabled and parsed.msg_type == InboundMessageType.CHAT and parsed.parent_id else None + ) + if pending: + return _Routing(pending["topic_id"], True, True, self._non_empty_str(pending.get("card_message_id"))) return _Routing(stored, True, False) topic_id: str | None = parsed.root_id or parsed.msg_id if parsed.chat_type == "p2p": topic_id = None if parsed.msg_type != InboundMessageType.CHAT: return _Routing(topic_id, False, False) - pending = self._consume_pending_clarification(parsed.chat_id, parsed.sender_id) + if self._card_interaction_enabled and parsed.chat_type == "p2p" and not candidates: + # Unthreaded P2P messages follow the current conversation, + # including /new. Only explicit card replies use the old alias. + return _Routing(None, False, False) + reply_to_ids = [parsed.parent_id] if self._card_interaction_enabled and parsed.parent_id else None + pending = self._consume_pending_clarification(parsed.chat_id, parsed.sender_id, reply_to_ids=reply_to_ids) if not (pending and self._non_empty_str(pending.get("topic_id"))): return _Routing(topic_id, False, False) attempt.pending = pending @@ -1024,7 +1241,7 @@ class FeishuChannel(Channel): thread_id = self._non_empty_str(attempt.pending.get("thread_id")) if thread_id: await store.set_thread_id(self.name, parsed.chat_id, thread_id, topic_id=pending_topic_id, user_id=parsed.sender_id) - return _Routing(pending_topic_id, False, True) + return _Routing(pending_topic_id, False, True, self._non_empty_str(attempt.pending.get("card_message_id"))) async def _resolve_routing_bounded(self, attempt: _RoutingAttempt) -> _Routing | StoreLookupUnresolved: """``_resolve_routing`` under the store deadline; no answer in time (or a store error) is *unresolved*. @@ -1183,6 +1400,8 @@ class FeishuChannel(Channel): } if source_preview: metadata[SOURCE_PREVIEW_METADATA_KEY] = source_preview + if routing.clarification_card_id: + metadata[_PENDING_CARD_METADATA_KEY] = routing.clarification_card_id inbound = self._make_inbound( chat_id=parsed.chat_id, @@ -1373,7 +1592,16 @@ class FeishuChannel(Channel): ) -> None: """Kick off Feishu side effects without delaying inbound dispatch.""" try: + clarification_card_id = inbound.metadata.pop(_PENDING_CARD_METADATA_KEY, None) inbound = await self._attach_connection_identity(inbound) + if self._card_interaction_enabled and not clarification_card_id and inbound.msg_type == InboundMessageType.CHAT: + # Ordinary answers retire only the current thread's buttons; + # older cards cannot redirect a fresh /new conversation. + thread_id = await lookup_thread_id(inbound, repo=self._connection_repo, store=self.config.get("channel_store")) + pending = self._consume_pending_clarification(inbound.chat_id, inbound.user_id, reply_to_ids=[f"feishu-card:{thread_id}"]) if thread_id else None + if pending: + clarification_card_id = pending.get("card_message_id") + inbound.metadata[RESOLVED_FROM_PENDING_CLARIFICATION_METADATA_KEY] = True reaction_message_ids = source_message_ids or [msg_id] for reaction_message_id in reaction_message_ids: reaction_task = asyncio.create_task(self._add_reaction(reaction_message_id, "OK")) @@ -1381,8 +1609,11 @@ class FeishuChannel(Channel): self._ensure_running_card_started(msg_id, metadata=inbound.metadata) if reservation is None: await self.bus.publish_inbound(inbound) + published = True else: - self._commit_reserved_inbound(reservation, inbound) + published = self._commit_reserved_inbound(reservation, inbound) + if published and clarification_card_id: + self._card_registry.discard_card(inbound.chat_id, inbound.user_id, clarification_card_id) finally: if reservation is not None: reservation.release() diff --git a/backend/app/channels/feishu_card_interaction.py b/backend/app/channels/feishu_card_interaction.py new file mode 100644 index 0000000000..5f85819fb8 --- /dev/null +++ b/backend/app/channels/feishu_card_interaction.py @@ -0,0 +1,247 @@ +"""Feishu interactive clarification cards — outbound half. + +When an ``ask_clarification`` interrupt carries options, the Feishu channel can +render the question as an interactive button card instead of plain text. This +module owns the two pieces of that path: + +- :class:`ClarificationCardRequest` — extracts the renderable request from an + outbound message's ``human_input_request`` metadata (``choice_with_other`` + mode only; free-text and form clarifications keep the plain-text path). +- :class:`PendingCardRegistry` — the in-process pending-request table that the + inbound half (card-action callbacks) validates clicks against. + +Design rules: + +- protocol request ids may repeat across runs; each rendered card gets a + separate random callback token, bound to the originator and consumed once. +- the registry is process-local by design: under multi-worker deployments a + callback landing on the wrong process finds no entry and must fail closed. +""" + +from __future__ import annotations + +import json +import logging +import secrets +import threading +import time +from dataclasses import dataclass +from typing import Any, TypedDict + +from app.channels.message_bus import HUMAN_INPUT_REQUEST_METADATA_KEY + +logger = logging.getLogger(__name__) + +# Feishu renders one action row legibly up to four buttons; beyond that the +# plain-text fallback reads better than a wrapped button pile. +MAX_CARD_OPTIONS = 4 + +# Mirrors PENDING_CLARIFICATION_TTL_SECONDS: a clarification unanswered for 30 +# minutes is abandoned, whether the answer path is typed text or a button. +PENDING_CARD_TTL_SECONDS = 30 * 60 + + +class PendingCardEntry(TypedDict): + request_id: str + source: str + thread_id: str + chat_id: str + originator_open_id: str + options: tuple[str, ...] + option_ids: tuple[str, ...] + created_at: float + card_message_id: str | None + # Routing identity the ChannelManager keys on; without these the synthetic + # inbound would open a new thread (groups) or be rejected as unbound. + topic_id: str | None + connection_id: str | None + owner_user_id: str | None + + +@dataclass(frozen=True, slots=True) +class ClarificationCardRequest: + """A button-renderable clarification extracted from outbound metadata.""" + + request_id: str + source: str + question: str + options: tuple[str, ...] + option_ids: tuple[str, ...] + callback_token: str + context: str | None = None + + @classmethod + def from_metadata(cls, metadata: dict[str, Any] | None) -> ClarificationCardRequest | None: + if not metadata: + return None + payload = metadata.get(HUMAN_INPUT_REQUEST_METADATA_KEY) + if not isinstance(payload, dict): + return None + if payload.get("kind") != "human_input_request": + return None + if payload.get("input_mode") != "choice_with_other": + return None + + request_id = payload.get("request_id") + source = payload.get("source") + question = payload.get("question") + context = payload.get("context") + raw_options = payload.get("options") + if not (isinstance(request_id, str) and request_id): + return None + if not (isinstance(source, str) and source): + return None + if not (isinstance(question, str) and question): + return None + if not isinstance(raw_options, list) or not raw_options or len(raw_options) > MAX_CARD_OPTIONS: + return None + + labels: list[str] = [] + option_ids: list[str] = [] + for index, raw in enumerate(raw_options, start=1): + if not isinstance(raw, dict): + return None + label = raw.get("label") + if not (isinstance(label, str) and label): + return None + option_id = raw.get("id") + if not (isinstance(option_id, str) and option_id): + option_id = f"option-{index}" + labels.append(label) + option_ids.append(option_id) + + return cls( + request_id=request_id, + source=source, + question=question, + options=tuple(labels), + option_ids=tuple(option_ids), + callback_token=secrets.token_urlsafe(32), + context=context if isinstance(context, str) and context.strip() else None, + ) + + +def build_button_card(request: ClarificationCardRequest) -> str: + """Render the clarification as a Feishu interactive card with one button per option.""" + content = f"{request.context}\n\n{request.question}" if request.context else request.question + card = { + "config": {"wide_screen_mode": True, "update_multi": True}, + "header": {"title": {"tag": "plain_text", "content": "需要你的选择"}}, + "elements": [ + {"tag": "markdown", "content": content}, + { + "tag": "action", + "actions": [ + { + "tag": "button", + "text": {"tag": "plain_text", "content": label}, + "type": "primary" if index == 1 else "default", + "value": { + "callback_token": request.callback_token, + "option_id": option_id, + "option_value": label, + }, + } + for index, (option_id, label) in enumerate(zip(request.option_ids, request.options), start=1) + ], + }, + ], + } + return json.dumps(card, ensure_ascii=False) + + +class PendingCardRegistry: + """Process-local table of unanswered button cards, keyed by callback token. + + Consumption is one-time: a resolved (or expired) entry is removed before + the caller acts on it, so replays see "unknown" and fail closed. + """ + + def __init__(self, *, ttl_seconds: float = PENDING_CARD_TTL_SECONDS, max_entries: int = 1000) -> None: + if max_entries < 1: + raise ValueError("max_entries must be positive") + self._ttl_seconds = ttl_seconds + self._max_entries = max_entries + self._entries: dict[str, PendingCardEntry] = {} + self._lock = threading.Lock() + + def register( + self, + *, + callback_token: str, + request_id: str, + source: str, + thread_id: str, + chat_id: str, + originator_open_id: str, + options: tuple[str, ...], + option_ids: tuple[str, ...], + card_message_id: str | None = None, + topic_id: str | None = None, + connection_id: str | None = None, + owner_user_id: str | None = None, + ) -> None: + entry: PendingCardEntry = { + "request_id": request_id, + "source": source, + "thread_id": thread_id, + "chat_id": chat_id, + "originator_open_id": originator_open_id, + "options": options, + "option_ids": option_ids, + "created_at": time.time(), + "card_message_id": card_message_id, + "topic_id": topic_id, + "connection_id": connection_id, + "owner_user_id": owner_user_id, + } + with self._lock: + expired = [token for token, pending in self._entries.items() if entry["created_at"] - pending["created_at"] > self._ttl_seconds] + for token in expired: + del self._entries[token] + if callback_token in self._entries: + raise ValueError("A pending card already owns this callback token") + if len(self._entries) >= self._max_entries: + del self._entries[next(iter(self._entries))] + self._entries[callback_token] = entry + + def consume_if_originator( + self, + callback_token: str, + operator_open_id: str, + *, + option_id: str, + option_value: str, + now: float | None = None, + ) -> tuple[PendingCardEntry | None, str | None]: + """Atomically validate a click and consume the entry on success. + + Returns ``(entry, None)`` for the originator clicking a registered + option; ``(None, "not_originator")`` for anyone else (entry kept — the + real originator may still click); ``(None, "unknown")`` for unknown or + expired tokens (entry dropped) and for option ID/value pairs the registry + never issued (entry kept — the card payload itself is not trusted). + """ + with self._lock: + entry = self._entries.get(callback_token) + if entry is None: + return None, "unknown" + created_at = entry.get("created_at") + now = time.time() if now is None else now + if not isinstance(created_at, (int, float)) or now - created_at > self._ttl_seconds: + del self._entries[callback_token] + logger.info("[Feishu] pending card request expired: request_id=%s", entry["request_id"]) + return None, "unknown" + if entry.get("originator_open_id") != operator_open_id: + return None, "not_originator" + if (option_id, option_value) not in zip(entry["option_ids"], entry["options"]): + return None, "unknown" + del self._entries[callback_token] + return entry, None + + def discard_card(self, chat_id: str, originator_open_id: str, card_message_id: str) -> None: + """Retire a card answered through the accepted plain-text path.""" + with self._lock: + tokens = [token for token, entry in self._entries.items() if (entry["chat_id"], entry["originator_open_id"], entry["card_message_id"]) == (chat_id, originator_open_id, card_message_id)] + for token in tokens: + del self._entries[token] diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index 5a8cb39efd..629e2956dc 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -29,6 +29,8 @@ from app.channels.commands import KNOWN_CHANNEL_COMMANDS from app.channels.connection_identity import lookup_thread_id from app.channels.dedupe_store import InboundDedupeStore, MemoryInboundDedupeStore from app.channels.message_bus import ( + HUMAN_INPUT_REQUEST_METADATA_KEY, + HUMAN_INPUT_RESPONSE_METADATA_KEY, INBOUND_FILE_CONTENT_KEY, PENDING_CLARIFICATION_METADATA_KEY, InboundMessage, @@ -46,6 +48,7 @@ from app.gateway.csrf_middleware import CSRF_COOKIE_NAME, CSRF_HEADER_NAME, gene from app.gateway.github import run_policy as _github_run_policy # noqa: F401 from app.gateway.internal_auth import create_internal_auth_headers from app.gateway.path_utils import resolve_outputs_confined_path +from deerflow.agents.human_input import read_human_input_response from deerflow.config.agents_config import list_custom_agents, load_agent_config from deerflow.config.paths import make_safe_user_id from deerflow.runtime import END_SENTINEL, StreamBridge @@ -754,6 +757,50 @@ def _has_current_turn_clarification(result: dict | list) -> bool: return False +def _extract_human_input_payload(result: dict | list) -> dict[str, Any] | None: + """Return the clarification's structured ``human_input`` payload, if present. + + The clarification middleware stashes it under the tool message's + ``artifact.human_input``. Only the current turn's trailing clarification + counts — a clarification answered earlier in the run must not leak into a + later final outbound. + """ + for msg in reversed(_current_turn_messages(result)): + if msg.get("type") == "tool" and msg.get("name") == "ask_clarification": + artifact = msg.get("artifact") + if isinstance(artifact, dict): + payload = artifact.get("human_input") + if isinstance(payload, dict): + return payload + return None + if msg.get("type") == "ai": + content = msg.get("content") + if isinstance(content, str) and content: + return None + if msg.get("tool_calls"): + return None + return None + + +def _response_metadata_with_clarification( + result: dict | list, + base_metadata: dict[str, Any], + *, + pending_clarification: bool, +) -> dict[str, Any]: + """``_response_metadata`` plus the clarification's structured payload. + + Channels that render richer clarification affordances (e.g. Feishu button + cards) read the payload from metadata; channels that do not simply ignore it. + """ + metadata = _response_metadata(base_metadata, pending_clarification=pending_clarification) + if pending_clarification: + payload = _extract_human_input_payload(result) + if payload is not None: + metadata[HUMAN_INPUT_REQUEST_METADATA_KEY] = payload + return metadata + + def _response_metadata(base_metadata: dict[str, Any], *, pending_clarification: bool = False) -> dict[str, Any]: metadata = _slim_metadata(base_metadata) if pending_clarification: @@ -985,7 +1032,7 @@ def _extract_artifacts(result: dict | list) -> list[str]: def _is_hidden_human_control_message(msg: Mapping[str, Any]) -> bool: - """Return whether a human message is an internal control message hidden from UI.""" + """Distinguish internal control messages from real hidden user answers.""" if msg.get("type") != "human": return False @@ -993,7 +1040,7 @@ def _is_hidden_human_control_message(msg: Mapping[str, Any]) -> bool: if not isinstance(additional_kwargs, Mapping): return False - return additional_kwargs.get("hide_from_ui") is True + return additional_kwargs.get("hide_from_ui") is True and read_human_input_response(additional_kwargs) is None def _format_artifact_text(artifacts: list[str]) -> str: @@ -1013,6 +1060,23 @@ def _unknown_command_reply(command: str | None = None) -> str: return f"Unknown command. Available commands: {available}" +def _human_input_message_with_channel_response(content: str, *, metadata: Mapping[str, Any], original_content: str | None = None, files: list[dict[str, Any]] | None = None) -> dict[str, Any]: + """``_human_input_message`` plus a channel-synthesized human-input answer. + + Card-driven channels (Feishu button clicks) carry the answer in inbound + metadata; it must land on the human message's additional_kwargs for the + Gateway's human_input protocol, hidden from transcripts like Web card + replies. + """ + message = _human_input_message(content, original_content=original_content, files=files) + response = metadata.get(HUMAN_INPUT_RESPONSE_METADATA_KEY) + if isinstance(response, Mapping): + additional_kwargs = message.setdefault("additional_kwargs", {}) + additional_kwargs[HUMAN_INPUT_RESPONSE_METADATA_KEY] = dict(response) + additional_kwargs["hide_from_ui"] = True + return message + + def _human_input_message(content: str, *, original_content: str | None = None, files: list[dict[str, Any]] | None = None) -> dict[str, Any]: # The Gateway records run input as sent, but add_messages gives an id-less # message a fresh uuid in the checkpoint. A web client reconnecting to the @@ -2899,7 +2963,7 @@ class ChannelManager: original_text = msg.text uploaded = await _ingest_inbound_files(thread_id, msg, user_id=storage_user_id) - human_message = _human_input_message(msg.text, original_content=original_text, files=uploaded or None) + human_message = _human_input_message_with_channel_response(msg.text, metadata=msg.metadata, original_content=original_text, files=uploaded or None) if self._channel_supports_streaming(msg.channel_name): await self._handle_streaming_chat( @@ -3022,7 +3086,7 @@ class ChannelManager: thread_ts=msg.thread_ts, connection_id=msg.connection_id, owner_user_id=msg.owner_user_id, - metadata=_response_metadata(msg.metadata, pending_clarification=pending_clarification), + metadata=_response_metadata_with_clarification(result, msg.metadata, pending_clarification=pending_clarification), ) logger.info("[Manager] publishing outbound message to bus: channel=%s, chat_id=%s", msg.channel_name, msg.chat_id) await self.bus.publish_outbound(outbound) @@ -3174,7 +3238,7 @@ class ChannelManager: thread_ts=msg.thread_ts, connection_id=msg.connection_id, owner_user_id=msg.owner_user_id, - metadata=_response_metadata(msg.metadata, pending_clarification=pending_clarification), + metadata=_response_metadata_with_clarification(result, msg.metadata, pending_clarification=pending_clarification), ) ) if stream_error is not None: diff --git a/backend/app/channels/message_bus.py b/backend/app/channels/message_bus.py index 2aa05a752c..2a476ca18f 100644 --- a/backend/app/channels/message_bus.py +++ b/backend/app/channels/message_bus.py @@ -17,6 +17,14 @@ logger = logging.getLogger(__name__) DEFAULT_INBOUND_QUEUE_MAXSIZE = 1000 PENDING_CLARIFICATION_METADATA_KEY = "pending_clarification" +# Full ``human_input`` payload (artifact of the ask_clarification tool message), +# attached to the final outbound so channels can render richer affordances +# (e.g. Feishu button cards) for option-based clarifications. +HUMAN_INPUT_REQUEST_METADATA_KEY = "human_input_request" +# A channel-synthesized human-input answer (e.g. a Feishu card button click), +# carried on the inbound so the manager can attach it to the human message's +# additional_kwargs for the Gateway's human_input protocol. +HUMAN_INPUT_RESPONSE_METADATA_KEY = "human_input_response" RESOLVED_FROM_PENDING_CLARIFICATION_METADATA_KEY = "resolved_from_pending_clarification" # Adapter-owned bytes may use this transient key while crossing the channel # boundary. ChannelManager consumes and removes it before persisting metadata. diff --git a/backend/tests/test_channel_hidden_human_input.py b/backend/tests/test_channel_hidden_human_input.py new file mode 100644 index 0000000000..f85dfce311 --- /dev/null +++ b/backend/tests/test_channel_hidden_human_input.py @@ -0,0 +1,131 @@ +"""Hidden user answers form turn boundaries; internal control messages do not.""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.channels.manager import ChannelManager, _extract_artifacts, _extract_response_text +from app.channels.message_bus import HUMAN_INPUT_RESPONSE_METADATA_KEY, InboundMessage, MessageBus +from app.channels.store import JsonChannelStore + + +def _hidden_answer() -> dict: + return { + "type": "human", + "content": "Alpha", + "additional_kwargs": { + "hide_from_ui": True, + "human_input_response": { + "version": 1, + "kind": "human_input_response", + "source": "ask_clarification", + "request_id": "clarification:call-1", + "response_kind": "option", + "option_id": "alpha", + "value": "Alpha", + }, + }, + } + + +def test_hidden_answer_stops_previous_clarification_text(): + result = { + "messages": [ + {"type": "human", "content": "Original question"}, + {"type": "tool", "name": "ask_clarification", "content": "Pick one"}, + _hidden_answer(), + ] + } + + assert _extract_response_text(result) == "" + + +def _file_response(content: str, path: str) -> dict: + return {"type": "ai", "content": content, "tool_calls": [{"name": "present_files", "args": {"filepaths": [path]}}]} + + +@pytest.mark.parametrize("response_kind", ["option", "text"]) +def test_hidden_answer_stops_previous_response_and_artifacts(response_kind): + answer = _hidden_answer() + response = answer["additional_kwargs"][HUMAN_INPUT_RESPONSE_METADATA_KEY] + response["response_kind"] = response_kind + if response_kind == "text": + response.pop("option_id") + result = {"messages": [{"type": "human", "content": "Original question"}, _file_response("Previous answer", "/old.txt"), answer], "artifacts": ["/old.txt"]} + + assert _extract_response_text(result) == "" + assert _extract_artifacts(result) == [] + + +def test_hidden_answer_keeps_current_response_and_artifacts(): + result = { + "messages": [ + {"type": "human", "content": "Original question"}, + _file_response("Previous answer", "/old.txt"), + _hidden_answer(), + _file_response("Current answer", "/new.txt"), + {"type": "human", "content": "Internal context", "additional_kwargs": {"hide_from_ui": True}}, + ] + } + + assert _extract_response_text(result) == "Current answer" + assert _extract_artifacts(result) == ["/new.txt"] + + +@pytest.mark.parametrize("invalid_response", [None, {}, "invalid", {"version": 2}, {"option_id": ""}, {"value": " "}]) +def test_invalid_hidden_answer_remains_a_control_message(invalid_response): + hidden = _hidden_answer() + if invalid_response is None: + hidden["additional_kwargs"].pop(HUMAN_INPUT_RESPONSE_METADATA_KEY) + elif isinstance(invalid_response, dict): + hidden["additional_kwargs"][HUMAN_INPUT_RESPONSE_METADATA_KEY].update(invalid_response) + if not invalid_response: + hidden["additional_kwargs"][HUMAN_INPUT_RESPONSE_METADATA_KEY] = {} + else: + hidden["additional_kwargs"][HUMAN_INPUT_RESPONSE_METADATA_KEY] = invalid_response + result = {"messages": [{"type": "human", "content": "Current question"}, _file_response("Current answer", "/current.txt"), hidden]} + + assert _extract_response_text(result) == "Current answer" + assert _extract_artifacts(result) == ["/current.txt"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failed", [False, True]) +async def test_card_answer_without_fresh_ai_preserves_terminal_result(tmp_path, monkeypatch, failed): + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + bus = MessageBus() + store = JsonChannelStore(path=tmp_path / "routing.json") + await store.set_thread_id("feishu", "chat-1", "thread-1", topic_id="topic-1") + manager = ChannelManager(bus=bus, store=store) + inbound = InboundMessage( + channel_name="feishu", chat_id="chat-1", user_id="ou-originator", topic_id="topic-1", text="Alpha", metadata={HUMAN_INPUT_RESPONSE_METADATA_KEY: _hidden_answer()["additional_kwargs"][HUMAN_INPUT_RESPONSE_METADATA_KEY]} + ) + client = MagicMock() + client.threads.get = AsyncMock(return_value={"thread_id": "thread-1"}) + client.threads.update = AsyncMock(return_value={"thread_id": "thread-1"}) + + async def stream(*args, **kwargs): + answer = dict(kwargs["input"]["messages"][0], type="human") + yield SimpleNamespace( + event="values", + data={"messages": [{"type": "human", "content": "Original question"}, _file_response("Previous answer", "/old.txt"), {"type": "tool", "name": "ask_clarification", "content": "Pick one"}, answer], "artifacts": ["/old.txt"]}, + ) + if failed: + yield SimpleNamespace(event="error", data={"name": "ModelError", "message": "offline failure"}) + + client.runs.stream = MagicMock(side_effect=stream) + manager._client = client + outbound = [] + + async def capture(message): + outbound.append(message) + + bus.subscribe_outbound(capture) + await manager._handle_chat(inbound) + + assert len(outbound) == 1 + assert outbound[0].is_final is True + assert outbound[0].text == ("An error occurred while processing your request. Please try again." if failed else "(No response from agent)") + assert outbound[0].artifacts == [] + assert outbound[0].attachments == [] diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index cd21ee0f06..c460262e31 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -13410,3 +13410,114 @@ def test_streaming_chat_never_publishes_hidden_memory_context(monkeypatch): assert [m.text for m in outbound_received] == ["All green. ▉", "All green."] _run(go()) + + +# _extract_human_input_payload / clarification metadata passthrough tests + + +class TestExtractHumanInputPayload: + def test_returns_payload_from_ask_clarification_artifact(self): + from app.channels.manager import _extract_human_input_payload + + payload = { + "version": 1, + "kind": "human_input_request", + "source": "ask_clarification", + "request_id": "clarification:call_1", + "question": "Pick one", + "input_mode": "choice_with_other", + "options": [{"id": "option-1", "label": "Alpha", "value": "Alpha"}], + } + result = { + "messages": [ + {"type": "human", "content": "hi"}, + { + "type": "tool", + "name": "ask_clarification", + "content": "Pick one", + "artifact": {"human_input": payload}, + }, + ] + } + assert _extract_human_input_payload(result) == payload + + def test_returns_none_without_artifact(self): + from app.channels.manager import _extract_human_input_payload + + result = {"messages": [{"type": "tool", "name": "ask_clarification", "content": "q"}]} + assert _extract_human_input_payload(result) is None + + def test_returns_none_for_non_clarification(self): + from app.channels.manager import _extract_human_input_payload + + result = {"messages": [{"type": "ai", "content": "done"}]} + assert _extract_human_input_payload(result) is None + + def test_ignores_previous_turn_clarification(self): + from app.channels.manager import _extract_human_input_payload + + payload = {"kind": "human_input_request", "request_id": "clarification:old"} + result = { + "messages": [ + {"type": "tool", "name": "ask_clarification", "content": "old", "artifact": {"human_input": payload}}, + {"type": "human", "content": "answer"}, + {"type": "ai", "content": "next"}, + ] + } + assert _extract_human_input_payload(result) is None + + +class TestResponseMetadataClarificationPassthrough: + def test_pending_clarification_carries_human_input_payload(self): + from app.channels.manager import _response_metadata_with_clarification + from app.channels.message_bus import HUMAN_INPUT_REQUEST_METADATA_KEY, PENDING_CLARIFICATION_METADATA_KEY + + payload = {"kind": "human_input_request", "request_id": "clarification:call_1", "input_mode": "choice_with_other"} + result = { + "messages": [ + {"type": "human", "content": "hi"}, + {"type": "tool", "name": "ask_clarification", "content": "q", "artifact": {"human_input": payload}}, + ] + } + metadata = _response_metadata_with_clarification(result, {"user_id": "ou_1"}, pending_clarification=True) + assert metadata[PENDING_CLARIFICATION_METADATA_KEY] is True + assert metadata[HUMAN_INPUT_REQUEST_METADATA_KEY] == payload + assert metadata["user_id"] == "ou_1" + + def test_non_clarification_outbound_never_carries_payload(self): + from app.channels.manager import _response_metadata_with_clarification + from app.channels.message_bus import HUMAN_INPUT_REQUEST_METADATA_KEY + + result = {"messages": [{"type": "ai", "content": "done"}]} + metadata = _response_metadata_with_clarification(result, {}, pending_clarification=False) + assert HUMAN_INPUT_REQUEST_METADATA_KEY not in metadata + + +class TestHumanInputResponseTranslation: + """A channel-synthesized human-input answer must land on the run's human + message as additional_kwargs, hidden from transcripts (Web card parity).""" + + def test_response_metadata_key_is_translated_to_additional_kwargs(self): + from app.channels.manager import _human_input_message_with_channel_response + from app.channels.message_bus import HUMAN_INPUT_RESPONSE_METADATA_KEY + + response = { + "version": 1, + "kind": "human_input_response", + "source": "feishu_card", + "request_id": "clarification:call_1", + "response_kind": "option", + "option_id": "option-1", + "value": "Alpha", + } + message = _human_input_message_with_channel_response("Alpha", metadata={HUMAN_INPUT_RESPONSE_METADATA_KEY: response}) + assert message["content"] == "Alpha" + assert message["additional_kwargs"][HUMAN_INPUT_RESPONSE_METADATA_KEY] == response + assert message["additional_kwargs"]["hide_from_ui"] is True + + def test_plain_message_unchanged(self): + from app.channels.manager import _human_input_message_with_channel_response + + message = _human_input_message_with_channel_response("hello", metadata={}) + assert message["content"] == "hello" + assert "additional_kwargs" not in message diff --git a/backend/tests/test_feishu_card_delivery.py b/backend/tests/test_feishu_card_delivery.py new file mode 100644 index 0000000000..4dfb8a804c --- /dev/null +++ b/backend/tests/test_feishu_card_delivery.py @@ -0,0 +1,626 @@ +"""Feishu card delivery and continuation through the real adapter/SDK seams.""" + +from __future__ import annotations + +import asyncio +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest + +from app.channels.feishu import FeishuChannel, _ParsedMessage, _RoutingAttempt +from app.channels.message_bus import HUMAN_INPUT_REQUEST_METADATA_KEY, HUMAN_INPUT_RESPONSE_METADATA_KEY, PENDING_CLARIFICATION_METADATA_KEY, InboundMessage, InboundMessageType, MessageBus, OutboundMessage + + +def clarification_payload(request_id: str = "clarification:call-1") -> dict: + return { + "version": 1, + "kind": "human_input_request", + "source": "ask_clarification", + "request_id": request_id, + "question": "Pick one", + "input_mode": "choice_with_other", + "options": [{"id": "alpha", "label": "Alpha", "value": "Alpha"}, {"id": "beta", "label": "Beta", "value": "Beta"}], + } + + +def clarification_outbound(*, request_id: str = "clarification:call-1", topic_id: str | None = "topic-1", thread_id: str = "thread-1", source_id: str = "source-1") -> OutboundMessage: + return OutboundMessage( + channel_name="feishu", + chat_id="chat-1", + thread_id=thread_id, + text="Pick one", + is_final=True, + thread_ts=source_id, + metadata={"user_id": "ou-originator", "topic_id": topic_id, PENDING_CLARIFICATION_METADATA_KEY: True, HUMAN_INPUT_REQUEST_METADATA_KEY: clarification_payload(request_id)}, + ) + + +@pytest.fixture +def card_channel(tmp_path): + from lark_oapi.api.im.v1 import CreateMessageRequest, CreateMessageRequestBody, PatchMessageRequest, PatchMessageRequestBody, ReplyMessageRequest, ReplyMessageRequestBody + + from app.channels.store import JsonChannelStore + + channel = FeishuChannel(MessageBus(), {"app_id": "offline-test", "app_secret": "offline-test", "card_interaction_enabled": True}) + channel.config["channel_store"] = JsonChannelStore(path=tmp_path / "cards.json") + channel._ReplyMessageRequest = ReplyMessageRequest + channel._ReplyMessageRequestBody = ReplyMessageRequestBody + channel._PatchMessageRequest = PatchMessageRequest + channel._PatchMessageRequestBody = PatchMessageRequestBody + channel._CreateMessageRequest = CreateMessageRequest + channel._CreateMessageRequestBody = CreateMessageRequestBody + channel._api_client = MagicMock() + channel._api_client.im.v1.message.reply.side_effect = lambda request: SimpleNamespace(success=lambda: True, data=SimpleNamespace(message_id=f"card-{request.message_id}")) + channel._api_client.im.v1.message.patch.return_value = SimpleNamespace(success=lambda: True) + channel._api_client.im.v1.message.create.return_value = SimpleNamespace(success=lambda: True) + channel._CallBackToast = SimpleNamespace + channel._P2CardActionTriggerResponse = SimpleNamespace + return channel + + +def sent_button(channel, index: int = 0) -> dict: + for call in reversed(channel._api_client.im.v1.message.mock_calls): + if call[0] not in {"reply", "patch"}: + continue + card = json.loads(call.args[0].request_body.content) + for element in card["elements"]: + if element["tag"] == "action": + return element["actions"][index]["value"] + raise AssertionError("No button card was delivered") + + +async def click_button(channel, value: dict): + channel._main_loop = asyncio.get_running_loop() + event = SimpleNamespace(event=SimpleNamespace(action=SimpleNamespace(value=value), operator=SimpleNamespace(open_id="ou-originator"))) + response = channel._on_card_action(event) + assert response.toast.type == "success" + inbound = await asyncio.wait_for(channel.bus.get_inbound(), timeout=2) + channel.bus.inbound_task_done() + return inbound + + +@pytest.mark.asyncio +@pytest.mark.parametrize("running_card", ["absent", "present", "patch_rejected"]) +@pytest.mark.parametrize( + ("context", "expected_content"), + [(None, "Pick one"), ("这将**永久删除**生产表中的数据。", "这将**永久删除**生产表中的数据。\n\nPick one")], +) +async def test_button_delivery_uses_real_sdk_requests(card_channel, running_card, context, expected_content): + channel = card_channel + if running_card != "absent": + channel._running_card_ids["source-1"] = "running-card" + if running_card == "patch_rejected": + channel._api_client.im.v1.message.patch.return_value = SimpleNamespace(success=lambda: False, code=230099, msg="offline rejection", get_log_id=lambda: "offline") + + message = clarification_outbound() + if context is not None: + message.metadata[HUMAN_INPUT_REQUEST_METADATA_KEY]["context"] = context + message.text = expected_content + await channel.send(message, _max_retries=1) + + if running_card == "present": + sdk_request = channel._api_client.im.v1.message.patch.call_args.args[0] + assert sdk_request.message_id == "running-card" + else: + sdk_request = channel._api_client.im.v1.message.reply.call_args.args[0] + assert sdk_request.message_id == "source-1" + assert sdk_request.request_body.msg_type == "interactive" + card = json.loads(sdk_request.request_body.content) + assert card["config"]["update_multi"] is True + markdown = next(element for element in card["elements"] if element["tag"] == "markdown") + assert markdown["content"] == expected_content + actions = next(element for element in card["elements"] if element["tag"] == "action")["actions"] + assert [action["value"]["option_id"] for action in actions] == ["alpha", "beta"] + assert [action["text"]["content"] for action in actions] == ["Alpha", "Beta"] + + +@pytest.mark.asyncio +async def test_plain_reply_keeps_markdown_wrapper(card_channel): + message = clarification_outbound() + message.metadata.pop(HUMAN_INPUT_REQUEST_METADATA_KEY) + await card_channel.send(message, _max_retries=1) + sdk_request = card_channel._api_client.im.v1.message.reply.call_args.args[0] + card = json.loads(sdk_request.request_body.content) + assert card["elements"] == [{"tag": "markdown", "content": "Pick one"}] + + +@pytest.mark.asyncio +async def test_button_answer_is_persisted_as_latest_human_input(card_channel): + from langchain_core.messages import HumanMessage + + from app.channels.manager import _human_input_message_with_channel_response + from deerflow.runtime.events.store.memory import MemoryRunEventStore + from deerflow.runtime.journal import RunJournal + + await card_channel.send(clarification_outbound(), _max_retries=1) + try: + inbound = await click_button(card_channel, sent_button(card_channel)) + posted = _human_input_message_with_channel_response(inbound.text, metadata=inbound.metadata) + hidden = HumanMessage(content=posted["content"], additional_kwargs=posted["additional_kwargs"]) + store = MemoryRunEventStore() + journal = RunJournal("run-1", "thread-1", store) + try: + journal.on_chat_model_start({}, [[HumanMessage(content="Original question"), hidden]], run_id=uuid4(), tags=["lead_agent"]) + await journal.flush() + events = await store.list_events("thread-1", "run-1") + human_events = [event for event in events if event["event_type"] == "llm.human.input"] + assert len(human_events) == 1 + assert human_events[0]["content"]["content"] == "Alpha" + assert inbound.metadata[HUMAN_INPUT_RESPONSE_METADATA_KEY]["source"] == "ask_clarification" + finally: + await journal.close(flush=True) + finally: + await card_channel._close_and_drain_threadsafe_futures() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("bound_identity", [False, True]) +@pytest.mark.parametrize("topic_id", [None, "topic-1"]) +@pytest.mark.parametrize("reset_conversation", [False, True]) +async def test_consecutive_card_answers_resume_the_same_thread(card_channel, tmp_path, monkeypatch, bound_identity, topic_id, reset_conversation): + from app.channels.manager import ChannelManager + from app.channels.store import JsonChannelStore + from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelCredentialCipher + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + store = JsonChannelStore(path=tmp_path / "routing.json") + card_channel.config["channel_store"] = store + message = clarification_outbound(topic_id=topic_id) + repo = None + try: + if bound_identity: + await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'connections.db'}", sqlite_dir=str(tmp_path)) + repo = ChannelConnectionRepository(get_session_factory(), cipher=ChannelCredentialCipher.from_key("offline-key")) + connection = await repo.upsert_connection(owner_user_id="owner-1", provider="feishu", external_account_id="ou-originator", workspace_id="chat-1") + message.connection_id = connection["id"] + message.owner_user_id = "owner-1" + card_channel._connection_repo = repo + await repo.set_thread_id(connection_id=connection["id"], owner_user_id="owner-1", provider="feishu", external_conversation_id="chat-1", thread_id="thread-1", external_topic_id=topic_id) + else: + await store.set_thread_id("feishu", "chat-1", "thread-1", topic_id=topic_id, user_id="ou-originator") + + manager = ChannelManager(bus=card_channel.bus, store=store, connection_repo=repo, require_bound_identity=bound_identity) + client = MagicMock() + client.threads.create = AsyncMock(return_value={"thread_id": "unexpected-new-thread"}) + client.threads.update = AsyncMock(return_value={"thread_id": "thread-1"}) + client.threads.get = AsyncMock(return_value={"thread_id": "thread-1"}) + + async def stream(*args, **kwargs): + messages = [{"type": "human", "content": "Alpha"}] + if client.runs.stream.call_count == 1: + messages.append({"type": "tool", "name": "ask_clarification", "content": "Pick again", "artifact": {"human_input": clarification_payload("clarification:call-2")}}) + else: + messages.append({"type": "ai", "content": "Done"}) + yield SimpleNamespace(event="values", data={"messages": messages, "artifacts": []}) + + client.runs.stream = MagicMock(side_effect=stream) + manager._client = client + final_messages = [] + + async def deliver(message): + if message.is_final: + final_messages.append(message) + await card_channel._on_outbound(message) + + card_channel.bus.subscribe_outbound(deliver) + await card_channel.send(message, _max_retries=1) + button = sent_button(card_channel) + if reset_conversation: + command = InboundMessage( + channel_name="feishu", + chat_id="chat-1", + user_id="ou-originator", + text="/new", + msg_type=InboundMessageType.COMMAND, + topic_id=topic_id, + connection_id=message.connection_id, + owner_user_id=message.owner_user_id, + workspace_id="chat-1", + metadata={"user_id": "ou-originator", "topic_id": topic_id}, + ) + await manager._handle_command(command) + first = await click_button(card_channel, button) + await manager._handle_chat(first) + assert final_messages[-1].metadata[PENDING_CLARIFICATION_METADATA_KEY] is True + second = await click_button(card_channel, sent_button(card_channel)) + await manager._handle_chat(second) + + assert client.threads.create.call_count == int(reset_conversation) + assert [call.args[0] for call in client.runs.stream.call_args_list] == ["thread-1", "thread-1"] + if reset_conversation: + assert await manager._lookup_thread_id(command) == "unexpected-new-thread" + for inbound in (first, second): + assert inbound.topic_id == inbound.metadata["topic_id"] + if bound_identity: + assert inbound.workspace_id == "chat-1" + assert inbound.connection_id == connection["id"] + assert inbound.owner_user_id == "owner-1" + for call in client.runs.stream.call_args_list: + posted = call.kwargs["input"]["messages"][0] + assert posted["additional_kwargs"]["hide_from_ui"] is True + assert posted["additional_kwargs"][HUMAN_INPUT_RESPONSE_METADATA_KEY]["source"] == "ask_clarification" + finally: + await card_channel._close_and_drain_threadsafe_futures() + if bound_identity: + await close_engine() + + +@pytest.mark.asyncio +async def test_reused_request_ids_do_not_cross_route_cards(card_channel): + first_message = clarification_outbound() + await card_channel.send(first_message, _max_retries=1) + first_button = sent_button(card_channel) + second_message = clarification_outbound(thread_id="thread-2", topic_id="topic-2", source_id="source-2") + await card_channel.send(second_message, _max_retries=1) + second_button = sent_button(card_channel) + try: + first = await click_button(card_channel, first_button) + second = await click_button(card_channel, second_button) + assert first.topic_id == "feishu-card:thread-1" + assert first.thread_ts == "card-source-1" + assert second.topic_id == "feishu-card:thread-2" + assert second.thread_ts == "card-source-2" + assert first_button["callback_token"] != second_button["callback_token"] + for inbound in (first, second): + assert inbound.metadata[HUMAN_INPUT_RESPONSE_METADATA_KEY]["request_id"] == first_message.metadata[HUMAN_INPUT_REQUEST_METADATA_KEY]["request_id"] + finally: + await card_channel._close_and_drain_threadsafe_futures() + + +def group_message(message_id: str) -> _ParsedMessage: + return _ParsedMessage(chat_id="chat-1", msg_id=message_id, sender_id="ou-originator", root_id=None, parent_id=None, feishu_thread_id=None, chat_type="group", text="Another question", files=[]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("with_store", [False, True]) +@pytest.mark.parametrize("other_pending_card", [False, True]) +async def test_answered_card_clears_only_its_text_continuation(card_channel, tmp_path, with_store, other_pending_card): + from app.channels.store import JsonChannelStore + + if with_store: + card_channel.config["channel_store"] = JsonChannelStore(path=tmp_path / "hints.json") + + async def route(message_id): + parsed = group_message(message_id) + return await card_channel._resolve_routing(_RoutingAttempt(parsed)) if with_store else card_channel._route_without_store(parsed) + + if other_pending_card: + await card_channel.send(clarification_outbound(request_id="clarification:other", thread_id="other-thread", topic_id="other-topic", source_id="other-source"), _max_retries=1) + await card_channel.send(clarification_outbound(), _max_retries=1) + try: + await click_button(card_channel, sent_button(card_channel)) + if other_pending_card: + remaining = await route("reply-to-other-card") + assert remaining.topic_id == "feishu-card:other-thread" + assert remaining.from_pending is True + independent = await route("independent-question") + assert independent.topic_id == "independent-question" + assert independent.from_pending is False + finally: + await card_channel._close_and_drain_threadsafe_futures() + + +@pytest.mark.asyncio +async def test_forged_option_id_does_not_consume_card(card_channel): + await card_channel.send(clarification_outbound(), _max_retries=1) + button = sent_button(card_channel) + forged = dict(button, option_id="beta") + card_channel._main_loop = asyncio.get_running_loop() + event = SimpleNamespace(event=SimpleNamespace(action=SimpleNamespace(value=forged), operator=SimpleNamespace(open_id="ou-originator"))) + try: + response = card_channel._on_card_action(event) + assert response.toast.type == "warning" + with pytest.raises(asyncio.QueueEmpty): + card_channel.bus.get_inbound_nowait() + accepted = await click_button(card_channel, button) + assert accepted.metadata[HUMAN_INPUT_RESPONSE_METADATA_KEY]["option_id"] == "alpha" + finally: + await card_channel._close_and_drain_threadsafe_futures() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("queue_state", ["full", "closed"]) +async def test_rejected_answer_keeps_text_fallback_and_card_unresolved(card_channel, queue_state): + from app.channels.message_bus import InboundMessage + + card_channel.bus = MessageBus(inbound_queue_maxsize=1) + await card_channel.send(clarification_outbound(), _max_retries=1) + if queue_state == "closed": + card_channel.bus.close_inbound() + else: + await card_channel.bus.publish_inbound(InboundMessage(channel_name="feishu", chat_id="busy", user_id="busy", text="Already admitted")) + button = sent_button(card_channel) + entry, reason = card_channel._card_registry.consume_if_originator(button["callback_token"], "ou-originator", option_id=button["option_id"], option_value=button["option_value"]) + assert reason is None and entry is not None + await card_channel._resolve_card_click(entry, option_id=button["option_id"], option_value=button["option_value"], operator_open_id="ou-originator") + card_channel._api_client.im.v1.message.patch.assert_not_called() + text_reply = card_channel._route_without_store(group_message("typed-answer")) + assert text_reply.topic_id == "feishu-card:thread-1" + assert text_reply.from_pending is True + if queue_state == "full": + assert card_channel.bus.get_inbound_nowait().text == "Already admitted" + card_channel.bus.inbound_task_done() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("explicit_reply", [False, True]) +async def test_accepted_text_answer_retires_its_button_card(card_channel, explicit_reply): + await card_channel.send(clarification_outbound(), _max_retries=1) + button = sent_button(card_channel) + parsed = group_message("typed-answer") + parsed.text = "Beta" + if explicit_reply: + parsed.root_id = "source-1" + parsed.parent_id = "card-source-1" + routing = await card_channel._resolve_routing(_RoutingAttempt(parsed)) + card_channel._main_loop = asyncio.get_running_loop() + card_channel._dispatch_inbound(parsed, routing) + try: + inbound = await asyncio.wait_for(card_channel.bus.get_inbound(), timeout=2) + card_channel.bus.inbound_task_done() + assert inbound.text == "Beta" + assert inbound.topic_id == "feishu-card:thread-1" + event = SimpleNamespace(event=SimpleNamespace(action=SimpleNamespace(value=button), operator=SimpleNamespace(open_id="ou-originator"))) + assert card_channel._on_card_action(event).toast.type == "warning" + with pytest.raises(asyncio.QueueEmpty): + card_channel.bus.get_inbound_nowait() + finally: + await card_channel._close_and_drain_threadsafe_futures() + await asyncio.gather(*card_channel._running_card_tasks.values(), return_exceptions=True) + + +@pytest.mark.asyncio +async def test_explicit_text_answer_targets_the_card_before_its_shared_topic(card_channel): + await card_channel.send(clarification_outbound(source_id="source-1"), _max_retries=1) + first_button = sent_button(card_channel) + await card_channel.send(clarification_outbound(thread_id="thread-2", source_id="source-2"), _max_retries=1) + second_button = sent_button(card_channel) + parsed = group_message("typed-second-answer") + parsed.text = "Beta" + parsed.root_id = "topic-1" + parsed.parent_id = "card-source-2" + routing = await card_channel._resolve_routing(_RoutingAttempt(parsed)) + card_channel._main_loop = asyncio.get_running_loop() + card_channel._dispatch_inbound(parsed, routing) + try: + inbound = await asyncio.wait_for(card_channel.bus.get_inbound(), timeout=2) + card_channel.bus.inbound_task_done() + assert inbound.topic_id == "feishu-card:thread-2" + event = SimpleNamespace(event=SimpleNamespace(action=SimpleNamespace(value=second_button), operator=SimpleNamespace(open_id="ou-originator"))) + assert card_channel._on_card_action(event).toast.type == "warning" + assert (await click_button(card_channel, first_button)).topic_id == "feishu-card:thread-1" + finally: + await card_channel._close_and_drain_threadsafe_futures() + await asyncio.gather(*card_channel._running_card_tasks.values(), return_exceptions=True) + + +@pytest.mark.asyncio +async def test_explicit_text_answer_keeps_the_new_conversation_route(card_channel): + from app.channels.manager import ChannelManager + + store = card_channel.config["channel_store"] + await store.set_thread_id("feishu", "chat-1", "thread-1", topic_id="topic-1", user_id="ou-originator") + await card_channel.send(clarification_outbound(), _max_retries=1) + manager = ChannelManager(bus=card_channel.bus, store=store) + client = MagicMock() + client.threads.create = AsyncMock(return_value={"thread_id": "thread-after-new"}) + client.threads.get = AsyncMock(return_value={"thread_id": "thread-1", "metadata": {}}) + client.threads.update = AsyncMock(return_value={"thread_id": "thread-1"}) + + async def stream(*args, **kwargs): + yield SimpleNamespace(event="values", data={"messages": [{"type": "human", "content": "Beta"}, {"type": "ai", "content": "Done"}], "artifacts": []}) + + client.runs.stream = MagicMock(side_effect=stream) + manager._client = client + card_channel.bus.subscribe_outbound(card_channel._on_outbound) + command = InboundMessage( + channel_name="feishu", + chat_id="chat-1", + user_id="ou-originator", + text="/new", + msg_type=InboundMessageType.COMMAND, + thread_ts="source-new", + topic_id="topic-1", + metadata={"user_id": "ou-originator", "message_id": "source-new", "topic_id": "topic-1"}, + ) + await manager._handle_command(command) + assert await manager._lookup_thread_id(command) == "thread-after-new" + parsed = group_message("typed-old-answer") + parsed.text = "Beta" + parsed.root_id = "topic-1" + parsed.parent_id = "card-source-1" + routing = await card_channel._resolve_routing(_RoutingAttempt(parsed)) + card_channel._main_loop = asyncio.get_running_loop() + card_channel._dispatch_inbound(parsed, routing) + try: + inbound = await asyncio.wait_for(card_channel.bus.get_inbound(), timeout=2) + card_channel.bus.inbound_task_done() + await manager._handle_chat(inbound) + assert client.runs.stream.call_args.args[0] == "thread-1" + assert await manager._lookup_thread_id(command) == "thread-after-new" + next_message = group_message("new-task") + next_message.root_id = "topic-1" + card_channel._dispatch_inbound(next_message, await card_channel._resolve_routing(_RoutingAttempt(next_message))) + next_inbound = await asyncio.wait_for(card_channel.bus.get_inbound(), timeout=2) + card_channel.bus.inbound_task_done() + await manager._handle_chat(next_inbound) + assert client.runs.stream.call_args.args[0] == "thread-after-new" + finally: + await card_channel._close_and_drain_threadsafe_futures() + await asyncio.gather(*card_channel._running_card_tasks.values(), return_exceptions=True) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("bound_identity", [False, True]) +@pytest.mark.parametrize("reset_conversation", [False, True]) +@pytest.mark.parametrize("topic_id", [None, "source-1"]) +async def test_ordinary_text_uses_current_conversation(card_channel, tmp_path, monkeypatch, bound_identity, reset_conversation, topic_id): + from app.channels.manager import ChannelManager + from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelCredentialCipher + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + store = card_channel.config["channel_store"] + message = clarification_outbound(topic_id=topic_id) + repo = None + try: + if bound_identity: + await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'connections.db'}", sqlite_dir=str(tmp_path)) + repo = ChannelConnectionRepository(get_session_factory(), cipher=ChannelCredentialCipher.from_key("offline-key")) + connection = await repo.upsert_connection(owner_user_id="owner-1", provider="feishu", external_account_id="ou-originator", workspace_id="chat-1") + message.connection_id = connection["id"] + message.owner_user_id = "owner-1" + card_channel._connection_repo = repo + await repo.set_thread_id(connection_id=connection["id"], owner_user_id="owner-1", provider="feishu", external_conversation_id="chat-1", thread_id="thread-1", external_topic_id=topic_id) + else: + await store.set_thread_id("feishu", "chat-1", "thread-1", topic_id=topic_id, user_id="ou-originator") + manager = ChannelManager(bus=card_channel.bus, store=store, connection_repo=repo, require_bound_identity=bound_identity) + client = MagicMock() + client.threads.create = AsyncMock(return_value={"thread_id": "thread-after-new"}) + client.threads.get = AsyncMock(return_value={"thread_id": "thread-1", "metadata": {}}) + client.threads.update = AsyncMock(return_value={"thread_id": "thread-1"}) + + async def stream(*args, **kwargs): + yield SimpleNamespace(event="values", data={"messages": [{"type": "human", "content": "Beta"}, {"type": "ai", "content": "Done"}], "artifacts": []}) + + client.runs.stream = MagicMock(side_effect=stream) + manager._client = client + card_channel.bus.subscribe_outbound(card_channel._on_outbound) + await card_channel.send(message, _max_retries=1) + old_button = sent_button(card_channel) + command = InboundMessage( + channel_name="feishu", + chat_id="chat-1", + user_id="ou-originator", + text="/new", + msg_type=InboundMessageType.COMMAND, + topic_id=topic_id, + connection_id=message.connection_id, + owner_user_id=message.owner_user_id, + workspace_id="chat-1", + metadata={"user_id": "ou-originator", "topic_id": topic_id}, + ) + if reset_conversation: + await manager._handle_command(command) + parsed = group_message("ordinary-answer") + parsed.chat_type = "group" if topic_id else "p2p" + parsed.root_id = topic_id + parsed.parent_id = "source-new" if topic_id and reset_conversation else None + parsed.text = "Beta" + card_channel._main_loop = asyncio.get_running_loop() + card_channel._dispatch_inbound(parsed, await card_channel._resolve_routing(_RoutingAttempt(parsed))) + inbound = await asyncio.wait_for(card_channel.bus.get_inbound(), timeout=2) + card_channel.bus.inbound_task_done() + await manager._handle_chat(inbound) + expected_thread = "thread-after-new" if reset_conversation else "thread-1" + assert client.runs.stream.call_args.args[0] == expected_thread + assert await manager._lookup_thread_id(command) == expected_thread + if reset_conversation: + old_answer = await click_button(card_channel, old_button) + await manager._handle_chat(old_answer) + assert client.runs.stream.call_args.args[0] == "thread-1" + assert await manager._lookup_thread_id(command) == "thread-after-new" + else: + event = SimpleNamespace(event=SimpleNamespace(action=SimpleNamespace(value=old_button), operator=SimpleNamespace(open_id="ou-originator"))) + assert card_channel._on_card_action(event).toast.type == "warning" + finally: + await card_channel._close_and_drain_threadsafe_futures() + await asyncio.gather(*card_channel._running_card_tasks.values(), return_exceptions=True) + if bound_identity: + await close_engine() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("field", ["callback_token", "option_id", "option_value", "operator"]) +async def test_malformed_callback_keeps_the_real_button_usable(card_channel, field): + await card_channel.send(clarification_outbound(), _max_retries=1) + button = sent_button(card_channel) + invalid = dict(button) + operator = "ou-originator" + if field == "operator": + operator = [] + else: + invalid[field] = [] + event = SimpleNamespace(event=SimpleNamespace(action=SimpleNamespace(value=invalid), operator=SimpleNamespace(open_id=operator))) + try: + assert card_channel._on_card_action(event).toast.type == "warning" + assert (await click_button(card_channel, button)).text == "Alpha" + finally: + await card_channel._close_and_drain_threadsafe_futures() + + +@pytest.mark.asyncio +async def test_failed_thread_alias_binding_falls_back_to_text(card_channel, tmp_path): + from app.channels.store import JsonChannelStore + + class FailingAliasStore(JsonChannelStore): + async def set_thread_id(self, *args, **kwargs): + if (kwargs.get("topic_id") or "").startswith("feishu-card:"): + raise RuntimeError("offline database failure") + return await super().set_thread_id(*args, **kwargs) + + card_channel.config["channel_store"] = FailingAliasStore(path=tmp_path / "failing.json") + await card_channel.send(clarification_outbound(), _max_retries=1) + request = card_channel._api_client.im.v1.message.reply.call_args.args[0] + assert json.loads(request.request_body.content)["elements"] == [{"tag": "markdown", "content": "Pick one"}] + + +@pytest.mark.asyncio +async def test_visible_buttons_work_while_auxiliary_mapping_is_slow(card_channel, tmp_path): + from app.channels.store import JsonChannelStore + + waiting = asyncio.Event() + release = asyncio.Event() + + class SlowMessageAliasStore(JsonChannelStore): + async def set_thread_id(self, *args, **kwargs): + if kwargs.get("topic_id") == "source-1": + waiting.set() + await release.wait() + return await super().set_thread_id(*args, **kwargs) + + card_channel.config["channel_store"] = SlowMessageAliasStore(path=tmp_path / "slow.json") + delivery = asyncio.create_task(card_channel.send(clarification_outbound(), _max_retries=1)) + try: + await asyncio.wait_for(waiting.wait(), timeout=2) + inbound = await click_button(card_channel, sent_button(card_channel)) + assert inbound.topic_id == "feishu-card:thread-1" + finally: + release.set() + await delivery + await card_channel._close_and_drain_threadsafe_futures() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("queue_state", ["full", "closed_reservation"]) +async def test_rejected_text_answer_keeps_the_button_usable(card_channel, queue_state): + from app.channels.feishu import _PENDING_CARD_METADATA_KEY + from app.channels.message_bus import InboundQueueFullError + + card_channel.bus = MessageBus(inbound_queue_maxsize=1) + await card_channel.send(clarification_outbound(), _max_retries=1) + button = sent_button(card_channel) + inbound = InboundMessage(channel_name="feishu", chat_id="chat-1", user_id="ou-originator", text="Beta", topic_id="feishu-card:thread-1", metadata={_PENDING_CARD_METADATA_KEY: "card-source-1"}) + reservation = None + try: + if queue_state == "full": + await card_channel.bus.publish_inbound(InboundMessage(channel_name="feishu", chat_id="busy", user_id="busy", text="Already admitted")) + with pytest.raises(InboundQueueFullError): + await card_channel._prepare_inbound("typed-answer", inbound) + card_channel.bus.get_inbound_nowait() + card_channel.bus.inbound_task_done() + else: + reservation = card_channel.bus.reserve_inbound(inbound) + card_channel.bus.close_inbound() + await card_channel._prepare_inbound("typed-answer", inbound, reservation=reservation) + entry, reason = card_channel._card_registry.consume_if_originator(button["callback_token"], "ou-originator", option_id=button["option_id"], option_value=button["option_value"]) + assert reason is None and entry is not None + assert _PENDING_CARD_METADATA_KEY not in inbound.metadata + finally: + if reservation: + reservation.release() + await asyncio.gather(*card_channel._running_card_tasks.values(), return_exceptions=True) diff --git a/backend/tests/test_feishu_card_interaction.py b/backend/tests/test_feishu_card_interaction.py new file mode 100644 index 0000000000..88b9451678 --- /dev/null +++ b/backend/tests/test_feishu_card_interaction.py @@ -0,0 +1,726 @@ +"""Tests for the Feishu interactive-card clarification module (outbound half). + +Rendering, ownership, one-time consumption and bounded registry lifecycle. +""" + +from __future__ import annotations + +import asyncio +import json +import time +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.channels.feishu_card_interaction import ( + MAX_CARD_OPTIONS, + ClarificationCardRequest, + PendingCardRegistry, + build_button_card, +) +from app.channels.message_bus import HUMAN_INPUT_RESPONSE_METADATA_KEY, InboundMessage, MessageBus +from app.channels.store import JsonChannelStore + + +def _metadata(payload: dict | None) -> dict: + return {} if payload is None else {"human_input_request": payload} + + +def _payload(**overrides) -> dict: + base = { + "version": 1, + "kind": "human_input_request", + "source": "ask_clarification", + "request_id": "clarification:call_1", + "question": "Pick one", + "input_mode": "choice_with_other", + "options": [ + {"id": "option-1", "label": "Alpha", "value": "Alpha"}, + {"id": "option-2", "label": "Beta", "value": "Beta"}, + ], + } + base.update(overrides) + return base + + +class TestClarificationCardRequestExtraction: + def test_absent_payload_returns_none(self): + assert ClarificationCardRequest.from_metadata(_metadata(None)) is None + + def test_free_text_mode_returns_none(self): + assert ClarificationCardRequest.from_metadata(_metadata(_payload(input_mode="free_text", options=[]))) is None + + def test_form_mode_returns_none(self): + assert ClarificationCardRequest.from_metadata(_metadata(_payload(version=2, input_mode="form", fields=[{"key": "x"}]))) is None + + def test_empty_options_returns_none(self): + assert ClarificationCardRequest.from_metadata(_metadata(_payload(options=[]))) is None + + def test_overflow_options_returns_none(self): + options = [{"id": f"option-{i}", "label": f"L{i}", "value": f"L{i}"} for i in range(1, MAX_CARD_OPTIONS + 2)] + assert ClarificationCardRequest.from_metadata(_metadata(_payload(options=options))) is None + + def test_choice_with_options_extracts(self): + req = ClarificationCardRequest.from_metadata(_metadata(_payload())) + assert req is not None + assert req.request_id == "clarification:call_1" + assert req.question == "Pick one" + assert req.options == ("Alpha", "Beta") + + def test_custom_option_ids_pass_through_to_buttons(self): + req = ClarificationCardRequest.from_metadata( + _metadata( + _payload( + options=[ + {"id": "approve", "label": "批准", "value": "批准"}, + {"id": "reject", "label": "拒绝", "value": "拒绝"}, + ] + ) + ) + ) + assert req is not None + card = json.loads(build_button_card(req)) + actions = next(e for e in card["elements"] if e["tag"] == "action")["actions"] + assert [a["value"]["option_id"] for a in actions] == ["approve", "reject"] + + def test_malformed_payload_returns_none(self): + assert ClarificationCardRequest.from_metadata(_metadata({"kind": "other"})) is None + assert ClarificationCardRequest.from_metadata({"human_input_request": "not-a-dict"}) is None + + +class TestBuildButtonCard: + def test_card_preserves_context_before_question(self): + req = ClarificationCardRequest.from_metadata(_metadata(_payload(question="是否继续?", context="这将**永久删除**生产表中的数据。\n请先确认已有备份。"))) + assert req is not None + + card = json.loads(build_button_card(req)) + markdown = next(element for element in card["elements"] if element["tag"] == "markdown") + assert markdown["content"] == "这将**永久删除**生产表中的数据。\n请先确认已有备份。\n\n是否继续?" + + @pytest.mark.parametrize("context", [None, "", " \t\n ", 42, ["background"]]) + def test_missing_or_invalid_context_keeps_question_only(self, context): + req = ClarificationCardRequest.from_metadata(_metadata(_payload(context=context))) + assert req is not None + + card = json.loads(build_button_card(req)) + markdown = next(element for element in card["elements"] if element["tag"] == "markdown") + assert markdown["content"] == "Pick one" + + def test_card_structure_binds_buttons_to_request(self): + req = ClarificationCardRequest.from_metadata(_metadata(_payload())) + assert req is not None + assert req.option_ids == ("option-1", "option-2") + card = json.loads(build_button_card(req)) + + assert card["config"]["update_multi"] is True + assert card["header"]["title"]["content"] + elements = card["elements"] + markdown = next(e for e in elements if e["tag"] == "markdown") + assert "Pick one" in markdown["content"] + + actions = next(e for e in elements if e["tag"] == "action")["actions"] + assert len(actions) == 2 + for i, (action, label) in enumerate(zip(actions, ("Alpha", "Beta")), start=1): + assert action["tag"] == "button" + assert action["text"]["content"] == label + assert action["value"]["callback_token"] == req.callback_token + assert action["value"]["option_id"] == f"option-{i}" + assert action["value"]["option_value"] == label + assert actions[0]["type"] == "primary" + assert actions[1]["type"] == "default" + + +class TestPendingCardRegistry: + def _entry(self, request_id: str = "clarification:call_1", **overrides) -> dict: + entry = { + "callback_token": "callback-1", + "request_id": request_id, + "source": "ask_clarification", + "thread_id": "thread-1", + "chat_id": "chat-1", + "originator_open_id": "ou_originator", + "options": ("Alpha", "Beta"), + "option_ids": ("option-1", "option-2"), + } + entry.update(overrides) + return entry + + def test_register_then_consume_returns_entry_once(self): + registry = PendingCardRegistry() + registry.register(**self._entry(topic_id="topic-1", connection_id="conn-1", owner_user_id="deer-user-1")) + entry, reason = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") + assert reason is None and entry is not None + assert entry["originator_open_id"] == "ou_originator" + assert entry["topic_id"] == "topic-1" + assert entry["connection_id"] == "conn-1" + assert entry["owner_user_id"] == "deer-user-1" + entry2, _ = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") + assert entry2 is None # one-time + + def test_same_request_id_keeps_independent_cards(self): + registry = PendingCardRegistry() + registry.register(**self._entry()) + registry.register(**self._entry(callback_token="callback-2", originator_open_id="ou_new")) + first, _ = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") + assert first is not None and first["originator_open_id"] == "ou_originator" + entry, _ = registry.consume_if_originator("callback-2", "ou_new", option_id="option-1", option_value="Alpha") + assert entry is not None + assert entry["originator_open_id"] == "ou_new" + + def test_duplicate_callback_token_cannot_replace_a_pending_card(self): + registry = PendingCardRegistry() + registry.register(**self._entry()) + with pytest.raises(ValueError, match="already owns"): + registry.register(**self._entry(request_id="clarification:other")) + entry, _ = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") + assert entry is not None and entry["request_id"] == "clarification:call_1" + + def test_register_prunes_expired_cards_without_clicks(self, monkeypatch): + clock = [100.0] + monkeypatch.setattr("app.channels.feishu_card_interaction.time.time", lambda: clock[0]) + registry = PendingCardRegistry(ttl_seconds=10) + registry.register(**self._entry()) + clock[0] += 11 + registry.register(**self._entry(callback_token="callback-2")) + assert len(registry._entries) == 1 + entry, _ = registry.consume_if_originator("callback-2", "ou_originator", option_id="option-1", option_value="Alpha") + assert entry is not None + + def test_capacity_retires_oldest_card_and_keeps_new_cards_usable(self): + registry = PendingCardRegistry(max_entries=2) + for index in range(3): + registry.register(**self._entry(callback_token=f"callback-{index}")) + oldest, _ = registry.consume_if_originator("callback-0", "ou_originator", option_id="option-1", option_value="Alpha") + assert oldest is None + for index in (1, 2): + entry, _ = registry.consume_if_originator(f"callback-{index}", "ou_originator", option_id="option-1", option_value="Alpha") + assert entry is not None + + +class TestFeishuChannelButtonCardWiring: + """Channel-level wiring: when the flag and payload line up, final + clarification outbounds go out as button cards; everything else is + byte-identical to today's plain-text path.""" + + def _channel(self, *, enabled: bool, tmp_path): + from unittest.mock import AsyncMock + + from app.channels.feishu import FeishuChannel + from app.channels.message_bus import MessageBus + + channel = FeishuChannel(MessageBus(), {"app_id": "test", "app_secret": "test", "card_interaction_enabled": enabled, "channel_store": JsonChannelStore(path=tmp_path / "routes.json")}) + channel._reply_card = AsyncMock(return_value="om_card_final") + channel._update_card = AsyncMock() + channel._add_reaction = AsyncMock() + channel._remember_thread_mapping = AsyncMock() + return channel + + def _clarification_outbound(self, *, with_options: bool = True): + from app.channels.message_bus import ( + HUMAN_INPUT_REQUEST_METADATA_KEY, + PENDING_CLARIFICATION_METADATA_KEY, + OutboundMessage, + ) + + metadata = { + PENDING_CLARIFICATION_METADATA_KEY: True, + "user_id": "ou_originator", + "topic_id": "topic-1", + "message_id": "om_src", + } + if with_options: + metadata[HUMAN_INPUT_REQUEST_METADATA_KEY] = { + "version": 1, + "kind": "human_input_request", + "source": "ask_clarification", + "request_id": "clarification:call_1", + "question": "Pick one", + "input_mode": "choice_with_other", + "options": [ + {"id": "option-1", "label": "Alpha", "value": "Alpha"}, + {"id": "option-2", "label": "Beta", "value": "Beta"}, + ], + } + else: + metadata[HUMAN_INPUT_REQUEST_METADATA_KEY] = { + "version": 1, + "kind": "human_input_request", + "source": "ask_clarification", + "request_id": "clarification:call_2", + "question": "Say anything", + "input_mode": "free_text", + } + return OutboundMessage( + channel_name="feishu", + chat_id="chat-1", + thread_id="thread-1", + text="Pick one", + is_final=True, + thread_ts="om_src", + metadata=metadata, + ) + + @pytest.mark.asyncio + async def test_flag_on_options_renders_button_card_and_registers(self, tmp_path): + channel = self._channel(enabled=True, tmp_path=tmp_path) + await channel._send_card_message(self._clarification_outbound()) + + card_json = channel._reply_card.call_args.args[1] + card = json.loads(card_json) + assert any(e.get("tag") == "action" for e in card["elements"]) + + token = next(e for e in card["elements"] if e["tag"] == "action")["actions"][0]["value"]["callback_token"] + entry, reason = channel._card_registry.consume_if_originator(token, "ou_originator", option_id="option-1", option_value="Alpha") + assert reason is None and entry is not None + assert entry["originator_open_id"] == "ou_originator" + assert entry["thread_id"] == "thread-1" + assert entry["card_message_id"] == "om_card_final" + + @pytest.mark.asyncio + async def test_flag_off_falls_back_to_plain_card(self, tmp_path): + channel = self._channel(enabled=False, tmp_path=tmp_path) + await channel._send_card_message(self._clarification_outbound()) + + # Plain-text path: markdown text, no raw-card bypass, nothing registered. + text_arg = channel._reply_card.call_args.args[1] + assert "Pick one" in text_arg + assert '"action"' not in text_arg + assert channel._reply_card.call_args.kwargs.get("raw_card_content") is not True + entry, _ = channel._card_registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") + assert entry is None + + @pytest.mark.asyncio + async def test_flag_on_free_text_falls_back_to_plain_card(self, tmp_path): + channel = self._channel(enabled=True, tmp_path=tmp_path) + await channel._send_card_message(self._clarification_outbound(with_options=False)) + + text_arg = channel._reply_card.call_args.args[1] + assert '"action"' not in text_arg + assert channel._reply_card.call_args.kwargs.get("raw_card_content") is not True + entry, _ = channel._card_registry.consume_if_originator("clarification:call_2", "ou_originator", option_id="option-1", option_value="x") + assert entry is None + + @pytest.mark.asyncio + async def test_running_card_is_updated_into_button_card(self, tmp_path): + channel = self._channel(enabled=True, tmp_path=tmp_path) + channel._running_card_ids["om_src"] = "om_running" + + await channel._send_card_message(self._clarification_outbound()) + + card_json = channel._update_card.call_args.args[1] + card = json.loads(card_json) + assert any(e.get("tag") == "action" for e in card["elements"]) + token = next(e for e in card["elements"] if e["tag"] == "action")["actions"][0]["value"]["callback_token"] + entry, _ = channel._card_registry.consume_if_originator(token, "ou_originator", option_id="option-1", option_value="Alpha") + assert entry is not None + assert entry["card_message_id"] == "om_running" + + +class TestConsumeIfOriginator: + """Atomic validate-and-consume for card clicks. Identity mismatch must NOT + consume the entry (the real originator may still click later).""" + + def _registry_with_entry(self): + registry = PendingCardRegistry() + registry.register( + callback_token="callback-1", + request_id="clarification:call_1", + source="ask_clarification", + thread_id="thread-1", + chat_id="chat-1", + originator_open_id="ou_originator", + options=("Alpha", "Beta"), + option_ids=("option-1", "option-2"), + card_message_id="om_card_1", + ) + return registry + + def test_originator_consumes_and_gets_entry(self): + registry = self._registry_with_entry() + entry, reason = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") + assert reason is None + assert entry is not None + assert entry["card_message_id"] == "om_card_1" + # replayed click: entry already gone + entry2, reason2 = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") + assert entry2 is None + assert reason2 == "unknown" + + def test_non_originator_rejected_without_consuming(self): + registry = self._registry_with_entry() + entry, reason = registry.consume_if_originator("callback-1", "ou_someone_else", option_id="option-1", option_value="Alpha") + assert entry is None + assert reason == "not_originator" + # the real originator can still click afterwards + entry2, reason2 = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") + assert reason2 is None + assert entry2 is not None + + def test_unknown_request_id(self): + registry = self._registry_with_entry() + entry, reason = registry.consume_if_originator("clarification:nope", "ou_originator", option_id="option-1", option_value="Alpha") + assert entry is None + assert reason == "unknown" + + def test_expired_entry_is_unknown(self): + registry = self._registry_with_entry() + future = time.time() + 31 * 60 + entry, reason = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha", now=future) + assert entry is None + assert reason == "unknown" + + def test_unknown_option_value_rejected_without_consuming(self): + registry = self._registry_with_entry() + entry, reason = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="注入值") + assert entry is None + assert reason == "unknown" + # the real button value still works afterwards + entry2, reason2 = registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") + assert reason2 is None and entry2 is not None + + +def _make_card_action_event(*, callback_token: str, option_value: str, option_id: str = "option-1", operator_open_id: str = "ou_originator"): + """Build a minimal P2CardActionTrigger-shaped stand-in.""" + from unittest.mock import MagicMock + + event = MagicMock() + event.event.action.value = {"callback_token": callback_token, "option_id": option_id, "option_value": option_value} + event.event.operator.open_id = operator_open_id + return event + + +class TestOnCardActionValidation: + """Sync layer of the card-action callback: validation + toast, no state change.""" + + def _channel(self): + from unittest.mock import MagicMock + + from app.channels.feishu import FeishuChannel + from app.channels.message_bus import MessageBus + + channel = FeishuChannel(MessageBus(), {"app_id": "t", "app_secret": "t", "card_interaction_enabled": True}) + channel._card_registry.register( + callback_token="callback-1", + request_id="clarification:call_1", + source="ask_clarification", + thread_id="thread-1", + chat_id="chat-1", + originator_open_id="ou_originator", + options=("Alpha", "Beta"), + option_ids=("option-1", "option-2"), + card_message_id="om_card_1", + ) + channel._main_loop = MagicMock() + channel._main_loop.is_running.return_value = True + channel._submit_threadsafe_coroutine = MagicMock(side_effect=lambda coroutine, *args, **kwargs: (coroutine.close(), True)[1]) + + class _Toast: + pass + + class _Resp: + pass + + channel._CallBackToast = _Toast + channel._P2CardActionTriggerResponse = _Resp + return channel + + def test_non_originator_gets_refusal_toast_and_entry_survives(self): + channel = self._channel() + resp = channel._on_card_action(_make_card_action_event(callback_token="callback-1", option_id="option-1", option_value="Alpha", operator_open_id="ou_other")) + assert resp.toast is not None + assert "发起人" in resp.toast.content + channel._submit_threadsafe_coroutine.assert_not_called() + # entry not consumed: originator can still click + entry, reason = channel._card_registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") + assert reason is None and entry is not None + + def test_unknown_request_id_gets_stale_toast(self): + channel = self._channel() + resp = channel._on_card_action(_make_card_action_event(callback_token="unknown", option_id="option-1", option_value="Alpha")) + assert resp.toast is not None + assert "失效" in resp.toast.content + channel._submit_threadsafe_coroutine.assert_not_called() + + def test_dispatch_failure_returns_warning_toast(self): + channel = self._channel() + channel._submit_threadsafe_coroutine = MagicMock(side_effect=lambda coroutine, *args, **kwargs: (coroutine.close(), False)[1]) + resp = channel._on_card_action(_make_card_action_event(callback_token="callback-1", option_id="option-1", option_value="Alpha")) + assert resp.toast is not None + assert resp.toast.type == "warning" + + def test_originator_gets_ack_toast_and_dispatches_async_resolution(self): + channel = self._channel() + resp = channel._on_card_action(_make_card_action_event(callback_token="callback-1", option_id="option-1", option_value="Alpha")) + assert resp.toast is not None + assert "Alpha" in resp.toast.content + channel._submit_threadsafe_coroutine.assert_called_once() + # one-time: the entry is consumed by the click + entry, reason = channel._card_registry.consume_if_originator("callback-1", "ou_originator", option_id="option-1", option_value="Alpha") + assert entry is None and reason == "unknown" + + +class TestResolveCardClick: + """Async layer: terminal card update + hidden synthetic inbound publication.""" + + @pytest.mark.asyncio + async def test_publishes_hidden_inbound_and_marks_card_resolved(self): + from unittest.mock import AsyncMock + + from app.channels.feishu import FeishuChannel + from app.channels.message_bus import HUMAN_INPUT_RESPONSE_METADATA_KEY, MessageBus + + bus = MessageBus() + channel = FeishuChannel(bus, {"app_id": "t", "app_secret": "t", "card_interaction_enabled": True}) + channel._update_card = AsyncMock() + + published = [] + + async def capture(msg): + published.append(msg) + return True + + channel._publish_inbound_or_drop = capture + + entry = { + "request_id": "clarification:call_1", + "source": "ask_clarification", + "thread_id": "thread-1", + "chat_id": "chat-1", + "originator_open_id": "ou_originator", + "options": ("Alpha", "Beta"), + "option_ids": ("option-1", "option-2"), + "card_message_id": "om_card_1", + "topic_id": "om_topic_1", + "connection_id": "conn-1", + "owner_user_id": "deer-user-1", + } + await channel._resolve_card_click(entry, option_id="option-1", option_value="Alpha", operator_open_id="ou_originator") + + # terminal card: no buttons, shows the choice + assert channel._update_card.await_count == 1 + terminal_text = channel._update_card.await_args.args[1] + assert "Alpha" in terminal_text + + # synthetic inbound: hidden human-input response routed at the card message, + # carrying the routing identity the manager keys on (topic + connection). + assert len(published) == 1 + inbound = published[0] + assert inbound.chat_id == "chat-1" + assert inbound.user_id == "ou_originator" + assert inbound.thread_ts == "om_card_1" + assert inbound.topic_id == "om_topic_1" + assert inbound.connection_id == "conn-1" + assert inbound.owner_user_id == "deer-user-1" + response = inbound.metadata[HUMAN_INPUT_RESPONSE_METADATA_KEY] + assert response["kind"] == "human_input_response" + assert response["response_kind"] == "option" + assert response["request_id"] == "clarification:call_1" + assert response["value"] == "Alpha" + assert inbound.metadata.get("hide_from_ui") is True + + @pytest.mark.asyncio + async def test_dropped_inbound_leaves_card_unresolved(self): + """If the inbound cannot be published, the card must not flip to a false resolved state.""" + from unittest.mock import AsyncMock + + from app.channels.feishu import FeishuChannel + from app.channels.message_bus import MessageBus + + channel = FeishuChannel(MessageBus(), {"app_id": "t", "app_secret": "t", "card_interaction_enabled": True}) + channel._update_card = AsyncMock() + channel._publish_inbound_or_drop = AsyncMock(return_value=False) + + entry = { + "request_id": "clarification:call_1", + "source": "ask_clarification", + "thread_id": "thread-1", + "chat_id": "chat-1", + "originator_open_id": "ou_originator", + "options": ("Alpha",), + "card_message_id": "om_card_1", + } + await channel._resolve_card_click(entry, option_id="option-1", option_value="Alpha", operator_open_id="ou_originator") + + channel._update_card.assert_not_awaited() + + +class TestCardClickEndToEndRouting: + """Pin the contract the reviewer flagged: the synthetic inbound must carry + the routing identity the manager's thread resolution keys on, so a card + click resumes the clarified thread instead of opening a new one.""" + + def test_card_answer_resumes_existing_thread(self, tmp_path, monkeypatch): + from app.channels.manager import ChannelManager + from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelCredentialCipher + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + + async def go(): + await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'channel-connections.db'}", sqlite_dir=str(tmp_path)) + repo = ChannelConnectionRepository(get_session_factory(), cipher=ChannelCredentialCipher.from_key("test-channel-key")) + conn = await repo.upsert_connection( + owner_user_id="deer-user-1", + provider="feishu", + external_account_id="ou_originator", + workspace_id="chat-1", + ) + # What the channel persisted when the button card was delivered: + # the card's topic maps to the clarified thread. + await repo.set_thread_id( + connection_id=conn["id"], + owner_user_id="deer-user-1", + provider="feishu", + external_conversation_id="chat-1", + thread_id="thread-existing", + external_topic_id="om_topic_1", + ) + + bus = MessageBus() + store = JsonChannelStore(path=tmp_path / "legacy-store.json") + manager = ChannelManager(bus=bus, store=store, connection_repo=repo) + mock_client = MagicMock() + mock_client.threads.create = AsyncMock(return_value={"thread_id": "thread-existing"}) + mock_client.threads.update = AsyncMock(return_value={"thread_id": "thread-existing"}) + mock_client.threads.get = AsyncMock(return_value={"thread_id": "thread-existing"}) + mock_client.runs.wait = AsyncMock(return_value={"messages": [{"type": "human", "content": "Alpha"}, {"type": "ai", "content": "resumed"}]}) + + # Feishu is a streaming channel: the run goes through runs.stream, + # whose first positional argument is the thread id. + class _Part: + def __init__(self, event, data): + self.event = event + self.data = data + + async def _stream(*args, **kwargs): + yield _Part("values", {"messages": [{"type": "human", "content": "Alpha"}, {"type": "ai", "content": "resumed"}], "artifacts": []}) + + mock_client.runs.stream = MagicMock(side_effect=lambda *args, **kwargs: _stream(*args, **kwargs)) + manager._client = mock_client + + # The synthetic inbound exactly as _resolve_card_click builds it. + inbound = InboundMessage( + channel_name="feishu", + chat_id="chat-1", + user_id="ou_originator", + owner_user_id="deer-user-1", + connection_id=conn["id"], + text="Alpha", + thread_ts="om_card_1", + topic_id="om_topic_1", + metadata={ + "hide_from_ui": True, + HUMAN_INPUT_RESPONSE_METADATA_KEY: { + "version": 1, + "kind": "human_input_response", + "source": "ask_clarification", + "request_id": "clarification:call_1", + "response_kind": "option", + "option_id": "option-1", + "value": "Alpha", + }, + }, + ) + await manager._handle_chat(inbound) + + # The clarified thread is reused — nothing is created anew. + mock_client.threads.create.assert_not_called() + assert mock_client.runs.stream.call_count == 1 + assert mock_client.runs.stream.call_args.args[0] == "thread-existing" + + # And the posted human message carries the hidden human-input response. + posted = mock_client.runs.stream.call_args.kwargs["input"]["messages"][0] + assert posted["additional_kwargs"]["hide_from_ui"] is True + assert posted["additional_kwargs"][HUMAN_INPUT_RESPONSE_METADATA_KEY]["value"] == "Alpha" + + try: + asyncio.run(go()) + finally: + asyncio.run(close_engine()) + + def test_card_answer_passes_bound_identity_admission(self, tmp_path, monkeypatch): + """Connections-enabled deployments (require_bound_identity=True) re-read the + binding with workspace_id; Feishu binds connections with workspace_id=chat_id, + so the synthetic inbound must carry it or every click is rejected as unbound.""" + from app.channels.manager import ChannelManager + from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelCredentialCipher + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + + async def go(): + await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'cc.db'}", sqlite_dir=str(tmp_path)) + repo = ChannelConnectionRepository(get_session_factory(), cipher=ChannelCredentialCipher.from_key("test-channel-key")) + conn = await repo.upsert_connection( + owner_user_id="deer-user-1", + provider="feishu", + external_account_id="ou_originator", + workspace_id="chat-1", + ) + await repo.set_thread_id( + connection_id=conn["id"], + owner_user_id="deer-user-1", + provider="feishu", + external_conversation_id="chat-1", + thread_id="thread-existing", + external_topic_id="om_topic_1", + ) + + bus = MessageBus() + store = JsonChannelStore(path=tmp_path / "legacy-store.json") + manager = ChannelManager(bus=bus, store=store, connection_repo=repo, require_bound_identity=True) + mock_client = MagicMock() + mock_client.threads.create = AsyncMock(return_value={"thread_id": "thread-existing"}) + mock_client.threads.update = AsyncMock(return_value={"thread_id": "thread-existing"}) + mock_client.threads.get = AsyncMock(return_value={"thread_id": "thread-existing"}) + + class _Part: + def __init__(self, event, data): + self.event = event + self.data = data + + async def _stream(*args, **kwargs): + yield _Part("values", {"messages": [{"type": "human", "content": "Alpha"}, {"type": "ai", "content": "resumed"}], "artifacts": []}) + + mock_client.runs.stream = MagicMock(side_effect=lambda *args, **kwargs: _stream(*args, **kwargs)) + manager._client = mock_client + + # Build the inbound through the channel's real click-resolution path, + # then hand it to the manager — this is the production seam. + from app.channels.feishu import FeishuChannel + + channel = FeishuChannel(MessageBus(), {"app_id": "t", "app_secret": "t", "card_interaction_enabled": True}) + channel._update_card = AsyncMock() + produced = [] + + async def capture(msg): + produced.append(msg) + return True + + channel._publish_inbound_or_drop = capture + entry = { + "request_id": "clarification:call_1", + "source": "ask_clarification", + "thread_id": "thread-existing", + "chat_id": "chat-1", + "originator_open_id": "ou_originator", + "options": ("Alpha", "Beta"), + "option_ids": ("option-1", "option-2"), + "card_message_id": "om_card_1", + "topic_id": "om_topic_1", + "connection_id": conn["id"], + "owner_user_id": "deer-user-1", + } + await channel._resolve_card_click(entry, option_id="option-1", option_value="Alpha", operator_open_id="ou_originator") + assert len(produced) == 1 + + await manager._handle_chat(produced[0]) + + mock_client.threads.create.assert_not_called() + assert mock_client.runs.stream.call_count == 1 + assert mock_client.runs.stream.call_args.args[0] == "thread-existing" + + try: + asyncio.run(go()) + finally: + asyncio.run(close_engine()) diff --git a/config.example.yaml b/config.example.yaml index dc30df1031..e8ad228d56 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -3076,6 +3076,11 @@ run_ownership: # app_secret: $FEISHU_APP_SECRET # # domain: https://open.feishu.cn # China (default) # # domain: https://open.larksuite.com # International +# # card_interaction_enabled: false # render option clarifications as clickable button +# # cards; requires subscribing card.action.trigger +# # under the console's callback-configuration tab +# # (separate from event subscriptions) and +# # publishing a new app version # # slack: # enabled: false