Repository navigation
feat(scheduler): push scheduled task results to IM channels via a notification outbox - #6135
Conversation
…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.
…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.
…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.
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>
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>
The previous BEGIN-only pattern left key material and the END marker in notifications. Co-authored-by: Cursor <cursoragent@cursor.com>
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>
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.
…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.
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).
Update migration 0015 test HEAD expectation to 0017_parked_attempts and add queue_timeout_seconds to the lifespan notification worker mock scheduler config.
Renumber notification outbox migrations to 0017/0018 after main 0016_subagent_batches. Update bootstrap HEAD expectations to 0018_parked_attempts.
Brings the branch up to main 5c67d61 (638 commits). Conflicts resolved: - scheduler/service.py: handle_run_completion keeps main's atomic complete_run (bytedance#5035). The notification enqueue runs after it, only when the completion was recorded, and reads the task title afterwards. - channels/wecom.py: keep both the transport-up flag and main's per-chat send locks (bytedance#5148). - persistence/models/__init__.py: register NotificationDeliveryRow next to the models main added. - tests: migration and bootstrap tests take main's computed head; the lifespan test file is main's plus the notification worker tests. - docs: main's restructured guidance plus the notification notes. The two outbox migrations still carry their old numbers here; the next commit re-chains them after main's head.
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.
- 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_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.
…vice starts The outbox wiring sat in the scheduler block and asked for the running channel service there. bytedance#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.
…he 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.
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.
… 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.
github-actions
Bot
added
area:backend
…otification outbox entry
willem-bd
left a comment
willem-bd
left a comment
There was a problem hiding this comment.
Reviewed d280625: solid outbox design overall — enqueue/delivery separation, atomic claim via guarded UPDATE..RETURNING, stale-sending reconciliation, bounded parking vs. counted retries, egress redaction, and the lifespan wiring (worker starts after the channel service, stops before it) all check out, with good test coverage of the multi-worker race and crash isolation. Two issues worth addressing before merge, both around state that changes between enqueue and delivery.
| channel = self._resolve_channel(provider) | ||
| if inspect.isawaitable(channel): | ||
| channel = await channel | ||
| if channel is None or not getattr(channel, "is_running", True): |
There was a problem hiding this comment.
Delivery never re-validates that the target's connection is still connected. The delivery-time recheck here only covers channel liveness (is_running), not the connection row, and the worker has no access to ChannelConnectionRepository. But delivery can legitimately happen long after enqueue: counted-retry backoff delays it by up to ~15 minutes, and the parking path below deliberately parks rows for up to ~24h (_CHANNEL_PARK_MAX_AGE / 96 x 900s). If the owner disconnects or the connection is revoked during that window (disconnect_connection only flips the connection to revoked; nothing fails or parks the already-enqueued rows), the worker still pushes the run outcome to the now-unbound external_account_id once the channel comes back — against explicit user intent.
Suggestion: re-check the connection at delivery time (the row already carries owner_user_id, provider, and target, so a scoped lookup is cheap and cannot cross owners), or have the disconnect/revoke flow fail or park pending outbox rows for that (provider, target).
| ack = await _send_once() | ||
| except ChannelUnavailable: | ||
| raise | ||
| except (ConnectionError, TimeoutError, OSError) as exc: |
There was a problem hiding this comment.
The SDK's closed-socket RuntimeError escapes the ChannelUnavailable wrapping and burns the retry budget. The comment at the is_connected probe says it directly: the SDK raises a plain RuntimeError when its socket is closed. That exception is not in (ConnectionError, TimeoutError, OSError), so whenever the SDK's is_connected is stale-True (disconnect callback hasn't fired yet, or the SDK reconnected silently but the socket dies again before the probe):
- in the optimistic branch,
_send_with_retryre-raises theRuntimeErrorafter 3 attempts — a single poll consumes 3 of the row's 5 counted attempts, so two polls during a transport outage settle the delivery as terminalfailed; - in the known-down probe branch, the single probe — whose stated purpose is to avoid spending budget — also fails as a counted attempt.
That is exactly the premature budget exhaustion this PR's park-vs-count design exists to prevent. Note the classification is cleaner than it looks: an errcode rejection is only raised after send_message returns an ACK dict, so any exception thrown by send_message is transport-level — RuntimeError from the send call itself could safely be wrapped as ChannelUnavailable (keeping the counted path for the explicit errcode branch).
|
@Totoro-qaq please resolve the conflicts with the main branch. |
…rk WeCom transport errors, count its rejections Review round 1 on bytedance#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.
…up; pin the rejection prefix through the retry branch and the real connection-row shape
|
@WillemJiang conflicts with The two review points are addressed in Delivery-time connection check. The worker now takes a SDK |
Fixes #4254
Continues #4843 by @GaoTianyuanhr. Opened as a separate PR, as discussed with @WillemJiang, because the original branch lives in a fork I cannot push to.
History
mainmoved on and the PR went back to conflicting; on 09-02 @WillemJiang asked for the conflicts to be resolved. No activity from the author since.4482d849: his 14 commits are unchanged,mainis merged on top, and my changes are separate commits. A note with the same summary is on feat(scheduler): push scheduled task results to IM channels via a notification outbox #4843.Co-authored-by: GaoTianyuanhr <GaoTianyuanhr@users.noreply.github.com>in the squash message; the design and most of the code are his.Why
Scheduled tasks run headlessly: their results only land in a web UI thread, so a completed or failed run goes unnoticed unless the user is watching the browser. #4254 agreed on pushing results to the task owner's bound IM identities through a durable outbox and an async delivery worker, with no synchronous IM calls in the completion hook.
What changed
What #4843 does (the full design is in its description):
run_completedon success,run_failedon error. Manual triggers and user interrupts stay silent.NotificationDeliveryWorkerpolls due deliveries and pushes a bounded markdown summary through the channel's proactive send path. WeCom implementssend_notification; providers without a push path record a failure instead of dropping silently.Result:line, read at delivery time; a lookup failure degrades to the skeleton text.notification_deliveriestable with an idempotency key on(task_run_id, event, provider, target), so hook replays never double-send.channel_connections.enabledis true.What this PR adds on top of #4843, in order:
main(2cf84a5f, 16 conflicted files; later syncs7cbe6abcand6e82a1bb, the last one resolving changelog link order and thebase.pybullet inbackend/app/channels/AGENTS.md).0027_notification_deliveriesand0028_parked_attempts, chained after0026_mcp_task_lease_tokens; the chain-head pin moved totest_migration_0027_0028_notification_outbox.py. feat(skills): add guarded evolution host APIs #5645 also uses 0027/0028; I left a note there, and whichever lands second renumbers (I offered to move to 0029/0030).83787bfe,59476203). After the clean merge the outbox wiring calledget_channel_service()inside the scheduler block, but fix(scheduler): reconcile stuck once tasks from committed run outcome #5035 had moved the channel service to start after the scheduler. The feature disabled itself at every real startup while the lifespan tests, which patch the accessor, stayed green. Now the enqueue side is wired when the scheduler starts, the delivery worker starts after the channel service, and a Gateway with the scheduler on but no channel service logs a "write-only outbox" warning and detaches the outbox instead of failing. Found by starting the real Gateway; covered by new lifespan tests that model the real accessor.59476203): a failed task lookup in the completion hook no longer drops the notification; the message falls back to the task id.7de065ab):send_notificationraisesChannelUnavailablewhen the SDK socket is not connected, so the row parks instead of burning retries. Test uses the real never-connected SDK client.failedfor a disconnected target, park on lookup failure, owner-scoped), sticky terminal outbox rows, and WeCom transportRuntimeErrors plusConnectionClosedparked while the SDK'sReply ack errorrejection still counts.backend/app/channels/AGENTS.md, and the root and backend files are byte-identical tomain.CHANGELOG_zh.md; README (en, zh) andbackend/docs/CONFIGURATION.mdsay which outcomes notify and how long parking lasts.Known limits, unchanged from #4843: WeCom is the only provider with a push path; every connected identity of the owner is notified, with no per-task switch; occurrences finalized by the recovery paths do not notify;
send_notificationdoes not use #5148's per-chat lock or byte cap.After review (willem-bd, round 1): the worker re-checks at send time, before the channel-liveness check, that the target is still a
connectedidentity of the owner and drops it as terminalfailedotherwise (lookup failure parks); terminal outbox rows ignore later status writes; the WeCom SDK's transportRuntimeErrors and thewebsocketsclose exception park the row, while itsReply ack errorrejection (how the SDK really reports a non-zeroerrcode) keeps counting.Surface area
backend/app, outbox persistence and two migrations indeerflow.persistencechannel_connections.enabledis trueValidation
cd backend && make lint: clean.python scripts/check_agent_guidance.py: 35 files, 0 errors.b787de32: 21,651 passed; the 10 failures are the known machine-specific ones (browser automation, Crawl4AI, fastCRW). At0370c02d, after the review fixes and the latestmainmerge: the scheduler, channel, notification, migration, lifespan and harness-boundary tests, 1,027 passed.channel_connections.enabled: true, DeepSeek: a scheduledoncerun wrote its outbox row about 5 ms after finishing and the worker parked it ("channel 'wecom' is not running"); a manual trigger wrote no row; the parked row was re-polled after its 15-minute park.run_failedrow of a failed run.AI assistance
Tool(s) used: Claude Code.
How you used it:
resolved the merge and renumbered the migrations;
found the startup-order break by running the real Gateway, and wrote the fixes and the tests;
ran the suite, the lint and the guidance check.
I've read and understand every line of this change and take responsibility for it — it's not unreviewed AI output.