mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-27 13:40:37 +00:00
feat(feishu): deliver scheduled-task notifications to bound users (#6753)
Co-authored-by: Totoro-qaq <279883115+Totoro-qaq@users.noreply.github.com>
This commit is contained in:
Totoro
GitHub
parent
338f8819a4
commit
4b36e680c3
@ -1082,6 +1082,8 @@ DINGTALK_CLIENT_SECRET=your_client_secret
|
||||
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.
|
||||
|
||||
Scheduled-task result and lifecycle notices can also be pushed to a bound Feishu/Lark user, without a new inbound message. They use the same notification outbox and localized summaries as WeCom. The app must be able to send to that user; see [Feishu scheduled-task notifications](backend/docs/IM_CHANNEL_CONNECTIONS.md#feishu-scheduled-task-notifications).
|
||||
|
||||
**WeChat Setup**
|
||||
|
||||
1. Enable the `wechat` channel in `config.yaml`.
|
||||
|
||||
@ -13,7 +13,7 @@ Cursor writes drain before cancellation; token expiry drains cursor/auth clearin
|
||||
- `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn) and `/agent` (`list` is owner-scoped; `use` creates a fresh thread and persists the Custom Agent selection under both the channel restart key and Web's canonical routing metadata so existing checkpoint lineage never changes agent across IM/Web), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout`
|
||||
Streaming `error` frames use the same failure path as transport exceptions. Publish the final outbound before releasing the inbound dedupe key, so provider redelivery can retry without overtaking the terminal reply. Log the frame's error name and message at WARNING with the thread ID; keep the synthetic error generic so payload details do not change reply selection.
|
||||
**What may be published from the stream is an allowlist, not a denylist** (`_accumulate_stream_text`): only assistant message types — LangChain serializes `AIMessage.type` as `"ai"` and `AIMessageChunk.type` as `"AIMessageChunk"`, plus the OpenAI-style `"assistant"` spelling for foreign runtimes — become displayable text. The previous rule rejected only payloads whose `type` contained `"tool"` and therefore published everything else, which leaked DeerFlow's hidden model context to every streaming IM channel: `DynamicContextMiddleware` injects recalled memory as a hidden `HumanMessage` (`type == "human"`) and rewrites the user's own turn into a new `HumanMessage`, `DurableContextMiddleware` injects a hidden `<durable_context_data>` `HumanMessage`, and LangGraph fans those state writes out on the `messages-tuple` stream. Proved live on a Buzz relay, which published a `<memory>` fact block and, in another run, a verbatim echo of the user's own message as the assistant's reply. Matching is by prefix (`ai` / `assistant`), never substring, because ordinary words contain `"ai"` (`chain`, `domain`). The message type is resolved by `_stream_payload_type`, which handles both the `model_dump()` shape DeerFlow's own gateway emits and LangChain's `to_json()` constructor shape (whose top-level `type` is the literal `"constructor"`, with the class name at the tail of the `id` path). A bare `str` payload is no longer accepted at all: it carries no type information, so it cannot be attributed to the assistant, and nothing in DeerFlow produces one (`runtime/serialization.py::serialize_messages_tuple` always emits `[message_dict, metadata]`).
|
||||
- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()` (or `_submit_threadsafe_coroutine_future()` to await the outcome through `asyncio.shield`): it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. Blocking SDK teardown in `stop()` (a client `close()`, a thread `join()`) must run off the event loop through `asyncio.to_thread`, since every run and channel on the Gateway shares that loop and `stop()` also serves `POST /api/channels/{name}/restart`; Slack's `SocketModeClient.close()` is a shielded, tracked task so a cancelled stop leaves it running and a retried stop awaits the same close. Anchored by `tests/blocking_io/test_channel_stop_offloop.py`. **Proactive push** (`Channel.send_notification`): the scheduled-task outbox pushes markdown to a bound identity with no inbound frame. The base default raises `NotImplementedError` (after a `ChannelUnavailable` check when not running). Scheduler/Settings read the `proactive_notifications` flag in `capabilities.py`; providers without it get no rows (legacy rows end terminal `failed`); `test_channel_capabilities.py` pins it to the registry and overrides. Adding push: implement it (return after the ACK), raise `ChannelUnavailable` for transient failures (parks, no retry used; rejections count), flip the flag. **Only WeCom pushes today** (`WSClient.send_message`); its SDK raises `RuntimeError` for a dead socket and a rejection alike, so only the `Reply ack error` prefix counts; other `RuntimeError`s, `ConnectionError`/`TimeoutError`/`OSError` and the `websockets` close exception park. Before sending, the worker re-checks the binding (owner-scoped `resolve_connections`, before liveness): a gone target is `mark_failed(terminal=True)`; a lookup failure parks. Terminal rows ignore later status writes. Scheduler side: `_on_finalization` enqueues each notice in the outcome's transaction, so recovery notifies once (unique key; idle ends use `idle:`); one message per occurrence (stop > pause > finish > run outcome); plain trials and interrupts stay silent; text (`scheduler/notification_text.py`) is localized, without IDs or links. Lifespan starts `NotificationDeliveryWorker` after the channel service (detaching enqueue if it cannot) and stops it first.
|
||||
- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()` (or `_submit_threadsafe_coroutine_future()` to await the outcome through `asyncio.shield`): it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. Blocking SDK teardown in `stop()` (a client `close()`, a thread `join()`) must run off the event loop through `asyncio.to_thread`, since every run and channel on the Gateway shares that loop and `stop()` also serves `POST /api/channels/{name}/restart`; Slack's `SocketModeClient.close()` is a shielded, tracked task so a cancelled stop leaves it running and a retried stop awaits the same close. Anchored by `tests/blocking_io/test_channel_stop_offloop.py`. **Proactive push** (`Channel.send_notification`): the scheduled-task outbox pushes markdown to a bound identity with no inbound frame. The base default raises `NotImplementedError` (after a `ChannelUnavailable` check when not running). Scheduler/Settings read the `proactive_notifications` flag in `capabilities.py`; providers without it get no rows (legacy rows end terminal `failed`); `test_channel_capabilities.py` pins it to the registry and overrides. Adding push: await ACK, raise `ChannelUnavailable` on transport failure (parks; rejections count), flip the flag. **WeCom/Feishu push**; Feishu uses owner-scoped p2p routing. WeCom uses `WSClient.send_message`; its SDK raises `RuntimeError` for a dead socket and a rejection alike, so only the `Reply ack error` prefix counts; other `RuntimeError`s, `ConnectionError`/`TimeoutError`/`OSError` and the `websockets` close exception park. Before sending, the worker re-checks the binding (owner-scoped `resolve_connections`, before liveness): a gone target is `mark_failed(terminal=True)`; a lookup failure parks. Terminal rows ignore later status writes. Scheduler side: `_on_finalization` enqueues each notice in the outcome's transaction, so recovery notifies once (unique key; idle ends use `idle:`); one message per occurrence (stop > pause > finish > run outcome); plain trials and interrupts stay silent; text (`scheduler/notification_text.py`) is localized, without IDs or links. Lifespan starts `NotificationDeliveryWorker` after the channel service (detaching enqueue if it cannot) and stops it first.
|
||||
- `wecom.py` - WeCom WebSocket channel with proactive `send_notification`; `_ws_transport_up` maps disconnects to `ChannelUnavailable` (no retry used). `capabilities.py` - SDK-free `CHANNEL_CAPABILITIES`.
|
||||
- `service.py` - Drain manager work before transports; retain timed-out instances for retry. Live Feishu/DingTalk/Telegram joins raise `ChannelStopTimeout`: retain, defer replacement/restart, log WARNING without traceback; wedged instances report not running. Reload returns snapshots; restart owns cached fallback/publication under the channel lock. `_stopping` fences starts/publication (including after pre-start I/O) until shutdown succeeds. Lock manager startup through stale cleanup; capture the shutdown generation before locking to fence queued starts. `stop()` stays unlocked to fence in-flight startup.
|
||||
- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, edits the "Working on it..." stream target in place via `editMessageText`, and can optionally send final Markdown replies as Rich Messages through `channels.telegram.rich_messages` — the Rich path only fires when `rich_messages` is on **and** the text contains a rich construct (fenced code, emphasis, a table, a task list, `<details>`, block math, or a link), so structured command/error replies (none of these) stay plain text and their newlines and `<placeholder>` tokens survive; `discord.py` runs `_on_message` on discord.py's private loop, so connection-repository work (identity lookup with the reservation commit, `/connect` binding) goes to the Gateway loop through `_submit_threadsafe_coroutine` — the SQLAlchemy engine is bound there — and bind replies return through `_run_on_discord_loop`; typing registers before that hand-off settles, and a failed hand-off skips the ack and stops the indicator it started unless another message still relies on it; `discord.py` registers typing loops before inbound handling yields and `_start_typing()` refuses work once `_running` is false; each tick sends the indicator with discord.py 2.x's `await channel.typing()` (`trigger_typing()` was removed in 2.0, and each typing loop logs its first failed tick at WARNING, later ones at DEBUG); because `stop()` runs on the main loop while typing tasks belong to `_discord_loop`, normal cross-thread cancellation, awaiting, and map cleanup are scheduled there with a bounded wait, while `_run_client()` drains tasks in its `finally` block before an exception/disconnect can make that loop unusable; an already-stopped foreign loop must never have its tasks awaited from the main loop; to that end every outbound cross-loop call (`send` / `send_file` / `_get_channel_or_thread`) goes through `_run_on_discord_loop`, which bounds the await (`DISCORD_OUTBOUND_TIMEOUT_SECONDS` 30s; `DISCORD_UPLOAD_TIMEOUT_SECONDS` 120s for file uploads, whose unbounded-size payload needs room for a slow uplink plus 429 retry-after) and fails fast with a `RuntimeError` when the client loop is missing or not running — a dead client becomes a logged send failure instead of a permanently hung `ChannelManager` worker — and `is_running` reports client-thread aliveness (like `feishu.py`) so `ensure_channel_ready` can restart the channel after `_run_client()` exits on a fatal error; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured, and overrides `receive_file` to download inbound images (`picture`/`richText`) and documents (`file`) by `downloadCode` into the thread uploads bucket, mirroring `feishu.py`)
|
||||
|
||||
@ -149,6 +149,14 @@ class Channel(ABC):
|
||||
raise ChannelUnavailable(f"channel '{self.name}' is not running")
|
||||
raise NotImplementedError(f"channel '{self.name}' does not support proactive notifications")
|
||||
|
||||
async def send_notification_for_owner(self, *, owner_user_id: str | None, target: str, text_markdown: str) -> None:
|
||||
"""Deliver to an already-verified identity, with owner context for routing.
|
||||
|
||||
Providers may resolve a private conversation from that owner's current
|
||||
bindings. The outbox target remains the external account identity.
|
||||
"""
|
||||
await self.send_notification(target=target, text_markdown=text_markdown)
|
||||
|
||||
# -- helpers -----------------------------------------------------------
|
||||
|
||||
async def _send_with_retry(
|
||||
|
||||
@ -18,7 +18,7 @@ CHANNEL_CAPABILITIES: dict[str, dict[str, bool]] = {
|
||||
"buzz": {"supports_streaming": True, "proactive_notifications": False},
|
||||
"dingtalk": {"supports_streaming": False, "proactive_notifications": False},
|
||||
"discord": {"supports_streaming": False, "proactive_notifications": False},
|
||||
"feishu": {"supports_streaming": True, "proactive_notifications": False},
|
||||
"feishu": {"supports_streaming": True, "proactive_notifications": True},
|
||||
"github": {"supports_streaming": False, "proactive_notifications": False},
|
||||
"qq": {"supports_streaming": False, "proactive_notifications": False},
|
||||
"slack": {"supports_streaming": False, "proactive_notifications": False},
|
||||
|
||||
@ -16,7 +16,7 @@ from concurrent.futures import Future
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Literal, TypeVar
|
||||
|
||||
from app.channels.base import Channel, ChannelStopTimeout
|
||||
from app.channels.base import Channel, ChannelStopTimeout, ChannelUnavailable
|
||||
from app.channels.commands import is_known_channel_command, strip_leading_mentions
|
||||
from app.channels.connection_identity import attach_connection_identity, lookup_thread_id
|
||||
from app.channels.feishu_card_interaction import (
|
||||
@ -432,6 +432,59 @@ class FeishuChannel(Channel):
|
||||
log_prefix="[Feishu]",
|
||||
)
|
||||
|
||||
async def send_notification(self, *, target: str, text_markdown: str) -> None:
|
||||
"""Send a static card to a bound user's open ID and await its ACK."""
|
||||
await self._send_notification_card(receive_id=target, receive_id_type="open_id", text_markdown=text_markdown)
|
||||
|
||||
async def send_notification_for_owner(self, *, owner_user_id: str | None, target: str, text_markdown: str) -> None:
|
||||
"""Prefer a verified private chat when direct user delivery is restricted."""
|
||||
if self._connection_repo is not None and owner_user_id:
|
||||
try:
|
||||
connections = await self._connection_repo.list_connections(owner_user_id)
|
||||
except Exception as exc:
|
||||
raise ChannelUnavailable("Feishu notification binding lookup is unavailable") from exc
|
||||
matching_binding = False
|
||||
for connection in connections:
|
||||
if connection.get("owner_user_id") != owner_user_id or connection.get("provider") != self.name or connection.get("status") != "connected" or connection.get("external_account_id") != target:
|
||||
continue
|
||||
matching_binding = True
|
||||
metadata = connection.get("metadata")
|
||||
if not isinstance(metadata, dict) or metadata.get("chat_type") != "p2p":
|
||||
continue
|
||||
chat_id = self._non_empty_str(metadata.get("chat_id"))
|
||||
if chat_id and chat_id == connection.get("workspace_id"):
|
||||
await self._send_notification_card(receive_id=chat_id, receive_id_type="chat_id", text_markdown=text_markdown)
|
||||
return
|
||||
if not matching_binding:
|
||||
raise RuntimeError("Feishu notification identity is no longer connected")
|
||||
# Legacy and group bindings do not prove that their chat is private.
|
||||
await self.send_notification(target=target, text_markdown=text_markdown)
|
||||
|
||||
async def _send_notification_card(self, *, receive_id: str, receive_id_type: Literal["open_id", "chat_id"], text_markdown: str) -> None:
|
||||
"""Send once and wait for the SDK ACK.
|
||||
|
||||
The notification outbox owns retries. Transport failures park its row;
|
||||
a platform rejection remains a counted delivery attempt.
|
||||
"""
|
||||
if not self.is_running or not self._api_client:
|
||||
raise ChannelUnavailable("Feishu channel is not running")
|
||||
|
||||
from requests.exceptions import ConnectionError as RequestsConnectionError
|
||||
from requests.exceptions import Timeout as RequestsTimeout
|
||||
|
||||
request = (
|
||||
self._CreateMessageRequest.builder()
|
||||
.receive_id_type(receive_id_type)
|
||||
.request_body(self._CreateMessageRequestBody.builder().receive_id(receive_id).msg_type("interactive").content(self._build_card_content(text_markdown)).build())
|
||||
.build()
|
||||
)
|
||||
try:
|
||||
response = await asyncio.to_thread(self._api_client.im.v1.message.create, request)
|
||||
except (RequestsConnectionError, RequestsTimeout, ConnectionError, TimeoutError, OSError) as exc:
|
||||
raise ChannelUnavailable("Feishu notification transport is unavailable") from exc
|
||||
if not response.success():
|
||||
raise RuntimeError(f"Feishu notification rejected: code={response.code}")
|
||||
|
||||
async def send_file(self, msg: OutboundMessage, attachment: ResolvedAttachment) -> bool:
|
||||
if not self._api_client:
|
||||
return False
|
||||
@ -1626,7 +1679,7 @@ class FeishuChannel(Channel):
|
||||
workspace_id=inbound.chat_id,
|
||||
)
|
||||
|
||||
async def _bind_connection_from_connect_code(self, *, message_id: str, chat_id: str, user_id: str, code: str) -> bool:
|
||||
async def _bind_connection_from_connect_code(self, *, message_id: str, chat_id: str, user_id: str, code: str, chat_type: str | None = None) -> bool:
|
||||
if self._connection_repo is None or not code:
|
||||
return False
|
||||
|
||||
@ -1647,6 +1700,7 @@ class FeishuChannel(Channel):
|
||||
metadata={
|
||||
"chat_id": chat_id,
|
||||
"message_id": message_id,
|
||||
"chat_type": chat_type,
|
||||
},
|
||||
status="connected",
|
||||
)
|
||||
@ -1767,6 +1821,7 @@ class FeishuChannel(Channel):
|
||||
chat_id=chat_id,
|
||||
user_id=sender_id,
|
||||
code=connect_code,
|
||||
chat_type=parsed.chat_type,
|
||||
),
|
||||
self._main_loop,
|
||||
name="bind_connection",
|
||||
|
||||
@ -271,7 +271,8 @@ class NotificationDeliveryWorker:
|
||||
enriched = dict(delivery)
|
||||
enriched["payload"] = {**(delivery.get("payload") or {}), "result_summary": summary}
|
||||
try:
|
||||
await channel.send_notification(
|
||||
await channel.send_notification_for_owner(
|
||||
owner_user_id=delivery.get("owner_user_id") or "",
|
||||
target=delivery.get("target") or "",
|
||||
text_markdown=render_notification_text(enriched, default_locale=self._default_locale),
|
||||
)
|
||||
|
||||
@ -1713,7 +1713,7 @@ stripped from client-supplied messages and state updates.
|
||||
**IM notices ("scheduled task updates").** With `channel_connections.enabled`,
|
||||
each item of `GET /api/channels/providers` carries
|
||||
`proactive_notifications` (bool): whether scheduled-task updates are pushed to
|
||||
that app. Only providers with proactive push (WeCom today) get notices; the
|
||||
that app. Only providers with proactive push (WeCom and Feishu/Lark today) get notices; the
|
||||
others get none, and Settings says so. An occurrence sends at most one message,
|
||||
in the owner's `locale` preference (see Account Preferences), else
|
||||
`channel_connections.notification_locale`. Notices carry no IDs and no links;
|
||||
|
||||
@ -566,7 +566,7 @@ Notes:
|
||||
- Manual trigger uses the same scheduled-task resource and run lifecycle.
|
||||
- Scheduled task definitions and task-run history are persisted in the application database.
|
||||
- With `channel_connections.enabled: true`, the scheduler sends IM notices ("scheduled task updates") to the task owner's connected identities. A delivery worker (same poll cadence as the scheduler) pushes them.
|
||||
- **Which apps:** only providers whose channel implements proactive push get notices; today that is WeCom (`proactive_notifications` in `app/channels/capabilities.py`, also returned by `GET /api/channels/providers`, where Settings shows "sent here" or "not available for this app yet"). Other providers get no outbox rows at all, and rows queued for them before the upgrade end once as `failed` without retries.
|
||||
- **Which apps:** only providers whose channel implements proactive push get notices; today those are WeCom and Feishu/Lark (`proactive_notifications` in `app/channels/capabilities.py`, also returned by `GET /api/channels/providers`, where Settings shows "sent here" or "not available for this app yet"). Other providers get no outbox rows at all, and rows queued for them before the upgrade end once as `failed` without retries.
|
||||
- **Events:** `run_completed`, `run_failed`, `run_unmet` (goal missed), `task_paused` (automatic pause after three missed goals), `task_stopped` (paused by the agent because its stop condition was met) and `task_finished` (all `max_runs` done or `end_at` reached).
|
||||
- **One message per occurrence:** when one occurrence produces several events, only the first of `task_stopped` > `task_paused` > `task_finished` > `run_failed` > `run_unmet` > `run_completed` is sent, and that message still says how the last run went ("The last run failed." / "didn't meet the goal"). A one-time task sends only its run's outcome, never a separate `task_finished`.
|
||||
- **Transactional:** notices are queued by the finalization observer in the same database transaction that records the outcome and are deduplicated per occurrence (or per end time for a task that ends between runs), so an occurrence finalized by crash or lease recovery notifies exactly once. Plain manual "run now" trials and interrupted runs stay silent; a trial that pauses or finishes the task still sends that notice. Occurrences that end without a finished run (launch error, queue timeout, interrupted by a restart) stay silent.
|
||||
|
||||
@ -448,6 +448,32 @@ Neither can make anything be *acted on*. The `allowed_users` allowlist and per-e
|
||||
|
||||
**Bound identity:** once a pubkey completes `/connect`, its inbound messages resolve to that connection and run under the bound DeerFlow user (memory, files, and artifacts land in that user's buckets). Bindings are scoped to the relay host, so the same pubkey on a different relay is a different identity and must bind separately.
|
||||
|
||||
### Feishu scheduled-task notifications
|
||||
|
||||
Feishu/Lark sends scheduled-task result and lifecycle notices as static Markdown
|
||||
cards to the task owner's currently bound user. The existing notification outbox
|
||||
provides retry, localization, redaction and send-time binding checks; Settings
|
||||
reports this through `proactive_notifications`. Manual trial runs stay silent.
|
||||
|
||||
The configured app needs bot messaging permission (`im:message:send_as_bot`, or
|
||||
its broader `im:message` alternative), and the recipient must be in its available
|
||||
scope. Complete `/connect <code>` in a private chat with the same app. The outbox
|
||||
keeps the application-scoped `open_id` as its identity; delivery prefers that
|
||||
owner's currently connected, explicitly recorded private-chat address. Group
|
||||
chats are never used for these personal notices. Older bindings without recorded
|
||||
chat type keep the `open_id` path; if Feishu rejects it with `230101`, bind again
|
||||
in private chat. Changing the app also requires rebinding.
|
||||
The existing WebSocket channel must be running; sends use the configured SDK's
|
||||
REST client, so no public callback or interactive-card subscription is needed.
|
||||
|
||||
For provider-backed acceptance, bind a test user, let a once task run on its
|
||||
schedule, and confirm the notice arrives without another chat message. Check
|
||||
that a platform rejection is recorded as a failed delivery and that unbinding
|
||||
prevents later sends. API acceptance is not a read receipt. The shared outbox
|
||||
can retry an ambiguously acknowledged send; it does not promise exactly-once
|
||||
messages at the provider. Record whether domestic Feishu or international Lark
|
||||
was actually exercised.
|
||||
|
||||
### QQ WebSocket setup (MVP)
|
||||
|
||||
QQ uses an outbound WebSocket connection for events and the QQ Open Platform
|
||||
|
||||
@ -35,10 +35,10 @@ def test_every_entry_declares_both_flags_as_bools():
|
||||
assert all(isinstance(value, bool) for value in entry.values()), provider
|
||||
|
||||
|
||||
def test_only_wecom_pushes_proactively_today():
|
||||
assert {provider for provider, entry in CHANNEL_CAPABILITIES.items() if entry["proactive_notifications"]} == {"wecom"}
|
||||
def test_wecom_and_feishu_push_proactively():
|
||||
assert {provider for provider, entry in CHANNEL_CAPABILITIES.items() if entry["proactive_notifications"]} == {"wecom", "feishu"}
|
||||
assert supports_proactive_notifications("wecom") is True
|
||||
assert supports_proactive_notifications("feishu") is False
|
||||
assert supports_proactive_notifications("feishu") is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("provider", ["", "unknown", "WeCom", None, 1])
|
||||
|
||||
@ -205,7 +205,7 @@ def test_get_providers_uses_existing_channels_config(tmp_path):
|
||||
"telegram": False,
|
||||
"slack": False,
|
||||
"discord": False,
|
||||
"feishu": False,
|
||||
"feishu": True,
|
||||
"dingtalk": False,
|
||||
"wechat": False,
|
||||
"wecom": True,
|
||||
|
||||
276
backend/tests/test_feishu_notifications.py
Normal file
276
backend/tests/test_feishu_notifications.py
Normal file
@ -0,0 +1,276 @@
|
||||
"""Proactive Feishu delivery uses a bound identity and waits for the SDK ACK."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import threading
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
from requests.exceptions import ConnectionError as RequestsConnectionError
|
||||
from requests.exceptions import Timeout as RequestsTimeout
|
||||
|
||||
from app.channels.base import Channel, ChannelUnavailable
|
||||
from app.channels.feishu import FeishuChannel
|
||||
from app.channels.message_bus import MessageBus
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def channel():
|
||||
from lark_oapi.api.im.v1 import CreateMessageRequest, CreateMessageRequestBody
|
||||
|
||||
instance = FeishuChannel(MessageBus(), {})
|
||||
instance._running = True
|
||||
instance._thread = MagicMock()
|
||||
instance._thread.is_alive.return_value = True
|
||||
instance._CreateMessageRequest = CreateMessageRequest
|
||||
instance._CreateMessageRequestBody = CreateMessageRequestBody
|
||||
instance._api_client = MagicMock()
|
||||
instance._api_client.im.v1.message.create.return_value = SimpleNamespace(success=lambda: True)
|
||||
return instance
|
||||
|
||||
|
||||
def _private_binding(**overrides):
|
||||
return {
|
||||
"owner_user_id": "owner-1",
|
||||
"provider": "feishu",
|
||||
"status": "connected",
|
||||
"external_account_id": "ou_bound_user",
|
||||
"workspace_id": "oc_private",
|
||||
"metadata": {"chat_id": "oc_private", "chat_type": "p2p"},
|
||||
**overrides,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_base_owner_notification_preserves_existing_provider_send(channel):
|
||||
channel.send_notification = AsyncMock()
|
||||
|
||||
await Channel.send_notification_for_owner(channel, owner_user_id="owner-1", target="ou_bound_user", text_markdown="Done")
|
||||
|
||||
channel.send_notification.assert_awaited_once_with(target="ou_bound_user", text_markdown="Done")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_owner_notification_uses_only_its_bound_private_chat(channel):
|
||||
channel._connection_repo = SimpleNamespace(list_connections=AsyncMock(return_value=[_private_binding(owner_user_id="another-owner"), _private_binding(metadata={"chat_id": "oc_private", "chat_type": "group"}), _private_binding()]))
|
||||
|
||||
await channel.send_notification_for_owner(owner_user_id="owner-1", target="ou_bound_user", text_markdown="Done")
|
||||
|
||||
channel._connection_repo.list_connections.assert_awaited_once_with("owner-1")
|
||||
request = channel._api_client.im.v1.message.create.call_args.args[0]
|
||||
assert request.receive_id_type == "chat_id"
|
||||
assert request.request_body.receive_id == "oc_private"
|
||||
assert json.loads(request.request_body.content)["elements"] == [{"tag": "markdown", "content": "Done"}]
|
||||
channel._api_client.im.v1.message.create.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"overrides",
|
||||
[
|
||||
{"owner_user_id": "another-owner"},
|
||||
{"provider": "wecom"},
|
||||
{"status": "revoked"},
|
||||
{"external_account_id": "ou_someone_else"},
|
||||
{"workspace_id": "oc_different_chat"},
|
||||
{"metadata": {"chat_id": "oc_private", "chat_type": "group"}},
|
||||
{"metadata": {"chat_id": "oc_private"}},
|
||||
{"metadata": {"chat_id": "", "chat_type": "p2p"}},
|
||||
],
|
||||
)
|
||||
async def test_owner_notification_never_promotes_unverified_chat_id(channel, overrides):
|
||||
channel._connection_repo = SimpleNamespace(list_connections=AsyncMock(return_value=[_private_binding(**overrides), _private_binding(metadata={"chat_id": "oc_private", "chat_type": "group"})]))
|
||||
|
||||
await channel.send_notification_for_owner(owner_user_id="owner-1", target="ou_bound_user", text_markdown="Done")
|
||||
|
||||
request = channel._api_client.im.v1.message.create.call_args.args[0]
|
||||
assert request.receive_id_type == "open_id"
|
||||
assert request.request_body.receive_id == "ou_bound_user"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("overrides", [None, {"status": "revoked"}, {"owner_user_id": "another-owner"}, {"provider": "wecom"}, {"external_account_id": "ou_someone_else"}])
|
||||
async def test_owner_notification_does_not_send_when_binding_is_no_longer_active(channel, overrides):
|
||||
rows = [] if overrides is None else [_private_binding(**overrides)]
|
||||
channel._connection_repo = SimpleNamespace(list_connections=AsyncMock(return_value=rows))
|
||||
|
||||
with pytest.raises(RuntimeError, match="no longer connected"):
|
||||
await channel.send_notification_for_owner(owner_user_id="owner-1", target="ou_bound_user", text_markdown="Done")
|
||||
|
||||
channel._api_client.im.v1.message.create.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_owner_notification_without_connection_repo_keeps_open_id(channel):
|
||||
await channel.send_notification_for_owner(owner_user_id="owner-1", target="ou_bound_user", text_markdown="Done")
|
||||
|
||||
request = channel._api_client.im.v1.message.create.call_args.args[0]
|
||||
assert request.receive_id_type == "open_id"
|
||||
assert request.request_body.receive_id == "ou_bound_user"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_owner_notification_parks_on_binding_lookup_error(channel):
|
||||
channel._connection_repo = SimpleNamespace(list_connections=AsyncMock(side_effect=OSError("private database detail")))
|
||||
|
||||
with pytest.raises(ChannelUnavailable) as caught:
|
||||
await channel.send_notification_for_owner(owner_user_id="owner-1", target="ou_bound_user", text_markdown="Done")
|
||||
|
||||
assert "private database detail" not in str(caught.value)
|
||||
channel._api_client.im.v1.message.create.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("chat_type", ["p2p", "group", None])
|
||||
async def test_binding_records_chat_type_from_inbound_event(channel, chat_type):
|
||||
channel._connection_repo = SimpleNamespace(consume_oauth_state=AsyncMock(return_value={"owner_user_id": "owner-1"}), upsert_connection=AsyncMock())
|
||||
channel._reply_card = AsyncMock()
|
||||
|
||||
assert await channel._bind_connection_from_connect_code(message_id="om_bind", chat_id="oc_bound", user_id="ou_bound_user", code="connect-code", chat_type=chat_type)
|
||||
|
||||
binding = channel._connection_repo.upsert_connection.call_args.kwargs
|
||||
assert binding["owner_user_id"] == "owner-1"
|
||||
assert binding["external_account_id"] == "ou_bound_user"
|
||||
assert binding["workspace_id"] == "oc_bound"
|
||||
assert binding["metadata"] == {"chat_id": "oc_bound", "message_id": "om_bind", "chat_type": chat_type}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bind_callback_passes_parsed_chat_type(channel):
|
||||
channel._connection_repo = MagicMock()
|
||||
channel._parse_message_event = MagicMock(return_value=SimpleNamespace(msg_id="om_bind", chat_id="oc_bound", sender_id="ou_bound_user", text="/connect connect-code", chat_type="p2p"))
|
||||
channel._main_loop = asyncio.get_running_loop()
|
||||
channel._bind_connection_from_connect_code = AsyncMock()
|
||||
pending = []
|
||||
|
||||
def capture(coroutine, *_args, **_kwargs):
|
||||
pending.append(coroutine)
|
||||
return True
|
||||
|
||||
channel._submit_threadsafe_coroutine = capture
|
||||
channel._on_message(object())
|
||||
assert len(pending) == 1
|
||||
await pending[0]
|
||||
|
||||
channel._bind_connection_from_connect_code.assert_awaited_once_with(message_id="om_bind", chat_id="oc_bound", user_id="ou_bound_user", code="connect-code", chat_type="p2p")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_notification_uses_bound_open_id_and_static_markdown_card(channel):
|
||||
text = "**任务已完成**\n结果已准备好。"
|
||||
|
||||
await channel.send_notification(target="ou_bound_user", text_markdown=text)
|
||||
|
||||
create = channel._api_client.im.v1.message.create
|
||||
create.assert_called_once()
|
||||
request = create.call_args.args[0]
|
||||
assert request.receive_id_type == "open_id"
|
||||
assert request.request_body.receive_id == "ou_bound_user"
|
||||
assert request.request_body.msg_type == "interactive"
|
||||
assert json.loads(request.request_body.content)["elements"] == [{"tag": "markdown", "content": text}]
|
||||
channel._api_client.im.v1.message.reply.assert_not_called()
|
||||
channel._api_client.im.v1.message.patch.assert_not_called()
|
||||
assert channel._running_card_ids == {}
|
||||
assert channel._pending_clarifications == {}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("unavailable", ["stopped", "missing_thread", "dead_thread", "missing_client"])
|
||||
async def test_notification_without_live_channel_parks_without_sending(channel, unavailable):
|
||||
create = channel._api_client.im.v1.message.create
|
||||
if unavailable == "stopped":
|
||||
channel._running = False
|
||||
elif unavailable == "missing_thread":
|
||||
channel._thread = None
|
||||
elif unavailable == "dead_thread":
|
||||
channel._thread.is_alive.return_value = False
|
||||
else:
|
||||
channel._api_client = None
|
||||
|
||||
with pytest.raises(ChannelUnavailable):
|
||||
await channel.send_notification(target="ou_bound_user", text_markdown="Done")
|
||||
|
||||
create.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("error_type", [RequestsConnectionError, RequestsTimeout, ConnectionError, TimeoutError, OSError])
|
||||
async def test_notification_transport_failures_park_without_local_retry(channel, error_type):
|
||||
create = channel._api_client.im.v1.message.create
|
||||
create.side_effect = error_type("private transport detail")
|
||||
|
||||
with pytest.raises(ChannelUnavailable) as caught:
|
||||
await channel.send_notification(target="ou_bound_user", text_markdown="Done")
|
||||
|
||||
create.assert_called_once()
|
||||
assert "private transport detail" not in str(caught.value)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_notification_platform_rejection_remains_a_counted_failure(channel):
|
||||
create = channel._api_client.im.v1.message.create
|
||||
create.return_value = SimpleNamespace(success=lambda: False, code=230013, msg="private provider detail")
|
||||
|
||||
with pytest.raises(RuntimeError, match="230013") as caught:
|
||||
await channel.send_notification(target="ou_bound_user", text_markdown="Done")
|
||||
|
||||
assert not isinstance(caught.value, ChannelUnavailable)
|
||||
assert "private provider detail" not in str(caught.value)
|
||||
create.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_notification_unexpected_sdk_error_is_not_disguised_as_transport(channel):
|
||||
error = ValueError("invalid request")
|
||||
channel._api_client.im.v1.message.create.side_effect = error
|
||||
|
||||
with pytest.raises(ValueError) as caught:
|
||||
await channel.send_notification(target="ou_bound_user", text_markdown="Done")
|
||||
|
||||
assert caught.value is error
|
||||
channel._api_client.im.v1.message.create.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("cancel", [False, True])
|
||||
async def test_notification_sdk_send_runs_offloop_and_waits_for_ack(channel, cancel):
|
||||
loop = asyncio.get_running_loop()
|
||||
loop_thread = threading.get_ident()
|
||||
started = asyncio.Event()
|
||||
finished = asyncio.Event()
|
||||
release = threading.Event()
|
||||
worker_threads = []
|
||||
|
||||
def send(request):
|
||||
worker_threads.append(threading.get_ident())
|
||||
loop.call_soon_threadsafe(started.set)
|
||||
try:
|
||||
if not release.wait(5):
|
||||
raise TimeoutError("test did not release send")
|
||||
return SimpleNamespace(success=lambda: True)
|
||||
finally:
|
||||
loop.call_soon_threadsafe(finished.set)
|
||||
|
||||
channel._api_client.im.v1.message.create.side_effect = send
|
||||
task = asyncio.create_task(channel.send_notification(target="ou_bound_user", text_markdown="Done"))
|
||||
try:
|
||||
await asyncio.wait_for(started.wait(), timeout=2)
|
||||
assert worker_threads != [loop_thread]
|
||||
assert not task.done()
|
||||
if cancel:
|
||||
task.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await task
|
||||
else:
|
||||
release.set()
|
||||
await asyncio.wait_for(task, timeout=2)
|
||||
finally:
|
||||
release.set()
|
||||
if worker_threads:
|
||||
await asyncio.wait_for(finished.wait(), timeout=2)
|
||||
if not task.done():
|
||||
task.cancel()
|
||||
await asyncio.gather(task, return_exceptions=True)
|
||||
@ -55,6 +55,11 @@ class FakeChannel:
|
||||
self.fail = fail
|
||||
self.unavailable = unavailable
|
||||
self.is_running = running
|
||||
self.owners = []
|
||||
|
||||
async def send_notification_for_owner(self, *, owner_user_id, target, text_markdown):
|
||||
self.owners.append(owner_user_id)
|
||||
await self.send_notification(target=target, text_markdown=text_markdown)
|
||||
|
||||
async def send_notification(self, *, target, text_markdown):
|
||||
if self.unavailable:
|
||||
@ -611,7 +616,7 @@ async def test_delivery_sends_skeleton_when_summary_is_missing():
|
||||
async def test_legacy_row_for_unsupported_provider_fails_terminally_without_send():
|
||||
# Rows queued before only push-capable providers got rows: they can never
|
||||
# be sent, so they end at once instead of using up five retries.
|
||||
repo = FakeDeliveryRepo([_delivery_row(provider="feishu", target="ou_123")])
|
||||
repo = FakeDeliveryRepo([_delivery_row(provider="slack", target="U123")])
|
||||
channel = FakeChannel()
|
||||
resolved_channels, checked_connections = [], []
|
||||
|
||||
@ -621,7 +626,7 @@ async def test_legacy_row_for_unsupported_provider_fails_terminally_without_send
|
||||
|
||||
def resolve_connections(owner):
|
||||
checked_connections.append(owner)
|
||||
return [_connection(provider="feishu", external_account_id="ou_123")]
|
||||
return [_connection(provider="slack", external_account_id="U123")]
|
||||
|
||||
worker = NotificationDeliveryWorker(delivery_repo=repo, resolve_channel=resolve_channel, resolve_connections=resolve_connections)
|
||||
|
||||
@ -633,6 +638,32 @@ async def test_legacy_row_for_unsupported_provider_fails_terminally_without_send
|
||||
assert repo.terminal_failures == [("delivery-1", "provider does not support proactive notifications")]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("binding_owner", ["user-1", "another-user"])
|
||||
async def test_feishu_notice_requires_its_owners_current_binding(binding_owner):
|
||||
repo = FakeDeliveryRepo([_delivery_row(provider="feishu", target="ou_bound_user")])
|
||||
channel = FakeChannel()
|
||||
worker = NotificationDeliveryWorker(
|
||||
delivery_repo=repo,
|
||||
resolve_channel=lambda provider: channel if provider == "feishu" else None,
|
||||
resolve_connections=lambda owner: [_connection(provider="feishu", external_account_id="ou_bound_user", owner_user_id=binding_owner)],
|
||||
)
|
||||
|
||||
await worker.run_once(now=datetime.now(UTC))
|
||||
|
||||
if binding_owner == "user-1":
|
||||
assert repo.sent == ["delivery-1"]
|
||||
assert channel.sent[0][0] == "ou_bound_user"
|
||||
assert "Scheduled task" in channel.sent[0][1]
|
||||
assert repo.failed == []
|
||||
assert channel.owners == ["user-1"]
|
||||
else:
|
||||
assert repo.sent == []
|
||||
assert channel.sent == []
|
||||
assert channel.owners == []
|
||||
assert repo.terminal_failures == [("delivery-1", "target is no longer a connected feishu identity of its owner")]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_task_stopped_is_enriched_with_summary():
|
||||
row = _delivery_row(event="task_stopped", payload={"payload_version": 2, "task_title": "Release check", "reason_code": "agent_stop", "run_status": "success", "stop_condition": "all items are ticked"})
|
||||
|
||||
@ -1330,6 +1330,7 @@ async def _notices(env):
|
||||
|
||||
async def _deliver_all(env, *, default_locale="en-US"):
|
||||
"""Send every queued notice through the real worker with a fake WeCom channel; return the texts."""
|
||||
from app.channels.base import Channel as BaseChannel
|
||||
from app.scheduler.notification_delivery import NotificationDeliveryWorker
|
||||
from deerflow.persistence.run import RunRepository
|
||||
|
||||
@ -1337,6 +1338,7 @@ async def _deliver_all(env, *, default_locale="en-US"):
|
||||
|
||||
class Channel:
|
||||
is_running = True
|
||||
send_notification_for_owner = BaseChannel.send_notification_for_owner
|
||||
|
||||
async def send_notification(self, *, target, text_markdown):
|
||||
sent.append(text_markdown)
|
||||
@ -1354,14 +1356,21 @@ async def _deliver_all(env, *, default_locale="en-US"):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unsupported_provider_gets_no_outbox_row(outbox):
|
||||
async def test_only_push_capable_connected_providers_get_outbox_rows(outbox):
|
||||
await _task(outbox)
|
||||
await _bind(outbox, "feishu", "ou_feishu")
|
||||
await _bind(outbox, "wecom", "wecom-user")
|
||||
await _bind(outbox, "slack", "U123")
|
||||
await _bind(outbox, "feishu", "ou_revoked", status="revoked")
|
||||
await _bind(outbox, "wecom", "revoked-user", status="revoked")
|
||||
occurrence_id, run_id = await _occurrence(outbox)
|
||||
assert await _complete(outbox, occurrence_id, run_id) is True
|
||||
assert [(row["event"], row["provider"], row["target"]) for row in await _notices(outbox)] == [("run_completed", "wecom", "wecom-user")]
|
||||
assert sorted((row["event"], row["provider"], row["target"]) for row in await _notices(outbox)) == [
|
||||
("run_completed", "feishu", "ou_feishu"),
|
||||
("run_completed", "wecom", "wecom-user"),
|
||||
]
|
||||
assert await _complete(outbox, occurrence_id, run_id) is False
|
||||
assert len(await _notices(outbox)) == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
||||
@ -10,7 +10,7 @@ import { cn } from "@/lib/utils";
|
||||
* Whether scheduled-task updates reach this app ("Scheduled task updates:
|
||||
* sent here" / "... available after you connect" / "... not available for
|
||||
* this app yet"). Only providers that
|
||||
* implement proactive push receive them (WeCom today). Renders nothing for
|
||||
* implement proactive push receive them (WeCom and Feishu/Lark today). Renders nothing for
|
||||
* an older backend that does not report `proactive_notifications`, rather
|
||||
* than claiming an app it cannot judge is unsupported.
|
||||
*/
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user