From 5ae95e6a344c7c28d760608e3a24312bb61ca03d Mon Sep 17 00:00:00 2001 From: Totoro Date: Tue, 13 Oct 2026 07:17:56 +0800 Subject: [PATCH] feat: add Feishu group notifications and fix mention caret placement (#6762) * feat(feishu): deliver scheduled-task notifications to bound users * feat: add Feishu group notifications and fix mention caret placement * fix(frontend): preserve group targets while capabilities load --------- Co-authored-by: Totoro-qaq <279883115+Totoro-qaq@users.noreply.github.com> --- README.md | 21 +- backend/app/channels/AGENTS.md | 2 +- backend/app/channels/base.py | 8 + backend/app/channels/feishu.py | 33 ++ backend/app/channels/notification_groups.py | 6 + backend/app/gateway/routers/features.py | 5 +- .../app/gateway/routers/scheduled_tasks.py | 34 ++ backend/app/gateway/scheduled_task_errors.py | 1 + .../app/scheduler/notification_delivery.py | 19 +- backend/app/scheduler/service.py | 16 +- backend/docs/API.md | 10 +- backend/docs/CONFIGURATION.md | 1 + backend/docs/IM_CHANNEL_CONNECTIONS.md | 33 ++ .../versions/0040_scheduled_group_notice.py | 28 ++ .../persistence/scheduled_tasks/model.py | 2 + .../persistence/scheduled_tasks/sql.py | 2 + backend/tests/test_features_router.py | 21 +- backend/tests/test_feishu_notifications.py | 80 ++- ...migration_0027_0028_notification_outbox.py | 2 +- ...st_migration_0029_scheduler_agent_tasks.py | 2 +- ...migration_0032_activity_and_task_events.py | 2 +- ..._migration_0034_run_event_seq_watermark.py | 2 +- .../test_migration_0035_login_throttle.py | 2 +- ...gration_0037_project_document_summaries.py | 6 +- ..._migration_0038_channel_thread_bindings.py | 2 +- ...t_migration_0040_scheduled_group_notice.py | 85 ++++ .../test_notification_delivery_worker.py | 76 +++ backend/tests/test_notification_groups.py | 28 ++ .../test_scheduled_task_group_targets.py | 185 +++++++ backend/tests/test_scheduled_task_service.py | 72 +++ contracts/scheduled_task_errors_contract.json | 1 + docs/full-stack-plugins.md | 27 ++ frontend/AGENTS.md | 43 +- .../src/components/workspace/input-box.tsx | 34 +- .../scheduled-tasks/bind-feishu-group.tsx | 146 ++++++ .../scheduled-tasks/notification-group.tsx | 161 +++++++ .../workspace/scheduled-tasks/task-detail.tsx | 9 + .../scheduled-tasks/task-form-dialog.tsx | 19 + frontend/src/core/features/api.ts | 5 + frontend/src/core/features/hooks.ts | 1 + frontend/src/core/i18n/locales/en-US.ts | 35 ++ frontend/src/core/i18n/locales/types.ts | 28 ++ frontend/src/core/i18n/locales/zh-CN.ts | 31 ++ frontend/src/core/scheduled-tasks/api.ts | 4 +- frontend/src/core/scheduled-tasks/errors.ts | 1 + .../scheduled-tasks/notification-groups.ts | 24 + frontend/src/core/scheduled-tasks/types.ts | 2 + .../tests/e2e/mention-caret-handoff.spec.ts | 30 ++ .../e2e/scheduled-notification-groups.spec.ts | 285 +++++++++++ frontend/tests/e2e/utils/mock-api.ts | 8 +- .../workspace/input-box-mentions.dom.test.tsx | 97 ++++ .../scheduled-tasks/task-detail.dom.test.tsx | 46 +- .../task-form-dialog.dom.test.tsx | 454 +++++++++++++++++- frontend/tests/unit/core/features/api.test.ts | 10 +- .../notification-groups.test.ts | 47 ++ 55 files changed, 2263 insertions(+), 71 deletions(-) create mode 100644 backend/app/channels/notification_groups.py create mode 100644 backend/packages/harness/deerflow/persistence/migrations/versions/0040_scheduled_group_notice.py create mode 100644 backend/tests/test_migration_0040_scheduled_group_notice.py create mode 100644 backend/tests/test_notification_groups.py create mode 100644 backend/tests/test_scheduled_task_group_targets.py create mode 100644 frontend/src/components/workspace/scheduled-tasks/bind-feishu-group.tsx create mode 100644 frontend/src/components/workspace/scheduled-tasks/notification-group.tsx create mode 100644 frontend/src/core/scheduled-tasks/notification-groups.ts create mode 100644 frontend/tests/e2e/mention-caret-handoff.spec.ts create mode 100644 frontend/tests/e2e/scheduled-notification-groups.spec.ts create mode 100644 frontend/tests/unit/core/scheduled-tasks/notification-groups.test.ts diff --git a/README.md b/README.md index 8ac9aee710..c3df609cad 100644 --- a/README.md +++ b/README.md @@ -1903,6 +1903,8 @@ Unsent Web UI composer drafts survive page reloads and switching between convers The Web UI composer supports `@` references at the cursor or through the `@` button. Search enabled skills, documents in the current project, and other conversations in a grouped picker; selected references appear inline alongside the draft text and can be removed with normal editing. Multiple skills can be selected in one message (up to 16 unique skills); selecting a checked skill again removes it, with each selection checked against the current user registry and agent allowlist. Project files are attached only after ingestion succeeds; conversation references require the `read_conversation` capability and retain its per-run limit. If capability discovery fails, saved conversation references remain intact and sending waits for a successful retry. Polishing preserves complete reference labels when instructions are reordered. The file group also offers local uploads. `/goal`, `/compact`, and legacy `/skill-name` input remain supported. +After picking an inline reference, typing continues immediately after it. Moving the cursor keeps the position you chose, including when editing text beside a reference. + The Web UI composer also supports browser-based voice dictation when the browser exposes the Web Speech API. The microphone button transcribes speech into the local draft only; DeerFlow receives only the transcribed text, while audio handling is delegated to the browser or operating system speech-recognition service according to that environment's policy. Users can review or edit the text before sending. Dictation keeps words separated when browsers trim transcript segments, preserves Chinese and Japanese joins, and attaches leading punctuation and percent signs without adding spaces. The Web UI displays a localized AI-generated-content disclaimer below the composer in both standard and custom-agent conversations, reminding users to verify important @@ -2808,7 +2810,7 @@ Current MVP capabilities: - Pause, resume, trigger, inspect history, and delete tasks - Search task titles or prompts, combined with status/type filters and the current thread scope. - Execute scheduled work through the normal DeerFlow run lifecycle -- When `channel_connections.enabled: true`, send scheduled task updates to the task owner's connected IM identities (outbox + delivery worker) on apps that support proactive push, which today is WeCom; Settings shows for each app whether updates are sent there, and other apps get none. Each occurrence sends at most one message: a finished, failed or goal-missed run, the automatic pause after three missed goals, a pause by the agent (its stop condition was met) or the task finishing (all `max_runs` done, `end_at` reached); when several apply, the pause or finish wins and still says how the last run went. A one-time task sends only its run's outcome. The message is queued in the same database transaction that records the outcome, so runs finalized after a crash or a lost lease notify exactly once. It reads on its own, in your web UI language (else `channel_connections.notification_locale`): the task title, what happened, a one-line result when the agent replied, and "Open DeerFlow → Scheduled tasks for details.", with no IDs and no links. Plain manual "run now" trials and interrupted runs stay silent, and so do occurrences that end without a finished run (launch error, queue timeout, interrupted by a restart). Channel/transport outages park deliveries without exhausting retries, for up to about a day; platform rejections retry for roughly 15 minutes before settling as `failed`. An identity you disconnect while a delivery is waiting is never pushed to: the row is dropped as `failed`. +- When `channel_connections.enabled: true`, send scheduled task updates to the task owner's connected IM identities (outbox + delivery worker) on apps that support proactive push, which today is WeCom and Feishu/Lark; Settings shows for each app whether updates are sent there, and other apps get none. Each occurrence queues at most one message per target: a finished, failed or goal-missed run, the automatic pause after three missed goals, a pause by the agent (its stop condition was met) or the task finishing (all `max_runs` done, `end_at` reached); when several apply, the pause or finish wins and still says how the last run went. A one-time task sends only its run's outcome. The message is queued in the same database transaction that records the outcome, so runs finalized after a crash or a lost lease do not enqueue a duplicate notice for the same target. It reads on its own, in your web UI language (else `channel_connections.notification_locale`): the task title, what happened, a one-line result when the agent replied, and "Open DeerFlow → Scheduled tasks for details.", with no IDs and no links. Plain manual "run now" trials and interrupted runs stay silent, and so do occurrences that end without a finished run (launch error, queue timeout, interrupted by a restart). Channel/transport outages park deliveries without exhausting retries, for up to about a day; platform rejections retry for roughly 15 minutes before settling as `failed`. An identity you disconnect while a delivery is waiting is never pushed to: the row is dropped as `failed`. - Browse execution history in pages of 50; older pages pause automatic refresh, with an explicit return to the latest runs. Counts appear only after a successful read; loading and failed reads are not reported as zero runs. Derived run summaries omit leading inline `` reasoning before limiting the text, so a long thought does not replace the actual answer in previews or scheduled notices. Original response history and literal tags in the answer are unchanged; previously stored run summaries are not rewritten. @@ -2828,6 +2830,23 @@ Enable background polling with `config.yaml -> scheduler.enabled`. Manual trigge In the task form, switching an hourly, daily, weekly, or monthly schedule to **Custom cron** starts from its current timing, including selected weekdays or day of the month. Switching back to Custom cron after preset edits uses those edits rather than an older expression. +### Optional Feishu group notifications + +In a task's create or edit form, **Also notify a Feishu group (optional)** adds +one chosen group to the existing personal notifications. Add the app bot to the +group, use **Bind a Feishu group** to get a command, send it while mentioning the +bot in that group, then refresh the list and select the group by name. The app +needs bot-send permission and `im:chat:read` to display its group names. + +While the form checks group-notification support, it shows a loading hint and +retains the saved selection. + +No group is selected automatically. Clearing the selection stops new group +notices; changing it affects future notices, while already queued notices keep +their original target. A disconnected binding never receives a waiting notice. +The first version is configured through the task form or REST API; conversation +schedule tools keep their existing inputs. See the [group delivery guide](backend/docs/IM_CHANNEL_CONNECTIONS.md#feishu-group-notifications). + ### Lifecycle, safety caps and stop conditions - The tasks page and the REST API (`POST` / `PATCH /api/scheduled-tasks`) accept the same per-run goal (`goal_objective`), safety cap (`max_runs`, `end_at`) and stop condition (`stop_condition`) as a conversation. Sending `null` in a PATCH clears any of these four; an `end_at` without a UTC offset is wall-clock time in the task's timezone. diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 27ba7e5598..6c4dae7766 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -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 `` `HumanMessage`, and LangGraph fans those state writes out on the `messages-tuple` stream. Proved live on a Buzz relay, which published a `` 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: 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. +- `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). `capabilities.py` gates enqueue/UI; unsupported legacy rows fail terminally (`test_channel_capabilities.py`). Await ACK; `ChannelUnavailable` parks transport failures, rejections count. Group targets use owner-bound connection IDs and a separate group hook; see [group delivery](../../docs/IM_CHANNEL_CONNECTIONS.md#feishu-group-notifications). **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, `
`, block math, or a link), so structured command/error replies (none of these) stay plain text and their newlines and `` 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`) diff --git a/backend/app/channels/base.py b/backend/app/channels/base.py index 5ac232359a..6e4482bb2d 100644 --- a/backend/app/channels/base.py +++ b/backend/app/channels/base.py @@ -157,6 +157,14 @@ class Channel(ABC): """ await self.send_notification(target=target, text_markdown=text_markdown) + async def send_group_notification_for_owner(self, *, owner_user_id: str | None, target: str, text_markdown: str) -> None: + """Push to an explicitly selected group binding, never a personal fallback. + + ``target`` is the owner's connection ID, not a provider chat ID. + Providers must revalidate that binding before sending to its group. + """ + raise NotImplementedError(f"channel '{self.name}' does not support group notifications") + # -- helpers ----------------------------------------------------------- async def _send_with_retry( diff --git a/backend/app/channels/feishu.py b/backend/app/channels/feishu.py index 085c70d6c4..0b1e578e1f 100644 --- a/backend/app/channels/feishu.py +++ b/backend/app/channels/feishu.py @@ -36,6 +36,7 @@ from app.channels.message_bus import ( OutboundMessage, ResolvedAttachment, ) +from app.channels.notification_groups import is_feishu_notification_group from app.channels.sandbox_files import sync_file_to_thread_sandbox from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths from deerflow.runtime.user_context import get_effective_user_id @@ -189,6 +190,7 @@ class FeishuChannel(Channel): self._CreateImageRequest = None self._CreateImageRequestBody = None self._GetMessageResourceRequest = None + self._GetChatRequest = None self._thread_lock = threading.Lock() # Per-chat FIFO of messages whose routing is deferred to the Gateway loop # (unresolved store lookups); guarded by ``_thread_lock``. A chat with a @@ -286,6 +288,7 @@ class FeishuChannel(Channel): CreateMessageRequest, CreateMessageRequestBody, Emoji, + GetChatRequest, GetMessageResourceRequest, PatchMessageRequest, PatchMessageRequestBody, @@ -318,6 +321,7 @@ class FeishuChannel(Channel): self._CreateImageRequest = CreateImageRequest self._CreateImageRequestBody = CreateImageRequestBody self._GetMessageResourceRequest = GetMessageResourceRequest + self._GetChatRequest = GetChatRequest app_id = self.config.get("app_id", "") app_secret = self.config.get("app_secret", "") @@ -485,6 +489,21 @@ class FeishuChannel(Channel): if not response.success(): raise RuntimeError(f"Feishu notification rejected: code={response.code}") + async def send_group_notification_for_owner(self, *, owner_user_id: str | None, target: str, text_markdown: str) -> None: + if self._connection_repo is None or not owner_user_id: + raise ChannelUnavailable("Feishu notification binding lookup is unavailable") + 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 + for connection in connections: + if connection.get("owner_user_id") != owner_user_id or connection.get("id") != target: + continue + if is_feishu_notification_group(provider=connection.get("provider"), status=connection.get("status"), workspace_id=connection.get("workspace_id"), metadata=connection.get("metadata")): + await self._send_notification_card(receive_id=connection["workspace_id"], receive_id_type="chat_id", text_markdown=text_markdown) + return + raise RuntimeError("Feishu notification group is no longer connected") + async def send_file(self, msg: OutboundMessage, attachment: ResolvedAttachment) -> bool: if not self._api_client: return False @@ -1679,6 +1698,19 @@ class FeishuChannel(Channel): workspace_id=inbound.chat_id, ) + async def _get_group_name(self, chat_id: str) -> str | None: + """Best-effort display name; missing read permission must not break binding.""" + if not self._api_client or not self._GetChatRequest: + return None + try: + request = self._GetChatRequest.builder().chat_id(chat_id).build() + response = await asyncio.to_thread(self._api_client.im.v1.chat.get, request) + if response.success(): + return self._non_empty_str(getattr(response.data, "name", None)) + except Exception: + logger.debug("[Feishu] group name lookup unavailable; binding without a display name") + return None + 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 @@ -1697,6 +1729,7 @@ class FeishuChannel(Channel): provider="feishu", external_account_id=user_id, workspace_id=chat_id, + workspace_name=await self._get_group_name(chat_id) if chat_type == "group" else None, metadata={ "chat_id": chat_id, "message_id": message_id, diff --git a/backend/app/channels/notification_groups.py b/backend/app/channels/notification_groups.py new file mode 100644 index 0000000000..ff54f9accf --- /dev/null +++ b/backend/app/channels/notification_groups.py @@ -0,0 +1,6 @@ +"""SDK-free checks for explicit Feishu group notification destinations.""" + + +def is_feishu_notification_group(*, provider: object, status: object, workspace_id: object, metadata: object) -> bool: + """Check the group shape; callers must separately verify owner and ID.""" + return provider == "feishu" and status == "connected" and isinstance(workspace_id, str) and bool(workspace_id.strip()) and isinstance(metadata, dict) and metadata.get("chat_type") == "group" and metadata.get("chat_id") == workspace_id diff --git a/backend/app/gateway/routers/features.py b/backend/app/gateway/routers/features.py index ede3fcc8a4..ac22c59310 100644 --- a/backend/app/gateway/routers/features.py +++ b/backend/app/gateway/routers/features.py @@ -73,6 +73,7 @@ class ScheduledTasksFeature(BaseModel): running: bool = Field(..., description="Whether this process's scheduler poller is running, so tasks run on schedule and new tasks can be created") tool_enabled: bool = Field(..., description="Whether chats can create and manage tasks and scheduled runs can stop their own schedule") min_interval_seconds: int = Field(..., description="Shortest interval and earliest one-time delay, in seconds") + group_notifications: bool = Field(..., description="Whether scheduled tasks support explicitly selected Feishu group notifications") class ThreadActivityFeature(BaseModel): @@ -143,11 +144,13 @@ def _scheduled_tasks_feature(request: Request, config: AppConfig) -> ScheduledTa service = getattr(state, "scheduled_task_service", None) running = bool(getattr(service, "is_running", False)) scheduler = getattr(config, "scheduler", None) + repository_available = getattr(state, "scheduled_task_repo", None) is not None return ScheduledTasksFeature( - available=getattr(state, "scheduled_task_repo", None) is not None, + available=repository_available, running=running, tool_enabled=running and scheduler_tools_enabled(config), min_interval_seconds=int(getattr(scheduler, "min_once_delay_seconds", 60)), + group_notifications=repository_available and bool(getattr(getattr(config, "channel_connections", None), "enabled", False)), ) diff --git a/backend/app/gateway/routers/scheduled_tasks.py b/backend/app/gateway/routers/scheduled_tasks.py index 8cb4de95d3..575f060803 100644 --- a/backend/app/gateway/routers/scheduled_tasks.py +++ b/backend/app/gateway/routers/scheduled_tasks.py @@ -9,6 +9,7 @@ from zoneinfo import ZoneInfo from fastapi import APIRouter, HTTPException, Query, Request from pydantic import AwareDatetime, BaseModel, Field +from app.channels.notification_groups import is_feishu_notification_group from app.gateway.authz import require_permission from app.gateway.deps import ( get_config, @@ -104,6 +105,32 @@ async def _task_response(repo, task: dict[str, Any]) -> dict[str, Any]: return {**task, "automatic_runs_used": used.get(task["id"], 0), "active_run_status": await repo.get_active_run_status(task["id"])} +async def _validate_notification_group(request: Request, connection_id: str | None, *, user_id: str) -> str | None: + """Resolve an explicitly selected group without changing personal delivery.""" + if connection_id is None: + return None + if not connection_id.strip() or len(connection_id) > 64: + raise scheduler_error(422, "invalid_notification_group", "Select a connected Feishu group belonging to this user") + + # Reuse the connection UI's current config and owner-scoped repository. + # Keep this REST-only: the conversation scheduling tools have no group field. + from app.gateway.routers import channel_connections + + try: + config = await channel_connections._get_channel_connections_config(request) + connections = await channel_connections._get_repository(request, config).list_connections(user_id) if config.enabled else [] + except Exception as exc: + raise scheduler_error(503, "scheduler_unavailable", "Notification group lookup is unavailable; retry later") from exc + if not isinstance(connections, list): + raise scheduler_error(503, "scheduler_unavailable", "Notification group lookup is unavailable; retry later") + for connection in connections: + if not isinstance(connection, dict) or connection.get("id") != connection_id or connection.get("owner_user_id") != user_id: + continue + if is_feishu_notification_group(provider=connection.get("provider"), status=connection.get("status"), workspace_id=connection.get("workspace_id"), metadata=connection.get("metadata")): + return connection_id + raise scheduler_error(422, "invalid_notification_group", "Select a connected Feishu group belonging to this user") + + class ScheduledTaskCreateRequest(BaseModel): # Types only: value rules (max_runs >= 1, lengths, end-time rules) raise # coded errors from the shared validation, never a FastAPI list detail. @@ -122,6 +149,7 @@ class ScheduledTaskCreateRequest(BaseModel): end_at: datetime | None = None # The user's "stop when ..." rule, stored in its own column. stop_condition: str | None = None + notification_group_connection_id: str | None = None class ScheduledTaskUpdateRequest(BaseModel): @@ -139,6 +167,7 @@ class ScheduledTaskUpdateRequest(BaseModel): max_runs: int | bool | None = None end_at: datetime | None = None stop_condition: str | None = None + notification_group_connection_id: str | None = None class ScheduledTaskResumeRequest(BaseModel): @@ -226,6 +255,7 @@ async def create_scheduled_task(request: Request, body: ScheduledTaskCreateReque assistant_resolver=resolve_scheduled_task_assistant_id, now=datetime.now(UTC), ) + definition["notification_group_connection_id"] = await _validate_notification_group(request, body.notification_group_connection_id, user_id=str(user.id)) # After validation, so a bad request still reports what to fix: a task # created while this process's poller is off would silently never run. if not is_scheduler_running(request): @@ -279,6 +309,8 @@ async def update_scheduled_task(task_id: str, request: Request, body: ScheduledT await _ensure_task_mutable(existing, repo) used = (await repo.automatic_runs_used_for([task_id])).get(task_id, 0) fields = body.model_dump(exclude_unset=True) + # Group destination edits are independent of schedule/goal lifecycle rules. + fields.pop("notification_group_connection_id", None) # Only a reused conversation needs the thread store (access check). reuses_thread = (fields.get("context_mode") or existing["context_mode"]) == "reuse_thread" updates = await validate_scheduled_task_update( @@ -291,6 +323,8 @@ async def update_scheduled_task(task_id: str, request: Request, body: ScheduledT used=used, now=datetime.now(UTC), ) + if "notification_group_connection_id" in body.model_fields_set: + updates["notification_group_connection_id"] = await _validate_notification_group(request, body.notification_group_connection_id, user_id=str(user.id)) try: updated = await repo.update( task_id, diff --git a/backend/app/gateway/scheduled_task_errors.py b/backend/app/gateway/scheduled_task_errors.py index 898b195089..5ed09c700b 100644 --- a/backend/app/gateway/scheduled_task_errors.py +++ b/backend/app/gateway/scheduled_task_errors.py @@ -34,6 +34,7 @@ SCHEDULER_UI_ERROR_CODES: frozenset[str] = frozenset( "invalid_goal", "goal_requires_fresh_thread", "invalid_stop_condition", + "invalid_notification_group", "invalid_max_runs", "end_at_in_past", "end_at_before_first_run", diff --git a/backend/app/scheduler/notification_delivery.py b/backend/app/scheduler/notification_delivery.py index 823b568228..f295763c4c 100644 --- a/backend/app/scheduler/notification_delivery.py +++ b/backend/app/scheduler/notification_delivery.py @@ -20,6 +20,7 @@ from typing import Any from app.channels.base import ChannelUnavailable from app.channels.capabilities import supports_proactive_notifications +from app.channels.notification_groups import is_feishu_notification_group from app.scheduler.notification_text import DEFAULT_NOTIFICATION_LOCALE, redact_egress_text, render_notification_text, wants_result_summary __all__ = ["NotificationDeliveryWorker", "redact_egress_text", "render_notification_text"] @@ -186,6 +187,7 @@ class NotificationDeliveryWorker: owner_user_id = delivery.get("owner_user_id") provider = delivery.get("provider") target = delivery.get("target") + is_group = (delivery.get("payload") or {}).get("target_type") == "group" if not owner_user_id or not provider or not target: return None try: @@ -212,6 +214,14 @@ class NotificationDeliveryWorker: row_owner = connection.get("owner_user_id") if row_owner is not None and row_owner != owner_user_id: continue + if is_group: + if ( + row_owner == owner_user_id + and connection.get("id") == target + and is_feishu_notification_group(provider=connection.get("provider"), status=connection.get("status"), workspace_id=connection.get("workspace_id"), metadata=connection.get("metadata")) + ): + return True + continue if connection.get("provider") == provider and connection.get("external_account_id") == target and connection.get("status") == "connected": return True return False @@ -219,6 +229,10 @@ class NotificationDeliveryWorker: async def _deliver(self, delivery: dict[str, Any]) -> None: delivery_id = delivery["id"] provider = delivery.get("provider") or "" + target_type = (delivery.get("payload") or {}).get("target_type") + if target_type not in (None, "group") or (target_type == "group" and provider != "feishu"): + await self._delivery_repo.mark_failed(delivery_id, claim_token=delivery.get("claim_token"), error="unsupported notification target type", terminal=True) + return if not supports_proactive_notifications(provider): # The scheduler enqueues only for providers with proactive push; # a row for any other provider predates that rule. It can never @@ -230,7 +244,7 @@ class NotificationDeliveryWorker: terminal=True, ) return - if self._resolve_connections is not None: + if self._resolve_connections is not None or target_type == "group": # Re-check the binding at delivery time, before anything else: a # target the owner has disconnected since enqueue must never # receive the push, however long the row waited, and it must not @@ -271,7 +285,8 @@ class NotificationDeliveryWorker: enriched = dict(delivery) enriched["payload"] = {**(delivery.get("payload") or {}), "result_summary": summary} try: - await channel.send_notification_for_owner( + send = channel.send_group_notification_for_owner if target_type == "group" else channel.send_notification_for_owner + await send( 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), diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py index a320a6dd57..1d3c1b3c2a 100644 --- a/backend/app/scheduler/service.py +++ b/backend/app/scheduler/service.py @@ -357,21 +357,27 @@ class ScheduledTaskService: for binding in await session.scalars(select(ChannelConnectionRow).where(ChannelConnectionRow.owner_user_id == task.user_id, ChannelConnectionRow.status == "connected")) if binding.provider and binding.external_account_id and supports_proactive_notifications(binding.provider) ] - if not bindings: + group_connection_id = getattr(task, "notification_group_connection_id", None) + if not bindings and not group_connection_id: return payload = _notice_payload(task, occurrence, event, reason, locale=await self._owner_locale(session, task.user_id)) task_run_id = occurrence.id if occurrence is not None else _idle_notice_key(task.id, lifecycle_anchor(task, occurrence, now=now)) - for binding in bindings: + recipients = [(binding.provider, binding.external_account_id, payload) for binding in bindings] + if group_connection_id: + # Keep the selected binding identity even if it was removed: the + # worker records an explicit failed delivery after its owner check. + recipients.append(("feishu", group_connection_id, {**payload, "target_type": "group"})) + for provider, target, notice_payload in recipients: await self._notification_repo.enqueue_in_session( session, task_id=task.id, task_run_id=task_run_id, run_id=occurrence.run_id if occurrence is not None else None, event=event, - provider=binding.provider, - target=binding.external_account_id, + provider=provider, + target=target, owner_user_id=task.user_id, - payload=payload, + payload=notice_payload, ) @staticmethod diff --git a/backend/docs/API.md b/backend/docs/API.md index ca106ec447..d33b9f4912 100644 --- a/backend/docs/API.md +++ b/backend/docs/API.md @@ -1578,15 +1578,21 @@ POST /api/scheduled-tasks/preview-cron | `max_runs` | integer \| null | Safety cap: automatic runs over the task's lifetime (trial runs excluded); at least 1. | | `end_at` | date-time \| null | Safety cap: no automatic run after this time. Without a UTC offset it is wall-clock time in the task's timezone. | | `stop_condition` | string \| null | The user's "stop when …" rule, stored in its own field. Whitespace collapses to single spaces; at most 500 characters; blank means none. It is never part of `prompt`: each launch appends it as an instruction to call `stop_scheduled_task` (or, with `scheduler.tool_enabled` off, to report a met rule). | +| `notification_group_connection_id` | string \| null | One additional connected Feishu group belonging to the task owner. Select a group binding returned by `GET /api/channels/connections`; it must record `chat_type=group` and a matching chat/workspace ID. Omitted in PATCH preserves the target; null clears it. Invalid choices return `invalid_notification_group` (422), lookup outages return `scheduler_unavailable` (503). | In a PATCH, sending `null` clears `goal_objective`, `max_runs`, `end_at` or -`stop_condition`; for every other field `null` means unchanged (`assistant_id: +`stop_condition` or `notification_group_connection_id`; for every other field `null` means unchanged (`assistant_id: null` restores `lead_agent`). PATCH may change `schedule_type` together with `schedule_spec`. Changing the goal, `prompt` or `stop_condition` starts a new count for the automatic pause after three unmet runs. A PATCH that changes the schedule of a finished task re-arms it (then its safety cap applies); a PATCH that only changes the cap leaves a finished task finished. +Group-target edits alone never change the schedule or re-arm a finished task. +Personal recipients are retained, and pending outbox rows keep their original +target. `/api/features.scheduled_tasks.group_notifications` reports support; +the conversation scheduling tool's inputs are unchanged in this version. + Task responses (list, get, create, PATCH, pause, resume) return the stored task plus `automatic_runs_used` (integer) and `active_run_status` (`queued` | `launching` | `running` | null). A recurring task's `status` stays `enabled` @@ -1677,7 +1683,7 @@ Three errors keep their old shape: a `403` from route permissions is the plain string `"Permission denied: "`, FastAPI's own `422` for malformed JSON or wrong types (for example `"max_runs": "abc"`) keeps its list `detail`, and the shared `503` `"Thread metadata store not available"` (a Gateway without -a thread store) stays a plain string. Codes the web UI translates: `invalid_request`, `invalid_schedule`, `invalid_schedule_type`, `invalid_timezone`, `interval_too_short`, `interval_too_long`, `once_in_past`, `once_too_soon`, `once_time_passed`, `invalid_context_mode`, `reuse_thread_requires_thread`, `thread_not_found`, `invalid_assistant`, `unknown_assistant`, `invalid_goal`, `goal_requires_fresh_thread`, `invalid_stop_condition`, `invalid_max_runs`, `end_at_in_past`, `end_at_before_first_run`, `frequent_requires_limit`, `max_runs_not_above_used`, `limits_exhausted`, `task_not_found`, `task_running`, `run_queued`, `task_changed`, `task_finished`, `task_quota_exceeded`, `scheduler_not_running`, `scheduler_unavailable` (503: this Gateway has no scheduled-task persistence), `trigger_failed`. Codes only the chat capability +a thread store) stays a plain string. Codes the web UI translates: `invalid_request`, `invalid_schedule`, `invalid_schedule_type`, `invalid_timezone`, `interval_too_short`, `interval_too_long`, `once_in_past`, `once_too_soon`, `once_time_passed`, `invalid_context_mode`, `reuse_thread_requires_thread`, `thread_not_found`, `invalid_assistant`, `unknown_assistant`, `invalid_goal`, `goal_requires_fresh_thread`, `invalid_stop_condition`, `invalid_notification_group`, `invalid_max_runs`, `end_at_in_past`, `end_at_before_first_run`, `frequent_requires_limit`, `max_runs_not_above_used`, `limits_exhausted`, `task_not_found`, `task_running`, `run_queued`, `task_changed`, `task_finished`, `task_quota_exceeded`, `scheduler_not_running`, `scheduler_unavailable` (503: this Gateway has no scheduled-task persistence), `trigger_failed`. Codes only the chat capability returns: `timezone_required`, `authentication_required`, `permission_denied`, `scheduler_tools_disabled`, `authority_expired`, `conversation_not_found`, `interactive_run_required`, `unsupported_action`, `unsupported_fields`, `task_id_required`, `note_not_verbatim`, `note_limit_reached`, `trial_requires_direct_request`, `no_stop_authority`, `occurrence_not_active`. The machine-readable list is `contracts/scheduled_task_errors_contract.json`. diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 3cb3359156..212e1e00db 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -567,6 +567,7 @@ Notes: - 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 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. + - **Optional Feishu group:** each task may select one additional owner-bound group through the task form or `notification_group_connection_id` in REST. No group is selected automatically; personal delivery is retained. The app needs `im:chat:read` for names, and a private binding never becomes a group destination. [Group setup and delivery](IM_CHANNEL_CONNECTIONS.md#feishu-group-notifications). - **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. diff --git a/backend/docs/IM_CHANNEL_CONNECTIONS.md b/backend/docs/IM_CHANNEL_CONNECTIONS.md index 74f5f45797..80125d6e0a 100644 --- a/backend/docs/IM_CHANNEL_CONNECTIONS.md +++ b/backend/docs/IM_CHANNEL_CONNECTIONS.md @@ -474,6 +474,39 @@ 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. +### Feishu group notifications + +Tasks may explicitly select one additional connected Feishu group. Personal +notifications keep their existing recipients. The task stores the chosen +`notification_group_connection_id`; a group notice stores that connection ID as +its outbox target with `payload.target_type="group"`. It never infers a group +destination from all of the owner's bindings. + +Add the app bot to a test or work group, grant `im:chat:read` for the group-name +lookup, and publish the app changes. In the task form choose **Bind a Feishu +group**, mention the bot and send the displayed `/connect` command in that +group, then refresh and select its name. Name lookup failure does not discard +the binding, but unnamed groups cannot be newly selected in the form: check +the read permission and bind again. API selection and each delivery verify the +task owner, connection ID, connected status, and recorded group/chat identity. +The provider sends to that verified `chat_id`; a lost binding cannot fall back +to another group or a private chat. A bot removed from the group produces a +normal platform rejection handled by the existing retry budget. + +Omitting the field in PATCH preserves it; `null` clears it. Destination-only +edits do not re-arm completed tasks or change the run budget. The destination +is fixed when each notification is enqueued; later task edits affect future +notices, and unbinding still blocks pending ones. Migration `0040` adds a nullable +reference with no backfill. Upgrade Gateway workers and the frontend together; +the form uses `/api/features.scheduled_tasks.group_notifications` to hide new +selection on older Gateways. The conversation scheduling tool does not set +group targets in this version. + +Real-provider acceptance should confirm that a selected task appears in the +specified group, an unselected task remains private, and removing the bot or +revoking its binding prevents a successful group send. Record the account and +tenant setup actually used; the outbox remains at-least-once. + ### QQ WebSocket setup (MVP) QQ uses an outbound WebSocket connection for events and the QQ Open Platform diff --git a/backend/packages/harness/deerflow/persistence/migrations/versions/0040_scheduled_group_notice.py b/backend/packages/harness/deerflow/persistence/migrations/versions/0040_scheduled_group_notice.py new file mode 100644 index 0000000000..a5df8ad9b0 --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/migrations/versions/0040_scheduled_group_notice.py @@ -0,0 +1,28 @@ +"""Add an optional Feishu group notification connection to scheduled tasks. + +Revision ID: 0040_scheduled_group_notice +Revises: 0039_user_disabled + +Legacy tasks keep NULL and their existing owner notifications. There is no +foreign key: disconnecting a group must retain the selected target so pending +delivery can report the loss instead of silently losing the destination. +""" + +from collections.abc import Sequence + +import sqlalchemy as sa + +from deerflow.persistence.migrations._helpers import safe_add_column, safe_drop_column + +revision: str = "0040_scheduled_group_notice" +down_revision: str | None = "0039_user_disabled" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + safe_add_column("scheduled_tasks", sa.Column("notification_group_connection_id", sa.String(64), nullable=True)) + + +def downgrade() -> None: + safe_drop_column("scheduled_tasks", "notification_group_connection_id") diff --git a/backend/packages/harness/deerflow/persistence/scheduled_tasks/model.py b/backend/packages/harness/deerflow/persistence/scheduled_tasks/model.py index fc14ce7328..96a7f1ec3f 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_tasks/model.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_tasks/model.py @@ -19,6 +19,8 @@ class ScheduledTaskRow(Base): goal_objective: Mapped[str | None] = mapped_column(Text, nullable=True) # The user's normalized "stop when ..." rule; never part of ``prompt``. stop_condition: Mapped[str | None] = mapped_column(Text, nullable=True) + # Optional additional Feishu group; keep the reference after disconnection. + notification_group_connection_id: Mapped[str | None] = mapped_column(String(64), nullable=True) max_runs: Mapped[int | None] = mapped_column(Integer, nullable=True) end_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) standing_notes: Mapped[list[str] | None] = mapped_column(JSON, nullable=True) diff --git a/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py b/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py index dc94487de3..7c910ec6cf 100644 --- a/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py +++ b/backend/packages/harness/deerflow/persistence/scheduled_tasks/sql.py @@ -149,6 +149,7 @@ class ScheduledTaskRepository: end_at: datetime | str | None = None, standing_notes: list[str] | None = None, stop_condition: str | None = None, + notification_group_connection_id: str | None = None, ) -> dict[str, Any]: if goal_objective is not None: normalize_goal_objective(goal_objective) @@ -166,6 +167,7 @@ class ScheduledTaskRepository: origin_thread_id=origin_thread_id, goal_objective=goal_objective, stop_condition=stop_condition, + notification_group_connection_id=notification_group_connection_id, max_runs=max_runs, end_at=_coerce_datetime(end_at), standing_notes=list(standing_notes) if standing_notes is not None else None, diff --git a/backend/tests/test_features_router.py b/backend/tests/test_features_router.py index e7f60733a4..d2336bdec3 100644 --- a/backend/tests/test_features_router.py +++ b/backend/tests/test_features_router.py @@ -37,6 +37,7 @@ def _app_with_config( scheduled_task_repo_available: bool = False, scheduler_running: bool = False, scheduler_tool_enabled: bool = False, + channel_connections_enabled: bool | None = None, ) -> FastAPI: app = FastAPI() app.state.scheduled_task_repo = object() if scheduled_task_repo_available else None @@ -62,6 +63,8 @@ def _app_with_config( ), scheduler=SimpleNamespace(enabled=scheduler_running, tool_enabled=scheduler_tool_enabled, min_once_delay_seconds=60), ) + if channel_connections_enabled is not None: + fake_config.channel_connections = SimpleNamespace(enabled=channel_connections_enabled) search_tool = SimpleNamespace(use=knowledge_search_provider) if knowledge_search_provider is not None else None fake_config.get_tool_config = lambda name: search_tool if name == "knowledge_search" else None app.dependency_overrides[get_config] = lambda: fake_config @@ -86,7 +89,7 @@ def test_features_reports_agents_api_enabled() -> None: "knowledge_base": { "scope_selection_enabled": False, }, - "scheduled_tasks": {"available": False, "running": False, "tool_enabled": False, "min_interval_seconds": 60}, + "scheduled_tasks": {"available": False, "running": False, "tool_enabled": False, "min_interval_seconds": 60, "group_notifications": False}, "thread_activity": {"available": False}, } @@ -109,7 +112,7 @@ def test_features_reports_agents_api_disabled() -> None: "knowledge_base": { "scope_selection_enabled": False, }, - "scheduled_tasks": {"available": False, "running": False, "tool_enabled": False, "min_interval_seconds": 60}, + "scheduled_tasks": {"available": False, "running": False, "tool_enabled": False, "min_interval_seconds": 60, "group_notifications": False}, "thread_activity": {"available": False}, } @@ -256,7 +259,19 @@ def test_features_reports_scheduled_tasks_process_state(repo: bool, running: boo with TestClient(app) as client: response = client.get("/api/features") assert response.status_code == 200 - assert response.json()["scheduled_tasks"] == {**expected, "min_interval_seconds": 60} + assert response.json()["scheduled_tasks"] == {**expected, "min_interval_seconds": 60, "group_notifications": False} + + +@pytest.mark.parametrize("repo", [False, True]) +@pytest.mark.parametrize("connections", [None, False, True]) +@pytest.mark.parametrize("running", [False, True]) +def test_features_group_notifications_require_repository_and_channel_connections(repo, connections, running): + app = _app_with_config(agents_api_enabled=True, scheduled_task_repo_available=repo, scheduler_running=running, channel_connections_enabled=connections) + with TestClient(app) as client: + response = client.get("/api/features") + assert response.status_code == 200 + # Existing tasks remain configurable even while the local poller is stopped. + assert response.json()["scheduled_tasks"]["group_notifications"] is (repo and connections is True) class _SqlRunStore: diff --git a/backend/tests/test_feishu_notifications.py b/backend/tests/test_feishu_notifications.py index c697ccef9c..43e2994c4a 100644 --- a/backend/tests/test_feishu_notifications.py +++ b/backend/tests/test_feishu_notifications.py @@ -19,7 +19,7 @@ from app.channels.message_bus import MessageBus @pytest.fixture def channel(): - from lark_oapi.api.im.v1 import CreateMessageRequest, CreateMessageRequestBody + from lark_oapi.api.im.v1 import CreateMessageRequest, CreateMessageRequestBody, GetChatRequest instance = FeishuChannel(MessageBus(), {}) instance._running = True @@ -27,6 +27,7 @@ def channel(): instance._thread.is_alive.return_value = True instance._CreateMessageRequest = CreateMessageRequest instance._CreateMessageRequestBody = CreateMessageRequestBody + instance._GetChatRequest = GetChatRequest instance._api_client = MagicMock() instance._api_client.im.v1.message.create.return_value = SimpleNamespace(success=lambda: True) return instance @@ -44,6 +45,83 @@ def _private_binding(**overrides): } +def _group_binding(**overrides): + return _private_binding(id="group-binding", workspace_id="oc_group", metadata={"chat_id": "oc_group", "chat_type": "group"}, **overrides) + + +@pytest.mark.asyncio +async def test_base_group_notification_is_unsupported_and_never_falls_back(channel): + channel.send_notification = AsyncMock() + with pytest.raises(NotImplementedError): + await Channel.send_group_notification_for_owner(channel, owner_user_id="owner-1", target="group-binding", text_markdown="Done") + channel.send_notification.assert_not_called() + + +@pytest.mark.asyncio +async def test_group_notification_uses_selected_binding_and_group_chat(channel): + channel._connection_repo = SimpleNamespace(list_connections=AsyncMock(return_value=[_private_binding(), _group_binding()])) + await channel.send_group_notification_for_owner(owner_user_id="owner-1", target="group-binding", text_markdown="Done") + channel._connection_repo.list_connections.assert_awaited_once_with("owner-1") + create = channel._api_client.im.v1.message.create + create.assert_called_once() + request = create.call_args.args[0] + assert request.receive_id_type == "chat_id" + assert request.request_body.receive_id == "oc_group" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "overrides", + [ + {"id": "different-binding"}, + {"owner_user_id": "another-owner"}, + {"provider": "wecom"}, + {"status": "revoked"}, + {"workspace_id": "oc_other"}, + {"metadata": {"chat_id": "oc_group", "chat_type": "p2p"}}, + {"metadata": {"chat_id": "oc_group"}}, + ], +) +async def test_group_notification_refuses_invalid_binding_without_private_fallback(channel, overrides): + binding = {**_group_binding(), **overrides} + channel._connection_repo = SimpleNamespace(list_connections=AsyncMock(return_value=[binding])) + with pytest.raises(RuntimeError, match="no longer connected"): + await channel.send_group_notification_for_owner(owner_user_id="owner-1", target="group-binding", text_markdown="Done") + channel._api_client.im.v1.message.create.assert_not_called() + + +@pytest.mark.asyncio +async def test_group_notification_binding_lookup_failure_parks(channel): + channel._connection_repo = SimpleNamespace(list_connections=AsyncMock(side_effect=OSError("private database detail"))) + with pytest.raises(ChannelUnavailable): + await channel.send_group_notification_for_owner(owner_user_id="owner-1", target="group-binding", text_markdown="Done") + channel._api_client.im.v1.message.create.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("outcome", ["success", "rejected", "exception"]) +async def test_group_binding_reads_name_offloop_without_blocking_binding_on_failure(channel, outcome): + channel._connection_repo = SimpleNamespace(consume_oauth_state=AsyncMock(return_value={"owner_user_id": "owner-1"}), upsert_connection=AsyncMock()) + channel._reply_card = AsyncMock() + loop_thread = threading.get_ident() + worker_threads = [] + + def get_name(request): + worker_threads.append(threading.get_ident()) + assert request.chat_id == "oc_group" + if outcome == "exception": + raise OSError("private transport detail") + return SimpleNamespace(success=lambda: outcome == "success", code=0 if outcome == "success" else 99991672, data=SimpleNamespace(name="Release team")) + + channel._api_client.im.v1.chat.get.side_effect = get_name + assert await channel._bind_connection_from_connect_code(message_id="om_bind", chat_id="oc_group", user_id="ou_bound_user", code="connect-code", chat_type="group") + binding = channel._connection_repo.upsert_connection.call_args.kwargs + assert worker_threads and worker_threads != [loop_thread] + assert binding.get("workspace_name") == ("Release team" if outcome == "success" else None) + assert binding["metadata"]["chat_type"] == "group" + assert binding["metadata"]["chat_id"] == "oc_group" + + @pytest.mark.asyncio async def test_base_owner_notification_preserves_existing_provider_send(channel): channel.send_notification = AsyncMock() diff --git a/backend/tests/test_migration_0027_0028_notification_outbox.py b/backend/tests/test_migration_0027_0028_notification_outbox.py index 24927cc02e..4518de296e 100644 --- a/backend/tests/test_migration_0027_0028_notification_outbox.py +++ b/backend/tests/test_migration_0027_0028_notification_outbox.py @@ -23,7 +23,7 @@ OUTBOX = "0027_notification_deliveries" PARKED = "0028_parked_attempts" SCHEDULER = "0029_scheduler_agent_tasks" CLAIM_TOKENS = "0030_notification_claim_tokens" -HEAD = "0039_user_disabled" +HEAD = "0040_scheduled_group_notice" PREVIOUS = "0026_mcp_task_lease_tokens" TABLE = "notification_deliveries" diff --git a/backend/tests/test_migration_0029_scheduler_agent_tasks.py b/backend/tests/test_migration_0029_scheduler_agent_tasks.py index 7dd5eacf61..643a5cc3ac 100644 --- a/backend/tests/test_migration_0029_scheduler_agent_tasks.py +++ b/backend/tests/test_migration_0029_scheduler_agent_tasks.py @@ -23,7 +23,7 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository REVISION = "0029_scheduler_agent_tasks" -CURRENT_HEAD = "0039_user_disabled" +CURRENT_HEAD = "0040_scheduled_group_notice" NEXT = "0030_notification_claim_tokens" PREVIOUS = "0028_parked_attempts" TASK_FIELDS = {"origin_thread_id", "goal_objective", "max_runs", "end_at", "standing_notes"} diff --git a/backend/tests/test_migration_0032_activity_and_task_events.py b/backend/tests/test_migration_0032_activity_and_task_events.py index f0775dde08..09bbce92c4 100644 --- a/backend/tests/test_migration_0032_activity_and_task_events.py +++ b/backend/tests/test_migration_0032_activity_and_task_events.py @@ -24,7 +24,7 @@ IDEMPOTENCY = "0036_run_idempotency_request" DOCUMENT_SUMMARIES = "0037_project_document_summaries" CHANNEL_BINDINGS = "0038_channel_thread_bindings" USER_DISABLED = "0039_user_disabled" -CURRENT_HEAD = "0039_user_disabled" +CURRENT_HEAD = "0040_scheduled_group_notice" PREVIOUS = "0031_scheduled_streak_boundary" TABLES = {"thread_read_markers", "thread_read_versions", "scheduled_task_events"} INDEXES = { diff --git a/backend/tests/test_migration_0034_run_event_seq_watermark.py b/backend/tests/test_migration_0034_run_event_seq_watermark.py index 2b4a39dfc5..3f5ae05607 100644 --- a/backend/tests/test_migration_0034_run_event_seq_watermark.py +++ b/backend/tests/test_migration_0034_run_event_seq_watermark.py @@ -24,7 +24,7 @@ IDEMPOTENCY = "0036_run_idempotency_request" DOCUMENT_SUMMARIES = "0037_project_document_summaries" CHANNEL_BINDINGS = "0038_channel_thread_bindings" USER_DISABLED = "0039_user_disabled" -CURRENT_HEAD = "0039_user_disabled" +CURRENT_HEAD = "0040_scheduled_group_notice" TABLE = "run_event_thread_seq" pytestmark = pytest.mark.asyncio diff --git a/backend/tests/test_migration_0035_login_throttle.py b/backend/tests/test_migration_0035_login_throttle.py index e0b98fdce0..245ff7e1f3 100644 --- a/backend/tests/test_migration_0035_login_throttle.py +++ b/backend/tests/test_migration_0035_login_throttle.py @@ -25,7 +25,7 @@ IDEMPOTENCY = "0036_run_idempotency_request" DOCUMENT_SUMMARIES = "0037_project_document_summaries" CHANNEL_BINDINGS = "0038_channel_thread_bindings" USER_DISABLED = "0039_user_disabled" -CURRENT_HEAD = "0039_user_disabled" +CURRENT_HEAD = "0040_scheduled_group_notice" PREVIOUS = "0034_run_event_seq_watermark" TABLE = "login_throttle" COLUMNS = {"ip", "fail_count", "locked_at", "lock_duration_seconds", "updated_at"} diff --git a/backend/tests/test_migration_0037_project_document_summaries.py b/backend/tests/test_migration_0037_project_document_summaries.py index 81f8672ef3..99c1419069 100644 --- a/backend/tests/test_migration_0037_project_document_summaries.py +++ b/backend/tests/test_migration_0037_project_document_summaries.py @@ -19,7 +19,8 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args REVISION = "0037_project_document_summaries" PREVIOUS = "0036_run_idempotency_request" CHANNEL_BINDINGS = "0038_channel_thread_bindings" -CURRENT_HEAD = "0039_user_disabled" +USER_DISABLED = "0039_user_disabled" +CURRENT_HEAD = "0040_scheduled_group_notice" pytestmark = pytest.mark.asyncio @@ -27,7 +28,8 @@ async def test_0037_chains_after_0036_and_is_single_head(): script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR)) assert script.get_heads() == [CURRENT_HEAD] assert script.get_revision(CHANNEL_BINDINGS).down_revision == REVISION - assert script.get_revision(CURRENT_HEAD).down_revision == CHANNEL_BINDINGS + assert script.get_revision(USER_DISABLED).down_revision == CHANNEL_BINDINGS + assert script.get_revision(CURRENT_HEAD).down_revision == USER_DISABLED assert script.get_revision(REVISION).down_revision == PREVIOUS # alembic_version.version_num is VARCHAR(32). assert len(REVISION) <= 32 diff --git a/backend/tests/test_migration_0038_channel_thread_bindings.py b/backend/tests/test_migration_0038_channel_thread_bindings.py index 40f37744b5..f628573b65 100644 --- a/backend/tests/test_migration_0038_channel_thread_bindings.py +++ b/backend/tests/test_migration_0038_channel_thread_bindings.py @@ -22,7 +22,7 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args REVISION = "0038_channel_thread_bindings" PREVIOUS = "0037_project_document_summaries" -CURRENT_HEAD = "0039_user_disabled" +CURRENT_HEAD = "0040_scheduled_group_notice" TABLE = "channel_thread_bindings" COLUMNS = {"key", "channel_name", "chat_id", "topic_id", "thread_id", "user_id", "created_at", "updated_at"} INDEX = "ix_channel_thread_bindings_channel_chat" diff --git a/backend/tests/test_migration_0040_scheduled_group_notice.py b/backend/tests/test_migration_0040_scheduled_group_notice.py new file mode 100644 index 0000000000..3c3748a03a --- /dev/null +++ b/backend/tests/test_migration_0040_scheduled_group_notice.py @@ -0,0 +1,85 @@ +"""The explicit group destination is additive and retains disconnected targets.""" + +import asyncio +import os +import uuid + +import pytest +import sqlalchemy as sa +from alembic import command +from alembic.script import ScriptDirectory +from sqlalchemy.ext.asyncio import create_async_engine +from support.postgres import asyncpg_test_url + +from deerflow.persistence import bootstrap +from deerflow.persistence.postgres_schema import build_asyncpg_connect_args + +REVISION = "0040_scheduled_group_notice" +PREVIOUS = "0039_user_disabled" +COLUMN = "notification_group_connection_id" + + +def test_group_notification_revision_has_one_head_after_0039(): + script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR)) + assert script.get_heads() == [REVISION] + assert script.get_revision(REVISION).down_revision == PREVIOUS + assert len(REVISION) <= 32 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("backend", ["sqlite", "postgres"]) +async def test_group_notification_migration_preserves_legacy_rows_and_retries(tmp_path, backend): + schema = None + if backend == "sqlite": + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'migration.db'}") + else: + uri = os.environ.get("TEST_POSTGRES_URI") + if not uri: + pytest.skip("requires TEST_POSTGRES_URI (real Postgres migration)") + schema = f"group_notice_{uuid.uuid4().hex}" + engine = create_async_engine(asyncpg_test_url(uri), connect_args=build_asyncpg_connect_args(schema)) + cfg = bootstrap._get_alembic_config(engine, postgres_schema=schema or "") + try: + if schema: + async with engine.begin() as conn: + await conn.execute(sa.text(f'CREATE SCHEMA "{schema}"')) + await asyncio.to_thread(command.upgrade, cfg, PREVIOUS) + async with engine.begin() as conn: + await conn.execute( + sa.text( + "INSERT INTO scheduled_tasks (id,user_id,context_mode,title,prompt,schedule_type,schedule_spec,timezone,status,overlap_policy,run_count,last_occurrence_seq,goal_objective,created_at,updated_at) " + "VALUES ('existing','owner','fresh_thread_per_run','Report','Write report','cron','{}','UTC','paused','enqueue',4,4,'report ready',CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)" + ) + ) + await asyncio.to_thread(command.upgrade, cfg, REVISION) + async with engine.begin() as conn: + columns = await conn.run_sync(lambda sync: {col["name"]: col for col in sa.inspect(sync).get_columns("scheduled_tasks")}) + assert columns[COLUMN]["nullable"] and columns[COLUMN]["default"] is None + foreign_keys = await conn.run_sync(lambda sync: sa.inspect(sync).get_foreign_keys("scheduled_tasks")) + assert all(COLUMN not in fk["constrained_columns"] for fk in foreign_keys) + row = (await conn.execute(sa.text(f"SELECT run_count,goal_objective,{COLUMN} FROM scheduled_tasks WHERE id='existing'"))).one() + assert tuple(row) == (4, "report ready", None) + await conn.execute(sa.text(f"UPDATE scheduled_tasks SET {COLUMN}='disconnected-group' WHERE id='existing'")) + + # Retry upgrade over an already-added column without changing its value. + await asyncio.to_thread(command.stamp, cfg, PREVIOUS) + await asyncio.to_thread(command.upgrade, cfg, REVISION) + async with engine.connect() as conn: + assert (await conn.execute(sa.text(f"SELECT {COLUMN} FROM scheduled_tasks WHERE id='existing'"))).scalar_one() == "disconnected-group" + + await asyncio.to_thread(command.downgrade, cfg, PREVIOUS) + async with engine.connect() as conn: + names = await conn.run_sync(lambda sync: {col["name"] for col in sa.inspect(sync).get_columns("scheduled_tasks")}) + assert COLUMN not in names + assert (await conn.execute(sa.text("SELECT run_count,goal_objective FROM scheduled_tasks WHERE id='existing'"))).one() == (4, "report ready") + # Retry downgrade after the column is gone, then upgrade once more. + await asyncio.to_thread(command.stamp, cfg, REVISION) + await asyncio.to_thread(command.downgrade, cfg, PREVIOUS) + await asyncio.to_thread(command.upgrade, cfg, REVISION) + async with engine.connect() as conn: + assert (await conn.execute(sa.text(f"SELECT {COLUMN} FROM scheduled_tasks WHERE id='existing'"))).scalar_one() is None + finally: + if schema: + async with engine.begin() as conn: + await conn.execute(sa.text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) + await engine.dispose() diff --git a/backend/tests/test_notification_delivery_worker.py b/backend/tests/test_notification_delivery_worker.py index 152b487a0b..3092d07385 100644 --- a/backend/tests/test_notification_delivery_worker.py +++ b/backend/tests/test_notification_delivery_worker.py @@ -56,6 +56,10 @@ class FakeChannel: self.unavailable = unavailable self.is_running = running self.owners = [] + self.group_sent = [] + + async def send_group_notification_for_owner(self, *, owner_user_id, target, text_markdown): + self.group_sent.append((owner_user_id, target, text_markdown)) async def send_notification_for_owner(self, *, owner_user_id, target, text_markdown): self.owners.append(owner_user_id) @@ -114,6 +118,78 @@ def _connection(*, provider="wecom", external_account_id="GaoZhiChao", status="c } +def _group_connection(**updates): + return {**_connection(provider="feishu", external_account_id="ou_owner"), "id": "group-binding", "workspace_id": "oc_group", "metadata": {"chat_id": "oc_group", "chat_type": "group"}, **updates} + + +@pytest.mark.asyncio +async def test_group_notice_uses_selected_binding_and_only_group_hook(): + repo = FakeDeliveryRepo([_delivery_row(provider="feishu", target="group-binding", payload={"target_type": "group", "run_status": "success"})]) + channel = FakeChannel() + worker = NotificationDeliveryWorker(delivery_repo=repo, resolve_channel=lambda provider: channel, resolve_connections=lambda owner: [_group_connection()]) + await worker.run_once(now=datetime.now(UTC)) + assert repo.sent == ["delivery-1"] + assert channel.group_sent[0][:2] == ("user-1", "group-binding") + assert channel.sent == [] + assert channel.owners == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "updates", + [ + None, + {"id": "another-binding"}, + {"owner_user_id": "another-owner"}, + {"status": "revoked"}, + {"provider": "wecom"}, + {"workspace_id": "oc_other"}, + {"metadata": {"chat_id": "oc_group", "chat_type": "p2p"}}, + {"metadata": {"chat_id": "oc_group"}}, + ], +) +async def test_group_notice_requires_current_exact_group_binding(updates): + repo = FakeDeliveryRepo([_delivery_row(provider="feishu", target="group-binding", payload={"target_type": "group"})]) + channel = FakeChannel() + rows = [] if updates is None else [_group_connection(**updates)] + worker = NotificationDeliveryWorker(delivery_repo=repo, resolve_channel=lambda provider: channel, resolve_connections=lambda owner: rows) + await worker.run_once(now=datetime.now(UTC)) + assert repo.sent == [] + assert len(repo.terminal_failures) == 1 + assert channel.sent == [] + assert channel.group_sent == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("provider,target_type", [("wecom", "group"), ("feishu", "unknown"), ("feishu", "private")]) +async def test_invalid_group_target_type_never_falls_back_to_personal_notice(provider, target_type): + repo = FakeDeliveryRepo([_delivery_row(provider=provider, target="ou_owner", payload={"target_type": target_type})]) + channel = FakeChannel() + worker = NotificationDeliveryWorker(delivery_repo=repo, resolve_channel=lambda provider: channel, resolve_connections=lambda owner: [_connection(provider=provider, external_account_id="ou_owner")]) + await worker.run_once(now=datetime.now(UTC)) + assert repo.sent == [] + assert len(repo.terminal_failures) == 1 + assert channel.sent == [] + assert channel.group_sent == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("resolver_missing", [False, True]) +async def test_group_notice_parks_when_binding_cannot_be_checked(resolver_missing): + repo = FakeDeliveryRepo([_delivery_row(provider="feishu", target="group-binding", payload={"target_type": "group"})]) + channel = FakeChannel() + + def unavailable(owner): + raise OSError("lookup unavailable") + + worker = NotificationDeliveryWorker(delivery_repo=repo, resolve_channel=lambda provider: channel, resolve_connections=None if resolver_missing else unavailable) + await worker.run_once(now=datetime.now(UTC)) + assert repo.sent == [] + assert repo.failed[0][2] is False + assert channel.sent == [] + assert channel.group_sent == [] + + def test_render_run_completed_text_has_no_raw_ids(): text = render_notification_text(_delivery_row(payload={"run_status": "success", "error": None, "task_id": "task-1", "task_title": "Daily digest"})) diff --git a/backend/tests/test_notification_groups.py b/backend/tests/test_notification_groups.py new file mode 100644 index 0000000000..ac49f86b35 --- /dev/null +++ b/backend/tests/test_notification_groups.py @@ -0,0 +1,28 @@ +"""The group predicate accepts only a recorded connected Feishu group.""" + +import pytest + + +@pytest.mark.parametrize( + "overrides,expected", + [ + ({}, True), + ({"provider": "wecom"}, False), + ({"status": "revoked"}, False), + ({"workspace_id": None}, False), + ({"workspace_id": ""}, False), + ({"workspace_id": " ", "metadata": {"chat_id": " ", "chat_type": "group"}}, False), + ({"workspace_id": "oc_other"}, False), + ({"metadata": None}, False), + ({"metadata": []}, False), + ({"metadata": {"chat_id": "oc_group"}}, False), + ({"metadata": {"chat_id": "oc_group", "chat_type": "p2p"}}, False), + ({"metadata": {"chat_id": "", "chat_type": "group"}}, False), + ({"metadata": {"chat_id": ["oc_group"], "chat_type": "group"}}, False), + ], +) +def test_feishu_group_predicate(overrides, expected): + from app.channels.notification_groups import is_feishu_notification_group + + value = {"provider": "feishu", "status": "connected", "workspace_id": "oc_group", "metadata": {"chat_id": "oc_group", "chat_type": "group"}, **overrides} + assert is_feishu_notification_group(**value) is expected diff --git a/backend/tests/test_scheduled_task_group_targets.py b/backend/tests/test_scheduled_task_group_targets.py new file mode 100644 index 0000000000..29d7c49cdf --- /dev/null +++ b/backend/tests/test_scheduled_task_group_targets.py @@ -0,0 +1,185 @@ +"""Owner-selected Feishu groups round-trip through the scheduled-task API.""" + +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from unittest.mock import AsyncMock +from uuid import UUID + +import httpx +import pytest +import pytest_asyncio +from _router_auth_helpers import make_authed_test_app +from _scheduled_rows import occurrence + +from app.gateway.auth.models import User +from app.gateway.auth_disabled import AUTH_SOURCE_SESSION +from app.gateway.routers import scheduled_tasks +from deerflow.config.channel_connections_config import ChannelConnectionsConfig +from deerflow.config.database_config import DatabaseConfig +from deerflow.persistence.channel_connections import ChannelConnectionRepository +from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config +from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository +from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow + +OWNER = "00000000-0000-0000-0000-000000000111" +OTHER = "00000000-0000-0000-0000-000000000222" +FIELD = "notification_group_connection_id" + + +@pytest_asyncio.fixture +async def env(tmp_path, monkeypatch): + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + sf = get_session_factory() + tasks, connections = ScheduledTaskRepository(sf), ChannelConnectionRepository(sf) + user = User(email="groups@example.com", password_hash="test", system_role="user", id=UUID(OWNER)) + app = make_authed_test_app(user_factory=lambda: user, bind_current_user=True) + + @app.middleware("http") + async def session_source(request, call_next): + request.state.auth_source = AUTH_SOURCE_SESSION + return await call_next(request) + + app.state.scheduled_task_repo = tasks + app.state.channel_connection_repo = connections + app.state.channel_connections_config = ChannelConnectionsConfig(enabled=True) + app.include_router(scheduled_tasks.router) + monkeypatch.setattr(scheduled_tasks, "is_scheduler_running", lambda _request: True) + monkeypatch.setattr(scheduled_tasks, "get_config", lambda: SimpleNamespace(scheduler=SimpleNamespace(min_once_delay_seconds=60))) + try: + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app, raise_app_exceptions=False), base_url="http://groups.test") as client: + yield SimpleNamespace(client=client, app=app, tasks=tasks, connections=connections, sf=sf) + finally: + await close_engine() + + +def _body(**extra): + return {"title": "Weekly report", "prompt": "Prepare the report", "schedule_type": "cron", "schedule_spec": {"cron": "0 9 * * 1"}, "timezone": "UTC", **extra} + + +async def _bind(env, **extra): + fields = {"owner_user_id": OWNER, "provider": "feishu", "external_account_id": "ou_member", "workspace_id": "oc_group", "metadata": {"chat_type": "group", "chat_id": "oc_group"}, **extra} + return await env.connections.upsert_connection(**fields) + + +async def _create(env, **extra): + response = await env.client.post("/api/scheduled-tasks", json=_body(**extra)) + assert response.status_code == 200, response.text + return response.json() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("selected", [False, True]) +async def test_group_destination_round_trips_create_get_and_list(env, selected): + group = await _bind(env) + target = group["id"] if selected else None + task = await _create(env, **({FIELD: target} if selected else {})) + assert task[FIELD] == target + assert (await env.client.get(f"/api/scheduled-tasks/{task['id']}")).json()[FIELD] == target + assert (await env.client.get("/api/scheduled-tasks")).json()[0][FIELD] == target + assert (await env.tasks.get(task["id"], user_id=OWNER))[FIELD] == target + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method", ["post", "patch"]) +@pytest.mark.parametrize( + "overrides", + [ + {"owner_user_id": OTHER}, + {"provider": "wecom"}, + {"status": "revoked"}, + {"metadata": {"chat_type": "p2p", "chat_id": "oc_group"}}, + {"metadata": {"chat_id": "oc_group"}}, + {"metadata": {"chat_type": "group", "chat_id": "oc_different"}}, + {"workspace_id": "", "metadata": {"chat_type": "group", "chat_id": ""}}, + ], +) +async def test_group_destination_rejects_unowned_or_unverified_bindings(env, method, overrides): + group = await _bind(env, **overrides) + existing = await _create(env) if method == "patch" else None + url = f"/api/scheduled-tasks/{existing['id']}" if existing else "/api/scheduled-tasks" + payload = {FIELD: group["id"]} if existing else _body(**{FIELD: group["id"]}) + response = await env.client.request(method, url, json=payload) + assert response.status_code == 422, response.text + assert response.json()["detail"]["code"] == "invalid_notification_group" + if existing: + assert (await env.tasks.get(existing["id"], user_id=OWNER))[FIELD] is None + else: + assert await env.tasks.list_by_user(OWNER) == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("target", ["unknown", "", " ", "x" * 65]) +async def test_invalid_group_identifier_is_coded_and_does_not_create_task(env, target): + response = await env.client.post("/api/scheduled-tasks", json=_body(**{FIELD: target})) + assert response.status_code == 422, response.text + assert response.json()["detail"]["code"] == "invalid_notification_group" + assert await env.tasks.list_by_user(OWNER) == [] + + +@pytest.mark.asyncio +async def test_title_patch_keeps_a_revoked_selection_and_null_clears_without_lookup(env, monkeypatch): + group = await _bind(env) + task = await _create(env, **{FIELD: group["id"]}) + assert await env.connections.disconnect_connection(connection_id=group["id"], owner_user_id=OWNER) + renamed = await env.client.patch(f"/api/scheduled-tasks/{task['id']}", json={"title": "Renamed"}) + assert renamed.status_code == 200, renamed.text + assert renamed.json()[FIELD] == group["id"] + lookup = AsyncMock(side_effect=OSError("lookup should not run")) + monkeypatch.setattr(env.connections, "list_connections", lookup) + cleared = await env.client.patch(f"/api/scheduled-tasks/{task['id']}", json={FIELD: None}) + assert cleared.status_code == 200, cleared.text + assert cleared.json()[FIELD] is None + lookup.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method", ["post", "patch"]) +async def test_group_lookup_failure_is_503_not_invalid_target(env, monkeypatch, method): + existing = await _create(env) if method == "patch" else None + monkeypatch.setattr(env.connections, "list_connections", AsyncMock(side_effect=OSError("private connection failure"))) + url = f"/api/scheduled-tasks/{existing['id']}" if existing else "/api/scheduled-tasks" + payload = {FIELD: "conn-group"} if existing else _body(**{FIELD: "conn-group"}) + response = await env.client.request(method, url, json=payload) + assert response.status_code == 503, response.text + assert response.json()["detail"]["code"] == "scheduler_unavailable" + assert "private connection failure" not in response.text + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ["completed", "paused", "enabled"]) +async def test_group_only_edit_preserves_schedule_goal_history_and_used_budget(env, status): + group = await _bind(env) + task = await _create(env, goal_objective="Finish the report", max_runs=1) + async with env.sf() as session: + row = await session.get(ScheduledTaskRow, task["id"]) + row.status = status + row.run_count = 7 + row.last_occurrence_seq = 9 + row.unmet_streak_after_seq = 4 + row.standing_notes = ["Preserve the report format"] + row.end_at = datetime.now(UTC) - timedelta(days=1) + if status == "completed": + row.next_run_at = None + session.add(occurrence("prior-run", task["id"], seq=9)) + await session.commit() + before = (await env.client.get(f"/api/scheduled-tasks/{task['id']}")).json() + response = await env.client.patch(f"/api/scheduled-tasks/{task['id']}", json={FIELD: group["id"]}) + assert response.status_code == 200, response.text + after = response.json() + assert after[FIELD] == group["id"] + for key in ("status", "next_run_at", "schedule_type", "schedule_spec", "timezone", "goal_objective", "max_runs", "end_at", "run_count", "automatic_runs_used", "standing_notes"): + assert after[key] == before[key], key + assert after["automatic_runs_used"] == 1 + async with env.sf() as session: + row = await session.get(ScheduledTaskRow, task["id"]) + assert row.last_occurrence_seq == 9 + assert row.unmet_streak_after_seq == 4 + + +@pytest.mark.asyncio +async def test_group_selection_requires_enabled_channel_connections(env): + group = await _bind(env) + env.app.state.channel_connections_config.enabled = False + response = await env.client.post("/api/scheduled-tasks", json=_body(**{FIELD: group["id"]})) + assert response.status_code == 422, response.text + assert response.json()["detail"]["code"] == "invalid_notification_group" diff --git a/backend/tests/test_scheduled_task_service.py b/backend/tests/test_scheduled_task_service.py index be3e6416dd..0bc2766c5a 100644 --- a/backend/tests/test_scheduled_task_service.py +++ b/backend/tests/test_scheduled_task_service.py @@ -1373,6 +1373,78 @@ async def test_only_push_capable_connected_providers_get_outbox_rows(outbox): assert len(await _notices(outbox)) == 2 +@pytest.mark.asyncio +async def test_selected_group_notice_is_additional_and_deduplicated(outbox): + from deerflow.persistence.channel_connections.model import ChannelConnectionRow + + await _task(outbox, notification_group_connection_id="selected-group") + await _bind(outbox, "feishu", "ou_feishu") + await _bind(outbox, "wecom", "wecom-user") + async with outbox.sf() as session: + session.add( + ChannelConnectionRow(id="selected-group", owner_user_id=NOTICE_OWNER, provider="feishu", status="connected", external_account_id="ou_feishu", workspace_id="oc_group", metadata_json={"chat_id": "oc_group", "chat_type": "group"}) + ) + await session.commit() + occurrence_id, run_id = await _occurrence(outbox) + assert await _complete(outbox, occurrence_id, run_id) is True + assert await _complete(outbox, occurrence_id, run_id) is False + notices = await _notices(outbox) + assert sorted((row["provider"], row["target"]) for row in notices) == [("feishu", "ou_feishu"), ("feishu", "selected-group"), ("wecom", "wecom-user")] + (group,) = [row for row in notices if row["target"] == "selected-group"] + assert group["payload"]["target_type"] == "group" + assert all("target_type" not in row["payload"] for row in notices if row is not group) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("revoked_binding", [False, True]) +async def test_selected_group_still_enqueues_if_binding_was_removed(outbox, revoked_binding): + from types import SimpleNamespace + + from sqlalchemy import select + + from app.scheduler.notification_delivery import NotificationDeliveryWorker + from deerflow.persistence.channel_connections.model import ChannelConnectionRow + from deerflow.persistence.notification_deliveries import NotificationDeliveryRow + + await _task(outbox, notification_group_connection_id="selected-group") + if revoked_binding: + async with outbox.sf() as session: + session.add( + ChannelConnectionRow( + id="selected-group", owner_user_id=NOTICE_OWNER, provider="feishu", status="revoked", external_account_id="ou_feishu", workspace_id="oc_group", metadata_json={"chat_id": "oc_group", "chat_type": "group"} + ) + ) + await session.commit() + occurrence_id, run_id = await _occurrence(outbox) + await _complete(outbox, occurrence_id, run_id) + notices = await _notices(outbox) + assert [(row["provider"], row["target"], row["payload"]["target_type"]) for row in notices] == [("feishu", "selected-group", "group")] + worker = NotificationDeliveryWorker(delivery_repo=outbox.deliveries, resolve_channel=lambda provider: SimpleNamespace(is_running=True), resolve_connections=outbox.service._connection_repo.list_connections) + await worker.run_once(now=datetime.now(UTC) + timedelta(days=1)) + async with outbox.sf() as session: + row = await session.scalar(select(NotificationDeliveryRow)) + assert row.status == "failed" + + +@pytest.mark.asyncio +async def test_group_binding_without_task_selection_keeps_personal_notice_only(outbox): + from deerflow.persistence.channel_connections.model import ChannelConnectionRow + + await _task(outbox) + async with outbox.sf() as session: + session.add( + ChannelConnectionRow( + id="group-not-selected", owner_user_id=NOTICE_OWNER, provider="feishu", status="connected", external_account_id="ou_feishu", workspace_id="oc_group", metadata_json={"chat_id": "oc_group", "chat_type": "group"} + ) + ) + await session.commit() + occurrence_id, run_id = await _occurrence(outbox) + await _complete(outbox, occurrence_id, run_id) + notices = await _notices(outbox) + assert [(row["provider"], row["target"]) for row in notices] == [("feishu", "ou_feishu")] + assert "target_type" not in notices[0]["payload"] + + @pytest.mark.asyncio async def test_auto_pause_sends_one_merged_notice(outbox): await _task(outbox, goal_objective="every item is checked") diff --git a/contracts/scheduled_task_errors_contract.json b/contracts/scheduled_task_errors_contract.json index fa3bbc0d9c..d17f344df4 100644 --- a/contracts/scheduled_task_errors_contract.json +++ b/contracts/scheduled_task_errors_contract.json @@ -19,6 +19,7 @@ "invalid_goal", "goal_requires_fresh_thread", "invalid_stop_condition", + "invalid_notification_group", "invalid_max_runs", "end_at_in_past", "end_at_before_first_run", diff --git a/docs/full-stack-plugins.md b/docs/full-stack-plugins.md index e9c6cf1cc1..5bc730766d 100644 --- a/docs/full-stack-plugins.md +++ b/docs/full-stack-plugins.md @@ -373,6 +373,33 @@ across a service restart. A changed asset revision or removed action fails expli and requires reload. Disabled/unloaded pages have no navigation entry after refresh; visiting an unavailable page directly does not mount a plugin. +## Frontend implementation invariants + +`core/extensions/` loads authenticated deployment-installed ES modules from `/api/plugins`. +Inline modules use authenticated fetch plus a released Blob URL. Manifest assets use +native credentialed module scripts, preserving relative imports and resource URLs. +Both honor the backend base and prefixes; transport and cache semantics are documented in +`docs/full-stack-plugins.md`. Host copy belongs in the typed locale dictionaries. +Conversation action factories, shapes and availability callbacks are guarded per plugin; +only validated value snapshots reach the toolbar/sidebar render paths. +`PluginNavigation` and the dynamic workspace extension route consume page declarations; +Capability Center defaults to the repository examples in `core/extensions/catalog.ts`, +merged by explicit namespace with runtime descriptors. Catalog-only entries are discovery +metadata, never module-loader inputs or proof of installation. Backend-only examples may +have no plugin descriptor; keep their runtime status unasserted. Details link to package +installation instructions. Keep the catalog aligned with `examples/deerflow-extension-*`. +Agent teams uses `community.agent-teams`; merge its installed descriptor into the +localized catalog entry without asserting runtime status for a catalog-only row. +Conversation action slots augment normal/custom-agent toolbars and sidebar menus without replacing native export or notification. +Plugin views use mount/dispose and abort signals; Shadow DOM is CSS isolation, not a sandbox. +Descriptors are user-keyed page snapshots, refreshed manually. Backend calls bind the plugin's +namespace, action allowlist and expected viewer identity. See `docs/full-stack-plugins.md`. + +Plugin page `openConversation(threadId)` resolves authenticated thread metadata +with `pathOfThread`; do not let plugins hardcode default-agent routes. The page's +abort signal fences late navigation after unmount/account changes. Synchronous +conversation-action callbacks reject Promise returns while consuming rejections. + ## Validation scope The bookmark E2E test runs the production frontend against real Python action handlers diff --git a/frontend/AGENTS.md b/frontend/AGENTS.md index 99f533e227..febafb7d32 100644 --- a/frontend/AGENTS.md +++ b/frontend/AGENTS.md @@ -57,6 +57,17 @@ Run status `unmet` identifies a finished occurrence whose scheduled goal was not `core/scheduled-tasks/goal-outcome.ts` maps goal verdicts and host reason codes for run history; show known codes as localized labels; unknown codes, raw run errors and the evaluator's reason stay behind the run row's Details. Check-failure codes are "unchecked", not a miss. `contracts/scheduled_goal_notes_contract.json` pins the host strings it and `run-error.ts` match. Scheduled-task views read state through the pure `core/scheduled-tasks` helpers (`status.ts`, `actions.ts`, `format.ts`, `describeTaskSchedule`, `errors.ts`) so list, detail and chat card agree. "Is a run active?" is `status === "running"` or `active_run_status` (a recurring task stays `enabled` while it runs). Default views show no IDs, ISO times, cron strings or enum names; errors localize by `detail.code` (`contracts/scheduled_task_errors_contract.json`) with raw text only behind Details. `tests/e2e/utils/readable.ts` and `tests/unit/helpers/readable.ts` assert this. The tasks page (`app/workspace/scheduled-tasks/page.tsx`) only composes `components/workspace/scheduled-tasks/*`: list with status tabs, detail (Runs / Stops when / goal / Does / notes / History), outcome notice, create/edit/duplicate dialog (PATCH sends only changed fields, `null` clears) and the renew dialog that a `limits_exhausted` Resume opens. `?task_id=` selects a task; `?thread_id=` scopes the list to one chat. +`notification_group_connection_id` adds one Feishu group without changing personal +notifications. The selector requires `features.scheduled_tasks.group_notifications` +(absent = false); otherwise a saved target is preserved but explicitly clearable. +While the capability query is pending, show loading without a clear control. +Reuse `useChannelConnections`: offer connected Feishu `group` bindings with matching +nonempty metadata chat/workspace IDs. Show workspace names; disable unnamed groups +and never display raw IDs. Loading/errors preserve selections. PATCH omits unchanged +groups, clears with `null`, and leaves schedules untouched; Duplicate copies the group. +`BindFeishuGroup` uses the connect API for a persistent, copyable `/connect` command +and explicitly refreshes the shared query. The general connection poll stops early +when a private binding already exists, so cannot wait for a new group. In chat, a `schedule_task` result becomes an `assistant:scheduled-task` group (`core/scheduled-tasks/tool-result.ts`, last card per task per turn; not a turn boundary) rendered by `ScheduledTaskCard`, which polls the task only while on screen. A human message with `additional_kwargs.deerflow_scheduled_origin` stays a `human` group with `scheduledOrigin`: `ScheduledRunPrompt` renders the origin's user-language parts, never the launched text, and it is not editable. Server-created threads: `core/threads/origin.ts` reads `metadata.deerflow_origin` (`contracts/thread_origin_contract.json`), then `channel_source`, then a legacy `scheduled_task_id`; provider names come from `threads.origin.providers`. `core/threads/activity.ts` polls `/api/thread-activity` (mount `useThreadActivity` once; unmounting resets its cursor; first poll only seeds; invalidate lists only on server-origin threads, truncation or a higher `read_version`) and `useMarkThreadRead` posts reads. `core/scheduled-tasks/events.ts` loads a chat's lifecycle events, not gated on its task list; `placeTaskEvents` puts each at the end of its `after_run_id` turn. The UI language is saved as the `locale` preference once per session and on each switch, only when the Gateway's preferences include `locale`. Thread rows (sidebar, `/workspace/chats`) show `ThreadOriginIcon` and, if `unread === true` and not the open thread, `ThreadUnreadDot` with `threads.unreadLabel` as row label. `WorkspaceSidebar` mounts `useThreadActivity`. Chat pages call `useMarkOpenThreadRead` once metadata loads (marks read on load, on an unread flip, on tab visible, and via its callback from `onFinish`). @@ -363,30 +374,10 @@ a model switch without changing the backend's direct legacy-request behavior. ## Full-stack plugin UI -`core/extensions/` loads authenticated deployment-installed ES modules from `/api/plugins`. -Inline modules use authenticated fetch plus a released Blob URL. Manifest assets use -native credentialed module scripts, preserving relative imports and resource URLs. -Both honor the backend base and prefixes; transport and cache semantics are documented in -`docs/full-stack-plugins.md`. Host copy belongs in the typed locale dictionaries. -Conversation action factories, shapes and availability callbacks are guarded per plugin; -only validated value snapshots reach the toolbar/sidebar render paths. -`PluginNavigation` and the dynamic workspace extension route consume page declarations; -Capability Center defaults to the repository examples in `core/extensions/catalog.ts`, -merged by explicit namespace with runtime descriptors. Catalog-only entries are discovery -metadata, never module-loader inputs or proof of installation. Backend-only examples may -have no plugin descriptor; keep their runtime status unasserted. Details link to package -installation instructions. Keep the catalog aligned with `examples/deerflow-extension-*`. -Agent teams uses `community.agent-teams`; merge its installed descriptor into the -localized catalog entry without asserting runtime status for a catalog-only row. -Conversation action slots augment normal/custom-agent toolbars and sidebar menus without replacing native export or notification. -Plugin views use mount/dispose and abort signals; Shadow DOM is CSS isolation, not a sandbox. -Descriptors are user-keyed page snapshots, refreshed manually. Backend calls bind the plugin's -namespace, action allowlist and expected viewer identity. See `docs/full-stack-plugins.md`. - -Plugin page `openConversation(threadId)` resolves authenticated thread metadata -with `pathOfThread`; do not let plugins hardcode default-agent routes. The page's -abort signal fences late navigation after unmount/account changes. Synchronous -conversation-action callbacks reject Promise returns while consuming rejections. +Before changing plugin loaders, pages or conversation actions, read the mandatory +[Frontend implementation invariants](../docs/full-stack-plugins.md#frontend-implementation-invariants). +Preserve authenticated, viewer-bound loading, native actions, localization and +mount/dispose plus abort fencing; Shadow DOM is not a security sandbox. ### Composer references @@ -404,5 +395,7 @@ sends, with retry on failure. Polish restores whole labels longest-first in one pass. Confirmed thread creation seeds metadata before migration; references stay locked during attachment. Deleting an object removes its context; undo restores references already attached to the draft. Labels grant no read authority. IME, -caret, thread changes and attachment fences apply to editor and picker. The `@` +caret, thread changes and attachment fences apply to editor and picker. Commit +mention text before restoring focus synchronously; a delayed callback must not +overwrite subsequent typing or caret movement. The `@` and attachment buttons share the picker. diff --git a/frontend/src/components/workspace/input-box.tsx b/frontend/src/components/workspace/input-box.tsx index 1759af7330..238b1beffc 100644 --- a/frontend/src/components/workspace/input-box.tsx +++ b/frontend/src/components/workspace/input-box.tsx @@ -1895,14 +1895,12 @@ export function InputBox({ [], ); const focusComposerAt = useCallback((offset: number) => { - requestAnimationFrame(() => { - const textarea = textareaRef.current; - if (textarea) { - textarea.focus(); - textarea.setSelectionRange(offset, offset); - } else if (inlineSkillTextRef.current) - focusReferenceAt(inlineSkillTextRef.current, offset); - }); + const textarea = textareaRef.current; + if (textarea) { + textarea.focus(); + textarea.setSelectionRange(offset, offset); + } else if (inlineSkillTextRef.current) + focusReferenceAt(inlineSkillTextRef.current, offset); }, []); const selectMention = useCallback( async (selection: MentionSelection) => { @@ -1938,9 +1936,11 @@ export function InputBox({ setConversationReferences((previous) => previous.filter((ref) => ref.threadId !== selectionId), ); - textInput.setInput(next); - setMentionQuery(null); - setMentionButtonOpen(false); + flushSync(() => { + textInput.setInput(next); + setMentionQuery(null); + setMentionButtonOpen(false); + }); focusComposerAt(Math.min(caret, next.length)); return; } @@ -2074,10 +2074,14 @@ export function InputBox({ latestDraftRef.current = null; clearComposerDraft(storage, draftKey); } - textInput.setInput(nextText); - setMentionQuery(null); - setMentionButtonOpen(false); - setMentionError(null); + // Commit the replacement editor before restoring its selection. Deferring + // to the next frame can misplace typing or overwrite a later caret move. + flushSync(() => { + textInput.setInput(nextText); + setMentionQuery(null); + setMentionButtonOpen(false); + setMentionError(null); + }); focusComposerAt(caret + inserted.length); if (selection.kind === "file") onReferenceFileAttached?.(); }, diff --git a/frontend/src/components/workspace/scheduled-tasks/bind-feishu-group.tsx b/frontend/src/components/workspace/scheduled-tasks/bind-feishu-group.tsx new file mode 100644 index 0000000000..2550706890 --- /dev/null +++ b/frontend/src/components/workspace/scheduled-tasks/bind-feishu-group.tsx @@ -0,0 +1,146 @@ +"use client"; + +import { useMutation, useQueryClient } from "@tanstack/react-query"; +import { useState } from "react"; + +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { connectChannelProvider } from "@/core/channels/api"; +import { channelConnectionsQueryKey } from "@/core/channels/hooks"; +import { useI18n } from "@/core/i18n/hooks"; + +import { fill } from "./shared"; + +/** Group binding needs its own command; a connected private chat is not completion. */ +export function BindFeishuGroup() { + const { t, locale } = useI18n(); + const labels = t.scheduledTasks.notificationGroup; + const queryClient = useQueryClient(); + const [visible, setVisible] = useState(false); + const [copied, setCopied] = useState(false); + const [copyFailed, setCopyFailed] = useState(false); + const [refreshing, setRefreshing] = useState(false); + const [expiresAt, setExpiresAt] = useState(null); + const binding = useMutation({ + mutationFn: () => connectChannelProvider("feishu"), + onSuccess: (result) => { + setCopied(false); + setCopyFailed(false); + setExpiresAt(Date.now() + result.expires_in * 1000); + }, + }); + const command = binding.data ? `/connect ${binding.data.code}` : ""; + const copy = async () => { + try { + await navigator.clipboard.writeText(command); + setCopied(true); + setCopyFailed(false); + } catch { + setCopyFailed(true); + } + }; + const refresh = async () => { + setRefreshing(true); + try { + await queryClient.invalidateQueries({ + queryKey: channelConnectionsQueryKey, + }); + } finally { + setRefreshing(false); + } + }; + + return ( +
+
+ + +
+ {visible && ( +
+

{labels.bindInstructions}

+ {binding.isPending && ( +

+ {labels.generating} +

+ )} + {binding.error && ( +

+ {labels.bindFailed} +

+ )} + {command && ( + <> + + {expiresAt !== null && ( +

+ {fill(labels.expiresAt, { + time: new Intl.DateTimeFormat(locale, { + hour: "2-digit", + minute: "2-digit", + }).format(expiresAt), + })} +

+ )} + + {copyFailed && ( +

+ {labels.copyFailed} +

+ )} + + )} +
+ + +
+
+ )} +
+ ); +} diff --git a/frontend/src/components/workspace/scheduled-tasks/notification-group.tsx b/frontend/src/components/workspace/scheduled-tasks/notification-group.tsx new file mode 100644 index 0000000000..5c8dcd6aeb --- /dev/null +++ b/frontend/src/components/workspace/scheduled-tasks/notification-group.tsx @@ -0,0 +1,161 @@ +"use client"; + +import { Button } from "@/components/ui/button"; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from "@/components/ui/select"; +import { useChannelConnections } from "@/core/channels/hooks"; +import { useScheduledTasksFeature } from "@/core/features/hooks"; +import { useI18n } from "@/core/i18n/hooks"; +import { + feishuNotificationGroups, + notificationGroupName, +} from "@/core/scheduled-tasks/notification-groups"; + +import { BindFeishuGroup } from "./bind-feishu-group"; + +const NO_GROUP = "__no_additional_group__"; + +type NotificationGroupSelectProps = { + id: string; + value: string; + onChange: (value: string) => void; +}; + +/** Older Gateways ignore unknown fields: do not offer a destination they cannot save. */ +export function NotificationGroupSelect(props: NotificationGroupSelectProps) { + const { groupNotifications, isLoading } = useScheduledTasksFeature(); + const { t } = useI18n(); + if (isLoading) { + return ( +

+ {t.scheduledTasks.notificationGroup.loading} +

+ ); + } + if (!groupNotifications && props.value) { + return ( +
+

+ {t.scheduledTasks.notificationGroup.savedUnavailable} +

+ +
+ ); + } + return groupNotifications ? ( + + ) : null; +} + +/** Loading or losing the list must never reset the form's saved selection. */ +function EnabledNotificationGroupSelect({ + id, + value, + onChange, +}: NotificationGroupSelectProps) { + const { t } = useI18n(); + const labels = t.scheduledTasks.notificationGroup; + const { connections, isLoading, error } = useChannelConnections(); + const groups = feishuNotificationGroups(connections); + const selected = groups.find((group) => group.id === value); + const selectedMissing = Boolean(value) && !selected; + const unnamed = groups.some((group) => !notificationGroupName(group)); + const hint = error + ? labels.loadFailed + : isLoading + ? labels.loading + : selectedMissing + ? labels.unavailableHint + : unnamed + ? labels.unnamedHint + : groups.length === 0 + ? labels.emptyHint + : null; + + return ( +
+ + +
+

{labels.personalHint}

+ {hint &&

{hint}

} +
+ +
+ ); +} + +/** Shares the existing connections query cache, without adding a polling loop. */ +export function NotificationGroupSummary({ + connectionId, +}: { + connectionId: string; +}) { + const { t } = useI18n(); + const labels = t.scheduledTasks.notificationGroup; + const { connections, isLoading, error } = useChannelConnections(); + const group = feishuNotificationGroups(connections).find( + (connection) => connection.id === connectionId, + ); + const name = notificationGroupName(group); + return ( +
+

+ {name ?? + (isLoading + ? labels.loading + : group + ? labels.unnamed + : labels.unavailable)} +

+

{labels.personalHint}

+ {error && ( +

{labels.loadFailed}

+ )} +
+ ); +} diff --git a/frontend/src/components/workspace/scheduled-tasks/task-detail.tsx b/frontend/src/components/workspace/scheduled-tasks/task-detail.tsx index 04cfa8ca66..d21f22e665 100644 --- a/frontend/src/components/workspace/scheduled-tasks/task-detail.tsx +++ b/frontend/src/components/workspace/scheduled-tasks/task-detail.tsx @@ -70,6 +70,7 @@ import { isTaskBusy, type ScheduledTask } from "@/core/scheduled-tasks/types"; import { cn } from "@/lib/utils"; import { DeleteTaskDialog } from "./delete-task-dialog"; +import { NotificationGroupSummary } from "./notification-group"; import { RenewLimitsDialog } from "./renew-limits-dialog"; import { RunRow } from "./run-row"; import { @@ -530,6 +531,14 @@ export function TaskDetail({ )} + {task.notification_group_connection_id && ( +
+ +
+ )} +

{stops.primary} diff --git a/frontend/src/components/workspace/scheduled-tasks/task-form-dialog.tsx b/frontend/src/components/workspace/scheduled-tasks/task-form-dialog.tsx index 8e138b010a..19a0f83a2c 100644 --- a/frontend/src/components/workspace/scheduled-tasks/task-form-dialog.tsx +++ b/frontend/src/components/workspace/scheduled-tasks/task-form-dialog.tsx @@ -57,6 +57,7 @@ import { RECIPES, type Recipe } from "@/core/scheduled-tasks/recipes"; import type { ScheduledTask } from "@/core/scheduled-tasks/types"; import { cn } from "@/lib/utils"; +import { NotificationGroupSelect } from "./notification-group"; import { DEFAULT_ASSISTANT_ID, fill, isFrequentSchedule } from "./shared"; export type TaskFormMode = "create" | "edit" | "duplicate"; @@ -83,6 +84,7 @@ export type TaskFormState = { assistantId: string; contextMode: ContextMode; chatId: string; + notificationGroupConnectionId: string; }; const DEFAULT_SCHEDULE: ScheduleValue = { @@ -133,6 +135,7 @@ export function initialFormState( assistantId: DEFAULT_ASSISTANT_ID, contextMode: "fresh_thread_per_run", chatId: "", + notificationGroupConnectionId: "", }, endAtDropped: false, }; @@ -160,6 +163,8 @@ export function initialFormState( assistantId: task.assistant_id ?? DEFAULT_ASSISTANT_ID, contextMode: task.context_mode, chatId: task.thread_id ?? "", + notificationGroupConnectionId: + task.notification_group_connection_id ?? "", }, endAtDropped: endAtPassed, }; @@ -269,6 +274,10 @@ export function createPayload( if (valid.endAt) payload.end_at = valid.endAt; const stop = normalizeText(state.stopCondition); if (stop) payload.stop_condition = stop; + if (state.notificationGroupConnectionId) { + payload.notification_group_connection_id = + state.notificationGroupConnectionId; + } return payload; } @@ -318,6 +327,10 @@ export function updatePayload( ? task.end_at != null : !task.end_at || minute(valid.endAt) !== minute(task.end_at); if (endChanged) updates.end_at = valid.endAt; + const notificationGroup = state.notificationGroupConnectionId || null; + if (notificationGroup !== (task.notification_group_connection_id ?? null)) { + updates.notification_group_connection_id = notificationGroup; + } return updates; } @@ -670,6 +683,12 @@ function TaskForm({

)} + set("notificationGroupConnectionId", value)} + /> +