feat(scheduler): push scheduled task results to IM channels via a notification outbox (#6135)

* feat(persistence): add notification delivery outbox for scheduled task IM push

New notification_deliveries table and repository (migration 0012) backing
the outbox pattern for issue #4254: idempotent enqueue keyed on task_run_id
plus event plus provider plus target, atomic claim of due rows, and
exponential-backoff retry state moving pending to sending to sent or failed.
Execution status stays in scheduled_task_runs; delivery status lives here.

* feat(scheduler): enqueue and deliver scheduled-task notifications over IM channels

Closes the loop for issue #4254. On run completion the scheduler writes
outbox rows for the task owner's connected channel identities (success and
failure events; interrupts stay silent), and a new NotificationDeliveryWorker
polls due rows and pushes them through the channel's proactive send path.
Channel gains send_notification; WeCom implements it via the WS client's
send_message with errcode checking, the base class raises for providers
without proactive push.

Delivery is best-effort and never shadows execution status: enqueue and send
failures are logged and recorded in the outbox, which retries with backoff.
Channel liveness is re-checked at delivery time, and the worker stops before
the channel service on shutdown. Feature activates only when channel
connections are enabled; no new dependencies.

Verified end to end against a live WeCom bot: scheduled run completion
produced an outbox row that was delivered in under one second.

* feat(scheduler): include scheduled run result summary in IM notifications

The v1 notification only carried the run outcome skeleton (status, task id,
run id). Completed runs now include the agent's final answer so the IM push
is actually useful headlessly.

- The delivery worker accepts an optional resolve_run_summary(run_id,
  user_id) resolver; for run_completed events it enriches the rendered text
  with a "Result:" line (bounded to 1000 chars). The lookup is best-effort:
  resolver failures or missing summaries degrade to the skeleton text and
  never block or fail the delivery.
- The summary is read at delivery time from runs.last_ai_message (the
  convenience column written on run completion), not snapshotted into the
  outbox payload; retried deliveries re-read the freshest value and the
  outbox row payload stays skeleton-only. The claimed outbox row is never
  mutated in place.
- Gateway wiring scopes the read to the outbox row's owner_user_id so a
  stale row cannot pull another user's run content.
- Failed runs keep the error-only skeleton: a partial answer on a failed
  run would be misleading.

Tests: 9 new worker cases (rendering incl. truncation, sync/async
resolvers, failure/missing fallbacks, no resolver call for failed runs).
Full scheduled + outbox + channels regression green (441 passed), ruff
check + format clean.

* fix(scheduler): address review feedback on notification delivery reliability

Review follow-ups for the scheduled-task IM notification outbox:

1. Crash recovery for "sending" rows: claim now stamps updated_at and the
   worker reconciles rows stuck in "sending" past a 10-minute lease back to
   "pending" before every claim. Per-row try/except isolates a crashing
   delivery from the rest of the claimed batch.
2. Double-send race under multiple workers: claim returns exactly the rows
   its own UPDATE...RETURNING flipped (guarded by status='pending'), so a
   rival worker's concurrent claim can no longer be handed out twice.
3. Restore non-ASCII comments and drop the BOM accidentally introduced in
   test_scheduled_task_service.py.
4. Manual "run now" completions no longer enqueue IM notifications; the
   user is already watching the UI. Missing trigger metadata fails safe
   towards notifying.
5. Channel-outage failures no longer consume the retry budget: they are
   parked with count_attempt=False on a flat 15-minute backoff so an
   hours-long outage cannot exhaust retries and drop the notification.

Tests: repository 16, worker 21, scheduled-task service 31; targeted
regression across notification/scheduler/channel suites: 511 passed.

* fix(scheduler): address round-2 review on IM notification delivery

Park WeCom transport/outage failures without burning retries, bound worker stop, include task titles, align enqueue with delivery, and document the IM push behavior.

Co-authored-by: Cursor <cursoragent@cursor.com>

* docs(scheduler): complete IM notification docs and drop unused list API

Add backend/channels/migrations AGENTS notes, CHANGELOG and config.example hints for the outbox push path, and remove the unused list_by_task_run repository method flagged in review.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(scheduler): address round-3 review on IM notification delivery

Cap channel-down parking by row age and parked attempts, redact IM egress
content, narrow WeCom transport exceptions, add lifespan wiring tests, and
run ruff format so lint-backend passes.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(scheduler): redact entire PEM blocks in IM egress text

The previous BEGIN-only pattern left key material and the END marker in notifications.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(persistence): align 0014 parked_attempts with bootstrap schema tests

Backfill existing rows with a temporary server default, then drop it so create_all matches Alembic, and move HEAD anchors to 0014.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(persistence): format notification tests and shorten alembic revision id

ruff format --check failed on test_notification_delivery_worker.py.
0015_notification_delivery_parked_attempts exceeds alembic_version.version_num
(VARCHAR 32) and truncates on Postgres CI; use 0015_parked_attempts instead.

* test: fix post-merge CI failures after migration renumber

Update migration 0015 test HEAD expectation to 0017_parked_attempts and add queue_timeout_seconds to the lifespan notification worker mock scheduler config.

* fix(persistence): chain the notification outbox migrations after 0026

Main's head moved from 0016 to 0026_mcp_task_lease_tokens, so the two
outbox revisions are renumbered and re-parented:

- 0017_notification_deliveries -> 0027_notification_deliveries
- 0018_parked_attempts -> 0028_parked_attempts

