mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
feat(channels): Feishu interactive clarification cards (#6670)
* feat(channels): render Feishu option clarifications as button cards When the agent's ask_clarification carries options, the Feishu channel now renders the question as an interactive card with one button per option instead of plain text. This is the outbound half: card construction plus a process-local pending registry that a later card-action callback handler will validate clicks against. - New feishu_card_interaction module owns request extraction, button card rendering, and the pending registry (30-minute TTL, matching the typed-reply clarification window) - Manager attaches the clarification's structured human_input payload to the final outbound's metadata so channels can render richer affordances without touching the Gateway protocol - Feishu channel renders the button card only when card_interaction_enabled is set (default off); free-text, form, and over-four-option clarifications keep the plain-text path - Button cards bypass the markdown wrapper via a raw_card_content flag on the card send/patch helpers; plain-text call signatures unchanged * feat(channels): close the Feishu button-card click loop The inbound half of interactive clarification cards: a click on an option button validates, resolves the clarification, and marks the card in place. - Register the card-action callback only when card_interaction_enabled is set; clicks arrive as WS EVENT frames through the SDK's callback processor map, and the synchronous return value becomes the toast ack - consume_if_originator atomically validates operator identity against the pending entry: non-originators get a refusal toast and the entry survives; unknown or expired requests are toast-only and dropped - A validated click consumes once, updates the card to a terminal resolved state, and injects a hidden inbound carrying the human_input_response so the run resumes on the originating thread with Web-card parity - Manager translates the inbound's human_input_response metadata into the human message's additional_kwargs with hide_from_ui set * docs(channels): document Feishu interactive clarification cards - README / README_zh: optional card_interaction_enabled setup, calling out that card.action.trigger lives under the console's callback configuration tab (separate from event subscriptions) and requires publishing a new app version - config.example.yaml: commented flag with the same console caveat - backend/AGENTS.md: module ownership note for feishu_card_interaction * fix(channels): close Feishu card-click routing and ack gaps from review Address review on #6670: - Carry the routing identity (topic_id, connection_id, owner_user_id) on the pending entry and the synthetic inbound, so clicks resolve to the clarified thread in topic-scoped chats and pass bound-identity admission instead of opening a new thread or being rejected - Validate option_value against the registered options at consume time: an unknown value is treated like a stale card and the entry is kept - Return a warning toast when a click cannot be dispatched, and flip the card to its resolved state only after the inbound is durably in the bus — never display a choice that was not recorded - Drop the dead mark_sent/consume registry APIs and pass the payload's own option ids through to the buttons - Add an end-to-end test through the manager's thread resolution pinning that a card answer reuses the clarified thread * fix(channels): carry workspace_id on the Feishu card-click inbound The connections-enabled admission path re-reads the binding with the message's workspace_id, and Feishu connections are bound with workspace_id = chat_id. The synthetic inbound set connection_id and owner_user_id but never workspace_id, so the re-read missed and every click was rejected as unbound after the entry was consumed. - Set inbound.workspace_id from the entry's chat_id (Feishu's chat-as-workspace binding convention) - Extend the card-click e2e to require_bound_identity=True, driven through the channel's real _resolve_card_click production * fix(channels): stabilize Feishu clarification delivery and routing Option cards could fail through the real reply SDK, resume another conversation, or leave a stale text hint after the answer was accepted. Unify raw-card delivery and persist trusted thread aliases before showing buttons. Keep protocol request IDs separate from bounded, one-time callback tokens; validate originators and option pairs, and retire only the accepted card's text/button state. Treat valid hidden human answers as turn boundaries so old replies and artifacts do not leak into the next answer. Cover consecutive clarifications and /new routing through real SDK request models, stores, and the run journal. Validation: 147 focused tests and 29,421 offline tests passed (405 skipped); strict blocking-I/O passed 245 tests (1 skipped). Ruff and guidance checks passed. * fix(channels): preserve clarification context in Feishu cards Button cards replaced the full clarification text but omitted its background, leaving users without the context needed to choose an option. Carry optional context from request metadata and render its original Markdown before the question. Keep question-only cards for absent, blank, or invalid context. Cover rendering and all real SDK delivery paths, and update both user guides. Validation: 156 focused tests, 245 strict blocking-I/O tests, Ruff, and guidance checks passed. The offline run recorded 29,381 passed, 49 failed, and 405 skipped; 48 import failures reproduced before the fix. All 49 cases passed on both versions in isolated environments, and the Feishu-to-export order check passed 13 tests.
This commit is contained in:
Storm
GitHub
parent
a1383a7953
commit
3f7a076972
@ -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**
|
||||
|
||||
|
||||
@ -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 配置**
|
||||
|
||||
|
||||
@ -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.
|
||||
|
||||
|
||||
@ -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 <name>` 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/<name>/chats/<thread_id>`; `lead_agent` omits it and stays on the ordinary chat route.
|
||||
|
||||
@ -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()
|
||||
|
||||
247
backend/app/channels/feishu_card_interaction.py
Normal file
247
backend/app/channels/feishu_card_interaction.py
Normal file
@ -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]
|
||||
@ -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:
|
||||
|
||||
@ -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.
|
||||
|
||||
131
backend/tests/test_channel_hidden_human_input.py
Normal file
131
backend/tests/test_channel_hidden_human_input.py
Normal file
@ -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 == []
|
||||
@ -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
|
||||
|
||||
626
backend/tests/test_feishu_card_delivery.py
Normal file
626
backend/tests/test_feishu_card_delivery.py
Normal file
@ -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)
|
||||
726
backend/tests/test_feishu_card_interaction.py
Normal file
726
backend/tests/test_feishu_card_interaction.py
Normal file
@ -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())
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user