diff --git a/CHANGELOG.md b/CHANGELOG.md index 71b63eadbe..1eda49d547 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -282,6 +282,12 @@ This release closes that milestone with **301 merged pull requests**. option that preserves an unexpired binding command; credentials are saved on the backend and only the WeChat channel restarts. New WeChat connections default to QR login, with manual token entry still available. ([#5582]) +- **scheduler:** Push scheduled-task outcomes to the owner's bound + IM identities via a durable `notification_deliveries` outbox and + `NotificationDeliveryWorker`. Activates only when `channel_connections.enabled` + is true and a channel service is running; manual triggers and interrupts stay + silent, and a target disconnected while its delivery waited is dropped. WeCom implements proactive `send_notification`; other providers fail + visibly in the outbox until they grow a push path. (issue #4254, [#4843], [#6135]) #### Auth & guardrails @@ -6925,6 +6931,7 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#4839]: https://github.com/bytedance/deer-flow/pull/4839 [#4840]: https://github.com/bytedance/deer-flow/pull/4840 [#4842]: https://github.com/bytedance/deer-flow/pull/4842 +[#4843]: https://github.com/bytedance/deer-flow/pull/4843 [#4844]: https://github.com/bytedance/deer-flow/pull/4844 [#4846]: https://github.com/bytedance/deer-flow/pull/4846 [#4848]: https://github.com/bytedance/deer-flow/pull/4848 @@ -7514,4 +7521,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6112]: https://github.com/bytedance/deer-flow/pull/6112 [#6132]: https://github.com/bytedance/deer-flow/pull/6132 [#6134]: https://github.com/bytedance/deer-flow/pull/6134 +[#6135]: https://github.com/bytedance/deer-flow/pull/6135 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index d03dd099fe..2afaac5f64 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -253,6 +253,13 @@ 过期与重试状态,以及一个保留未过期绑定指令的“再次扫描”选项; 凭据保存在后端,且只有微信渠道会重启。 新的微信连接默认走扫码登录,手动输入 token 仍然可用。([#5582]) +- **scheduler:** 把定时任务的运行结果推送到任务所有者已绑定的 IM 身份: + 完成钩子把结果写入持久化的 `notification_deliveries` outbox, + 由 `NotificationDeliveryWorker` 负责投递。仅当 `channel_connections.enabled` + 为 true 且渠道服务在运行时生效;手动触发和中断不推送,等待投递期间被解绑的 + 身份也不再推送。目前只有企业微信实现了主动推送 `send_notification`, + 其它渠道在实现之前会在 outbox 中显式失败。 + (issue #4254, [#4843], [#6135]) #### 认证与防护 @@ -5717,6 +5724,7 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子 [#4839]: https://github.com/bytedance/deer-flow/pull/4839 [#4840]: https://github.com/bytedance/deer-flow/pull/4840 [#4842]: https://github.com/bytedance/deer-flow/pull/4842 +[#4843]: https://github.com/bytedance/deer-flow/pull/4843 [#4844]: https://github.com/bytedance/deer-flow/pull/4844 [#4846]: https://github.com/bytedance/deer-flow/pull/4846 [#4848]: https://github.com/bytedance/deer-flow/pull/4848 @@ -6306,3 +6314,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子 [#6112]: https://github.com/bytedance/deer-flow/pull/6112 [#6132]: https://github.com/bytedance/deer-flow/pull/6132 [#6134]: https://github.com/bytedance/deer-flow/pull/6134 +[#6135]: https://github.com/bytedance/deer-flow/pull/6135 diff --git a/README.md b/README.md index fc24188edb..f3bff36458 100644 --- a/README.md +++ b/README.md @@ -2241,6 +2241,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`, push a summary to the task owner's connected IM identities when a scheduled run finishes as success or failed (outbox + delivery worker). Manual "run now" and interrupts stay silent, and so do occurrences that end without a finished run (launch error, queue timeout, restart recovery). 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`. Proactive push is currently implemented for WeCom; other connected providers enqueue but fail until they grow a `send_notification` path. - 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. **Filter execution history through the API** @@ -2253,7 +2254,7 @@ Current MVP limits: - No conversation-created `schedule_task` tool yet - No text-only notification jobs -- No channel or GitHub dispatch targets +- No channel or GitHub dispatch targets (result push above is not a dispatch target) Enable background polling with `config.yaml -> scheduler.enabled`. Manual trigger uses the same scheduled-task resource and execution path. diff --git a/README_zh.md b/README_zh.md index ac57847627..22ceb3b0fd 100644 --- a/README_zh.md +++ b/README_zh.md @@ -921,6 +921,7 @@ DeerFlow 现在在 workspace 里内置了一个一等的定时任务(scheduled - 当某次执行处于 `queued`、`launching` 或 `running` 时冻结任务定义,避免持久化的执行意外换用新的 prompt、thread 或调度;将任务切换为暂停或删除任务会取消已在等待的执行,而 `launching`/`running` 执行结束后才能重试这些变更;显式手动触发在调度已暂停时仍可等待并执行,且不会自动恢复调度 - 支持暂停、恢复、手动触发、查看历史和删除任务 - 定时任务通过正常的 DeerFlow run 生命周期执行 +- 当 `channel_connections.enabled: true` 时,定时运行以成功或失败结束时,会把摘要推送到任务所有者已绑定的 IM 身份(outbox + 投递 worker)。手动「立即运行」和中断不推送;没有跑完一次运行就结束的执行记录(启动失败、排队超时、重启恢复)也不推送。渠道/传输宕机时会停车且不耗尽重试,最长约一天;平台拒绝大约重试 15 分钟后落为 `failed`。等待期间被你解绑的身份不会再收到推送:该条投递直接落为 `failed`。目前主动推送由企业微信实现;其它已绑定渠道会入队,但在实现 `send_notification` 前会失败。 - 按每页 50 条浏览执行历史;历史页暂停自动刷新,可随时返回最新记录。 仅在读取成功后显示条数,加载中或失败不会误显示为零条。 **通过 API 筛选执行历史** @@ -933,7 +934,7 @@ DeerFlow 现在在 workspace 里内置了一个一等的定时任务(scheduled - 暂时还没有可在对话中创建任务的 `schedule_task` 工具 - 没有纯文本通知任务 -- 没有渠道或 GitHub 分发目标 +- 没有渠道或 GitHub 分发目标(上面的结果推送不是分发目标) 通过 `config.yaml -> scheduler.enabled` 开启后台轮询。手动触发使用同样的 scheduled-task 资源和执行路径。 diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 35589d6f1d..01437f9f01 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -10,7 +10,8 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk - `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` A swallowed streaming failure publishes its final outbound before releasing the inbound dedupe key, so a provider redelivery can retry without overtaking the terminal reply. **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()`: 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`. +- `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()`: 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`): used by the scheduled-task notification outbox to push markdown to a bound external identity with no inbound frame. The base default raises `NotImplementedError` (after a `ChannelUnavailable` check when not running) so unsupported providers fail visibly in the outbox. Connectivity failures must raise `ChannelUnavailable` so the delivery worker parks without consuming retries; platform rejections of a well-formed send remain ordinary exceptions. **Only WeCom implements proactive push today** (`wecom.py` via `WSClient.send_message`); the SDK raises `RuntimeError` from that call for a dead socket and for a platform 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 via `resolve_connections` (`ChannelConnectionRepository.list_connections`, owner-scoped, ahead of the channel-liveness check): a target no longer `connected` for the row's owner is finalized with `mark_failed(terminal=True)`; a lookup failure parks. Terminal outbox rows (`sent`/`failed`) ignore later status writes. Scheduler side (`app/scheduler/`): `handle_run_completion` enqueues only after `complete_run` recorded a success or failed outcome; manual triggers, interrupts and recovery-finalized occurrences stay silent. Gateway lifespan wires the enqueue side with the scheduler, then starts `NotificationDeliveryWorker` after the channel service (the scheduler starts first), detaching the enqueue side if delivery cannot start; the worker stops before the channel service. +- `wecom.py` - WeCom WebSocket channel; implements `send_notification` for the outbox worker. Tracks `_ws_transport_up` across disconnect/reconnect so transport failures become `ChannelUnavailable` rather than burning the counted retry budget. - `service.py` - Manages lifecycle of all configured channels from `config.yaml`. Shutdown closes manager admission first and keeps transports alive until every manager worker/follow-up watcher has exited. A successful manager stop therefore owns no live handler; if the Gateway's outer timeout cancels shutdown, the service retains its channel objects and global singleton so unfinished resources are not detached and cleanup can be retried. - `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` registers typing loops before inbound handling yields and `_start_typing()` refuses work once `_running` is false; 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`) - `buzz.py` - Buzz (Nostr relay) implementation: one NIP-42-authenticated websocket, pubkey-allowlist + mention/DM/thread-follow gating, streaming replies via in-place kind-40003 edits; requires the `buzz` dependency extra. Its durable seen-event replay guard uses async `aseen()` / `arecord()` / `aflush()` boundaries: initial JSON reads and coalesced atomic writes run off the Gateway event loop, and `stop()` awaits the final flush before returning. **Subscription model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-subscription-model)): the relay fans kind-9 chat events out **only** to channel-scoped subscriptions, proved against a live relay — `REQ {"kinds":[9]}` is accepted and answered with `EOSE` but never receives an event (the connector authenticated and then sat silent forever), `REQ {"kinds":[9],"#h":[uuid]}` works, and a multi-value `#h` matches nothing, so it is strictly one REQ per channel (the same shape as Buzz's own `buzz-acp` harness). Every connection therefore rebuilds three kinds of subscription after NIP-42 auth: `buzz-discovery` (`{"kinds":[39000]}`, a historical query returning exactly the channels this identity belongs to, one stored event each then `EOSE` — adding `#p` returns zero, do not "narrow" it), `buzz-membership` (`{"kinds":[44100,44101],"#p":[us],"since":}`, the relay-signed member-added/member-removed notifications whose `p` tag names the affected member and `h` tag the channel), and one `buzz-chat-` per discovered channel. Chat subscriptions open as each kind-39000 arrives; the discovery `EOSE` is the completeness barrier that retries any that failed and warns when discovery found nothing. A kind-44100 for our pubkey subscribes to the new channel **live** (then re-issues discovery so its name/type reach the DM-detection cache, but only when that channel's metadata is actually missing — an unconditional refresh is how a burst of 44100s multiplied into one discovery pass each); a kind-44101 issues `buzz_nostr.close_frame` for exactly that channel's subscription and drops its metadata. **The membership subscription is scoped to LIVE events** and this is load-bearing: buzz-relay *stores* 44100/44101 and serves history newest-first (default limit 2000), so an unscoped filter replayed the whole membership history on every connect — every stored add read as live, re-running discovery once each (M+1 discovery passes × N stored kind-39000 events per connect, observed live as two `channel discovery complete` lines and one channel logged `` because the 44100 path subscribed before its metadata arrived), re-subscribing channels we have since been removed from, and letting a stored 44101 transiently unsubscribe a channel we are still in. `since` is anchored at the moment the socket opened (`_session_started_at`) minus `MEMBERSHIP_LOOKBACK_SECONDS` (60) of slack, which covers both relay clock skew and a membership change published *during* the connect/auth handshake; the slack can only cost an idempotent replay of the last minute. **A relay `CLOSED` frame is recovered, not merely forgotten** — every subscription on the socket fails silently when dropped, so `_handle_closed` re-issues it, bounded by `MAX_RESUBSCRIBE_ATTEMPTS` (3) per subscription id per connection *and per auth epoch* (`_resubscribe_attempts` is reset on session start, on re-auth, and by `stop()`, so pre-auth rejections — which the auth branch already recovers wholesale — never spend the authenticated session's budget). The first retry is immediate (the common case is a one-off hiccup); later ones back off 1s then 2s, awaited inline in the read loop rather than in a background task that could outlive its own socket. **`auth-required:` before this socket has completed its NIP-42 handshake is the expected bootstrap sequence, not a refusal**, and `_handle_closed` short-circuits it ahead of every recovery path: the connector opens its control REQs immediately in case the relay serves unauthenticated reads, a closed relay answers `auth-required:` plus an `AUTH` challenge, and the auth branch re-opens everything. That case is logged at DEBUG and consumes neither the permanent-refusal branch nor the retry budget (a chat subscription is still dropped from `_chat_subscriptions`, since it genuinely is not subscribed and discovery is what re-opens it). Treating it as permanent produced an operator-facing warning claiming discovery/membership tracking was DOWN in the same run where discovery then completed, every channel was subscribed, and a brand-new channel's kind-44100 was picked up one second later. The boundary is the per-socket `_auth_completed` flag, set once the signed AUTH event has been sent and cleared on session entry, session exit, and `stop()`; the same reason *after* that stays loud, and says the subscription is down until the relay's next AUTH challenge or the next reconnect rather than borrowing the non-auth wording. Then `_is_transient_close` decides whether to retry at all: NIP-01/NIP-42 `auth-required:`/`restricted:`/`blocked:`/`mute:`/`invalid:`/`pow:` prefixes and buzz-relay's own removal/revocation prose are permanent (do not fight the relay over a channel that is no longer ours; a post-auth `auth-required:` is recovered by the AUTH branch, not by re-issuing), `rate-limited:`/`error:`/no-reason-at-all and anything unrecognized are transient — the default resolves toward "keep listening" because going silently deaf is the failure this exists to remove, and the attempt budget bounds a wrong guess. A chat `CLOSED` is only ever recovered for a channel already in `_chat_subscriptions`: a `CLOSED` is relay-supplied, so acting on an unknown one would let a relay induce a subscription just by naming a channel. Every subscription that goes unlistened is logged at WARNING, never INFO. Subscription ids are deterministic per channel precisely so one can be replaced or closed without disturbing the others on the socket, and `_chat_subscriptions` is per-socket state cleared on session end, on re-auth (a pre-auth REQ may have been rejected), and by `stop()`. Three bounds on remote-fed state: `MAX_CACHED_CHANNELS` (512) caps the kind-39000 metadata cache, the watermark map, and the resubscribe-attempt map; `MAX_CHANNEL_SUBSCRIPTIONS` (256, well under buzz-relay's own 1024-per-connection ceiling) caps live chat subscriptions — at the cap new channels are refused and named in a warning rather than evicting a working subscription. **Known bound (documented, not fixed):** the relay caps historical delivery at 2000 events per subscription, newest-first, even with a `since`, so >2000 unread messages in a *single* channel across a disconnect loses the oldest — the relay never sends them and the watermark advances past them. That is the one remaining path that can skip; everything else fails toward replay. **Trust model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-trust-model)): every inbound `EVENT` is authenticated at the single `handle_relay_frame` choke point — the NIP-01 id is recomputed from the delivered payload and the BIP-340 Schnorr signature verified against the claimed `pubkey` (`buzz_nostr.verify_event`, pure and total: malformed input returns `False`, never raises) — so `ev["pubkey"]`, the authorization principal for both the allowlist and the `/connect` bind, cannot be forged by a relay the DeerFlow operator does not run. What remains trusted is the *authorship* of kind-39000 channel metadata: any member can sign one, and because per-channel subscriptions are now driven by discovery, a forged kind-39000 has two effects rather than one — it can mark a channel `type: "dm"` (relaxing `require_mention` for that channel) **and** it can induce a chat subscription for a channel of the forger's choosing, since the channels we listen to are exactly the channels we hold metadata for. Neither makes anything be *acted on*: `allowed_users` and per-event signature verification are independent gates, so an induced subscription only means the relay reads its own traffic back to a subscriber that drops it, bounded by `MAX_CHANNEL_SUBSCRIPTIONS` (which refuses rather than evicts, so it cannot displace a real channel). Same for a forged kind-44100, except its `p` tag is re-checked locally so it must at least name us. Closing this needs a configured trusted relay pubkey, which `relay_url` is not. `allowed_users` is deny-by-default (empty = nobody, unlike siblings' empty = everyone), so `start()` logs a WARNING when it is empty and each drop logs at DEBUG. The resubscribe cursor (`since`) is **per channel**, advances only for events that were actually processed, and never past `now + MAX_FUTURE_SKEW_SECONDS`, because it is peer-supplied (`created_at`) and a single future-dated event otherwise made the connector permanently deaf. Per channel rather than global is the safety-critical half: subscriptions are per channel, so one shared cursor is the newest event seen in *any* channel and a busy channel would drag it past a quiet channel's unread messages, skipping them on the next reconnect — measured on a live relay, three channels of one identity sat ~28h apart. Per-channel cursors can only ever cost duplicate delivery (absorbed by the manager's `event_id` dedupe), and an evicted cursor degrades to "no `since`", i.e. the relay's default backlog — both fail toward replay, never toward a miss. Streaming tracks every oversize chunk index (`_stream_targets` for chunk 0, `_stream_tails` for the rest), since the manager republishes cumulative text and reposting `chunks[1:]` per update flooded the channel; all of it is per-connection state cleared by `stop()`, and the remote-fed kind-39000 cache is capped at `MAX_CACHED_CHANNELS`. **`send()` refuses outright to publish text carrying a hidden model-context wrapper** (``, ``, `` — `_HIDDEN_CONTEXT_MARKERS`), logging at ERROR and clearing the stream bookkeeping on a blocked `is_final`. This is defense in depth behind the manager's allowlist, and it lives here rather than in a sibling connector because on Buzz a leak is permanent: every streaming update is an immutable public Nostr event, so a corrective edit only changes what clients render while the original leaked event stays on the relay. Matching is on the literal opening tag, so a reply that merely talks about memory is still published diff --git a/backend/app/channels/base.py b/backend/app/channels/base.py index 58e633cbd5..3b6b6b6e51 100644 --- a/backend/app/channels/base.py +++ b/backend/app/channels/base.py @@ -30,6 +30,15 @@ logger = logging.getLogger(__name__) T = TypeVar("T") +class ChannelUnavailable(Exception): + """Local connectivity failure: the channel cannot accept traffic right now. + + Distinct from a platform rejection of a well-formed send. Delivery workers + park outbox rows with ``count_attempt=False`` on this error so a transport + outage cannot exhaust the retry budget. + """ + + @dataclass(eq=False, slots=True) class _ThreadsafeSubmission: coroutine: Coroutine[Any, Any, Any] @@ -101,6 +110,25 @@ class Channel(ABC): """ return False + async def send_notification(self, *, target: str, text_markdown: str) -> None: + """Push a proactive notification to an external identity (issue #4254). + + Unlike ``send``, there is no inbound frame to reply to: *target* is a + platform identity (e.g. the ``external_account_id`` recorded by the + bind flow). Used by the notification delivery worker to deliver + scheduled-task outcomes. Channels without proactive push support keep + the default, which raises so the delivery outbox records a failure + instead of silently dropping the notification. + + Connectivity preconditions should raise :class:`ChannelUnavailable` + (not a generic ``RuntimeError``) so the outbox can park the row without + consuming its retry budget. Platform rejections of a well-formed send + remain ordinary exceptions that count as attempts. + """ + if not self.is_running: + raise ChannelUnavailable(f"channel '{self.name}' is not running") + raise NotImplementedError(f"channel '{self.name}' does not support proactive notifications") + # -- helpers ----------------------------------------------------------- async def _send_with_retry( diff --git a/backend/app/channels/wecom.py b/backend/app/channels/wecom.py index 154b232d40..b762c6f774 100644 --- a/backend/app/channels/wecom.py +++ b/backend/app/channels/wecom.py @@ -8,7 +8,7 @@ import logging from collections.abc import Awaitable, Callable from typing import Any, cast -from app.channels.base import Channel +from app.channels.base import Channel, ChannelUnavailable from app.channels.commands import is_known_channel_command from app.channels.connection_identity import attach_connection_identity from app.channels.message_bus import ( @@ -21,6 +21,23 @@ from app.channels.message_bus import ( logger = logging.getLogger(__name__) +# Exceptions that mean the transport failed, not the message. ``websockets`` +# is the SDK's socket layer and forwards its close exception unchanged +# through ``send_message``; it is not an OSError subclass. +_TRANSPORT_EXCEPTIONS: tuple[type[BaseException], ...] = (ConnectionError, TimeoutError, OSError) +try: + from websockets.exceptions import ConnectionClosed as _WsConnectionClosed +except Exception: # pragma: no cover - absent only when the SDK itself is missing + pass +else: + _TRANSPORT_EXCEPTIONS = (*_TRANSPORT_EXCEPTIONS, _WsConnectionClosed) + +# The SDK surfaces a platform rejection of a sent frame (non-zero ``errcode`` +# in the ACK) as ``RuntimeError`` with this prefix, from the same call that +# raises ``RuntimeError`` for a closed socket. The prefix is what tells a +# verdict about the message apart from a transport failure. +_WECOM_ACK_REJECTION_PREFIX = "Reply ack error" + def _file_md5(path: str) -> str: md5_hasher = hashlib.md5() @@ -102,6 +119,11 @@ class WeComChannel(Channel): self._lifecycle_lock = asyncio.Lock() self._ws_frames: dict[str, dict[str, Any]] = {} self._ws_stream_ids: dict[str, str] = {} + # Optimistic until the SDK reports a disconnect: start() hands us a + # live client that is connecting/reconnecting. Cleared on disconnect + # and restored after a successful send so proactive push can park + # without burning the outbox retry budget during a transport outage. + self._ws_transport_up = False self._ws_send_locks: dict[str, asyncio.Lock] = {} self._ws_send_lock_users: dict[str, int] = {} self._ws_send_locks_guard = asyncio.Lock() @@ -191,6 +213,7 @@ class WeComChannel(Channel): self._ws_task = asyncio.create_task(self._ws_client.connect()) self._ws_task.add_done_callback(self._on_ws_task_done) + self._ws_transport_up = True self._running = True self.bus.subscribe_outbound(self._on_outbound) logger.info("WeCom channel started") @@ -210,6 +233,7 @@ class WeComChannel(Channel): logger.error("WeCom WebSocket error: %s", error) def _on_ws_disconnected(self, *args: Any) -> None: + self._ws_transport_up = False detail = f" ({args[0]})" if args else "" logger.warning("WeCom WebSocket disconnected%s; SDK will attempt to reconnect", detail) @@ -246,6 +270,7 @@ class WeComChannel(Channel): async def stop(self) -> None: async with self._lifecycle_lock: self._running = False + self._ws_transport_up = False self.bus.unsubscribe_outbound(self._on_outbound) ws_client = self._ws_client ws_task = self._ws_task @@ -283,6 +308,79 @@ class WeComChannel(Channel): return logger.warning("[WeCom] send called but WebSocket client is not available") + async def send_notification(self, *, target: str, text_markdown: str) -> None: + """Proactively push markdown to a bound WeCom identity (issue #4254). + + Goes through ``WSClient.send_message`` (no inbound frame required) + and waits for the platform ACK, so an ACK-less call is never treated + as success. + + Transport / not-connected failures raise :class:`ChannelUnavailable` + so the outbox parks without consuming retries. Platform rejections + count against the budget: the SDK reports a non-zero ``errcode`` as + ``RuntimeError("Reply ack error: ...")`` from ``send_message`` itself, + so they are told apart from its transport ``RuntimeError`` by that + prefix. The ``errcode`` check on a returned ACK dict below is kept as + a guard for an SDK that returns the frame instead of raising. + """ + if not self._running or not self._ws_client: + raise ChannelUnavailable("WeCom channel is not connected") + # Ask the SDK first when it already knows the socket is closed: that + # fails fast and skips the retry/sleep storm. The except branch below + # still parks a closed socket the SDK only discovers during the send. + if getattr(self._ws_client, "is_connected", True) is False: + self._ws_transport_up = False + raise ChannelUnavailable("WeCom websocket is not connected") + body = {"msgtype": "markdown", "markdown": {"content": text_markdown}} + + async def _send_once() -> Any: + return await self._ws_client.send_message(target, body) + + try: + if self._ws_transport_up: + ack = await self._send_with_retry( + _send_once, + max_retries=3, + log_prefix="[WeCom]", + operation_name="notification send", + ) + else: + # Known disconnect: one probe so a silent SDK reconnect can + # recover, without the 3× retry/sleep storm burning shutdown + # and poll time while the socket is still down. + ack = await _send_once() + except ChannelUnavailable: + raise + except RuntimeError as exc: + if str(exc).startswith(_WECOM_ACK_REJECTION_PREFIX): + # The platform answered and said no (bad chatid, user out of + # range, ...). The frame went out and the ACK came back on + # this socket, so the transport is up; the rejection itself + # counts like any other deterministic failure. + self._ws_transport_up = True + raise + # Everything else the SDK raises as RuntimeError is transport-side: + # a socket that is not open, a connection dropped while an ACK was + # pending. Park the outbox row until the SDK reconnects instead of + # charging the delivery a counted attempt per poll for a fault + # that is not its own. + self._ws_transport_up = False + raise ChannelUnavailable(f"WeCom transport unavailable: {exc}") from exc + except _TRANSPORT_EXCEPTIONS as exc: + # Dead/flapping transport, including the socket layer's own close + # exception, which the SDK forwards unchanged. Same treatment. + self._ws_transport_up = False + raise ChannelUnavailable(f"WeCom transport unavailable: {exc}") from exc + errcode = None + if isinstance(ack, dict): + errcode = ack.get("errcode") + ack_body = ack.get("body") + if errcode is None and isinstance(ack_body, dict): + errcode = ack_body.get("errcode") + if errcode not in (None, 0): + raise RuntimeError(f"WeCom send_message rejected with errcode={errcode}") + self._ws_transport_up = True + async def _on_outbound(self, msg: OutboundMessage) -> None: if msg.channel_name != self.name: return diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index af5c27fb8f..ce6e48f576 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -393,6 +393,77 @@ async def _shutdown_startup_trash_sweep(app: FastAPI) -> None: logger.exception("Startup trash sweep failed during shutdown") +def _scheduled_task_notification_repos(startup_config: AppConfig): + """Return ``(connection_repo, notification_repo)`` for the scheduled-run outbox (issue #4254). + + Both are None unless channel connections are enabled, because the completion + hook resolves the task owner's bound IM identities through them. + """ + connection_config = getattr(startup_config, "channel_connections", None) + if connection_config is None or not getattr(connection_config, "enabled", False): + return None, None + + from deerflow.persistence.channel_connections import ChannelConnectionRepository + from deerflow.persistence.engine import get_session_factory + from deerflow.persistence.notification_deliveries import NotificationDeliveryRepository + + session_factory = get_session_factory() + if session_factory is None: + return None, None + return ChannelConnectionRepository(session_factory), NotificationDeliveryRepository(session_factory) + + +async def _start_scheduled_task_notification_delivery(app: FastAPI, startup_config: AppConfig, notification_repo, connection_repo) -> None: + """Start the delivery side of the scheduled-run outbox, or switch its enqueue side off. + + The scheduler only enqueues; this worker polls due rows and pushes them through + the owning channel's proactive send path. Enqueue and delivery stay active + together: writing outbox rows with no worker or channel service would leave a + silent backlog, so the enqueue side is detached when delivery cannot start. + """ + scheduled_task_service = getattr(app.state, "scheduled_task_service", None) + if scheduled_task_service is None or notification_repo is None: + return + + from app.channels.service import get_channel_service + + channel_service = get_channel_service() + if channel_service is None: + logger.warning("channel_connections.enabled but no channel service is running; disabling scheduled-task IM notification enqueue to avoid a write-only outbox") + scheduled_task_service.detach_notification_outbox() + return + + try: + from app.scheduler.notification_delivery import NotificationDeliveryWorker + from deerflow.persistence.engine import get_session_factory + from deerflow.persistence.run import RunRepository + + run_repo = RunRepository(get_session_factory()) + + async def resolve_run_summary(run_id: str, user_id: str | None) -> str | None: + # Scoped to the outbox row's owner so a stale row + # can never pull another user's run content. + row = await run_repo.get(run_id, user_id=user_id) + return row.get("last_ai_message") if row else None + + worker = NotificationDeliveryWorker( + delivery_repo=notification_repo, + resolve_channel=channel_service.get_channel, + resolve_run_summary=resolve_run_summary, + # Delivery can run up to a day after enqueue; the worker re-checks + # that the target is still one of the owner's connected identities. + # Both repositories come from the same session factory, so the + # connection repository is present whenever the outbox one is. + resolve_connections=connection_repo.list_connections, + poll_interval_seconds=startup_config.scheduler.poll_interval_seconds, + ) + await worker.start() + except Exception: + scheduled_task_service.detach_notification_outbox() + raise + app.state.notification_delivery_worker = worker + + @asynccontextmanager async def _runtime_with_mcp_pool_shutdown(app: FastAPI, startup_config: AppConfig) -> AsyncGenerator[None, None]: """Close pooled MCP transports after runtime producers have stopped.""" @@ -546,6 +617,15 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: # already-started file worker drains before the runtime is torn down. app.state.startup_trash_sweep_task = asyncio.create_task(_run_startup_trash_sweep(app, startup_config)) + # Enqueue side of the scheduled-run notification outbox (issue #4254). + # It only needs the durable table, so it is wired with the scheduler; + # the delivery worker starts after the channel service, further down. + notification_connection_repo = scheduled_notification_repo = None + try: + notification_connection_repo, scheduled_notification_repo = _scheduled_task_notification_repos(startup_config) + except Exception: + logger.exception("Failed to prepare the scheduled-task notification outbox") + try: from app.gateway.services import launch_scheduled_thread_run from app.scheduler import ScheduledTaskService @@ -561,6 +641,8 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: queue_timeout_seconds=startup_config.scheduler.queue_timeout_seconds, multi_instance=startup_config.scheduler.multi_instance, run_lease_grace_seconds=startup_config.run_ownership.grace_seconds, + connection_repo=notification_connection_repo, + notification_repo=scheduled_notification_repo, ) app.state.scheduled_task_service = scheduled_task_service if startup_config.scheduler.enabled: @@ -595,6 +677,14 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: except Exception: logger.exception("No IM channels configured or channel service failed to start") + # Delivery side of the scheduled-run notification outbox (issue #4254). + # It starts here rather than next to the scheduler: the scheduler starts + # before the channel service, and delivery needs that service running. + try: + await _start_scheduled_task_notification_delivery(app, startup_config, scheduled_notification_repo, notification_connection_repo) + except Exception: + logger.exception("Failed to start scheduled-task IM notification delivery") + from app.gateway.services import launch_mcp_task_notification_run from app.mcp_tasks import McpTaskService from deerflow.config.extensions_config import ExtensionsConfig @@ -683,6 +773,23 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: except Exception: logger.exception("Failed to close OIDC service") + # Stop the notification delivery worker BEFORE the channel service: it + # sends through running channels, so letting channels die first would + # turn every in-flight delivery into a spurious failure/retry. + if getattr(app.state, "notification_delivery_worker", None) is not None: + try: + await asyncio.wait_for( + app.state.notification_delivery_worker.stop(), + timeout=_SHUTDOWN_HOOK_TIMEOUT_SECONDS, + ) + except TimeoutError: + logger.warning( + "Notification delivery worker shutdown exceeded %.1fs; proceeding with worker exit.", + _SHUTDOWN_HOOK_TIMEOUT_SECONDS, + ) + except Exception: + logger.exception("Failed to stop notification delivery worker") + # Stop channel service on shutdown (bounded to prevent worker hang) try: from app.channels.service import stop_channel_service diff --git a/backend/app/scheduler/notification_delivery.py b/backend/app/scheduler/notification_delivery.py new file mode 100644 index 0000000000..8bfe0bbb2d --- /dev/null +++ b/backend/app/scheduler/notification_delivery.py @@ -0,0 +1,334 @@ +"""Outbox delivery worker for scheduled-task IM notifications (issue #4254). + +The completion hook (``ScheduledTaskService._enqueue_run_notifications``) +only writes durable outbox rows; this worker owns the actual IM send. +Execution state and delivery state stay separated: the worker moves outbox +rows through ``pending -> sending -> sent|failed`` and never touches run or +task rows. Retries with backoff live in the repository (``mark_failed``); +the worker just claims what is due and reports outcomes. +""" + +from __future__ import annotations + +import asyncio +import inspect +import logging +import re +from collections.abc import Callable +from datetime import UTC, datetime, timedelta +from typing import Any + +from app.channels.base import ChannelUnavailable + +logger = logging.getLogger(__name__) + +# Resolves a provider name (e.g. "wecom") to a running Channel instance, or +# None when the channel is not configured/running. Sync or async callables +# are both accepted (``ChannelService.get_channel`` is sync). +ChannelResolver = Callable[[str], Any] + +# Resolves the run's final answer for a completed run, keyed +# ``(run_id, owner_user_id)``, or None when no summary is available. Sync or +# async callables are both accepted; failures fall back to the skeleton text. +SummaryResolver = Callable[[str, str | None], Any] + +# Lists an owner's channel connections (``ChannelConnectionRepository +# .list_connections``), each a dict with ``provider``, ``external_account_id`` +# and ``status``. Sync or async callables are both accepted. The worker uses +# it to confirm, at send time, that the delivery's target is still bound. +ConnectionResolver = Callable[[str], Any] + +_ERROR_TEXT_LIMIT = 500 +_SUMMARY_TEXT_LIMIT = 1000 + +# A claimed row stuck in "sending" longer than this is considered orphaned +# (process died between claim and the final status write) and flipped back +# to pending. Generous on purpose: a healthy send plus status write finishes +# in seconds, and resetting too eagerly would double-send live deliveries. +_STALE_SENDING_TIMEOUT_SECONDS = 600 + +# Matches Gateway ``_SHUTDOWN_HOOK_TIMEOUT_SECONDS``: this worker sits in +# front of external IM I/O, so an unbounded join would stall the whole +# lifespan shutdown path the channel-service bound was added to protect. +_STOP_TIMEOUT_SECONDS = 5.0 + + +def _truncate(text: str, *, limit: int = _ERROR_TEXT_LIMIT) -> str: + return text if len(text) <= limit else text[: limit - 1] + "\u2026" + + +_EGRESS_REDACTION = "[redacted]" +# Best-effort scrub before run content leaves for a third-party IM platform. +# Patterns mirror the high-confidence secret detectors in skillscan. +_EGRESS_SECRET_PATTERNS: tuple[re.Pattern[str], ...] = ( + re.compile( + r"-----BEGIN [A-Z0-9 ]*PRIVATE KEY-----.*?-----END [A-Z0-9 ]*PRIVATE KEY-----", + re.DOTALL, + ), + re.compile(r"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b"), + re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}\b"), + re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{20,}\b"), + re.compile(r"\bsk-[A-Za-z0-9]{20,}\b"), + re.compile(r"(?im)\b(token|password|passwd|api[_-]?key|secret|credential)s?\b\s*[:=]\s*[\"']?([^\"'\s#]+)"), +) + + +def redact_egress_text(text: str) -> str: + """Scrub likely secrets from text before it is pushed to external IM.""" + if not text: + return text + redacted = text + for pattern in _EGRESS_SECRET_PATTERNS: + if pattern.groups: + redacted = pattern.sub(lambda match: f"{match.group(1)}={_EGRESS_REDACTION}", redacted) + else: + redacted = pattern.sub(_EGRESS_REDACTION, redacted) + return redacted + + +def render_notification_text(delivery: dict[str, Any]) -> str: + """Render a bounded markdown summary of the run outcome for IM push.""" + payload = delivery.get("payload") or {} + event = delivery.get("event") or "" + task_label = payload.get("task_title") or payload.get("task_id") or delivery.get("task_id") + if event == "run_failed": + lines = ["**Scheduled task failed**", f"Task: `{task_label}`"] + # Do not forward raw error text to external IM: scheduled runs can + # surface hostnames, paths, and token fragments in tracebacks. Users + # can open the workspace for the full detail. + lines.append("See the DeerFlow workspace for error details.") + else: + lines = ["**Scheduled task completed**", f"Task: `{task_label}`"] + # The result summary belongs to a successful outcome only: on failed + # runs a partial answer would be misleading, so the error line above + # stays the sole detail. + summary = payload.get("result_summary") + if isinstance(summary, str) and summary.strip(): + safe_summary = redact_egress_text(summary.strip()) + lines.append(f"Result: {_truncate(safe_summary, limit=_SUMMARY_TEXT_LIMIT)}") + if delivery.get("run_id"): + lines.append(f"Run: `{delivery['run_id']}`") + return "\n".join(lines) + + +class NotificationDeliveryWorker: + """Polls the notification outbox and pushes due deliveries over IM.""" + + def __init__( + self, + *, + delivery_repo, + resolve_channel: ChannelResolver, + poll_interval_seconds: int = 5, + batch_size: int = 10, + resolve_run_summary: SummaryResolver | None = None, + resolve_connections: ConnectionResolver | None = None, + stale_sending_timeout_seconds: int = _STALE_SENDING_TIMEOUT_SECONDS, + stop_timeout_seconds: float = _STOP_TIMEOUT_SECONDS, + ) -> None: + self._delivery_repo = delivery_repo + self._resolve_channel = resolve_channel + self._resolve_run_summary = resolve_run_summary + self._resolve_connections = resolve_connections + self._poll_interval_seconds = poll_interval_seconds + self._batch_size = batch_size + self._stale_sending_timeout_seconds = stale_sending_timeout_seconds + self._stop_timeout_seconds = stop_timeout_seconds + self._task: asyncio.Task | None = None + self._stop = asyncio.Event() + + async def start(self) -> None: + if self._task is not None: + return + self._stop.clear() + self._task = asyncio.create_task(self._run_loop()) + + async def stop(self) -> None: + if self._task is None: + return + self._stop.set() + task = self._task + self._task = None + try: + await asyncio.wait_for(task, timeout=self._stop_timeout_seconds) + except TimeoutError: + logger.warning( + "Notification delivery worker stop exceeded %.1fs; cancelling in-flight poll", + self._stop_timeout_seconds, + ) + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + async def run_once(self, *, now: datetime) -> None: + await self._recover_stale_sending(now) + rows = await self._delivery_repo.claim_due_deliveries(now=now, limit=self._batch_size) + for row in rows: + try: + await self._deliver(row) + except Exception: + # One poisoned row must not abort the loop and strand the + # rest of the claimed batch in "sending". Best-effort + # mark_failed; if even that raises, the stale reset above + # reclaims the row on a later poll. + logger.exception("Notification delivery %s crashed; isolating from the rest of the batch", row.get("id")) + try: + await self._delivery_repo.mark_failed(row["id"], error="delivery crashed before completion") + except Exception: + logger.warning("Could not mark crashed delivery %s as failed; stale reset will recover it", row.get("id"), exc_info=True) + + async def _recover_stale_sending(self, now: datetime) -> None: + """Lease-style reconciliation, run before every claim (the first poll + after startup is covered too). Mirrors how the scheduler recovers + stale active runs.""" + try: + reset = await self._delivery_repo.reset_stale_sending_rows( + now=now, + timeout=timedelta(seconds=self._stale_sending_timeout_seconds), + ) + if reset: + logger.info("Recovered %s notification deliveries stuck in 'sending'", reset) + except Exception: + logger.warning("Failed to reset stale sending rows; retrying next poll", exc_info=True) + + async def _resolve_summary(self, delivery: dict[str, Any]) -> str | None: + """Best-effort lookup of the run's final answer at delivery time. + + Read at delivery time, not enqueue time, so retried deliveries see + the freshest value and the outbox payload stays skeleton-only. Any + failure degrades to the skeleton notification instead of blocking it. + """ + if self._resolve_run_summary is None: + return None + run_id = delivery.get("run_id") + if not run_id: + return None + try: + summary = self._resolve_run_summary(run_id, delivery.get("owner_user_id")) + if inspect.isawaitable(summary): + summary = await summary + except Exception: + logger.warning("Failed to resolve run summary for run %s; sending skeleton notification", run_id, exc_info=True) + return None + if isinstance(summary, str) and summary.strip(): + return summary + return None + + async def _target_still_connected(self, delivery: dict[str, Any]) -> bool | None: + """Return whether the delivery's ``(provider, target)`` is still a + ``connected`` identity of its owner, or None when that could not be + determined (no resolver wired, or the lookup failed). + + The outbox row was written while the identity was connected, but + delivery can run up to a day later (counted retries, then parking), + and ``disconnect_connection`` only flips the connection row. The + lookup is scoped to the row's own ``owner_user_id`` so it cannot + match another owner's binding of the same external account. + """ + if self._resolve_connections is None: + return None + owner_user_id = delivery.get("owner_user_id") + provider = delivery.get("provider") + target = delivery.get("target") + if not owner_user_id or not provider or not target: + return None + try: + connections = self._resolve_connections(owner_user_id) + if inspect.isawaitable(connections): + connections = await connections + except Exception: + logger.warning("Failed to check channel connections for delivery %s; parking it", delivery.get("id"), exc_info=True) + return None + if connections is None: + connections = [] + if not isinstance(connections, (list, tuple)): + # An unexpected shape is a lookup problem, not evidence that the + # binding is gone: iterating a dict or a str would "find" nothing + # and drop the row for good. + logger.warning("Channel connection lookup for delivery %s returned %s, not a list; parking it", delivery.get("id"), type(connections).__name__) + return None + for connection in connections: + if not isinstance(connection, dict): + continue + # The resolver is called with the row's owner; a row that names a + # different owner is skipped as well, so a resolver that is not + # owner-scoped still cannot make another user's binding count. + row_owner = connection.get("owner_user_id") + if row_owner is not None and row_owner != owner_user_id: + continue + if connection.get("provider") == provider and connection.get("external_account_id") == target and connection.get("status") == "connected": + return True + return False + + async def _deliver(self, delivery: dict[str, Any]) -> None: + delivery_id = delivery["id"] + provider = delivery.get("provider") or "" + if self._resolve_connections is not None: + # 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 + # sit in parking for a day either when the channel is also down + # (disconnecting a provider revokes its connections and stops its + # channel together). A lookup failure parks instead, so a + # transient store error neither drops nor sends the row. + still_connected = await self._target_still_connected(delivery) + if still_connected is False: + await self._delivery_repo.mark_failed( + delivery_id, + error=f"target is no longer a connected {provider} identity of its owner", + terminal=True, + ) + return + if still_connected is None: + await self._delivery_repo.mark_failed(delivery_id, error="could not verify the target's channel connection", count_attempt=False) + return + # Re-check channel liveness at delivery time, not enqueue time: the + # channel may have been disabled or disconnected after the outbox row + # was written, and the row stays retryable for when it comes back. + channel = self._resolve_channel(provider) + if inspect.isawaitable(channel): + channel = await channel + if channel is None or not getattr(channel, "is_running", True): + # Channel outage is not the delivery's fault: park the row + # without consuming its retry budget so it survives an + # hours-long outage and delivers once the channel returns. + await self._delivery_repo.mark_failed(delivery_id, error=f"channel '{provider}' is not running", count_attempt=False) + return + enriched = delivery + if delivery.get("event") == "run_completed": + summary = await self._resolve_summary(delivery) + if summary is not None: + # Copy before enriching: the claimed row must not be mutated + # in place (its payload is the durable outbox snapshot). + enriched = dict(delivery) + enriched["payload"] = {**(delivery.get("payload") or {}), "result_summary": summary} + try: + await channel.send_notification( + target=delivery.get("target") or "", + text_markdown=render_notification_text(enriched), + ) + except ChannelUnavailable as exc: + await self._delivery_repo.mark_failed(delivery_id, error=str(exc), count_attempt=False) + return + except Exception as exc: + await self._delivery_repo.mark_failed(delivery_id, error=str(exc)) + return + await self._delivery_repo.mark_sent(delivery_id) + + async def _run_loop(self) -> None: + while not self._stop.is_set(): + try: + await self.run_once(now=datetime.now(UTC)) + except Exception: + # A transient DB error must not kill the poller task for the + # rest of the process life (same policy as the scheduler). + logger.exception("Notification delivery poll failed; retrying next interval") + try: + await asyncio.wait_for( + self._stop.wait(), + timeout=self._poll_interval_seconds, + ) + except TimeoutError: + continue diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py index cc7905d892..e4eb40d018 100644 --- a/backend/app/scheduler/service.py +++ b/backend/app/scheduler/service.py @@ -38,6 +38,8 @@ class ScheduledTaskService: queue_timeout_seconds: int = 3600, multi_instance: bool = False, run_lease_grace_seconds: int = 10, + connection_repo=None, + notification_repo=None, ) -> None: self._task_repo = task_repo self._task_run_repo = task_run_repo @@ -48,11 +50,25 @@ class ScheduledTaskService: self._queue_timeout_seconds = queue_timeout_seconds self._multi_instance = multi_instance self._run_lease_grace_seconds = run_lease_grace_seconds + # Notification outbox wiring (issue #4254): both repos must be present + # for the feature to be active; the completion hook only enqueues, the + # delivery worker sends. Either being None keeps legacy behavior. + self._connection_repo = connection_repo + self._notification_repo = notification_repo self._lease_owner = f"{socket.gethostname()}:{uuid.uuid4().hex}" self._task: asyncio.Task | None = None self._stop = asyncio.Event() self._skip_next_lease_reconciliation = False + def detach_notification_outbox(self) -> None: + """Stop enqueueing run notifications (issue #4254). + + The Gateway calls this when the delivery side cannot start, so the outbox + does not fill with rows nothing will send. + """ + self._connection_repo = None + self._notification_repo = None + async def run_once(self, *, now: datetime) -> None: if self._multi_instance: if self._skip_next_lease_reconciliation: @@ -540,7 +556,7 @@ class ScheduledTaskService: if terminal_status is None: return - await self._task_repo.complete_run( + completed = await self._task_repo.complete_run( task_id, user_id=user_id, task_run_id=task_run_id, @@ -549,6 +565,102 @@ class ScheduledTaskService: error=error, finished_at=datetime.now(UTC), ) + if not completed: + # The occurrence was not this run's to complete (missing row, another + # run, another owner): nothing was recorded, so nothing is announced. + return + + if metadata.get("scheduled_trigger") == "manual": + # A manual "run now" happens with the user watching the UI; the + # IM push would only echo what they already see. Missing + # metadata fails safe towards notifying. + return + + if self._notification_repo is None or self._connection_repo is None: + return + # complete_run commits the outcome atomically without returning the task, + # so the title is read afterwards. A task deleted meanwhile stays silent. + # A failed read must not shadow the committed outcome or drop the push: + # the title is optional and the message falls back to the task id. + task_title = None + try: + task = await self._task_repo.get(task_id, user_id=user_id) + except Exception: + logger.exception("[Scheduler] failed to load task %s for notifications", task_id) + else: + if task is None: + return + task_title = task.get("title") + + await self._enqueue_run_notifications( + task_id=task_id, + task_run_id=task_run_id, + run_id=record.run_id, + user_id=user_id, + terminal_status=terminal_status, + error=error, + task_title=task_title, + ) + + async def _enqueue_run_notifications( + self, + *, + task_id: str, + task_run_id: str, + run_id: str | None, + user_id: str, + terminal_status: str, + error: str | None, + task_title: str | None = None, + ) -> None: + """Write durable outbox rows for the run outcome (issue #4254). + + Best-effort by design: a notification failure is logged and swallowed + so delivery problems can never shadow the execution status written + above. Interrupted runs are user-initiated cancels and intentionally + produce no notification. + """ + if self._notification_repo is None or self._connection_repo is None: + return + event = {"success": "run_completed", "failed": "run_failed"}.get(terminal_status) + if event is None: + return + try: + connections = await self._connection_repo.list_connections(user_id) + except Exception: + logger.exception("[Scheduler] failed to list channel connections for notifications") + return + payload = { + "run_status": terminal_status, + "error": error, + "task_id": task_id, + "task_title": task_title, + } + for connection in connections: + if connection.get("status") != "connected": + continue + provider = connection.get("provider") + target = connection.get("external_account_id") + if not provider or not target: + continue + try: + await self._notification_repo.enqueue( + task_id=task_id, + task_run_id=task_run_id, + run_id=run_id, + event=event, + provider=provider, + target=target, + owner_user_id=user_id, + payload=payload, + ) + except Exception: + logger.exception( + "[Scheduler] failed to enqueue notification task_run=%s provider=%s target=%s", + task_run_id, + provider, + target, + ) async def start(self) -> None: if self._task is not None: diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 587f29e825..11e85ab2e2 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -520,6 +520,7 @@ Notes: - Create/update accept `once`, `cron`, and `interval`. Interval uses `schedule_spec.every_seconds` (UTC `now + N`, no missed-beat catch-up). N is at least `min_once_delay_seconds` (default 60) and at most 30 days. - Manual trigger uses the same scheduled-task resource and run lifecycle. - Scheduled task definitions and task-run history are persisted in the application database. +- With `channel_connections.enabled: true`, the scheduler enqueues IM outcome notifications for the task owner's connected identities. A delivery worker (same poll cadence as the scheduler) pushes them. Only runs that finish as success or failed notify: manual triggers and interrupts stay silent, and so do occurrences that end without a finished run (launch error, queue timeout, restart recovery). Channel/transport outages park rows without consuming the retry budget, for up to about a day; platform rejections exhaust ~15 minutes of counted retries then settle `failed`. The worker re-checks the binding right before sending: a target the owner has disconnected since enqueue is dropped as `failed`, never pushed. Only WeCom currently implements proactive `send_notification`. ### Agent Storage diff --git a/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md b/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md index 123851c176..ac01e4aec5 100644 --- a/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md @@ -25,7 +25,8 @@ The empty-DB path keeps using `create_all` because `Base.metadata` is the only a `0019_thread_incarnations` → `0022_scheduled_occurrence_seq` → `0023_run_change_seq` → `0023_user_preferences` → `0024_project_documents` → `0025_repair_run_change_seq` → -`0026_mcp_task_lease_tokens` (current head). The preference +`0026_mcp_task_lease_tokens` → `0027_notification_deliveries` → +`0028_parked_attempts` (current head). The preference revision adds a separate owner/key table with a cascading users foreign key and does not alter users; the project-documents revision adds a new owner-scoped shelf table, and the MCP lease-token revision adds two nullable token columns to @@ -165,6 +166,8 @@ on installs that never enabled it. The convention is: - `migrations/versions/0024_project_documents.py` — creates the `project_documents` shelf table (id/project_id/user_id/name/stored_relpath/sha256/size_bytes, nullable promotion provenance and trash fields, timestamps) with indexes on project_id, user_id, sha256 and trashed_at; no DB-level foreign key on project_id by design (project delete trashes the shelf inside its own transaction). New table, so the bootstrap forward-compat floor is unchanged; chains after `0023_user_preferences` (renumbered from 0023 after the rebase) - `migrations/versions/0025_repair_run_change_seq.py` — heals databases that skipped `0023_run_change_seq` because it was inserted ahead of the already-shipped `0023_user_preferences` (#5516): re-applies the guarded `run_change_clock` table, `runs.change_seq` column, and cursor indexes on upgrade; no-ops on healthy shapes; chains after `0024_project_documents`. Its downgrade is intentionally a no-op: the repaired objects belong to ancestor `0023_run_change_seq`, remain required at 0024, and must retain their existing change positions. Only the original 0023 downgrade removes them. `tests/test_run_change_repair_history.py` reconstructs both pre-insertion published descendants and verifies historical upgrade, unchanged healthy positions, and usable run-store writes after downgrade and re-upgrade - `migrations/versions/0026_mcp_task_lease_tokens.py` — chains after `0025_repair_run_change_seq` and adds nullable `mcp_tasks.lease_token` / `notification_lease_token` columns so every poll, cancel, and notification mutation can be fenced to the exact claim generation +- `migrations/versions/0027_notification_deliveries.py` — creates the scheduled-task IM notification outbox (`notification_deliveries`) with idempotency on `(task_run_id, event, provider, target)`; chains after `0026_mcp_task_lease_tokens`. Consumed by `ScheduledTaskService` enqueue + `NotificationDeliveryWorker` (issue #4254); no HTTP read surface yet +- `migrations/versions/0028_parked_attempts.py` — adds `notification_deliveries.parked_attempts` so channel-down parking is capped; backfills with a temporary `server_default="0"` then drops it so the durable schema matches `create_all` (ORM Python-side `default=0` only). Chains after `0027_notification_deliveries` - `persistence/bootstrap.py` — `bootstrap_schema(engine, backend=...)`, the three-branch provisioning decision, locked revision validation, and the narrow 0019 forward-compatibility exception - `extensions/loader.py::load_extensions` — registers each spec's `table_prefix` with `register_extension_table_prefix()` - Tests: `tests/test_persistence_bootstrap.py` (branches), `tests/test_persistence_bootstrap_concurrency.py` (concurrency), `tests/test_persistence_bootstrap_regression.py` (issue #3682), `tests/test_persistence_migrations_env.py` (filter, including extension-owned tables), `tests/test_extension_loader.py::TestTablePrefixRegistration` (spec-to-filter wiring), `tests/blocking_io/test_persistence_bootstrap.py` (asyncio.to_thread anchor), `tests/test_migration_0004_run_ownership_dedupe.py` + `tests/test_migration_0007_scheduled_run_active_dedupe.py` (dedupe-before-unique-index pre-steps), `tests/test_migration_0025_repair_run_change_seq.py` (issue #5516 skipped-revision heal) diff --git a/backend/packages/harness/deerflow/persistence/migrations/versions/0027_notification_deliveries.py b/backend/packages/harness/deerflow/persistence/migrations/versions/0027_notification_deliveries.py new file mode 100644 index 0000000000..64a3cb8a82 --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/migrations/versions/0027_notification_deliveries.py @@ -0,0 +1,72 @@ +"""Scheduled-task notification delivery outbox (issue #4254). + +Revision ID: 0027_notification_deliveries +Revises: 0026_mcp_task_lease_tokens +Create Date: 2026-08-16 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0027_notification_deliveries" +down_revision: str | Sequence[str] | None = "0026_mcp_task_lease_tokens" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + if inspector.has_table("notification_deliveries"): + # Idempotent: a DB whose full-metadata create_all already provisioned + # the table (e.g. a fresh DB) must not have it re-created here. + return + op.create_table( + "notification_deliveries", + sa.Column("id", sa.String(length=64), nullable=False), + sa.Column("task_id", sa.String(length=64), nullable=False), + sa.Column("task_run_id", sa.String(length=64), nullable=False), + sa.Column("run_id", sa.String(length=64), nullable=True), + sa.Column("event", sa.String(length=64), nullable=False), + sa.Column("provider", sa.String(length=32), nullable=False), + sa.Column("target", sa.String(length=128), nullable=False), + sa.Column("owner_user_id", sa.String(length=64), nullable=False), + sa.Column("status", sa.String(length=32), nullable=False), + sa.Column("attempts", sa.Integer(), nullable=False), + sa.Column("max_attempts", sa.Integer(), nullable=False), + sa.Column("payload_json", sa.JSON(), nullable=False), + sa.Column("last_error", sa.Text(), nullable=True), + sa.Column("available_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("sent_at", sa.DateTime(timezone=True), nullable=True), + sa.PrimaryKeyConstraint("id", name="pk_notification_deliveries"), + # Idempotency key from the issue design: one notification per + # (task_run_id, event, provider, target), enforced at the DB layer. + sa.UniqueConstraint( + "task_run_id", + "event", + "provider", + "target", + name="uq_notification_delivery_run_event_target", + ), + ) + op.create_index("ix_notification_deliveries_task_id", "notification_deliveries", ["task_id"], unique=False) + op.create_index("ix_notification_deliveries_task_run_id", "notification_deliveries", ["task_run_id"], unique=False) + op.create_index("ix_notification_deliveries_owner_user_id", "notification_deliveries", ["owner_user_id"], unique=False) + op.create_index("ix_notification_deliveries_due", "notification_deliveries", ["status", "available_at"], unique=False) + + +def downgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + if inspector.has_table("notification_deliveries"): + op.drop_index("ix_notification_deliveries_due", table_name="notification_deliveries") + op.drop_index("ix_notification_deliveries_owner_user_id", table_name="notification_deliveries") + op.drop_index("ix_notification_deliveries_task_run_id", table_name="notification_deliveries") + op.drop_index("ix_notification_deliveries_task_id", table_name="notification_deliveries") + op.drop_table("notification_deliveries") diff --git a/backend/packages/harness/deerflow/persistence/migrations/versions/0028_parked_attempts.py b/backend/packages/harness/deerflow/persistence/migrations/versions/0028_parked_attempts.py new file mode 100644 index 0000000000..bbab36db62 --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/migrations/versions/0028_parked_attempts.py @@ -0,0 +1,51 @@ +"""Add parked_attempts counter for notification delivery parking cap. + +Revision ID: 0028_parked_attempts +Revises: 0027_notification_deliveries +Create Date: 2026-08-19 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0028_parked_attempts" +down_revision: str | Sequence[str] | None = "0027_notification_deliveries" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + if not inspector.has_table("notification_deliveries"): + return + columns = {col["name"] for col in inspector.get_columns("notification_deliveries")} + if "parked_attempts" not in columns: + # Temporary server_default backfills existing rows. Dropped below so + # the durable schema matches create_all (ORM Python-side default=0). + with op.batch_alter_table("notification_deliveries") as batch: + batch.add_column( + sa.Column("parked_attempts", sa.Integer(), nullable=False, server_default="0"), + ) + with op.batch_alter_table("notification_deliveries") as batch: + batch.alter_column( + "parked_attempts", + existing_type=sa.Integer(), + existing_nullable=False, + server_default=None, + ) + + +def downgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + if not inspector.has_table("notification_deliveries"): + return + columns = {col["name"] for col in inspector.get_columns("notification_deliveries")} + if "parked_attempts" not in columns: + return + op.drop_column("notification_deliveries", "parked_attempts") diff --git a/backend/packages/harness/deerflow/persistence/models/__init__.py b/backend/packages/harness/deerflow/persistence/models/__init__.py index 56da9abda7..a3ff39fa2d 100644 --- a/backend/packages/harness/deerflow/persistence/models/__init__.py +++ b/backend/packages/harness/deerflow/persistence/models/__init__.py @@ -25,6 +25,7 @@ from deerflow.persistence.feedback.model import FeedbackRow from deerflow.persistence.managed_subagents.model import ManagedSubagentRow from deerflow.persistence.mcp_tasks.model import McpTaskRow from deerflow.persistence.models.run_event import RunEventRow +from deerflow.persistence.notification_deliveries.model import NotificationDeliveryRow from deerflow.persistence.personal_access_tokens.model import PersonalAccessTokenRow from deerflow.persistence.projects.model import ProjectDocumentRow, ProjectRow from deerflow.persistence.run.model import RunChangeClockRow, RunRow @@ -44,6 +45,7 @@ __all__ = [ "FeedbackRow", "McpTaskRow", "ManagedSubagentRow", + "NotificationDeliveryRow", "PersonalAccessTokenRow", "ProjectDocumentRow", "ProjectRow", diff --git a/backend/packages/harness/deerflow/persistence/notification_deliveries/__init__.py b/backend/packages/harness/deerflow/persistence/notification_deliveries/__init__.py new file mode 100644 index 0000000000..a60d9a3544 --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/notification_deliveries/__init__.py @@ -0,0 +1,9 @@ +"""Scheduled-task notification delivery outbox persistence (issue #4254).""" + +from deerflow.persistence.notification_deliveries.model import NotificationDeliveryRow +from deerflow.persistence.notification_deliveries.sql import NotificationDeliveryRepository + +__all__ = [ + "NotificationDeliveryRepository", + "NotificationDeliveryRow", +] diff --git a/backend/packages/harness/deerflow/persistence/notification_deliveries/model.py b/backend/packages/harness/deerflow/persistence/notification_deliveries/model.py new file mode 100644 index 0000000000..e3032c2b2a --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/notification_deliveries/model.py @@ -0,0 +1,66 @@ +"""ORM model for the scheduled-task notification delivery outbox (issue #4254). + +A row records one pending IM notification for a scheduled-task outcome. The +completion hook only enqueues; a separate delivery worker claims due rows and +sends them. Execution status (``scheduled_task_runs``) and delivery status +(this table) are deliberately independent: a failed delivery never changes the +run outcome, and a re-fired completion hook cannot duplicate a notification +because ``(task_run_id, event, provider, target)`` is unique. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from sqlalchemy import JSON, DateTime, Index, Integer, String, Text, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from deerflow.persistence.base import Base + + +def _utc_now() -> datetime: + return datetime.now(UTC) + + +class NotificationDeliveryRow(Base): + __tablename__ = "notification_deliveries" + + id: Mapped[str] = mapped_column(String(64), primary_key=True) + task_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + task_run_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + run_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + event: Mapped[str] = mapped_column(String(64), nullable=False) + provider: Mapped[str] = mapped_column(String(32), nullable=False) + target: Mapped[str] = mapped_column(String(128), nullable=False) + owner_user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + + # pending -> sending -> sent | failed. A failure with retries remaining + # returns the row to pending with a backoff ``available_at``. + status: Mapped[str] = mapped_column(String(32), nullable=False, default="pending") + attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + # Incremented on channel-down / transport parking (``count_attempt=False``) + # so permanently absent channels eventually reach a terminal ``failed`` + # state instead of retrying forever. + parked_attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + max_attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=5) + + # Bounded result snapshot written at enqueue time (summary text, run + # status, error); the delivery worker never re-reads agent state. + payload_json: Mapped[dict] = mapped_column(JSON, default=dict) + last_error: Mapped[str | None] = mapped_column(Text, nullable=True) + + available_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=_utc_now) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=_utc_now) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=_utc_now, onupdate=_utc_now) + sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + __table_args__ = ( + UniqueConstraint( + "task_run_id", + "event", + "provider", + "target", + name="uq_notification_delivery_run_event_target", + ), + Index("ix_notification_deliveries_due", "status", "available_at"), + ) diff --git a/backend/packages/harness/deerflow/persistence/notification_deliveries/sql.py b/backend/packages/harness/deerflow/persistence/notification_deliveries/sql.py new file mode 100644 index 0000000000..b1e324b29e --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/notification_deliveries/sql.py @@ -0,0 +1,279 @@ +"""SQL repository for the scheduled-task notification delivery outbox. + +The completion hook calls :meth:`NotificationDeliveryRepository.enqueue` +(idempotent), and the delivery worker uses ``claim_due_deliveries`` / +``mark_sent`` / ``mark_failed``. Claim flips rows ``pending -> sending`` +inside one transaction guarded by ``status = 'pending'``, so concurrent +workers (multi-pod deployments) cannot double-send the same row. +""" + +from __future__ import annotations + +import logging +import uuid +from datetime import UTC, datetime, timedelta +from typing import Any + +from sqlalchemy import select, update +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from deerflow.persistence.notification_deliveries.model import NotificationDeliveryRow + +logger = logging.getLogger(__name__) + +# Exponential backoff base/cap for retries: 60s, 120s, 240s, 480s, ... +# capped at 15 minutes. Bounded so a broken channel still drains the outbox +# to "failed" within roughly an hour at max_attempts=5. +_RETRY_BASE_SECONDS = 60 +_RETRY_MAX_SECONDS = 900 + +# Flat re-poll interval for rows parked because the owning channel is not +# running. These failures do not consume the retry budget, so the row can +# wait out an hours-long channel outage and still deliver once it returns. +_CHANNEL_DOWN_RETRY_SECONDS = 900 + +# Parking is unbounded in count by default; cap it so a permanently absent +# provider (removed from config, never starts) settles to ``failed`` instead +# of retrying every 15 minutes for the life of the deployment. Age and +# parked-attempt limits are both checked — whichever trips first wins. +_CHANNEL_PARK_MAX_AGE = timedelta(days=2) +# 96 × 15min ≈ 24h of parking at the flat backoff interval. +_CHANNEL_PARK_MAX_ATTEMPTS = 96 + +# Rows in these states are finished; later status writes are ignored. +_TERMINAL_DELIVERY_STATUSES = frozenset({"sent", "failed"}) + + +class NotificationDeliveryRepository: + """Persistence facade for the notification delivery outbox.""" + + def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: + self.session_factory = session_factory + + @staticmethod + def _new_id() -> str: + return uuid.uuid4().hex + + @staticmethod + def _to_dict(row: NotificationDeliveryRow) -> dict[str, Any]: + data = row.to_dict() + data["payload"] = data.pop("payload_json") or {} + return data + + async def enqueue( + self, + *, + task_id: str, + task_run_id: str, + event: str, + provider: str, + target: str, + owner_user_id: str, + payload: dict[str, Any] | None = None, + run_id: str | None = None, + available_at: datetime | None = None, + max_attempts: int = 5, + ) -> dict[str, Any]: + """Insert one delivery; a duplicate (task_run_id, event, provider, + target) returns the existing row instead of raising.""" + row = NotificationDeliveryRow( + id=self._new_id(), + task_id=task_id, + task_run_id=task_run_id, + run_id=run_id, + event=event, + provider=provider, + target=target, + owner_user_id=owner_user_id, + payload_json=dict(payload or {}), + max_attempts=max_attempts, + ) + if available_at is not None: + row.available_at = available_at + try: + async with self.session_factory() as session: + session.add(row) + await session.commit() + await session.refresh(row) + return self._to_dict(row) + except IntegrityError: + existing = await self._find_by_idempotency_key( + task_run_id=task_run_id, + event=event, + provider=provider, + target=target, + ) + if existing is None: + raise + return self._to_dict(existing) + + async def _find_by_idempotency_key( + self, + *, + task_run_id: str, + event: str, + provider: str, + target: str, + ) -> NotificationDeliveryRow | None: + async with self.session_factory() as session: + result = await session.execute( + select(NotificationDeliveryRow).where( + NotificationDeliveryRow.task_run_id == task_run_id, + NotificationDeliveryRow.event == event, + NotificationDeliveryRow.provider == provider, + NotificationDeliveryRow.target == target, + ) + ) + return result.scalar_one_or_none() + + async def claim_due_deliveries(self, *, now: datetime, limit: int) -> list[dict[str, Any]]: + """Atomically claim due pending rows by flipping them to ``sending``. + + The UPDATE is guarded by ``status = 'pending'`` and returns its own + result set via ``RETURNING`` (SQLite >= 3.35, Postgres), so the rows + handed back are exactly the rows THIS transaction flipped. A separate + re-SELECT would also match rows a concurrent worker flipped in the + same window and hand them out twice -- the multi-pod double-send the + docstring above rules out. + """ + if limit <= 0: + return [] + async with self.session_factory() as session: + due_ids = ( + ( + await session.execute( + select(NotificationDeliveryRow.id) + .where( + NotificationDeliveryRow.status == "pending", + NotificationDeliveryRow.available_at <= now, + ) + .order_by(NotificationDeliveryRow.available_at, NotificationDeliveryRow.created_at) + .limit(limit) + ) + ) + .scalars() + .all() + ) + if not due_ids: + return [] + claimed = ( + ( + await session.execute( + update(NotificationDeliveryRow) + .where( + NotificationDeliveryRow.id.in_(due_ids), + NotificationDeliveryRow.status == "pending", + ) + .values(status="sending", updated_at=now) + .returning(NotificationDeliveryRow) + ) + ) + .scalars() + .all() + ) + await session.commit() + return [self._to_dict(row) for row in claimed] + + async def reset_stale_sending_rows(self, *, now: datetime, timeout: timedelta) -> int: + """Flip rows stuck in ``sending`` past ``timeout`` back to ``pending``. + + A row gets stranded there when the process dies between claim and the + final ``mark_sent``/``mark_failed`` write (SIGKILL/OOM), or when that + write itself raises. Without this reconciliation such rows sit in + ``sending`` forever. The accepted price is a possible duplicate push + for a delivery that actually went out right before the crash -- the + at-least-once trade-off documented on the outbox. The timeout is + measured from ``updated_at``, which claim stamps at flip time. + """ + cutoff = now - timeout + async with self.session_factory() as session: + result = await session.execute( + update(NotificationDeliveryRow) + .where( + NotificationDeliveryRow.status == "sending", + NotificationDeliveryRow.updated_at <= cutoff, + ) + .values(status="pending", updated_at=now) + ) + await session.commit() + return result.rowcount + + async def mark_sent(self, delivery_id: str) -> dict[str, Any]: + async with self.session_factory() as session: + row = await session.get(NotificationDeliveryRow, delivery_id) + if row is None: + raise LookupError(f"notification delivery {delivery_id} not found") + if row.status in _TERMINAL_DELIVERY_STATUSES: + # A late write after the row was already finalized (a crash + # handler racing a terminal mark_failed, or a stale claimant) + # must not reopen or relabel it. + logger.info("notification delivery %s is already %s; ignoring late mark_sent", delivery_id, row.status) + return self._to_dict(row) + row.status = "sent" + row.sent_at = datetime.now(UTC) + await session.commit() + await session.refresh(row) + return self._to_dict(row) + + async def mark_failed( + self, + delivery_id: str, + *, + error: str | None = None, + count_attempt: bool = True, + terminal: bool = False, + ) -> dict[str, Any]: + """Record a failed attempt; reschedule with backoff while retries + remain, otherwise finalize the row as ``failed``. + + With ``count_attempt=False`` the failure is parked instead: the retry + budget stays intact and the row returns on a flat long backoff. Used + when the owning channel is not running -- an outage is not the + delivery's fault and must not be able to exhaust its retries. Parking + is capped by row age and ``parked_attempts`` so permanently absent + channels eventually settle to ``failed``. + + With ``terminal=True`` the row is finalized as ``failed`` at once, + whatever its remaining budget. Used when the delivery can never + become valid again, such as a target the owner has since + disconnected: neither waiting nor retrying would make sending it + right. + """ + async with self.session_factory() as session: + row = await session.get(NotificationDeliveryRow, delivery_id) + if row is None: + raise LookupError(f"notification delivery {delivery_id} not found") + if row.status in _TERMINAL_DELIVERY_STATUSES: + # Terminal states are sticky: a counted or parked mark_failed + # arriving after a terminal one must not flip the row back to + # pending and send it later after all. + logger.info("notification delivery %s is already %s; ignoring late mark_failed", delivery_id, row.status) + return self._to_dict(row) + row.last_error = error + if terminal: + row.status = "failed" + elif not count_attempt: + now = datetime.now(UTC) + created_at = row.created_at + if created_at.tzinfo is None: + created_at = created_at.replace(tzinfo=UTC) + park_expired = (now - created_at) >= _CHANNEL_PARK_MAX_AGE or row.parked_attempts >= _CHANNEL_PARK_MAX_ATTEMPTS + if park_expired: + row.status = "failed" + row.last_error = error or "channel did not return before parking limit" + else: + row.parked_attempts += 1 + row.status = "pending" + row.available_at = now + timedelta(seconds=_CHANNEL_DOWN_RETRY_SECONDS) + else: + row.attempts += 1 + if row.attempts >= row.max_attempts: + row.status = "failed" + else: + delay = min(_RETRY_BASE_SECONDS * 2 ** (row.attempts - 1), _RETRY_MAX_SECONDS) + row.status = "pending" + row.available_at = datetime.now(UTC) + timedelta(seconds=delay) + await session.commit() + await session.refresh(row) + return self._to_dict(row) diff --git a/backend/tests/test_gateway_lifespan_shutdown.py b/backend/tests/test_gateway_lifespan_shutdown.py index 44edb3a9b7..2ecb2b8de7 100644 --- a/backend/tests/test_gateway_lifespan_shutdown.py +++ b/backend/tests/test_gateway_lifespan_shutdown.py @@ -751,3 +751,172 @@ def test_lifespan_closes_mcp_sessions_created_during_run_drain(): await pool.close_all() asyncio.run(scenario()) + + +# ── notification delivery worker wiring (issue #4254) ─────────────────────── + + +@asynccontextmanager +async def _langgraph_with_scheduled_repos(app, _startup_config): + app.state.scheduled_task_repo = MagicMock() + app.state.scheduled_task_run_repo = MagicMock() + yield + + +def _notification_startup_config(*, channel_connections_enabled: bool = True): + from deerflow.config.channel_connections_config import ChannelConnectionsConfig + + return SimpleNamespace( + log_level="INFO", + memory=SimpleNamespace(token_counting="char", enabled=False, shutdown_flush_timeout_seconds=5.0), + scheduler=SimpleNamespace( + enabled=False, + poll_interval_seconds=5, + lease_seconds=30, + max_concurrent_runs=1, + multi_instance=False, + queue_timeout_seconds=3600, + ), + run_ownership=SimpleNamespace(grace_seconds=30), + channel_connections=ChannelConnectionsConfig.model_validate({"enabled": channel_connections_enabled}), + ) + + +async def _run_lifespan_with_notification_worker(*, channel_service_available: bool, visible_only_after_start: bool = False): + from app.gateway.app import lifespan + + app = FastAPI() + startup_config = _notification_startup_config() + close_oidc_service = AsyncMock() + stop_channel_service = AsyncMock() + fake_service = MagicMock() + fake_service.get_status = MagicMock(return_value={}) + worker_start = AsyncMock() + worker_stop = AsyncMock() + worker_instance = MagicMock() + worker_instance.start = worker_start + worker_instance.stop = worker_stop + scheduled_service = MagicMock() + scheduled_service.start = AsyncMock() + scheduled_service.stop = AsyncMock() + session_factory = MagicMock() + shutdown_events: list[str] = [] + + channel_service_started = False + + async def fake_start(_startup_config, **_kwargs): + nonlocal channel_service_started + channel_service_started = True + return fake_service + + def fake_get_channel_service(): + # The real accessor returns None until start_channel_service() has run. + if not channel_service_available or (visible_only_after_start and not channel_service_started): + return None + return fake_service + + async def record_worker_stop(): + shutdown_events.append("worker") + + async def record_channel_stop(): + shutdown_events.append("channels") + + worker_stop.side_effect = record_worker_stop + stop_channel_service.side_effect = record_channel_stop + worker_factory = MagicMock(return_value=worker_instance) + + with ( + patch("app.gateway.app.get_app_config", return_value=startup_config), + patch("app.gateway.app.get_gateway_config", return_value=MagicMock(host="x", port=0)), + patch("app.gateway.app.langgraph_runtime", _langgraph_with_scheduled_repos), + patch("app.gateway.app._ensure_admin_user", AsyncMock()), + patch("deerflow.skills.projection.ensure_public_skill_projection"), + patch("app.gateway.app.auth.close_oidc_service", close_oidc_service), + patch("app.channels.service.start_channel_service", side_effect=fake_start), + patch("app.channels.service.get_channel_service", side_effect=fake_get_channel_service), + patch("app.channels.service.stop_channel_service", stop_channel_service), + patch("deerflow.persistence.engine.get_session_factory", return_value=session_factory), + patch("app.scheduler.ScheduledTaskService", return_value=scheduled_service), + patch("app.scheduler.notification_delivery.NotificationDeliveryWorker", worker_factory), + patch("deerflow.persistence.run.RunRepository"), + ): + async with lifespan(app): + worker_on_app = getattr(app.state, "notification_delivery_worker", None) + worker_instance.factory_kwargs = worker_factory.call_args.kwargs if worker_factory.call_args else None + return worker_start, worker_stop, stop_channel_service, worker_on_app, shutdown_events, scheduled_service + + +def test_lifespan_skips_notification_worker_when_channel_service_missing(caplog) -> None: + caplog.set_level(logging.WARNING, logger="app.gateway.app") + worker_start, worker_stop, stop_channel_service, worker_on_app, _shutdown_events, scheduled_service = asyncio.run(_run_lifespan_with_notification_worker(channel_service_available=False)) + + worker_start.assert_not_awaited() + assert worker_on_app is None + assert any("write-only outbox" in record.message for record in caplog.records) + # The enqueue side was wired with the scheduler; without delivery it is switched off again. + scheduled_service.detach_notification_outbox.assert_called_once_with() + stop_channel_service.assert_awaited_once() + + +def test_lifespan_starts_notification_worker_when_enqueue_and_channel_are_wired() -> None: + worker_start, worker_stop, stop_channel_service, worker_on_app, shutdown_events, scheduled_service = asyncio.run(_run_lifespan_with_notification_worker(channel_service_available=True)) + + worker_start.assert_awaited_once() + assert worker_on_app is not None + # The worker re-checks the target's connection at delivery time, so it + # must be handed the connection repository's lookup, not left unwired. + resolve_connections = worker_on_app.factory_kwargs["resolve_connections"] + assert resolve_connections is not None + assert getattr(resolve_connections, "__name__", None) == "list_connections" + scheduled_service.detach_notification_outbox.assert_not_called() + worker_stop.assert_awaited_once() + stop_channel_service.assert_awaited_once() + assert shutdown_events.index("worker") < shutdown_events.index("channels") + + +def test_lifespan_wires_notifications_after_the_channel_service_starts() -> None: + # The lifespan starts the scheduler before the channel service (#5035), so the + # outbox has to look the channel service up after that start. + worker_start, worker_stop, _stop_channel_service, worker_on_app, _shutdown_events, scheduled_service = asyncio.run(_run_lifespan_with_notification_worker(channel_service_available=True, visible_only_after_start=True)) + + worker_start.assert_awaited_once() + assert worker_on_app is not None + scheduled_service.detach_notification_outbox.assert_not_called() + worker_stop.assert_awaited_once() + + +def test_lifespan_detaches_the_outbox_when_the_delivery_worker_cannot_start() -> None: + from app.gateway.app import lifespan + + app = FastAPI() + scheduled_service = MagicMock() + scheduled_service.start = AsyncMock() + scheduled_service.stop = AsyncMock() + worker_instance = MagicMock() + worker_instance.start = AsyncMock(side_effect=RuntimeError("worker cannot start")) + fake_service = MagicMock() + fake_service.get_status = MagicMock(return_value={}) + + async def run() -> None: + with ( + patch("app.gateway.app.get_app_config", return_value=_notification_startup_config()), + patch("app.gateway.app.get_gateway_config", return_value=MagicMock(host="x", port=0)), + patch("app.gateway.app.langgraph_runtime", _langgraph_with_scheduled_repos), + patch("app.gateway.app._ensure_admin_user", AsyncMock()), + patch("deerflow.skills.projection.ensure_public_skill_projection"), + patch("app.gateway.app.auth.close_oidc_service", AsyncMock()), + patch("app.channels.service.start_channel_service", AsyncMock(return_value=fake_service)), + patch("app.channels.service.get_channel_service", return_value=fake_service), + patch("app.channels.service.stop_channel_service", AsyncMock()), + patch("deerflow.persistence.engine.get_session_factory", return_value=MagicMock()), + patch("app.scheduler.ScheduledTaskService", return_value=scheduled_service), + patch("app.scheduler.notification_delivery.NotificationDeliveryWorker", return_value=worker_instance), + patch("deerflow.persistence.run.RunRepository"), + ): + async with lifespan(app): + assert getattr(app.state, "notification_delivery_worker", None) is None + + asyncio.run(run()) + + # Startup survives, and nothing keeps enqueueing rows that no worker would send. + scheduled_service.detach_notification_outbox.assert_called_once_with() diff --git a/backend/tests/test_migration_0026_mcp_task_lease_tokens.py b/backend/tests/test_migration_0026_mcp_task_lease_tokens.py index 0f22205684..0499b69668 100644 --- a/backend/tests/test_migration_0026_mcp_task_lease_tokens.py +++ b/backend/tests/test_migration_0026_mcp_task_lease_tokens.py @@ -1,9 +1,8 @@ """Migration tests for 0026_mcp_task_lease_tokens. Adds the nullable per-claim token columns ``McpTaskRepository`` uses to fence -poll, cancel, and notification mutations to the exact claim generation. This -file owns the chain-head pin, moved on from -``test_migration_0025_repair_run_change_seq`` with this revision. +poll, cancel, and notification mutations to the exact claim generation. The +chain-head pin moved on to ``test_migration_0027_0028_notification_outbox``. """ from __future__ import annotations @@ -13,6 +12,7 @@ import asyncio import pytest import sqlalchemy as sa from alembic import command +from alembic.script import ScriptDirectory from sqlalchemy.ext.asyncio import create_async_engine from deerflow.persistence import bootstrap @@ -24,8 +24,12 @@ PREVIOUS = "0025_repair_run_change_seq" TOKEN_COLUMNS = {"lease_token", "notification_lease_token"} -async def test_0026_is_the_chain_head(): - assert bootstrap._get_head_revision() == REVISION +async def test_0026_chains_into_the_single_head(): + script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR)) + assert len(script.get_heads()) == 1 + # Later migrations may advance the head without removing this revision. + assert REVISION in {revision.revision for revision in script.walk_revisions()} + assert script.get_revision(REVISION).down_revision == PREVIOUS async def test_0026_adds_nullable_claim_tokens_and_downgrades(tmp_path): diff --git a/backend/tests/test_migration_0027_0028_notification_outbox.py b/backend/tests/test_migration_0027_0028_notification_outbox.py new file mode 100644 index 0000000000..418ee36170 --- /dev/null +++ b/backend/tests/test_migration_0027_0028_notification_outbox.py @@ -0,0 +1,73 @@ +"""Migration tests for the scheduled-task notification outbox (issue #4254). + +``0027_notification_deliveries`` creates the outbox table and +``0028_parked_attempts`` adds its parking counter. This file owns the chain-head +pin, moved on from ``test_migration_0026_mcp_task_lease_tokens``. +""" + +from __future__ import annotations + +import asyncio + +import pytest +import sqlalchemy as sa +from alembic import command +from alembic.script import ScriptDirectory +from sqlalchemy.ext.asyncio import create_async_engine + +from deerflow.persistence import bootstrap + +pytestmark = pytest.mark.asyncio + +OUTBOX = "0027_notification_deliveries" +PARKED = "0028_parked_attempts" +PREVIOUS = "0026_mcp_task_lease_tokens" +TABLE = "notification_deliveries" + + +async def test_0028_is_the_chain_head(): + assert bootstrap._get_head_revision() == PARKED + + +async def test_outbox_revisions_chain_after_0026(): + script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR)) + assert len(script.get_heads()) == 1 + assert script.get_revision(OUTBOX).down_revision == PREVIOUS + assert script.get_revision(PARKED).down_revision == OUTBOX + # alembic_version.version_num is VARCHAR(32); a longer id fails on Postgres. + assert max(len(OUTBOX), len(PARKED)) <= 32 + + +async def test_outbox_revisions_upgrade_and_downgrade(tmp_path): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'outbox.db'}") + cfg = bootstrap._get_alembic_config(engine) + + async def outbox_columns() -> dict[str, dict] | None: + async with engine.connect() as conn: + + def read(sync): + inspector = sa.inspect(sync) + if not inspector.has_table(TABLE): + return None + return {column["name"]: column for column in inspector.get_columns(TABLE)} + + return await conn.run_sync(read) + + try: + await asyncio.to_thread(bootstrap._upgrade, cfg, PREVIOUS) + assert await outbox_columns() is None + + await asyncio.to_thread(bootstrap._upgrade, cfg, OUTBOX) + columns = await outbox_columns() + assert columns is not None and "parked_attempts" not in columns + + await asyncio.to_thread(bootstrap._upgrade, cfg, "head") + columns = await outbox_columns() + assert columns["parked_attempts"]["nullable"] is False + # The backfill default is dropped so the schema matches create_all. + assert columns["parked_attempts"]["default"] is None + + await asyncio.to_thread(command.downgrade, cfg, PREVIOUS) + assert await outbox_columns() is None + finally: + await engine.dispose() diff --git a/backend/tests/test_notification_deliveries_repository.py b/backend/tests/test_notification_deliveries_repository.py new file mode 100644 index 0000000000..d9ed0669c7 --- /dev/null +++ b/backend/tests/test_notification_deliveries_repository.py @@ -0,0 +1,334 @@ +"""Tests for the scheduled-task notification delivery outbox (issue #4254). + +The outbox is the durable queue between scheduled-task completion and IM +delivery: the completion hook only enqueues, a separate delivery worker +claims and sends. Execution status and delivery status stay independent. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest + +from deerflow.persistence.notification_deliveries import NotificationDeliveryRepository, NotificationDeliveryRow +from deerflow.persistence.notification_deliveries.sql import ( + _CHANNEL_PARK_MAX_AGE, + _CHANNEL_PARK_MAX_ATTEMPTS, +) + + +@pytest.fixture +async def repo(tmp_path): + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + + url = f"sqlite+aiosqlite:///{tmp_path / 'deliveries.db'}" + await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)) + try: + yield NotificationDeliveryRepository(get_session_factory()) + finally: + await close_engine() + + +def _enqueue_kwargs(**overrides): + kwargs = { + "task_id": "task-1", + "task_run_id": "run-1", + "event": "run_completed", + "provider": "wecom", + "target": "GaoZhiChao", + "owner_user_id": "alice", + "payload": {"summary": "daily report ready"}, + } + kwargs.update(overrides) + return kwargs + + +class TestNotificationDeliveryRepository: + @pytest.mark.anyio + async def test_enqueue_creates_pending_delivery(self, repo): + delivery = await repo.enqueue(**_enqueue_kwargs()) + + assert delivery["status"] == "pending" + assert delivery["attempts"] == 0 + assert delivery["provider"] == "wecom" + assert delivery["target"] == "GaoZhiChao" + assert delivery["payload"] == {"summary": "daily report ready"} + + @pytest.mark.anyio + async def test_enqueue_is_idempotent_per_run_event_target(self, repo): + # (task_run_id, event, provider, target) is the idempotency key the + # issue design calls for: a retried completion hook must not enqueue + # a second notification for the same outcome. + first = await repo.enqueue(**_enqueue_kwargs()) + second = await repo.enqueue(**_enqueue_kwargs(payload={"summary": "different"})) + + assert second["id"] == first["id"] + assert second["payload"] == {"summary": "daily report ready"} + claimed = await repo.claim_due_deliveries(now=datetime.now(UTC), limit=10) + assert len(claimed) == 1 + + @pytest.mark.anyio + async def test_enqueue_allows_distinct_events_and_targets(self, repo): + await repo.enqueue(**_enqueue_kwargs()) + await repo.enqueue(**_enqueue_kwargs(event="run_failed")) + await repo.enqueue(**_enqueue_kwargs(target="bob")) + + claimed = await repo.claim_due_deliveries(now=datetime.now(UTC), limit=10) + assert len(claimed) == 3 + + @pytest.mark.anyio + async def test_claim_returns_only_due_pending_rows(self, repo): + due = await repo.enqueue(**_enqueue_kwargs(task_run_id="run-due")) + await repo.enqueue(**_enqueue_kwargs(task_run_id="run-future", available_at=datetime.now(UTC) + timedelta(hours=1))) + + claimed = await repo.claim_due_deliveries(now=datetime.now(UTC), limit=10) + + assert [row["id"] for row in claimed] == [due["id"]] + + @pytest.mark.anyio + async def test_claim_marks_rows_in_flight_so_second_claim_is_empty(self, repo): + await repo.enqueue(**_enqueue_kwargs()) + + first = await repo.claim_due_deliveries(now=datetime.now(UTC), limit=10) + second = await repo.claim_due_deliveries(now=datetime.now(UTC), limit=10) + + assert len(first) == 1 + assert second == [] + + @pytest.mark.anyio + async def test_claim_respects_limit(self, repo): + for index in range(3): + await repo.enqueue(**_enqueue_kwargs(task_run_id=f"run-{index}")) + + claimed = await repo.claim_due_deliveries(now=datetime.now(UTC), limit=2) + + assert len(claimed) == 2 + + @pytest.mark.anyio + async def test_claim_excludes_rows_already_claimed_by_another_worker(self, repo): + """The claim result set must be exactly the rows THIS transaction + flipped. Replay the multi-worker race deterministically: right after + this worker's due-id SELECT, a rival worker commits a claim on one of + the same rows; the stale id must not come back in the result set, + otherwise multi-instance deployments double-send + (``scheduler.multi_instance`` is a supported mode).""" + from sqlalchemy import select + + await repo.enqueue(**_enqueue_kwargs(task_run_id="run-mine")) + await repo.enqueue(**_enqueue_kwargs(task_run_id="run-stolen", target="bob")) + + real_factory = repo.session_factory + + async def _rival_steals_one(): + async with real_factory() as session: + row = (await session.execute(select(NotificationDeliveryRow).where(NotificationDeliveryRow.status == "pending").order_by(NotificationDeliveryRow.created_at, NotificationDeliveryRow.id).limit(1))).scalars().one() + row.status = "sending" + await session.commit() + + fired = {"done": False} + + class _InterceptSession: + """Fires the rival claim right after the first SELECT of the + intercepted claim transaction.""" + + def __init__(self, session): + self._session = session + + async def __aenter__(self): + await self._session.__aenter__() + return self + + async def __aexit__(self, *exc_info): + return await self._session.__aexit__(*exc_info) + + async def execute(self, stmt, *args, **kwargs): + result = await self._session.execute(stmt, *args, **kwargs) + if not fired["done"] and getattr(stmt, "is_select", False): + fired["done"] = True + await _rival_steals_one() + return result + + def __getattr__(self, name): + return getattr(self._session, name) + + racing_repo = NotificationDeliveryRepository(lambda: _InterceptSession(real_factory())) + claimed = await racing_repo.claim_due_deliveries(now=datetime.now(UTC), limit=10) + + assert fired["done"] is True + # Exactly one row remains for this worker; the rival-owned row must + # not be in the result set even though the due-id view still saw it. + assert len(claimed) == 1 + + @pytest.mark.anyio + async def test_claim_stamps_updated_at_for_stale_detection(self, repo): + """Claim must write ``updated_at`` explicitly: a Core UPDATE does not + fire ORM ``onupdate``, and the stale-``sending`` recovery measures + its timeout from the moment the row was claimed.""" + await repo.enqueue(**_enqueue_kwargs(), available_at=datetime.now(UTC) - timedelta(hours=2)) + claim_at = datetime.now(UTC) + + claimed = await repo.claim_due_deliveries(now=claim_at, limit=1) + + updated_at = claimed[0]["updated_at"] + if updated_at.tzinfo is None: + updated_at = updated_at.replace(tzinfo=UTC) + assert abs((updated_at - claim_at).total_seconds()) < 1 + + @pytest.mark.anyio + async def test_reset_stale_sending_rows_flips_back_only_timed_out_rows(self, repo): + """A row claimed then orphaned (crash between claim and the final + status write) must become claimable again once the timeout passes -- + otherwise it sits in ``sending`` forever.""" + stuck = await repo.enqueue(**_enqueue_kwargs(task_run_id="run-stuck")) + await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1) + + # Freshly claimed rows are not stale yet. + assert await repo.reset_stale_sending_rows(now=datetime.now(UTC), timeout=timedelta(minutes=10)) == 0 + + assert await repo.reset_stale_sending_rows(now=datetime.now(UTC) + timedelta(minutes=11), timeout=timedelta(minutes=10)) == 1 + + reclaimed = await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(minutes=11), limit=1) + assert [row["id"] for row in reclaimed] == [stuck["id"]] + + @pytest.mark.anyio + async def test_reset_stale_sending_rows_ignores_terminal_and_pending_rows(self, repo): + await repo.enqueue(**_enqueue_kwargs(task_run_id="run-sent")) + await repo.enqueue(**_enqueue_kwargs(task_run_id="run-pending")) + claimed = await repo.claim_due_deliveries(now=datetime.now(UTC), limit=1) + await repo.mark_sent(claimed[0]["id"]) + + reset = await repo.reset_stale_sending_rows(now=datetime.now(UTC) + timedelta(hours=1), timeout=timedelta(minutes=10)) + + assert reset == 0 + + @pytest.mark.anyio + async def test_mark_sent_finalises_delivery(self, repo): + delivery = await repo.enqueue(**_enqueue_kwargs()) + + updated = await repo.mark_sent(delivery["id"]) + + assert updated["status"] == "sent" + assert updated["sent_at"] is not None + + @pytest.mark.anyio + async def test_mark_failed_reschedules_with_backoff_while_retries_remain(self, repo): + delivery = await repo.enqueue(**_enqueue_kwargs()) + before = datetime.now(UTC) + + updated = await repo.mark_failed(delivery["id"], error="wecom: rate limited") + + assert updated["status"] == "pending" + assert updated["attempts"] == 1 + assert updated["last_error"] == "wecom: rate limited" + retry_at = updated["available_at"] + if retry_at.tzinfo is None: + retry_at = retry_at.replace(tzinfo=UTC) + assert retry_at > before + + @pytest.mark.anyio + async def test_mark_failed_terminal_finalises_at_once_whatever_the_budget(self, repo): + # Used when the delivery can never become valid again (the owner + # disconnected the target): no backoff, no parking, budget untouched. + delivery = await repo.enqueue(**_enqueue_kwargs()) + + updated = await repo.mark_failed(delivery["id"], error="target is no longer connected", terminal=True) + + assert updated["status"] == "failed" + assert updated["attempts"] == 0 + assert updated["parked_attempts"] == 0 + assert updated["last_error"] == "target is no longer connected" + assert await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(days=2), limit=10) == [] + + @pytest.mark.anyio + async def test_terminal_states_are_sticky(self, repo): + # A late write from a crash handler or a stale claimant must neither + # reopen a finished row nor relabel it. + dropped = await repo.enqueue(**_enqueue_kwargs()) + await repo.mark_failed(dropped["id"], error="target is no longer connected", terminal=True) + after_late_failure = await repo.mark_failed(dropped["id"], error="delivery crashed before completion") + after_late_sent = await repo.mark_sent(dropped["id"]) + assert after_late_failure["status"] == "failed" + assert after_late_sent["status"] == "failed" + assert after_late_sent["last_error"] == "target is no longer connected" + assert after_late_sent["attempts"] == 0 + + sent = await repo.enqueue(**_enqueue_kwargs(target="other-target")) + await repo.mark_sent(sent["id"]) + after_late_failure = await repo.mark_failed(sent["id"], error="late failure", count_attempt=False) + assert after_late_failure["status"] == "sent" + assert after_late_failure["last_error"] is None + assert await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(days=2), limit=10) == [] + + @pytest.mark.anyio + async def test_mark_failed_exhausts_retries(self, repo): + delivery = await repo.enqueue(**_enqueue_kwargs()) + + updated = None + for _ in range(delivery["max_attempts"]): + claimed = await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(days=1), limit=1) + assert len(claimed) == 1 + updated = await repo.mark_failed(claimed[0]["id"], error="boom") + + assert updated["status"] == "failed" + # A failed (exhausted) row is never claimed again. + assert await repo.claim_due_deliveries(now=datetime.now(UTC) + timedelta(days=1), limit=1) == [] + + @pytest.mark.anyio + async def test_mark_failed_without_counting_attempt_preserves_budget_until_park_cap(self, repo): + """Channel-down failures are parked, not counted: the row keeps its + full retry budget until parking expires, then settles to failed.""" + delivery = await repo.enqueue(**_enqueue_kwargs()) + before = datetime.now(UTC) + + updated = await repo.mark_failed(delivery["id"], error="channel 'wecom' is not running", count_attempt=False) + + assert updated["status"] == "pending" + assert updated["attempts"] == 0 + assert updated["parked_attempts"] == 1 + retry_at = updated["available_at"] + if retry_at.tzinfo is None: + retry_at = retry_at.replace(tzinfo=UTC) + assert retry_at >= before + timedelta(minutes=10) + + for _ in range(_CHANNEL_PARK_MAX_ATTEMPTS - 1): + updated = await repo.mark_failed(updated["id"], error="channel still down", count_attempt=False) + assert updated["status"] == "pending" + assert updated["attempts"] == 0 + + updated = await repo.mark_failed(updated["id"], error="channel still down", count_attempt=False) + assert updated["status"] == "failed" + assert updated["attempts"] == 0 + + @pytest.mark.anyio + async def test_mark_failed_parking_expires_by_row_age(self, repo): + delivery = await repo.enqueue(**_enqueue_kwargs()) + async with repo.session_factory() as session: + row = await session.get(NotificationDeliveryRow, delivery["id"]) + row.created_at = datetime.now(UTC) - _CHANNEL_PARK_MAX_AGE - timedelta(minutes=1) + await session.commit() + + updated = await repo.mark_failed(delivery["id"], error="channel 'wecom' is not running", count_attempt=False) + + assert updated["status"] == "failed" + assert updated["attempts"] == 0 + + @pytest.mark.anyio + async def test_unique_constraint_rejects_manual_duplicate(self, repo): + from sqlalchemy.exc import IntegrityError + + await repo.enqueue(**_enqueue_kwargs()) + with pytest.raises(IntegrityError): + async with repo.session_factory() as session: + session.add( + NotificationDeliveryRow( + id="manual-duplicate", + task_id="task-1", + task_run_id="run-1", + event="run_completed", + provider="wecom", + target="GaoZhiChao", + owner_user_id="alice", + ) + ) + await session.commit() diff --git a/backend/tests/test_notification_delivery_worker.py b/backend/tests/test_notification_delivery_worker.py new file mode 100644 index 0000000000..465da93cc7 --- /dev/null +++ b/backend/tests/test_notification_delivery_worker.py @@ -0,0 +1,830 @@ +"""Tests for the notification delivery worker (issue #4254). + +The worker consumes the notification outbox written by the scheduled-task +completion hook and pushes each delivery through the owning channel's +proactive send path. It must never touch execution state. +""" + +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime + +import pytest + +from app.channels.base import ChannelUnavailable +from app.channels.message_bus import MessageBus +from app.channels.wecom import WeComChannel +from app.scheduler.notification_delivery import NotificationDeliveryWorker, redact_egress_text, render_notification_text + + +class FakeDeliveryRepo: + def __init__(self, rows=None): + self.rows = list(rows or []) + self.claims = [] + self.sent = [] + self.failed = [] + self.terminal_failures = [] + self.resets = [] + self.events = [] + + async def reset_stale_sending_rows(self, *, now, timeout): + self.resets.append((now, timeout)) + self.events.append("reset") + return 0 + + async def claim_due_deliveries(self, *, now, limit): + claimed, self.rows = self.rows[:limit], self.rows[limit:] + self.claims.append((now, limit)) + self.events.append("claim") + return claimed + + async def mark_sent(self, delivery_id): + self.sent.append(delivery_id) + + async def mark_failed(self, delivery_id, *, error=None, count_attempt=True, terminal=False): + self.failed.append((delivery_id, error, count_attempt)) + if terminal: + self.terminal_failures.append((delivery_id, error)) + + +class FakeChannel: + def __init__(self, *, fail=False, unavailable=False, running=True): + self.sent = [] + self.fail = fail + self.unavailable = unavailable + self.is_running = running + + async def send_notification(self, *, target, text_markdown): + if self.unavailable: + raise ChannelUnavailable("channel transport down") + if self.fail: + raise RuntimeError("platform rejected the push") + self.sent.append((target, text_markdown)) + + +def _delivery_row( + *, + delivery_id="delivery-1", + event="run_completed", + provider="wecom", + target="GaoZhiChao", + payload=None, +): + return { + "id": delivery_id, + "task_id": "task-1", + "task_run_id": "task-run-1", + "run_id": "run-1", + "event": event, + "provider": provider, + "target": target, + "owner_user_id": "user-1", + "status": "sending", + "payload": payload if payload is not None else {"run_status": "success", "error": None, "task_id": "task-1"}, + } + + +def _make_worker(repo, channel=None, resolve_run_summary=None, resolve_connections=None): + channels = {} + if channel is not None: + channels["wecom"] = channel + return NotificationDeliveryWorker( + delivery_repo=repo, + resolve_channel=lambda provider: channels.get(provider), + poll_interval_seconds=5, + resolve_run_summary=resolve_run_summary, + resolve_connections=resolve_connections, + ) + + +def _connection(*, provider="wecom", external_account_id="GaoZhiChao", status="connected", owner_user_id="user-1"): + return { + "id": f"conn-{provider}-{external_account_id}", + "owner_user_id": owner_user_id, + "provider": provider, + "external_account_id": external_account_id, + "status": status, + } + + +def test_render_run_completed_text_includes_task_and_run(): + text = render_notification_text(_delivery_row()) + + assert "completed" in text.lower() + assert "task-1" in text + assert "run-1" in text + + +def test_render_prefers_task_title_over_id(): + text = render_notification_text(_delivery_row(payload={"run_status": "success", "error": None, "task_id": "task-1", "task_title": "Daily digest"})) + + assert "Daily digest" in text + task_line = next(line for line in text.splitlines() if line.startswith("Task:")) + assert "task-1" not in task_line + + +def test_render_run_failed_text_does_not_forward_raw_error(): + secret = "sk-" + ("x" * 40) + text = render_notification_text( + _delivery_row( + event="run_failed", + payload={"run_status": "failed", "error": secret, "task_id": "task-1"}, + ) + ) + + assert "failed" in text.lower() + assert secret not in text + assert "workspace" in text.lower() + assert "Error:" not in text + + +def test_render_truncates_unbounded_error_text(): + text = render_notification_text( + _delivery_row( + event="run_failed", + payload={"run_status": "failed", "error": "x" * 5000, "task_id": "task-1"}, + ) + ) + + assert len(text) < 1000 + assert "Error:" not in text + + +def test_redact_egress_text_scrubs_seeded_secret(): + secret = "sk-" + ("S" * 40) + redacted = redact_egress_text(f"answer used {secret} here") + + assert secret not in redacted + assert "[redacted]" in redacted + + +def test_redact_egress_text_scrubs_entire_pem_block(): + pem_body = "ABCDEFSECRETKEYBODY" + pem = "-----BEGIN PRIVATE KEY-----\n" + pem_body + "\n-----END PRIVATE KEY-----" + redacted = redact_egress_text(f"key follows\n{pem}\nend") + + assert pem_body not in redacted + assert "BEGIN PRIVATE KEY" not in redacted + assert "END PRIVATE KEY" not in redacted + assert "[redacted]" in redacted + + +def test_render_redacts_entire_pem_block_in_result_summary(): + pem_body = "ABCDEFSECRETKEYBODY" + pem = "-----BEGIN PRIVATE KEY-----\n" + pem_body + "\n-----END PRIVATE KEY-----" + text = render_notification_text(_delivery_row(payload={"run_status": "success", "error": None, "task_id": "task-1", "result_summary": pem})) + + assert pem_body not in text + assert "END PRIVATE KEY" not in text + assert "[redacted]" in text + + +def test_render_redacts_secret_in_result_summary(): + secret = "ghp_" + ("a" * 36) + text = render_notification_text(_delivery_row(payload={"run_status": "success", "error": None, "task_id": "task-1", "result_summary": f"token={secret}"})) + + assert secret not in text + assert "[redacted]" in text + + +@pytest.mark.asyncio +async def test_run_once_delivers_claimed_row_and_marks_sent(): + repo = FakeDeliveryRepo([_delivery_row()]) + channel = FakeChannel() + worker = _make_worker(repo, channel) + + await worker.run_once(now=datetime.now(UTC)) + + assert repo.sent == ["delivery-1"] + assert repo.failed == [] + assert len(channel.sent) == 1 + target, text = channel.sent[0] + assert target == "GaoZhiChao" + assert "task-1" in text + + +@pytest.mark.asyncio +async def test_run_once_marks_failed_when_channel_is_not_running(): + repo = FakeDeliveryRepo([_delivery_row()]) + worker = _make_worker(repo, channel=None) + + await worker.run_once(now=datetime.now(UTC)) + + assert repo.sent == [] + assert len(repo.failed) == 1 + delivery_id, error, count_attempt = repo.failed[0] + assert delivery_id == "delivery-1" + assert "wecom" in error + # A channel that is simply not running must not burn the retry budget: + # the row comes back on a flat long backoff until the channel returns. + assert count_attempt is False + + +@pytest.mark.asyncio +async def test_run_once_parks_when_registered_channel_reports_not_running(): + repo = FakeDeliveryRepo([_delivery_row()]) + worker = _make_worker(repo, FakeChannel(running=False)) + + await worker.run_once(now=datetime.now(UTC)) + + assert repo.sent == [] + _, error, count_attempt = repo.failed[0] + assert "not running" in error + assert count_attempt is False + + +@pytest.mark.asyncio +async def test_run_once_parks_without_counting_channel_unavailable(): + repo = FakeDeliveryRepo([_delivery_row()]) + worker = _make_worker(repo, FakeChannel(unavailable=True)) + + await worker.run_once(now=datetime.now(UTC)) + + assert repo.sent == [] + delivery_id, error, count_attempt = repo.failed[0] + assert delivery_id == "delivery-1" + assert "transport down" in error + assert count_attempt is False + + +@pytest.mark.asyncio +async def test_run_once_sends_when_target_is_still_connected(): + repo = FakeDeliveryRepo([_delivery_row()]) + channel = FakeChannel() + seen = [] + + async def resolve_connections(owner_user_id): + seen.append(owner_user_id) + return [_connection(), _connection(provider="slack", external_account_id="U1")] + + worker = _make_worker(repo, channel, resolve_connections=resolve_connections) + + await worker.run_once(now=datetime.now(UTC)) + + assert seen == ["user-1"] + assert repo.sent == ["delivery-1"] + assert repo.failed == [] + assert repo.terminal_failures == [] + + +@pytest.mark.asyncio +async def test_run_once_drops_delivery_whose_target_was_disconnected(): + # The owner unbound the identity after the row was enqueued (the row can + # wait up to a day in parking). The push must not go out, and there is + # nothing to wait for: the row is finalized, not parked or retried. + repo = FakeDeliveryRepo([_delivery_row()]) + channel = FakeChannel() + worker = _make_worker(repo, channel, resolve_connections=lambda owner: [_connection(status="revoked")]) + + await worker.run_once(now=datetime.now(UTC)) + + assert channel.sent == [] + assert repo.sent == [] + assert repo.terminal_failures == [("delivery-1", "target is no longer a connected wecom identity of its owner")] + + +@pytest.mark.asyncio +async def test_run_once_ignores_another_owners_binding_of_the_same_target(): + # The resolver is called with the row's own owner, and a returned row that + # names a different owner is skipped too, so a resolver that is not + # owner-scoped still cannot make another user's connected binding of the + # same external account count for this delivery. + repo = FakeDeliveryRepo([_delivery_row()]) + channel = FakeChannel() + calls = [] + + def resolve_connections(owner_user_id): + calls.append(owner_user_id) + return [_connection(owner_user_id="user-OTHER")] + + worker = _make_worker(repo, channel, resolve_connections=resolve_connections) + + await worker.run_once(now=datetime.now(UTC)) + + assert calls == ["user-1"] + assert channel.sent == [] + assert repo.sent == [] + assert len(repo.terminal_failures) == 1 + + +@pytest.mark.asyncio +async def test_run_once_parks_when_connection_lookup_returns_an_unexpected_shape(): + # A dict or a str is iterable and would "find" nothing; that is a lookup + # problem to park on, not proof that the binding is gone. + repo = FakeDeliveryRepo([_delivery_row()]) + channel = FakeChannel() + worker = _make_worker(repo, channel, resolve_connections=lambda owner: {"items": [_connection()]}) + + await worker.run_once(now=datetime.now(UTC)) + + assert channel.sent == [] + delivery_id, error, count_attempt = repo.failed[0] + assert "could not verify" in error + assert count_attempt is False + assert repo.terminal_failures == [] + + +@pytest.mark.asyncio +async def test_run_once_drops_a_disconnected_target_even_while_the_channel_is_down(): + # Disconnecting a provider revokes its connections and stops its channel + # together. The binding check runs first, so the row is finalized instead + # of parking for a day behind the channel-down branch. + repo = FakeDeliveryRepo([_delivery_row()]) + worker = _make_worker(repo, FakeChannel(running=False), resolve_connections=lambda owner: [_connection(status="revoked")]) + + await worker.run_once(now=datetime.now(UTC)) + + assert repo.sent == [] + assert len(repo.terminal_failures) == 1 + assert all(count_attempt is True for _, _, count_attempt in repo.failed) # no park entry was written + + +@pytest.mark.asyncio +async def test_run_once_parks_when_connection_lookup_fails(): + # A store error must neither drop the row nor send against unknown intent. + repo = FakeDeliveryRepo([_delivery_row()]) + channel = FakeChannel() + + async def resolve_connections(owner_user_id): + raise RuntimeError("database unavailable") + + worker = _make_worker(repo, channel, resolve_connections=resolve_connections) + + await worker.run_once(now=datetime.now(UTC)) + + assert channel.sent == [] + assert repo.sent == [] + delivery_id, error, count_attempt = repo.failed[0] + assert delivery_id == "delivery-1" + assert "could not verify" in error + assert count_attempt is False + assert repo.terminal_failures == [] + + +@pytest.mark.asyncio +async def test_run_once_checks_the_connection_before_resolving_the_summary(): + # The run content must not be read for a push that will not be sent. + repo = FakeDeliveryRepo([_delivery_row()]) + channel = FakeChannel() + summary_calls = [] + + async def resolve_run_summary(run_id, user_id): + summary_calls.append(run_id) + return "final answer" + + worker = _make_worker(repo, channel, resolve_run_summary=resolve_run_summary, resolve_connections=lambda owner: []) + + await worker.run_once(now=datetime.now(UTC)) + + assert summary_calls == [] + assert channel.sent == [] + + +@pytest.mark.asyncio +async def test_run_once_marks_failed_when_send_raises(): + repo = FakeDeliveryRepo([_delivery_row()]) + worker = _make_worker(repo, FakeChannel(fail=True)) + + await worker.run_once(now=datetime.now(UTC)) + + assert repo.sent == [] + delivery_id, error, count_attempt = repo.failed[0] + assert delivery_id == "delivery-1" + assert "platform rejected" in error + # A real send error does count against the retry budget. + assert count_attempt is True + + +@pytest.mark.asyncio +async def test_stop_cancels_when_poller_exceeds_timeout(): + repo = FakeDeliveryRepo([]) + worker = NotificationDeliveryWorker( + delivery_repo=repo, + resolve_channel=lambda _provider: None, + poll_interval_seconds=60, + stop_timeout_seconds=0.05, + ) + + async def _hang_forever(): + await asyncio.Event().wait() + + worker._stop.clear() + worker._task = asyncio.create_task(_hang_forever()) + + await worker.stop() + + assert worker._task is None + + +@pytest.mark.asyncio +async def test_run_once_without_due_rows_is_noop(): + repo = FakeDeliveryRepo([]) + worker = _make_worker(repo, FakeChannel()) + + await worker.run_once(now=datetime.now(UTC)) + + assert repo.sent == [] + assert repo.failed == [] + + +# --------------------------------------------------------------------------- +# Crash recovery (review follow-up): a row exploding mid-delivery must not +# strand the rest of the batch in "sending", and rows orphaned between claim +# and the final status write come back through the stale reset. +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_run_once_recovers_stale_sending_rows_before_claiming(): + repo = FakeDeliveryRepo([_delivery_row()]) + worker = _make_worker(repo, FakeChannel()) + now = datetime.now(UTC) + + await worker.run_once(now=now) + + assert repo.events == ["reset", "claim"] + assert repo.resets[0][0] == now + + +@pytest.mark.asyncio +async def test_run_once_isolates_crashes_so_rest_of_batch_still_delivers(): + """Replay the reviewer's worst case: row one crashes even its own + mark_failed (e.g. the DB blip that killed the status write), which used + to abort the loop and strand every following claimed row in + ``sending``.""" + + class ExplodingRepo(FakeDeliveryRepo): + async def mark_failed(self, delivery_id, **kwargs): + if delivery_id == "delivery-1": + raise RuntimeError("db gone") + await super().mark_failed(delivery_id, **kwargs) + + class TargetFailChannel(FakeChannel): + async def send_notification(self, *, target, text_markdown): + if target == "GaoZhiChao": + raise RuntimeError("platform rejected") + await super().send_notification(target=target, text_markdown=text_markdown) + + repo = ExplodingRepo( + [ + _delivery_row(delivery_id="delivery-1"), + _delivery_row(delivery_id="delivery-2", target="bob"), + ] + ) + worker = _make_worker(repo, TargetFailChannel()) + + await worker.run_once(now=datetime.now(UTC)) + + # Row one is left for the stale reset to recover; row two must not be + # sacrificed with it. + assert repo.sent == ["delivery-2"] + assert repo.failed == [] + + +@pytest.mark.asyncio +async def test_start_stop_lifecycle(): + repo = FakeDeliveryRepo([]) + worker = _make_worker(repo, FakeChannel()) + + await worker.start() + await worker.stop() + # Idempotent stop mirrors the scheduler service contract. + await worker.stop() + + +# --------------------------------------------------------------------------- +# Result summary enrichment (issue #4254): completed runs carry the agent's +# final answer into the IM push; failed runs keep the error-only skeleton. +# --------------------------------------------------------------------------- + + +def test_render_includes_result_summary_for_completed_runs(): + text = render_notification_text(_delivery_row(payload={"run_status": "success", "error": None, "task_id": "task-1", "result_summary": "The answer is 42"})) + + assert "The answer is 42" in text + + +def test_render_truncates_long_result_summary(): + text = render_notification_text(_delivery_row(payload={"run_status": "success", "error": None, "task_id": "task-1", "result_summary": "x" * 5000})) + + # Bounded on the wire: 999 chars of summary + the ellipsis marker. + result_line = next(line for line in text.splitlines() if line.startswith("Result: ")) + assert len(result_line) == len("Result: ") + 1000 + assert result_line.endswith("\u2026") + + +def test_render_ignores_result_summary_for_failed_runs(): + text = render_notification_text( + _delivery_row( + event="run_failed", + payload={"run_status": "failed", "error": "boom", "task_id": "task-1", "result_summary": "partial answer"}, + ) + ) + + assert "partial answer" not in text + + +@pytest.mark.asyncio +async def test_delivery_enriches_completed_run_with_summary(): + row = _delivery_row() + repo = FakeDeliveryRepo([row]) + channel = FakeChannel() + seen = [] + + def resolver(run_id, user_id): + seen.append((run_id, user_id)) + return "The answer is 42" + + worker = _make_worker(repo, channel, resolve_run_summary=resolver) + + await worker.run_once(now=datetime.now(UTC)) + + assert seen == [("run-1", "user-1")] + assert repo.sent == ["delivery-1"] + assert "The answer is 42" in channel.sent[0][1] + # The claimed outbox row itself must not be mutated in place. + assert "result_summary" not in row["payload"] + + +@pytest.mark.asyncio +async def test_delivery_supports_async_summary_resolver(): + repo = FakeDeliveryRepo([_delivery_row()]) + channel = FakeChannel() + + async def resolver(run_id, user_id): + return "async answer" + + worker = _make_worker(repo, channel, resolve_run_summary=resolver) + + await worker.run_once(now=datetime.now(UTC)) + + assert repo.sent == ["delivery-1"] + assert "async answer" in channel.sent[0][1] + + +@pytest.mark.asyncio +async def test_delivery_skips_resolver_for_failed_runs(): + repo = FakeDeliveryRepo([_delivery_row(event="run_failed", payload={"run_status": "failed", "error": "boom", "task_id": "task-1"})]) + channel = FakeChannel() + called = [] + + def resolver(run_id, user_id): + called.append(run_id) + return "should not appear" + + worker = _make_worker(repo, channel, resolve_run_summary=resolver) + + await worker.run_once(now=datetime.now(UTC)) + + assert called == [] + assert repo.sent == ["delivery-1"] + + +@pytest.mark.asyncio +async def test_delivery_falls_back_to_skeleton_when_resolver_raises(): + repo = FakeDeliveryRepo([_delivery_row()]) + channel = FakeChannel() + + def resolver(run_id, user_id): + raise RuntimeError("db unavailable") + + worker = _make_worker(repo, channel, resolve_run_summary=resolver) + + await worker.run_once(now=datetime.now(UTC)) + + # A summary lookup failure must never block the notification itself. + assert repo.sent == ["delivery-1"] + assert repo.failed == [] + assert "task-1" in channel.sent[0][1] + + +@pytest.mark.asyncio +async def test_delivery_sends_skeleton_when_summary_is_missing(): + repo = FakeDeliveryRepo([_delivery_row()]) + channel = FakeChannel() + worker = _make_worker(repo, channel, resolve_run_summary=lambda run_id, user_id: None) + + await worker.run_once(now=datetime.now(UTC)) + + assert repo.sent == ["delivery-1"] + text = channel.sent[0][1] + assert "task-1" in text + assert "Result:" not in text + + +class FakeWsClient: + def __init__(self, *, errcode=0): + self.calls = [] + self.errcode = errcode + + async def send_message(self, chatid, body): + self.calls.append((chatid, body)) + return {"errcode": self.errcode, "errmsg": "ok"} + + +def _wecom_channel(ws_client=None): + channel = WeComChannel(bus=MessageBus(), config={}) + channel._ws_client = ws_client + channel._running = True + channel._ws_transport_up = ws_client is not None + return channel + + +@pytest.mark.asyncio +async def test_wecom_send_notification_pushes_markdown(): + ws_client = FakeWsClient() + channel = _wecom_channel(ws_client) + + await channel.send_notification(target="GaoZhiChao", text_markdown="**done**") + + assert len(ws_client.calls) == 1 + chatid, body = ws_client.calls[0] + assert chatid == "GaoZhiChao" + assert body == {"msgtype": "markdown", "markdown": {"content": "**done**"}} + + +@pytest.mark.asyncio +async def test_wecom_send_notification_raises_on_platform_error(): + ws_client = FakeWsClient(errcode=853000) + channel = _wecom_channel(ws_client) + + with pytest.raises(RuntimeError, match="853000"): + await channel.send_notification(target="GaoZhiChao", text_markdown="**done**") + + +@pytest.mark.asyncio +async def test_wecom_send_notification_raises_when_not_connected(): + channel = _wecom_channel(None) + channel._running = True + + with pytest.raises(ChannelUnavailable, match="not connected"): + await channel.send_notification(target="GaoZhiChao", text_markdown="**done**") + + +@pytest.mark.asyncio +async def test_wecom_send_notification_parks_when_the_sdk_socket_is_closed(): + # The real SDK client raises a plain RuntimeError when its socket is not open, + # which would count against the retry budget instead of parking the row. + aibot = pytest.importorskip("aibot") + ws_client = aibot.WSClient(aibot.WSClientOptions(bot_id="bot", secret="secret")) + assert ws_client.is_connected is False + channel = _wecom_channel(ws_client) + + with pytest.raises(ChannelUnavailable, match="not connected"): + await channel.send_notification(target="GaoZhiChao", text_markdown="**done**") + + assert channel._ws_transport_up is False + + +@pytest.mark.asyncio +async def test_wecom_send_notification_wraps_transport_errors_as_unavailable(): + class BrokenWsClient: + async def send_message(self, chatid, body): + raise ConnectionError("websocket closed") + + channel = _wecom_channel(BrokenWsClient()) + channel._ws_transport_up = False # single probe; no retry/sleep storm + + with pytest.raises(ChannelUnavailable, match="transport unavailable"): + await channel.send_notification(target="GaoZhiChao", text_markdown="**done**") + + assert channel._ws_transport_up is False + + +@pytest.mark.asyncio +async def test_wecom_send_notification_parks_on_the_sdk_runtime_error(): + # The SDK's send path raises plain RuntimeError for a closed socket, a + # connection dropped with replies pending and a full reply queue; none is + # a ConnectionError/OSError. With is_connected still (stale) True this used + # to escape the ChannelUnavailable wrapping and count against the budget. + class StaleSocketWsClient: + is_connected = True + + def __init__(self): + self.calls = 0 + + async def send_message(self, chatid, body): + self.calls += 1 + raise RuntimeError("WebSocket not connected, unable to send data") + + ws_client = StaleSocketWsClient() + channel = _wecom_channel(ws_client) + channel._ws_transport_up = False # single probe; no retry/sleep storm + + with pytest.raises(ChannelUnavailable, match="transport unavailable"): + await channel.send_notification(target="GaoZhiChao", text_markdown="**done**") + + assert ws_client.calls == 1 + assert channel._ws_transport_up is False + + +@pytest.mark.asyncio +async def test_wecom_send_notification_counts_the_sdk_ack_rejection(): + # The SDK never returns a non-zero errcode ACK as a dict: _handle_reply_ack + # turns it into RuntimeError("Reply ack error: ...") from send_message. The + # socket is fine, so this must count against the budget, not park. + class RejectingWsClient: + is_connected = True + + async def send_message(self, chatid, body): + raise RuntimeError("Reply ack error: errcode=60111, errmsg=invalid userid") + + channel = _wecom_channel(RejectingWsClient()) + channel._ws_transport_up = False # single probe; no retry/sleep storm + + with pytest.raises(RuntimeError, match="errcode=60111"): + await channel.send_notification(target="GaoZhiChao", text_markdown="**done**") + + # The ACK came back on this socket, so the transport is known to be up. + assert channel._ws_transport_up is True + + +@pytest.mark.asyncio +async def test_wecom_send_notification_rejection_prefix_survives_the_retry_branch(monkeypatch): + # With the transport believed up, the send runs under _send_with_retry; the + # prefix check relies on the helper re-raising the SDK's own exception + # object rather than wrapping it. + import app.channels.base as base_module + + async def no_sleep(_delay): + return None + + monkeypatch.setattr(base_module.asyncio, "sleep", no_sleep) + + class RejectingWsClient: + is_connected = True + + def __init__(self): + self.calls = 0 + + async def send_message(self, chatid, body): + self.calls += 1 + raise RuntimeError("Reply ack error: errcode=60111, errmsg=invalid userid") + + ws_client = RejectingWsClient() + channel = _wecom_channel(ws_client) + assert channel._ws_transport_up is True + + with pytest.raises(RuntimeError, match="^Reply ack error: errcode=60111"): + await channel.send_notification(target="GaoZhiChao", text_markdown="**done**") + + assert ws_client.calls == 3 + assert channel._ws_transport_up is True + + +@pytest.mark.asyncio +async def test_wecom_send_notification_parks_on_the_socket_layer_close_exception(): + websockets_exceptions = pytest.importorskip("websockets.exceptions") + websockets_frames = pytest.importorskip("websockets.frames") + + class ClosingWsClient: + is_connected = True + + async def send_message(self, chatid, body): + close = websockets_frames.Close(1006, "abnormal closure") + raise websockets_exceptions.ConnectionClosedError(close, None) + + channel = _wecom_channel(ClosingWsClient()) + channel._ws_transport_up = False # single probe; no retry/sleep storm + + with pytest.raises(ChannelUnavailable, match="transport unavailable"): + await channel.send_notification(target="GaoZhiChao", text_markdown="**done**") + + assert channel._ws_transport_up is False + + +@pytest.mark.asyncio +async def test_wecom_send_notification_propagates_deterministic_errors(): + class InvalidTargetWsClient: + async def send_message(self, chatid, body): + raise ValueError("invalid target") + + channel = _wecom_channel(InvalidTargetWsClient()) + + with pytest.raises(ValueError, match="invalid target"): + await channel.send_notification(target="GaoZhiChao", text_markdown="**done**") + + +@pytest.mark.asyncio +async def test_binding_check_reads_the_real_connection_repository_shape(tmp_path): + # The worker matches on the dict keys the real repository emits; a key + # rename there would otherwise fail every delivery silently. + from deerflow.persistence.channel_connections import ChannelConnectionRepository + from deerflow.persistence.engine import close_engine, get_session_factory, init_engine + + await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path / 'conn.db'}", sqlite_dir=str(tmp_path)) + try: + conn_repo = ChannelConnectionRepository(get_session_factory()) + bound = await conn_repo.upsert_connection(owner_user_id="user-1", provider="wecom", external_account_id="GaoZhiChao") + repo = FakeDeliveryRepo([_delivery_row()]) + worker = _make_worker(repo, FakeChannel(), resolve_connections=conn_repo.list_connections) + + assert await worker._target_still_connected(_delivery_row()) is True + # Another owner's row never counts for this delivery. + assert await worker._target_still_connected({**_delivery_row(), "owner_user_id": "user-2"}) is False + + assert await conn_repo.disconnect_connection(connection_id=bound["id"], owner_user_id="user-1") is True + assert await worker._target_still_connected(_delivery_row()) is False + finally: + await close_engine() diff --git a/backend/tests/test_scheduled_task_service.py b/backend/tests/test_scheduled_task_service.py index af03ae64da..1d563d3648 100644 --- a/backend/tests/test_scheduled_task_service.py +++ b/backend/tests/test_scheduled_task_service.py @@ -529,6 +529,7 @@ def _once_task_row(task_id="task-once", status="running"): "thread_id": None, "context_mode": "fresh_thread_per_run", "assistant_id": "lead_agent", + "title": "Once summary", "prompt": "Summarize thread", "schedule_type": "once", "schedule_spec": {"run_at": "2026-07-02T01:00:00+00:00"}, @@ -537,7 +538,7 @@ def _once_task_row(task_id="task-once", status="running"): } -def _completion_record(status, *, task_id="task-once", error=None): +def _completion_record(status, *, task_id="task-once", error=None, trigger="scheduled"): return RunRecord( run_id="run-x", thread_id="thread-x", @@ -547,6 +548,7 @@ def _completion_record(status, *, task_id="task-once", error=None): metadata={ "scheduled_task_id": task_id, "scheduled_task_run_id": "task-run-x", + "scheduled_trigger": trigger, }, user_id="user-1", error=error, @@ -1187,3 +1189,308 @@ async def test_manual_trigger_proceeds_when_global_budget_available(): assert result["outcome"] == "launched" assert len(launched) == 1 + + +# --------------------------------------------------------------------------- +# Notification enqueue on completion (issue #4254): the hook writes durable +# outbox rows; a separate delivery worker does the actual IM send. +# --------------------------------------------------------------------------- + + +class DummyConnectionRepo: + def __init__(self, connections): + self.connections = connections + + async def list_connections(self, owner_user_id): + return [dict(connection, owner_user_id=owner_user_id) for connection in self.connections] + + +class DummyNotificationRepo: + def __init__(self, *, fail=False): + self.enqueued = [] + self.fail = fail + + async def enqueue(self, **kwargs): + if self.fail: + raise RuntimeError("delivery backend unavailable") + self.enqueued.append(kwargs) + return {"id": f"delivery-{len(self.enqueued)}", **kwargs} + + +def _make_notifying_service(task_repo, run_repo, *, connection_repo, notification_repo): + return ScheduledTaskService( + task_repo=task_repo, + task_run_repo=run_repo, + launch_run=lambda **_kwargs: None, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_runs=3, + connection_repo=connection_repo, + notification_repo=notification_repo, + ) + + +@pytest.mark.asyncio +async def test_completion_enqueues_run_completed_for_connected_channels(): + task_repo = DummyTaskRepo([_once_task_row()]) + run_repo = DummyRunRepo() + connection_repo = DummyConnectionRepo( + [ + {"provider": "wecom", "external_account_id": "GaoZhiChao", "status": "connected"}, + ] + ) + notification_repo = DummyNotificationRepo() + service = _make_notifying_service( + task_repo, + run_repo, + connection_repo=connection_repo, + notification_repo=notification_repo, + ) + + await service.handle_run_completion(_completion_record(RunStatus.success)) + + assert len(notification_repo.enqueued) == 1 + enqueued = notification_repo.enqueued[0] + assert enqueued["event"] == "run_completed" + assert enqueued["provider"] == "wecom" + assert enqueued["target"] == "GaoZhiChao" + assert enqueued["task_id"] == "task-once" + assert enqueued["task_run_id"] == "task-run-x" + assert enqueued["run_id"] == "run-x" + assert enqueued["owner_user_id"] == "user-1" + assert enqueued["payload"]["run_status"] == "success" + assert enqueued["payload"]["error"] is None + assert enqueued["payload"]["task_title"] == "Once summary" + + +@pytest.mark.asyncio +async def test_completion_enqueues_run_failed_event_on_error(): + task_repo = DummyTaskRepo([_once_task_row()]) + run_repo = DummyRunRepo() + connection_repo = DummyConnectionRepo( + [ + {"provider": "wecom", "external_account_id": "GaoZhiChao", "status": "connected"}, + ] + ) + notification_repo = DummyNotificationRepo() + service = _make_notifying_service( + task_repo, + run_repo, + connection_repo=connection_repo, + notification_repo=notification_repo, + ) + + await service.handle_run_completion(_completion_record(RunStatus.error, error="boom")) + + assert len(notification_repo.enqueued) == 1 + enqueued = notification_repo.enqueued[0] + assert enqueued["event"] == "run_failed" + assert enqueued["payload"]["run_status"] == "failed" + assert enqueued["payload"]["error"] == "boom" + + +@pytest.mark.asyncio +async def test_completion_does_not_notify_for_interrupted_runs(): + task_repo = DummyTaskRepo([_once_task_row()]) + run_repo = DummyRunRepo() + connection_repo = DummyConnectionRepo( + [ + {"provider": "wecom", "external_account_id": "GaoZhiChao", "status": "connected"}, + ] + ) + notification_repo = DummyNotificationRepo() + service = _make_notifying_service( + task_repo, + run_repo, + connection_repo=connection_repo, + notification_repo=notification_repo, + ) + + await service.handle_run_completion(_completion_record(RunStatus.interrupted)) + + # An interrupt is a user-initiated cancel; the user is already aware. + assert notification_repo.enqueued == [] + + +@pytest.mark.asyncio +async def test_completion_does_not_notify_for_manual_triggers(): + """A manual "run now" happens with the user watching the UI; pushing the + same outcome to IM is noise, so only scheduled firings notify.""" + task_repo = DummyTaskRepo([_once_task_row()]) + run_repo = DummyRunRepo() + connection_repo = DummyConnectionRepo( + [ + {"provider": "wecom", "external_account_id": "GaoZhiChao", "status": "connected"}, + ] + ) + notification_repo = DummyNotificationRepo() + service = _make_notifying_service( + task_repo, + run_repo, + connection_repo=connection_repo, + notification_repo=notification_repo, + ) + + await service.handle_run_completion(_completion_record(RunStatus.success, trigger="manual")) + + # Execution status is still recorded; only the IM push is suppressed. + assert task_repo.completions[-1][1]["status"] == "success" + assert notification_repo.enqueued == [] + + +@pytest.mark.asyncio +async def test_completion_still_notifies_when_trigger_metadata_is_missing(): + """Fail safe: an untagged completion (older rows, unexpected paths) keeps + notifying -- dropping a push silently is worse than a rare extra one.""" + task_repo = DummyTaskRepo([_once_task_row()]) + run_repo = DummyRunRepo() + connection_repo = DummyConnectionRepo( + [ + {"provider": "wecom", "external_account_id": "GaoZhiChao", "status": "connected"}, + ] + ) + notification_repo = DummyNotificationRepo() + service = _make_notifying_service( + task_repo, + run_repo, + connection_repo=connection_repo, + notification_repo=notification_repo, + ) + + record = _completion_record(RunStatus.success) + record.metadata.pop("scheduled_trigger") + await service.handle_run_completion(record) + + assert len(notification_repo.enqueued) == 1 + + +@pytest.mark.asyncio +async def test_completion_skips_unusable_connections(): + task_repo = DummyTaskRepo([_once_task_row()]) + run_repo = DummyRunRepo() + connection_repo = DummyConnectionRepo( + [ + {"provider": "wecom", "external_account_id": "revoked-user", "status": "revoked"}, + {"provider": "wecom", "external_account_id": None, "status": "connected"}, + {"provider": "wecom", "external_account_id": "GaoZhiChao", "status": "connected"}, + ] + ) + notification_repo = DummyNotificationRepo() + service = _make_notifying_service( + task_repo, + run_repo, + connection_repo=connection_repo, + notification_repo=notification_repo, + ) + + await service.handle_run_completion(_completion_record(RunStatus.success)) + + assert [item["target"] for item in notification_repo.enqueued] == ["GaoZhiChao"] + + +@pytest.mark.asyncio +async def test_completion_notification_failure_does_not_break_status_writes(): + task_repo = DummyTaskRepo([_once_task_row()]) + run_repo = DummyRunRepo() + connection_repo = DummyConnectionRepo( + [ + {"provider": "wecom", "external_account_id": "GaoZhiChao", "status": "connected"}, + ] + ) + notification_repo = DummyNotificationRepo(fail=True) + service = _make_notifying_service( + task_repo, + run_repo, + connection_repo=connection_repo, + notification_repo=notification_repo, + ) + + await service.handle_run_completion(_completion_record(RunStatus.success)) + + # Execution status is written regardless: delivery is best-effort and must + # never shadow the run outcome (execution/delivery status separation). + assert task_repo.completions[-1][1]["status"] == "success" + + +def _notifying_service_with_wecom(task_repo, notification_repo): + connection_repo = DummyConnectionRepo([{"provider": "wecom", "external_account_id": "GaoZhiChao", "status": "connected"}]) + return _make_notifying_service(task_repo, DummyRunRepo(), connection_repo=connection_repo, notification_repo=notification_repo) + + +@pytest.mark.asyncio +async def test_detached_outbox_stops_the_enqueue(): + # The Gateway detaches the outbox when the delivery side cannot start. + task_repo = DummyTaskRepo([_once_task_row()]) + notification_repo = DummyNotificationRepo() + service = _notifying_service_with_wecom(task_repo, notification_repo) + + await service.handle_run_completion(_completion_record(RunStatus.success)) + assert [row["event"] for row in notification_repo.enqueued] == ["run_completed"] + + service.detach_notification_outbox() + await service.handle_run_completion(_completion_record(RunStatus.success)) + + assert len(notification_repo.enqueued) == 1 + assert len(task_repo.completions) == 2 + + +@pytest.mark.asyncio +async def test_completion_that_was_not_recorded_does_not_notify(): + # complete_run returns False for a missing occurrence, another run's occurrence or another owner. + class RejectingTaskRepo(DummyTaskRepo): + async def complete_run(self, task_id, **kwargs): + await super().complete_run(task_id, **kwargs) + return False + + task_repo = RejectingTaskRepo([_once_task_row()]) + notification_repo = DummyNotificationRepo() + + await _notifying_service_with_wecom(task_repo, notification_repo).handle_run_completion(_completion_record(RunStatus.success)) + + assert len(task_repo.completions) == 1 + assert notification_repo.enqueued == [] + + +@pytest.mark.asyncio +async def test_completion_for_a_task_deleted_meanwhile_does_not_notify(): + task_repo = DummyTaskRepo([]) + notification_repo = DummyNotificationRepo() + + await _notifying_service_with_wecom(task_repo, notification_repo).handle_run_completion(_completion_record(RunStatus.success)) + + assert task_repo.completions[-1][1]["status"] == "success" + assert notification_repo.enqueued == [] + + +@pytest.mark.asyncio +async def test_completion_still_notifies_when_the_task_read_fails(): + class UnreadableTaskRepo(DummyTaskRepo): + async def get(self, task_id, *, user_id): + raise RuntimeError("database unavailable") + + task_repo = UnreadableTaskRepo([_once_task_row()]) + notification_repo = DummyNotificationRepo() + + await _notifying_service_with_wecom(task_repo, notification_repo).handle_run_completion(_completion_record(RunStatus.success)) + + assert task_repo.completions[-1][1]["status"] == "success" + # The title is optional: the push still goes out and falls back to the task id. + assert [row["payload"]["task_title"] for row in notification_repo.enqueued] == [None] + + +@pytest.mark.asyncio +async def test_completion_without_the_outbox_does_not_read_the_task(): + class CountingTaskRepo(DummyTaskRepo): + reads = 0 + + async def get(self, task_id, *, user_id): + self.reads += 1 + return await super().get(task_id, user_id=user_id) + + task_repo = CountingTaskRepo([_once_task_row()]) + service = _make_notifying_service(task_repo, DummyRunRepo(), connection_repo=None, notification_repo=None) + + await service.handle_run_completion(_completion_record(RunStatus.success)) + + assert task_repo.completions[-1][1]["status"] == "success" + assert task_repo.reads == 0 diff --git a/config.example.yaml b/config.example.yaml index 068be4033a..6cf19e4cec 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -2693,10 +2693,15 @@ agent_storage: # when every Gateway instance shares Postgres, run ownership heartbeats, and # database-backed run events; startup recovery then preserves live peer runs. # +# When channel_connections.enabled is true and IM channels are running, the +# same poll_interval_seconds also drives the notification delivery worker that +# pushes scheduled-run outcomes to the owner's bound IM identities (WeCom +# proactive push today). Manual "run now" and interrupts stay silent. +# # scheduler: # enabled: false # Master switch for the background poller # multi_instance: false # Opt into lease-aware recovery across Gateway instances -# poll_interval_seconds: 5 # How often to scan for due tasks +# poll_interval_seconds: 5 # How often to scan for due tasks (+ IM outbox delivery) # lease_seconds: 120 # Claim lease; a crashed process's task becomes reclaimable after this # max_concurrent_runs: 3 # Global cap on launching/running scheduled runs across multi-instance Pods # queue_timeout_seconds: 3600 # Maximum durable queue wait before an occurrence fails @@ -2827,6 +2832,9 @@ run_ownership: # - Telegram uses a deep link when `bot_username` is configured. # - Slack, Discord, Feishu, DingTalk, WeChat, and WeCom use `/connect ` # through the already-running bot/app. +# - When enabled (and channels are running), scheduled-task completions/failures +# also enqueue IM pushes to the owner's connected identities (see scheduler +# section above). Manual triggers and interrupts do not notify. # # channel_connections: # enabled: false