Only the revision ids, parents and the migration guide change.

* fix(scheduler): adapt the notification hook and its docs to main

- tests: the completion tests assert on the atomic complete_run record;
  new cases cover a completion that was not recorded, a task deleted
  meanwhile, a failed task read, and the outbox being off.
- AGENTS.md: drop the root note and shorten the backend note to a pointer;
  the two notes pushed two effective guidance chains past the 96 KiB limit.
- CHANGELOG: move the entry from the released 2.1.0 section to Unreleased
  and add its link references.

* test(persistence): move the chain-head pin to the outbox migrations

test_migration_0026 pinned 0026 as the head, so it failed once the outbox
revisions chained after it. It now checks chain membership like the older
migration tests, and a new test file owns the head pin, checks the parents
and the 32-character revision id limit, and runs the upgrade and downgrade.

* fix(gateway): wire scheduled-task notifications after the channel service starts

The outbox wiring sat in the scheduler block and asked for the running
channel service there. #5035 moved the channel service start to after the
scheduler, and the two changes merged without a conflict, so on every
startup the lookup returned None and the feature disabled itself with the
"write-only outbox" warning. The lifespan tests patched the lookup to a
constant and kept passing.

The wiring now runs in its own step after the channel service start. The
scheduler is built as on main, and ScheduledTaskService gains
attach_notification_outbox() so the enqueue side is enabled once the
delivery worker is running. The guard, the worker and the shutdown order
are unchanged.

Tests: the lifespan helper can model the real accessor (None until the
channel service has started); a new case fails without this fix.

* fix(scheduler): enqueue from scheduler start and keep the push when the title read fails

Follow-up to the startup-order fix, from review.

- The enqueue side is wired with the scheduler again (it only needs the
  durable table), and only the delivery worker waits for the channel
  service. Attaching it later left a window after the scheduler started in
  which a finished run wrote no outbox row. If the channel service is
  missing or the worker cannot start, the enqueue side is detached, so the
  outbox still never becomes write-only.
- A failed task read after complete_run no longer drops the notification:
  the title is optional and the message falls back to the task id. A task
  deleted meanwhile still stays silent.

* fix(channels): park WeCom notifications while the SDK socket is closed

The WeCom SDK raises a plain RuntimeError when its socket is not open, and
send_notification only mapped ConnectionError, TimeoutError and OSError to
ChannelUnavailable. A disconnect therefore spent the counted retry budget
and settled the row as failed after about 15 minutes, although the docs
say transport outages park without consuming retries.

send_notification now asks the client's is_connected first. The new test
uses a real, never-connected SDK client: it raised RuntimeError after three
retries before, and parks now.

* docs(scheduler): describe which outcomes notify; mirror the changelog entry

- README (en/zh) and CONFIGURATION.md: only runs that finish as success or
  failed notify; occurrences that end without a finished run (launch error,
  queue timeout, restart recovery) do not. Parking lasts up to about a day.
- CHANGELOG: cite the issue as text like the rest of the file, and add the
  entry and link reference to CHANGELOG_zh.md.
- AGENTS.md: drop the backend pointer so this branch adds nothing to the
  guidance chains closest to their size limit; the scheduler-side rules
  and the lifespan order now live in app/channels/AGENTS.md.

* docs(changelog): cite #6135 next to #4843 for the notification outbox entry

* fix(scheduler,channels): re-check the IM binding at delivery time; park WeCom transport errors, count its rejections

Review round 1 on #6135.

The delivery worker now takes resolve_connections (the Gateway wires
ChannelConnectionRepository.list_connections) and, before the channel
liveness check and the summary read, requires the row's owner to still
hold a connected (provider, external_account_id). A disconnected target
is finalized with mark_failed(terminal=True); a lookup failure or an
unexpected result shape parks the row. Terminal outbox rows ignore
later status writes.

WeCom: the SDK raises RuntimeError from send_message for a socket that
is not open and for a dropped connection, but also for a platform
rejection ("Reply ack error: errcode=..."). Only that prefix keeps
counting; other RuntimeErrors, ConnectionError/TimeoutError/OSError and
websockets' ConnectionClosed park the row as ChannelUnavailable.

* fix(channels): treat a WeCom ACK rejection as proof the transport is up; pin the rejection prefix through the retry branch and the real connection-row shape

---------

Co-authored-by: GaoTianyuanhr <75788039+GaoTianyuanhr@users.noreply.github.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: GaoTianyuanhr <GaoTianyuanhr@users.noreply.github.com>
Co-authored-by: Totoro-qaq <279883115+Totoro-qaq@users.noreply.github.com>
This commit is contained in:
Totoro 2026-10-01 17:59:30 +08:00 committed by GitHub
parent 67db3d883c
commit b57fa6fb9e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
25 changed files with 2920 additions and 13 deletions

View File

@ -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

View File

@ -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

View File

@ -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.

View File

@ -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 资源和执行路径。

File diff suppressed because one or more lines are too long

View File

@ -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(

View File

@ -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

View File

@ -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

View File

@ -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

View File

@ -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:

View File

@ -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

View File

@ -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)

View File

@ -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")

View File

@ -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")

View File

@ -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",

View File

@ -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",
]

View File

@ -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"),
)

View File

@ -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)

View File

@ -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()

View File

@ -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):

View File

@ -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()

View File

@ -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()

View File

@ -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()

View File

@ -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

View File

@ -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 <code>`
# 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