Repository navigation
feat(scheduler): push scheduled task results to IM channels via a notification outbox - #4843
GaoTianyuanhr wants to merge 14 commits into
Conversation
|
|
github-actions
Bot
added
area:backend
…k IM push New notification_deliveries table and repository (migration 0012) backing the outbox pattern for issue bytedance#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.
…r IM channels Closes the loop for issue bytedance#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.
c410904 to
9b8974b
Compare
…ions 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.
willem-bd
left a comment
willem-bd
left a comment
There was a problem hiding this comment.
Overall this is a well-structured outbox implementation: enqueue/send separation, a real idempotency key, execution/delivery status isolation, bounded notification text, and thorough tests. Findings below, ranked. The first two are durability/concurrency gaps in the outbox itself; the third is an encoding regression in an existing test file.
Also worth a look (no inline thread): notification_deliveries rows are never cleaned up — one sent/failed row per run per connected channel accumulates forever; a retention sweep would be a cheap follow-up.
…ability 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.
|
@willem-bd All 5 comments have been addressed. Ready for re-review when you have a chance. Thanks! |
willem-bd
left a comment
willem-bd
left a comment
There was a problem hiding this comment.
Follow-up review at 7305e11. Verified all five round-1 findings are properly fixed: per-row crash isolation plus stale-sending reconciliation in the worker, the RETURNING-based claim that closes the multi-worker double-send window (with a deterministic race test), the UTF-8 BOM/mojibake fully restored in test_scheduled_task_service.py, the manual-trigger suppression guard, and attempt-free parking for channel-down failures. The comments below are the residual suggestions: the WeCom "not connected" path still burns the counted retry budget (the one outage shape the parking fix does not reach), the worker's stop() is unbounded while the adjacent channel-service shutdown is deliberately bounded, and the new user-facing IM-push behavior ships without the README/docs note the repo's documentation policy asks for.
| retry on rejection instead of treating an ACK-less call as success. | ||
| """ | ||
| if not self._ws_client: | ||
| raise RuntimeError("WeCom channel is not connected") |
There was a problem hiding this comment.
[suggestion] A registered-but-disconnected WeCom channel still burns the counted retry budget — this is the one outage shape the new parking fix doesn't reach.
When the channel object exists but its WS client is down, _deliver resolves it as "running" and send_notification raises here, which goes through mark_failed(..., count_attempt=True): 5 attempts over ~15 minutes of backoff, then the row settles failed permanently. That is exactly the "WeCom reconnect outlasts the window" case the round-1 thread named, but the count_attempt=False parking only triggers when get_channel returns None — and get_channel is a plain dict lookup (self._channels.get(name)), so a stopped/disconnected channel that stays registered still comes back non-None (Channel.is_running exists but isn't consulted; the docstring even says "running channel instance").
Two cheap options: raise a dedicated ChannelUnavailable-style exception here (and from base.send_notification's connectivity preconditions) that _deliver catches with count_attempt=False, or check channel.is_running in _deliver before sending. Either way the not-connected local condition gets the same park semantics as channel-object-missing, instead of silently exhausting the budget.
| if self._task is None: | ||
| return | ||
| self._stop.set() | ||
| await self._task |
There was a problem hiding this comment.
[suggestion] stop() awaits the poller task with no bound, while the adjacent channel-service shutdown is bounded specifically "to prevent worker hang".
stop() sets the event and awaits self._task, but the loop only observes _stop between iterations: if a delivery is stuck inside channel.send_notification (a WS stall that never errors, or _send_with_retry's sleeps compounding across a batch), shutdown blocks the lifespan until it resolves. Since this worker is stopped right before the bounded stop_channel_service() in the gateway lifespan, one hung send now stalls the whole shutdown path the bounding was added to protect. Consider asyncio.wait_for(self._task, timeout=...) with a cancel-and-await fallback, mirroring the neighboring shutdown discipline (the scheduler service's own stop() has the same shape, but it doesn't sit in front of external I/O the way this worker does).
| notification_connection_repo = None | ||
| notification_repo = None | ||
| connection_config = getattr(startup_config, "channel_connections", None) | ||
| if connection_config is not None and getattr(connection_config, "enabled", False): |
There was a problem hiding this comment.
[suggestion] This is a user-facing behavior change gated here, but no README/docs update ships in the PR.
With channel_connections.enabled: true, scheduled-task outcomes now silently push to every connected IM identity of the task owner — users have no way to discover this from the docs, nor that manual "run now" and interrupts stay silent, that a channel outage parks deliveries while platform rejections retry only ~15 minutes before settling failed, and that only WeCom currently implements the proactive path (other providers will log failures). The repo's documentation policy (AGENTS.md) asks for a README/docs paragraph in the same change set for user-facing changes; a short note under the scheduled-tasks section covering the gating and the failure semantics would cover it.
RykerFeng
left a comment
RykerFeng
left a comment
There was a problem hiding this comment.
Deep review — the outbox design, execution/delivery separation, idempotency key, atomic multi-worker claim, and crash isolation are all solid and well tested. My main concern is one reliability guarantee that doesn't actually hold for the only provider that implements the push (WeCom). Details inline; the first comment is the important one.
| channel = self._resolve_channel(provider) | ||
| if inspect.isawaitable(channel): | ||
| channel = await channel | ||
| if channel is None: | ||
| # 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) |
There was a problem hiding this comment.
Key finding: the "channel-outage failures don't consume the retry budget" guarantee doesn't hold for WeCom.
This count_attempt=False branch is the only path that protects the retry budget, and it fires only when resolve_channel returns None — i.e. the channel is disabled/stopped/never registered. But a WeCom WebSocket disconnect does not deregister the channel: WeComChannel._on_ws_disconnected only logs ("SDK will attempt to reconnect") and does not null _ws_client or remove the channel from ChannelService. So during a WS outage:
get_channel("wecom")still returns the live channel object (notNone), so we skip this branch;send_notification'sif not self._ws_clientguard doesn't trip (the client object is still there, only the transport is down);send_messageon the dead connection fails → retried 3× by_send_with_retry→ raises → caught below and marked failed with the defaultcount_attempt=True.
So the most common real outage (channel running, transport flapping) takes the budget-burning path. With backoff 60→120→240→480s and max_attempts=5, ~15 minutes of WS downtime settles the notification to failed and drops it — directly contradicting the "an hours-long outage cannot exhaust retries" claim in the PR description, on the one provider that actually delivers.
Suggestion: distinguish connectivity errors from explicit platform rejections and route connectivity failures through count_attempt=False too (e.g. have send_notification raise a typed "not connected / transport error" that the worker parks without counting).
| text_markdown=render_notification_text(enriched), | ||
| ) | ||
| except Exception as exc: | ||
| await self._delivery_repo.mark_failed(delivery_id, error=str(exc)) |
There was a problem hiding this comment.
This is the branch a WeCom WS disconnect actually lands on (see the key-finding comment above): the send raises, and mark_failed is called with the default count_attempt=True, consuming a retry. A transient transport outage therefore burns the budget instead of being parked. Consider inspecting exc (or a typed exception from send_notification) and passing count_attempt=False for connectivity-class failures.
| 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); | ||
| the platform ACK's ``errcode`` is checked so the delivery outbox can | ||
| retry on rejection instead of treating an ACK-less call as success. | ||
| """ | ||
| if not self._ws_client: | ||
| raise RuntimeError("WeCom channel is not connected") | ||
| body = {"msgtype": "markdown", "markdown": {"content": text_markdown}} | ||
| ack = await self._send_with_retry( | ||
| lambda: self._ws_client.send_message(target, body), | ||
| max_retries=3, | ||
| log_prefix="[WeCom]", | ||
| operation_name="notification send", | ||
| ) | ||
| 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}") |
There was a problem hiding this comment.
send_notification collapses two very different failures into the same generic RuntimeError: (a) transport down / WS disconnected (transient, not the delivery's fault) and (b) platform rejected the message via errcode (a real, budget-worthy failure). The worker can't tell them apart, so both count against the retry budget. To make the outage protection above effective, raise a distinguishable error (or return a status) for the not-connected / transport-failure case so the worker can park it with count_attempt=False. Note if not self._ws_client only catches a fully torn-down client, not a live client whose WebSocket has dropped.
| if self._task is None: | ||
| return | ||
| self._stop.set() | ||
| await self._task |
There was a problem hiding this comment.
Secondary: stop() does a bare await self._task with no upper bound, while the gateway deliberately bounds channel shutdown with asyncio.wait_for(stop_channel_service(), timeout=_SHUTDOWN_HOOK_TIMEOUT_SECONDS). If stop() is called mid-batch, this awaits the whole in-flight batch, each delivery of which can spend seconds inside _send_with_retry (retries + 1s/2s sleeps). Consider wrapping the join in wait_for for symmetry with the channel-service shutdown, so a slow/hung send can't stall process exit.
RykerFeng
left a comment
RykerFeng
left a comment
There was a problem hiding this comment.
A few more follow-ups from a deeper pass — one usability issue that directly undercuts the feature's goal (A), plus two architecture-consistency notes (B, C). None block merge, but A is worth folding in alongside the WeCom reliability fix.
| except Exception: | ||
| logger.exception("[Scheduler] failed to list channel connections for notifications") | ||
| return | ||
| payload = {"run_status": terminal_status, "error": error, "task_id": task_id} |
There was a problem hiding this comment.
(A) The notification shows only the raw task UUID, not the task title — this undercuts the feature's own goal. The payload here carries task_id (an opaque hex UUID) but not the human-readable title, so the rendered IM message reads Task: \a1b2c3d4…`` and the user can't tell which task completed. That contradicts the PR's stated motivation ("so the IM push is actually useful headlessly").
The fix is nearly free: handle_run_completion has already loaded the full row via task = await self._task_repo.get(...) a few lines above, and scheduled_tasks has a title column (String(255)). Add it to the payload, e.g.:
payload = {"run_status": terminal_status, "error": error, "task_id": task_id, "task_title": task.get("title")}and have render_notification_text prefer task_title over the id in the Task: line.
| notification_delivery_worker = NotificationDeliveryWorker( | ||
| delivery_repo=notification_repo, | ||
| resolve_channel=delivery_channel_service.get_channel, | ||
| resolve_run_summary=resolve_run_summary, | ||
| ) | ||
| await notification_delivery_worker.start() |
There was a problem hiding this comment.
(B) The delivery worker's poll interval is hard-coded (defaults to 5s) and not wired to config, unlike its sibling pollers. The scheduler is constructed with poll_interval_seconds=startup_config.scheduler.poll_interval_seconds and the MCP task worker with mcp_tasks_config.poll_interval_seconds, but NotificationDeliveryWorker(...) here passes no poll_interval_seconds, so it silently takes the 5s default. That means it issues two DB queries (reset_stale_sending_rows + claim_due_deliveries) every 5s forever, even when the outbox is empty — permanent idle load, and a break from the "poll cadence is configurable" convention the other two pollers follow. Consider passing a config-driven interval (reusing the scheduler's is fine).
| if notification_repo is not None: | ||
| # Delivery side of the outbox (issue #4254): polls due | ||
| # rows and pushes them through the owning channel's | ||
| # proactive send path. Only started when the enqueue side | ||
| # is wired and a channel service exists to resolve | ||
| # providers against. | ||
| from app.channels.service import get_channel_service | ||
| from app.scheduler.notification_delivery import NotificationDeliveryWorker | ||
|
|
||
| delivery_channel_service = get_channel_service() | ||
| if delivery_channel_service is not None: |
There was a problem hiding this comment.
(C) Enqueue and delivery have asymmetric activation conditions, leaving a "write-only" hole. The worker only starts when notification_repo is not None and get_channel_service() is not None (this block). But the enqueue side (ScheduledTaskService._enqueue_run_notifications) activates purely on both repos being present. So in a config where channel_connections.enabled=true but no channel service is running, the scheduler keeps writing pending rows into notification_deliveries that no worker will ever drain — they accumulate silently. Suggest aligning the two sides (gate enqueue on the same condition, or at least log a startup warning when the outbox is wired but no worker/channel service exists).
RykerFeng
left a comment
RykerFeng
left a comment
There was a problem hiding this comment.
One necessity/bloat note on a newly added method.
| await session.refresh(row) | ||
| return self._to_dict(row) | ||
|
|
||
| async def list_by_task_run(self, task_run_id: str) -> list[dict[str, Any]]: |
There was a problem hiding this comment.
list_by_task_run is new in this PR but has no production caller — it's only referenced by its own unit test. For contrast, every sibling repo's list_* method is wired to something real: ScheduledTaskRunRepository.list_by_task / ScheduledTaskRepository.list_by_user / list_by_user_and_thread back routes in routers/scheduled_tasks.py, and ChannelConnectionRepository.list_connections backs both the channel-connections router and the scheduler hook. This one backs nothing.
Since the outbox has no read/query surface yet (no router, no service reads it), the method is speculative. Suggest dropping it (and its test) until an actual consumer exists — or, if it's meant to back a planned deliveries-status endpoint, a one-line note saying so would make the intent clear. Not a blocker, just flagging unused surface area added in this change.
RykerFeng
left a comment
RykerFeng
left a comment
There was a problem hiding this comment.
Two more necessity/over-generality notes (non-blocking).
| # 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): |
There was a problem hiding this comment.
Over-general sync/async dual path. Both _resolve_channel (here) and _resolve_run_summary are await-ed through an inspect.isawaitable(...) guard so they accept either sync or async callables. But each injection point is fixed at wiring time, not variable at runtime: production get_channel is always sync (ChannelService.get_channel), and production resolve_run_summary is always async (the closure in app.py). So of the two guards, the summary one matches its real (async) resolver, while this channel one only ever takes the sync branch — and the async branch here isn't covered by any test (they all inject lambdas). The isawaitable cost is negligible, so this isn't urgent, but it's generality with no current consumer; a fixed contract per resolver (async summary, sync channel) would be simpler to read. Flagging as speculative surface rather than something that must change.
| poll_interval_seconds: int = 5, | ||
| batch_size: int = 10, | ||
| resolve_run_summary: SummaryResolver | None = None, | ||
| stale_sending_timeout_seconds: int = _STALE_SENDING_TIMEOUT_SECONDS, |
There was a problem hiding this comment.
These three tunables are never set in production — all defaults. poll_interval_seconds, batch_size, and stale_sending_timeout_seconds are constructor knobs, but the only instantiation (app.py) passes none of them, so they always take the defaults and are exercised solely by test injection. Test-injectable params are fine, but combined with the earlier note that the poll interval isn't wired to config, this reads as half-finished configurability: the knobs look tunable but nothing production-side tunes them. Suggest either wiring poll_interval_seconds (at least) to a config value, or trimming the ones that exist purely for tests if you don't intend to expose them.
|
This PR adds a user-facing capability (scheduled-task outcomes pushed to bound IM identities) and new architecture (the notification_deliveries outbox table, migration 0013, and the NotificationDeliveryWorker), but git diff --name-only shows no docs were touched — no README.md, no AGENTS.md, no config.example.yaml, no CHANGELOG. The root AGENTS.md Cross-Cutting Conventions make this a same-change-set requirement: Documentation update policy — keep docs in sync with code: update README.md for user-facing changes and the relevant AGENTS.md for development/architecture changes in the same change set. Concretely, I'd expect at least: backend/AGENTS.md — document the new outbox table + delivery worker in the persistence/migrations and scheduler/channels sections, and note the new Channel.send_notification contract (default raises; only WeCom implements it). |
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>
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>
|
Thanks @willem-bd and @18062706139fcz for the follow-up review ? addressed across e225b2c and a1e2f65. Reliability (WeCom / retry budget)
Shutdown
Usability / wiring
Docs
Cleanup
Tests: notification delivery worker + repository coverage (run in the gateway container). Ready for another look when you have a chance. Thanks! |
willem-bd
left a comment
willem-bd
left a comment
There was a problem hiding this comment.
Round-3 review at a1e2f65. Everything from rounds 1-2 is verified fixed at this head: the WeCom transport-outage path now raises ChannelUnavailable (parked via count_attempt=False, _ws_transport_up tracking), stop() is bounded with a cancel fallback, the README/README_zh/CHANGELOG/AGENTS.md/config docs shipped, task_title reaches the rendered push, the worker's poll interval is wired to scheduler.poll_interval_seconds, the enqueue side is disabled when no channel service is running (no write-only outbox), and list_by_task_run is gone. Encoding in test_scheduled_task_service.py is clean again. No new critical or important issues found; two residual suggestions below (unbounded parking, and untested lifespan wiring).
| # 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 |
There was a problem hiding this comment.
[suggestion] Parked rows can never reach a terminal state — a permanently absent channel retries forever.
count_attempt=False parking is unbounded in both count and time: unlike counted failures, a parked row has no path to failed. The transient-outage case is exactly right, but consider the permanent one — a provider whose channel is removed from config.yaml (or fails to start) while other channels keep the service alive: get_channel(provider) returns None forever, so every future scheduled run for owners still connected to that provider enqueues a row that retries every 15 minutes for the life of the deployment. With no read surface on notification_deliveries (yet), that backlog is invisible apart from claim churn in the logs.
Suggestion: bound parking by row age — e.g. once created_at is older than a day or two, finalize the row as failed with last_error noting the channel never returned — or keep a separate cap on parked attempts. That preserves the outage-survival property the README documents while giving permanently-gone channels a settled, observable terminal state.
| # 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: |
There was a problem hiding this comment.
[suggestion] The new lifespan wiring — both the startup gating and this shutdown block — ships without any test coverage.
Neither branch of the gating (repos nulled with the warning when no channel service is running; the worker started only when both sides are wired) nor this stop-before-channels block is exercised by any test. That gating is what enforces the "no write-only outbox" invariant the backend AGENTS.md note now documents, and this block is the reason a worker in front of external IM I/O gets the same bounded-shutdown discipline as the channel service — both are exactly the kind of invariant a later refactor could silently break.
tests/test_gateway_lifespan_shutdown.py already drives the real lifespan with patched channel services, so both are cheap to pin there: e.g. assert no NotificationDeliveryWorker is started (and the warning fires) when channel_connections.enabled is true but start_channel_service yields no service, and that the worker's stop() is awaited before stop_channel_service() on shutdown.
| except Exception as exc: | ||
| # Dead/flapping transport: not the delivery's fault. Park the | ||
| # outbox row until the SDK reconnects instead of burning retries. | ||
| self._ws_transport_up = False | ||
| raise ChannelUnavailable(f"WeCom transport unavailable: {exc}") from exc |
There was a problem hiding this comment.
New edge introduced by this fix: the blanket except Exception -> ChannelUnavailable can turn a deterministic send error into an immortal outbox row.
The transport-outage fix is correct and I confirmed it resolves the earlier WS-drop budget-burn. But this catch-all funnels every non-ChannelUnavailable exception from send_message into ChannelUnavailable, which the worker parks with count_attempt=False (never counted, retried every ~15min forever). That's right for transport/connectivity failures — but it also swallows deterministic failures that aren't the transport's fault and will never succeed on retry: e.g. the SDK raising ValueError/TypeError on a malformed target or body, a serialization error, or an SDK-version mismatch. Those become undying rows that re-attempt every 15 minutes indefinitely and never settle to failed.
Note the asymmetry: a platform rejection surfaced as an errcode does count against the budget (the RuntimeError below), but the same class of permanent error surfaced as a raised exception now does not. So whether a permanent failure is bounded depends on how the SDK happens to report it.
Suggestion: narrow the wrap to connectivity-class exceptions (e.g. ConnectionError/TimeoutError/OSError-family, or whatever the aibot SDK raises on a dead socket) and let genuinely unexpected exceptions fall through as counted failures — so a deterministic error still drains to failed instead of parking forever. Low severity (needs the SDK to report a deterministic error as a raise rather than an errcode), but it's a gap this narrowing opened up.
|
Discussion — do scheduled-task IM notifications need content redaction before egress? @willem-bd flagging this for your judgment, since it's a policy call rather than a clear defect. What goes out.
Between read and send, the only transform is Why it might matter. Scheduled runs are headless and periodic; their prompts/outputs can touch internal data (hostnames, paths, customer data, connection strings in a traceback, token fragments in an error). Once pushed to a third-party IM, it's effectively irreversible (platform-side caching/indexing/retention), and there's no human in the loop the way an interactive reply has. Honest caveat (why I'm calling it a discussion, not a bug). The normal channel reply path doesn't apply generic secret-scrubbing to agent content either, so the underlying "agent output may contain sensitive text" risk isn't unique to this PR. Two things still make this path worth a deliberate decision:
For context, the repo already treats "filter before egress" as a real concern in places (e.g. Options if you decide it's worth addressing (ordered by cost): (a) on failure, send only status + task/run id and let the user open the UI for the error detail; (b) run |
RykerFeng
left a comment
RykerFeng
left a comment
There was a problem hiding this comment.
Per the discussion above (redaction of run content in IM notifications) — noting this has been decided as in-scope, here is the concrete change request, anchored to the two egress points.
| error = payload.get("error") | ||
| if error: | ||
| lines.append(f"Error: {_truncate(str(error))}") | ||
| 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(): | ||
| lines.append(f"Result: {_truncate(summary.strip(), limit=_SUMMARY_TEXT_LIMIT)}") |
There was a problem hiding this comment.
Change requested (decided in the discussion thread): redact run content before it leaves for the external IM platform.
Both outbound lines carry raw run content with only _truncate applied — no scrub/mask:
Error: {error}(line 62) fromrecord.errorResult: {result_summary}(line 70) fromruns.last_ai_message
Scheduled runs are headless/periodic and can surface internal data (hostnames, paths, connection strings in a traceback, token fragments), and once pushed to a third-party IM it's effectively irreversible. The Error: line in particular is a new egress this feature introduces — the normal manager flow only logs record.error, it doesn't forward it to IM.
Ask: run error and result_summary through a redaction pass before rendering. The repo already has secret utilities in deerflow.runtime.secret_context (config/metadata-oriented) and a content-filter precedent in buzz.py (_HIDDEN_CONTEXT_MARKERS) — reuse/extend rather than inventing a new mechanism. If a full content scrub is out of scope for this PR, the minimal acceptable version is to drop the raw Error: body (send status + task/run id only, user opens the UI for detail), since that's the highest-risk / lowest-value line. Please also cover it with a test asserting a seeded secret does not appear in render_notification_text output.
|
@GaoTianyuanhr please fix the lint error. |
|
@GaoTianyuanhr there are some conflict with the main branch, do you mind resolve them? |
Windows git could not detach HEAD for rebase, so this resolves the PR conflicts via merge. Renumber the scheduled-task IM outbox migrations to 0014/0015 so they chain after 0013_mcp_task_notifications.
|
Thanks @WillemJiang — conflicts with Windows git on this machine could not detach |
|
@GaoTianyuanhr, please fix the lint error in the latest change. |
…ion 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.
|
Thanks @WillemJiang — lint is fixed in The Postgres unit-test errors were Alembic |
Renumber notification outbox migrations to 0016/0017 after main 0014_managed_subagents and 0015_scheduled_task_enqueue. Resolve merge conflicts in AGENTS docs, persistence bootstrap tests, and config examples while keeping scheduled-task IM push (issue bytedance#4254).
|
@WillemJiang @18062706139fcz — latest \origin/main\ is merged and pushed (\980c4b0). Merge / migrations
|
Update migration 0015 test HEAD expectation to 0017_parked_attempts and add queue_timeout_seconds to the lifespan notification worker mock scheduler config.
|
@WillemJiang @18062706139fcz - CI follow-up fix pushed in 9de2cfe. Two post-merge backend-unit-tests failures, both test-only:
Both tests pass locally. Thanks! |
|
@GaoTianyuanhr please resolve the conflicts with the main branch. |
Renumber notification outbox migrations to 0017/0018 after main 0016_subagent_batches. Update bootstrap HEAD expectations to 0018_parked_attempts.
|
@WillemJiang @18062706139fcz - Conflicts with main are resolved and pushed in 4482d84. Main added 0016_subagent_batches, so the notification outbox migrations are renumbered to 0017_notification_deliveries and 0018_parked_attempts. Bootstrap HEAD expectations updated accordingly. Local tests: migration/bootstrap suite + notification worker + gateway lifespan (82 passed). Thanks! |
The PR's migration-coordination comment states each migration file carries the note; the file did not. Adds it: numbering was generated against main head 0016 alongside bytedance#5078 and bytedance#4843; whoever merges first keeps the slot, the others renumber on rebase (revision/down_revision plus the bootstrap head assertions).
…5041) * feat(auth): add personal access tokens for programmatic API access (#4849) Backend-first implementation of the PAT contract from #4849: show-once dfp_ tokens bound to their owning user (AUTH_SOURCE_PAT, is_internal=false), digest-only storage (migration 0017), strict credential precedence (invalid Bearer is a 401, never cookie fallback), CSRF double-submit skipped only for Bearer requests while auth-endpoint origin checks still run, scopes intersecting the authz route permissions, session-auth-only PAT management and password changes, and throttled best-effort last_used_at stamps. * fix(auth): harden PAT scope boundary and schema parity from adversarial review Independent review of the initial draft found: (1) scopes only constrained the threads/runs permission axis while admin routes treated a PAT as its (possibly admin) owner — is_admin_user now rejects PAT callers outright since no scope grants admin capability; (2) the model declared a column UNIQUE constraint while migration 0017 created a named unique index, so downgrade failed on create_all-bootstrapped DBs — both now use the named unique index; (3) auth-disabled mode is an operator override and now stays ahead of the Bearer check so a stray Authorization header cannot 401 an E2E sandbox; plus wiring the previously-unused constants, bounding the last_used_at stamp cache, and four new tests (middleware-level expiry, expires_in_days, admin-capability rejection with session control, and the auth-disabled precedence). * docs(api): document personal access tokens for programmatic API access * fix(auth): close PAT security boundaries from review (default-deny routes, extension admin suppression) P1-1: scope intersection only constrains @require_permission routes, so undecorated mutation routes (DELETE /api/memory, POST /api/agents, Lark credential switching, channel config) accepted a PAT holding a single read scope. AuthMiddleware now enforces a default-deny route policy in auth/pat.py: PAT requests are admitted only to the thread/run lifecycle routes the v1 scopes govern; everything else answers 403 regardless of scopes. Session-cookie callers are unaffected. P1-2: the extension principal resolver projected is_admin/roles from the raw system_role, so an admin-owned PAT passed deerflow_extension_api.require_admin on contributed routes despite the documented no-admin guarantee. The projection is now PAT-aware and suppresses every admin signal for PAT callers, mirroring deps.is_admin_user. Both fixes carry regression tests (route outside policy 403 + session control; production resolver admin suppression), and API.md documents the default-deny boundary. * fix(auth): enforce PAT scopes on stateless run entry and harden decorator Follow-up hardening from an independent audit of the P1 fixes: - POST /api/runs/stream and /api/runs/wait were the only allowlisted run entrypoints without @require_permission, so a threads:read-only PAT could still start runs (same bug class as P1-1, now closed): both now carry @require_permission("runs", "create"). POST /api/threads and POST /api/threads/search gain threads:write / threads:read for the same reason. Authorization-disabled deployments see no change (the permission set resolves to all permissions). - require_permission now binds the wrapped signature to locate a positionally-passed request before injecting the test stub, fixing 'got multiple values for argument' on direct positional unit-test calls. - API.md: the intro PAT example used GET /api/models, which the new default-deny policy 403s — replaced with GET /api/threads; the default-deny route list now spells out method sets. Regression test: threads:read-only PAT is 403 on the decorated stateless entry while a runs:create PAT passes. * fix(auth): address review P2s (empty Authorization header, PAT name trimming, API example) - CSRFMiddleware treats an explicitly empty Authorization header as present (is None), so an invalid credential always reaches AuthMiddleware's uniform 401 instead of a CSRF 403 that varies by method/CSRF state. Regression: empty-header request dies at auth. - PATCreateRequest strips the name and rejects whitespace-only values before token generation; created names are stored trimmed. - API.md intro PAT example now uses the implemented POST /api/threads/search endpoint (GET /api/threads does not exist). - AGENTS.md trimmed back under the guidance soft budget after the upstream merge. * fix(auth): tighten PAT route policy to implemented methods only The allowlist admitted GET /api/threads, a method no router implements. Pre-authorizing a dead method weakens the default-deny boundary: a future GET collection route added without a permission decorator would become PAT-reachable without an explicit policy change. Restrict the rule to POST, fix the stale GET description in API.md's PAT constraints, and document the default-deny boundary accurately in the gateway AGENTS.md guidance (only the threads/runs allowlist is PAT-reachable; every other authenticated route 403s PAT callers). Audited every remaining rule against the mounted routers: all other method+path entries map to real routes. Regression: test_pat_policy_does_not_pre_authorize_unimplemented_methods. * test(auth): guarantee the negative digest test mutates the token token[:-1] + "X" is identical to the original whenever the generated token already ends in X (1/62), making the negative digest assertion fail intermittently. Choose the replacement character based on the existing tail so the mutated token always differs. * fix(auth): require runs:cancel for cancel-then-stream requests stream_existing_run is gated at runs:read so action-less stream joins work with read-only credentials, but its ?action=interrupt|rollback branch cancels the run — a separate permission. A runs:read-only PAT passed both the PAT route policy and the route decorator and could interrupt or roll back an active run, bypassing the runs:cancel scope. Decorators cannot express query-parameter-conditional permissions, so the check lives in require_cancel_permission_when_action(), applied at the top of the handler. Regression drives the real helper through the production middleware: runs:read-only PAT + action is 403, the same token joins action-less, runs:read+cancel passes, session control unaffected. * docs(changelog): add the PAT feature entry * docs(readme): add personal access tokens section Repo documentation-update policy requires user-facing features to update README.md in the same changeset; the PAT feature previously touched only backend/docs/API.md and the gateway AGENTS.md. * fix(auth): require runs:cancel for mutating multitask strategies All five run-creation entrypoints were gated only by runs:create, but RunCreateRequest.multitask_strategy accepts interrupt/rollback and start_run forwards it to create_or_reject, which terminates an already-active run. A runs:create-only PAT could therefore kill an existing run through a create request, bypassing runs:cancel. Decorators cannot express body-parameter-conditional permissions, and per-route checks leave the same hole for the next entrypoint, so the gate lives in start_run itself — the single choke point every run-creation path (HTTP routes and internal launchers) flows through. Regenerate launches pass multitask_strategy="reject" and are unaffected; requests without a stamped auth context (internal/test compositions) skip the gate. The check is the shared authz.require_cancel_permission_if primitive; require_cancel_permission_when_action now delegates to it, so every request dimension that carries cancel capability (query action, body strategy) flows through one gate. Regression drives the real middleware stack: runs:create-only PAT + interrupt/rollback is 403 with the exact detail, reject (explicit and default) stays available, runs:create+cancel passes, session control unaffected; a source anchor pins the gate inside start_run. * fix(runs): keep observer joins from applying creator cancel-on-disconnect sse_consumer's finally block applied the record's on_disconnect=cancel policy on ANY consumer's disconnect. The join surfaces (GET /join and the action-less GET/POST stream join) feed it the existing RunRecord, so anyone with thread read access — including a runs:read-only PAT — could cancel a locally-owned running run simply by closing the SSE connection, without runs:cancel. The policy expresses the creator's intent for their own connection; an observer's disconnect must never be read as that intent. sse_consumer gains apply_on_disconnect (default True). The two join surfaces pass False; the creating endpoints (thread-scoped and stateless create-and-stream) keep the creator semantics unchanged. wait_for_run_completion needs no change: its callers are creator-side or post-explicit-cancel paths only. Regression exercises a real generator close — the same machinery Starlette drives on client disconnect — against the production sse_consumer: creator stream disconnect cancels, observer join disconnect does not; a wiring anchor pins both join call sites and the creator defaults. API.md documents the cancel-capability constraint (this fix plus the action/strategy gates) in PAT Constraints. * test(auth): pin the multitask gate behaviorally; state wait invariant Independent adversarial review of the round-5 fixes found the P1-a regression only mirror-pinned: the source anchor could be satisfied by a comment, and deleting the gate from start_run would not fail the suite. This drives the production start_run directly — a create-only auth context gets 403 with the exact detail for interrupt, and a reject request with no cancel permission at all proceeds past the gate (never a permission 403). Also documents wait_for_run_completion's creator-side invariant (every caller is the creating endpoint or post-explicit-cancel) so a future observer wiring thinks twice before reusing it — the one-caller- away variant of the observer-disconnect P1. * docs(changelog): correct the PAT entry's digest and route-policy description The entry said HMAC digests (the implementation stores SHA-256 digests, as documented in API.md and pinned by the repository tests) and claimed the route policy admits 'implemented stateless endpoints' (it admits the thread/run lifecycle routes, narrowing further by scopes). Also notes the cancel-capability gate now covering action and multitask strategies. * fix(auth): enumerate the PAT runs route policy per implemented subroute The runs subtree rule was a GET|POST /runs(/.*)? wildcard — it pre-authorized every current and future subroute under /runs, including methods the router never implemented (e.g. GET /runs/stream), which is the same latent default-deny weakening the threads collection rule was tightened for: a future route added under /runs would become PAT-reachable without an explicit policy change. The wildcard is replaced with six segment-precise rules covering exactly the 14 implemented method+path combinations; the {run_id} slot necessarily matches any single segment, so the POST-only collection names (stream, wait, regenerate, edit-regenerate) are excluded from the GET run-id rule via negative lookahead — no dead method stays pre-authorized. Behavior for implemented routes is unchanged. test_pat_runs_policy_admits_exactly_the_mounted_routes derives the expected set from the mounted thread_runs router instead of a hand-maintained list: every implemented GET/POST route under /runs must be admitted, routes in this router outside the subtree stay denied, and representative unimplemented neighbors are denied — so adding a route under /runs now fails CI until it is explicitly allowlisted, and a removed route leaves a dead rule visible. API.md's PAT constraints list the enumerated routes and drops a feedback mention that belonged to the stateless /api/runs axis. * docs(migration): add the 0017 renumbering coordination note to 0017 The PR's migration-coordination comment states each migration file carries the note; the file did not. Adds it: numbering was generated against main head 0016 alongside #5078 and #4843; whoever merges first keeps the slot, the others renumber on rebase (revision/down_revision plus the bootstrap head assertions). * fix(auth): pad base62 tokens to a fixed 43-char width int.from_bytes discards leading zero bytes, so the unpadded encoder returned a variable-length body — empty for all-zero input, and shorter than 40 characters for any draw below 62**39 (~1 in 14.5M), leaving test_generate_pat_token_format probabilistically flaky and the token body without stable width (review round 6, P3). _base62 now left-pads with "0" to _base62_width(len(data)) — the exact integer digit count (62^43 > 2^256 > 62^42, so 43 for 32 bytes). The format test asserts the exact fixed width instead of a probabilistic floor, and a new unit test pins the all-zero, leading-zero-byte, and max-value edges deterministically.
|
@WillemJiang @18062706139fcz - Friendly check-in on this PR. Conflicts with main were resolved in 4482d84 (notification migrations renumbered to 0017/0018 after 0016_subagent_batches). Local migration/bootstrap + notification tests passed at that head. I noticed main has moved ahead again since then, and the fork CI workflows on 4482d84 still show action_required. Happy to sync main once more and push if that would help - just let me know. Thanks for the earlier reviews! |
|
@GaoTianyuanhr Sorry for the delay. Please resolve the conflicts so we can merge the PR. |
|
@GaoTianyuanhr What sits on top of your work:
Checked on a real Gateway (DeepSeek, no IM channel configured): a scheduled @WillemJiang would it be easier to push this onto the PR branch (maintainer edits are enabled), so it lands as #4843 under the author's name? If a separate PR is simpler for you, I can open one that keeps these commits. |
…otification outbox entry
|
Update: the carried branch is now its own PR, #6135, opened as a separate PR after checking with @WillemJiang (I cannot push to this fork). Your 14 commits are in it unchanged, the description keeps the history of this PR, and I have asked for @GaoTianyuanhr thank you for the design and the implementation; the outbox, the idempotency key and the parking logic are all yours, and they held up in a real Gateway without changes. If you are back, please review #6135 or tell me if you would rather carry it yourself and I will close mine. |
…ification 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>
|
The issue has been addressed by #6135 |
`markdown
Fixes #4254
Why
Scheduled tasks run headlessly: their results only land in a web UI thread, so
unless the user happens to watch the browser, a completed (or failed) run goes
unnoticed. This PR implements the design agreed on in #4254: results are pushed
to the task owner's bound IM identities through a durable outbox and an async
delivery worker — no synchronous IM calls in the completion hook.
What changed
state enqueues one notification delivery per connected channel identity of
the task owner (
run_completedon success,run_failedon error;user-initiated interrupts stay silent).
NotificationDeliveryWorkerpolls due deliveries and pushes a boundedmarkdown summary through the channel's proactive send path. WeCom implements
it via the WS client's
send_messagewitherrcodechecking; providerswithout proactive push record a failure instead of silently dropping.
Result:line, read fromruns.last_ai_messageat delivery time(best-effort: a lookup failure degrades to the skeleton text and never
blocks the delivery). Failed runs keep the error-only skeleton.
15min, 5 attempts), then settle as
failed. Execution status is writtenfirst and is never shadowed by delivery problems.
notification_deliveriestable (migration 0013, auto-applied atstartup) with an idempotency key on
(task_run_id, event, provider, target)so hook replays never double-send.
channel_connections.enabledis true; no new config keys, no newdependencies.
Surface area
backend/app, outbox persistence indeerflow.persistenceValidation
ruff check+ruff format --checkclean on every changed file.scheduled task service incl. enqueue hook (31), delivery worker incl. WeCom
push and result-summary enrichment (19), plus the full channel + scheduler
regression batch (441) — all green.
create_alland the new migration covered by theexisting bootstrap regression tests (HEAD anchor bumped to 0013).
v1 skeleton push: completion produced the outbox row below and the
notification arrived in under one second:
run_completedwecom/ bound usersent/008:16:25.24→08:16:26.13Result-summary push: after a gateway restart, a completed scheduled run
delivered the same skeleton plus the run's final assistant text as a
Result:line in WeCom.AI assistance
Tool(s) used: Lingma (Qoder IDE AI coding assistant)
How you used it: AI drafted the implementation and tests under my
direction (TDD: tests first per slice); I reviewed every line, ran all
validation, and performed the live end-to-end verification myself.