feat(drain): in-process stream drain; external /worker demoted to optional safety net

- notify::stream_drain shared core + tokio::spawned run_stream_drain (10s loop,
  consumer INPROC_CONSUMER), plus immediate one-shot drain after each
  successful push_reconcile_delta; /worker route unchanged, still a no-op
  when the shared consumer group already claimed a message.
- 3 new stream_drain unit tests on MemoryState (success/commit/ack,
  retry-to-DLQ, pre-committed short-circuit): 111 passed / 0 failed / 4 ignored.
- Doc sync in the same change: README §3.8-3.11, design/.zh §5.4/§10.5/§11.2,
  reference/.zh §9.3, opengaps/.zh item #4 + deployment-defects section,
  charter/.zh test baseline + C-NO-LONG-CONN + invariant 21, retired/.zh
  push-register lock anchor drift (notify.rs:1805-1825), cron-drain.sh header.
- GATE-P0 green (fmt/check/clippy/tests 111/0/4); GATE-DOCS 6/6 PASS
  (reachability only, not semantic correctness).

Co-Authored-By: Claude Code <noreply@anthropic.com>

README.md

@@ -220,8 +220,9 @@ curl -s "https://api.telegram.org/bot<TOKEN>/getWebhookInfo"
 ### 3.8 Turn on processing
 
 Flip **Enable business processing** at the top of the page. This is the master switch: with it off,
-webhooks are accepted but `/worker` and `/reconcile` refuse to drain, so nothing reaches Telegram.
+webhooks are accepted but the in-process drain skips each tick and `/worker` and `/reconcile`
-With it on, incoming mail and messages are enqueued and the worker drains them.
+refuse to drain, so nothing reaches Telegram. With it on, incoming mail and messages are
+enqueued and drained in-process (see §3.10).
 
 ### 3.9 Verify end to end
 
@@ -235,10 +236,19 @@ With it on, incoming mail and messages are enqueued and the worker drains them.
    confirmation; the summary only goes out after you confirm, and only if the LLM section is
    enabled.
 
-### 3.10 Schedule the drain
+### 3.10 Drain timing: in-process, with an external safety net
 
-Push delivery can be delayed or missed; an external scheduler is the safety net. On any machine
+Delivery does not wait on an external scheduler. Each backend instance runs an in-process
-that can reach your origin, every 5–10 minutes:
+background drain: every 10 seconds it reads the `stalwart:*` delivery streams from Redis (short
+request–response round trips only, no long connection — `C-NO-LONG-CONN`), and a successful
+Stalwart push reconcile triggers one immediate drain right after it, so push-delivered mail
+reaches Telegram in seconds. The drain shares the same Redis consumer group and the per-message
+committed/in-flight leases as the route-based `POST /worker`, so a concurrent trigger is a no-op
+rather than a double send — the same holds for multiple instances.
+
+The external scheduler remains an **optional safety net** for what the in-process loop cannot
+see: pushes Stalwart never delivered, or an instance that was down when its mail arrived. On any
+machine that can reach your origin, every 5–10 minutes:
 
 ```sh
 scripts/cron-drain.sh --once \
@@ -256,7 +266,9 @@ must reach the backend origin directly — see §6.3 for why.
 
 - **Nothing arrives after saving.** Check the enable toggle (step 3.8), then
   `getWebhookInfo` (step 3.7). A `last_error_message` about a 401 means the webhook secret does not
-  match what Telegram sends — re-save section 02 and re-register.
+  match what Telegram sends — re-save section 02 and re-register. Mail that was enqueued is picked
+  up by the in-process drain within about 10 seconds without cron; mail that was never enqueued
+  needs `/reconcile` (step 3.10) to rescan.
 - **`/ready` answers 503.** That body is just an error code — it does not name the failing
   component. When `/ready` returns 200, its JSON has per-field booleans (`jmap`, `telegram`,
   `redis`). Common causes are an unreachable `jmap_session_url`, a wrong app password, or

docs/charter.md

@@ -41,7 +41,7 @@ A JMAP email notification service deployed on Docker + externally hosted Redis:
 | key type | secrecy | 0.8 |
 | encryption | ring | 0.17 |
 | Log | tracing + tracing-subscriber | 0.1 / 0.3 |
-| Scheduling | External cron → `POST /reconcile` | — |
+| Scheduling | In-process stream drain (`notify::run_stream_drain`); external cron `POST /reconcile` + `/worker` as an optional safety net | — |
 | Front-end | Native HTML + single-file JS + single-file CSS, **No build tools, no frameworks** | — |
 | Load balancer | Cloudflare Worker (Pure JS, zero dependencies) | — |
 
@@ -69,7 +69,7 @@ A JMAP email notification service deployed on Docker + externally hosted Redis:
 18. **The load balancer only transparently transmits and does not make decisions**: Worker does not parse the request body, does not verify the business logic, and does not issue `Retry-After`.
 19. **No retry for AI results**: Failure of LLM analysis will be regarded as "no notification", no retry, and no downgrade to no AI notification.
 20. **Stateless process**: Two container instances are safe to run concurrently; no local writes.
-21. **No long polling scheduling**: Scheduling is triggered by external cron `POST /reconcile`.
+21. **No long polling scheduling**: No scheduler may be a long connection. Reconciliation is triggered by short `POST /reconcile` requests (external cron, optional safety net); the stream drain is an in-process loop of short Redis round trips.
 22. **AI analysis is not persistent**: The analysis results and intermediate states other than the authorized state are not recorded.
 23. **JMAP capability dynamic discovery**: Hard-coded JMAP capability list is prohibited and must be read at runtime.
 
@@ -118,7 +118,7 @@ cd /home/okabe/Repo/messageweave && docker run --rm --user 1000:1000 \
   bash -lc 'export PATH=/usr/local/cargo/bin:$PATH; cargo fmt --all -- --check && cargo check --locked && cargo clippy --locked --all-targets -- -D warnings && cargo test --locked 2>&1 | tail -12'
 ```
 
-**Current baseline: 108 passed / 0 failed / 4 ignored** (4 cursor/walk tests + 2 `/api/status` Host-header tests (label echo + null fallback)), retested 2026-10-16; ignored: `real_server_tests::session_list_and_read_smoke` and `debug::tests::debug_config_reports_timezone_of_business_configured_app` require external real JMAP Server credentials, `real_redis_ttl_tests::ttl_claim_dedup_sets_the_exactly_requested_expiry` and `real_redis_ttl_tests::ttl_consent_and_retry_landing_on_real_redis` require `REDIS_TEST_URL`).
+**Current baseline: 111 passed / 0 failed / 4 ignored** (3 `stream_drain` unit tests covering success/commit/ack, retry-to-DLQ, and the pre-committed short-circuit, on top of the 108 baseline), retested 2026-10-16; ignored: `real_server_tests::session_list_and_read_smoke` and `debug::tests::debug_config_reports_timezone_of_business_configured_app` require external real JMAP Server credentials, `real_redis_ttl_tests::ttl_claim_dedup_sets_the_exactly_requested_expiry` and `real_redis_ttl_tests::ttl_consent_and_retry_landing_on_real_redis` require `REDIS_TEST_URL`).
 
 **Note**: The bash `-lc` script of `docker run` must be wrapped in **single quotes**. Using double quotes will first expand `$PWD` / `$PATH` on the host, and cargo will not be found in the container.
 

docs/charter.zh-CN.md

@@ -41,7 +41,7 @@
 | 密钥类型 | secrecy | 0.8 |
 | 加密 | ring | 0.17 |
 | 日志 | tracing + tracing-subscriber | 0.1 / 0.3 |
-| 调度 | 外部 cron → `POST /reconcile` | — |
+| 调度 | 进程内 stream drain（`notify::run_stream_drain`）；外部 cron `POST /reconcile` + `/worker` 为可选安全网 | — |
 | 前端 | 原生 HTML + 单文件 JS + 单文件 CSS，**无构建工具、无框架** | — |
 | 负载均衡器 | Cloudflare Worker（纯 JS，零依赖） | — |
 
@@ -69,7 +69,7 @@
 18. **负载均衡器只透传、不决策**：Worker 不解析请求体、不校验业务逻辑、不下发 `Retry-After`。
 19. **AI 结果不重试**：LLM 分析失败即视为「不通知」，不重试、不降级为无 AI 通知。
 20. **无状态进程**：两个容器实例并发安全；无本地写。
-21. **无长轮询调度**：调度由外部 cron 触发 `POST /reconcile`。
+21. **无长轮询调度**：调度器不得是长连接。对账由短 `POST /reconcile` 请求触发（外部 cron，可选安全网）；stream drain 是进程内循环的一批短 Redis 往返。
 22. **AI 分析不持久化**：分析结果与授权状态之外的中间态不落盘。
 23. **JMAP capability 动态发现**：禁止硬编码 JMAP 能力列表，必须运行时读取。
 
@@ -118,7 +118,7 @@ cd /home/okabe/Repo/messageweave && docker run --rm --user 1000:1000 \
   bash -lc 'export PATH=/usr/local/cargo/bin:$PATH; cargo fmt --all -- --check && cargo check --locked && cargo clippy --locked --all-targets -- -D warnings && cargo test --locked 2>&1 | tail -12'
 ```
 
-**当前基线：106 passed / 0 failed / 4 ignored**（4 个游标解析与 walk 分页单测 + 2 个 `/api/status` Host 头单测（标签回显 + null 回退））后复测，2026-10-14；ignored：`real_server_tests::session_list_and_read_smoke` 与 `debug::tests::debug_config_reports_timezone_of_business_configured_app` 需外部真实 JMAP 服务器凭据，`real_redis_ttl_tests::ttl_claim_dedup_sets_the_exactly_requested_expiry` 与 `real_redis_ttl_tests::ttl_consent_and_retry_landing_on_real_redis` 需 `REDIS_TEST_URL`）。
+**当前基线：111 passed / 0 failed / 4 ignored**（在 108 基线上新增 3 个 `stream_drain` 单测，覆盖成功/提交/ack、重试转 DLQ、已提交短路），2026-10-16 复测；ignored：`real_server_tests::session_list_and_read_smoke` 与 `debug::tests::debug_config_reports_timezone_of_business_configured_app` 需外部真实 JMAP 服务器凭据，`real_redis_ttl_tests::ttl_claim_dedup_sets_the_exactly_requested_expiry` 与 `real_redis_ttl_tests::ttl_consent_and_retry_landing_on_real_redis` 需 `REDIS_TEST_URL`）。
 
 **注意**：`docker run` 的 bash `-lc` 脚本必须用**单引号**包裹。用双引号会先在宿主机展开 `$PWD` / `$PATH`，容器内找不到 cargo。
 

docs/design.md

@@ -234,8 +234,8 @@ Candidate comparison (`teloxide` / `grammers` / old `telegram-bot`), 6 reasons w
   1. Load configuration (`config::Config`).
   2. Construct `JmapService` (`Client::connect` completes session parsing, account_id caching, mailbox role→id mapping warm-up); sinceState is restored from external Redis (`MOD-SINCESTATE`).
   3. Start the **HTTP portal** (axum, single port `PORT`): `/webhook/tg`, `/push/jmap`, `/reconcile`, `/healthz`, `/ready`; three of the write paths first pass the `SAF-AUTH-*` portal authentication (fail-closed, §7.3), `/healthz`, `/ready` For the public probe (`SAF-PROBE-PUBLIC`); `/ready` has done end-to-end detection (configuration + Redis + outbound read-only detection JMAP session `GET` and TG `getMe`, each 3s, parallel, about 3s at worst).
-  4. Start **Redis Streams worker** (background task) to consume Push events → `Email/changes` → Notification → Send to TG → Push sinceState → XACK.
+  4. Spawn an **in-process stream drain** (`notify::run_stream_drain`, `tokio::spawn`) that every `ttl::INPROC_DRAIN_INTERVAL_SECONDS` (10 s) drains the Redis Streams worker queue (`stalwart:jmap`, `stalwart:telegram`) → `Email/changes` → Notification → Send to TG → advance sinceState → XACK. After each successful push reconcile, the same drain runs once immediately so freshly enqueued mail is delivered without waiting for the next tick. The drain reuses the shared consumer group, so an overlapping `/worker` trigger is a no-op rather than a double send (`MOD-STREAMS-GROUP`, `MOD-DEDUP`).
-  5. **Does not hold any long connections and does not build self-timers** (`C-NO-LONG-CONN`); reconciliation is triggered by **external HTTPS Cron** `/reconcile` (`FLOW-RECONCILE`).
+  5. **Does not hold any long connections** (`C-NO-LONG-CONN`): every drain iteration is a batch of short Redis round trips, the same shape as the route-based `POST /worker` call — no WebSocket, SSE, long polling, or persistent upstream connection. The drain is an in-process periodic loop, not a network long connection. Reconciliation is additionally triggered by **external HTTPS Cron** `/reconcile` (`FLOW-RECONCILE`); the external scheduler is now an optional safety net (see `scripts/cron-drain.sh`), not the only drain path.
 - Notification sending and command processing share `Arc<JmapService>`, and internal `tokio::sync::RwLock` protects the variable cache; cross-request status always falls to external Redis (`C-REDIS-ONLY-STATE`).
 - **Multiple instances and load balancing (deployment form, `ARCH-LB-WORKER`)**: Since the state is all external Redis (`C-REDIS-ONLY-STATE`) and the delivery is idempotent (`MOD-DEDUP`), **the same image can be instantiated across multiple serverless platforms**, free Cloudflare Worker is used as the only entrance and failover (`C-LB-SINGLE-REG-URL`); `/reconcile` Use Redis to lock a single instance (`SAF-RECONCILE-LOCK`), and Streams to automatically amortize using the same consumer group (`MOD-STREAMS-GROUP`). See README.md §6.1 for details. **Field/channel logic does not need to be changed. **
 - **Production redline (`C-NO-DB` / `C-NO-LOCAL-WRITE` / `C-LOG-STDOUT-ONLY` / `SAF-LOG-PURITY` / `C-NO-STATEFUL-RECOVERY`, see docs/charter.md §3 / docs/reference.md §9.1)**:
@@ -260,7 +260,7 @@ Stalwart JMAP Push → POST /push/jmap (StateChange{Email/EmailDelivery: new_sta
 ```
 > Key: `Email/changes` lists the new email id in `created` to avoid full `email_query`; `sinceState` is stored in external Redis (`C-REDIS-ONLY-STATE`). When Redis is lost, it will be reconstructed from JMAP by reconciliation, and the **fact source is in JMAP**.
 > Notifications only contain sender/subject/time (+number of attachments), the text never appears in the notification (see `docs/charter.md §3` Security Boundary `SAF-NOTIFY-META`).
-> No long connection: Push is a short request callback, reconciliation is triggered by **external HTTPS Cron** (`C-NO-LONG-CONN`), and is not a self-sustaining timer in the container.
+> No long connection (`C-NO-LONG-CONN`): Push is a short request callback, and the in-process stream drain is a periodic loop of short Redis round trips — no WebSocket, SSE, long polling, or any other network long connection. Reconciliation may also be triggered by **external HTTPS Cron** as an optional safety net.
 
 ### 5.5 Data flow: command `/read <seq>` (long mail branch)
 ```
@@ -624,12 +624,13 @@ The design target form once included six additional variants: `Jmap(#[from] jmap
 ### 10.4 Phase 3.5: LLM facade + rollback (1.5d, not executed)
 Not implemented, see `docs/retired.md`.
 
-### 10.5 Implemented capabilities: real-time push (Push callback + Streams worker + external Cron reconciliation)
+### 10.5 Implemented capabilities: real-time push (Push callback + in-process stream drain + external Cron safety net)
 `src/notify.rs` is a single tiled file of nearly 2000 lines, with no submodules (`notify::push_handler` / `notify::worker` / `notify::reconcile` / `mod_streams` / `mod_sincestate` do not exist in the following table):
 - `register_push` (`POST /api/push/register`): Accepts an explicit HTTPS callback URL and creates a subscription; writes back `verificationCode` with JMAP `PushSubscription/set` when receiving a Stalwart push.
 - `jmap_push` (`POST /push/jmap`): Verify subscription ID + verificationCode → remove duplicates → enter Redis Streams → 2xx.
-- `worker` (`POST /worker`): Redis Streams consume → `Email/changes` increment → notify → advance sinceState → XACK.
+- In-process stream drain (`notify::run_stream_drain`, `tokio::spawn` loop in `src/main.rs`): every `ttl::INPROC_DRAIN_INTERVAL_SECONDS` (10 s), plus once after each successful push reconcile, drains the Redis Streams worker queue (`stalwart:jmap`, `stalwart:telegram`) → `Email/changes` increment → notify → advance sinceState → XACK. Short Redis round trips only (`C-NO-LONG-CONN`); no new network long connection.
-- `reconcile` (`POST /reconcile`): External HTTPS Cron trigger (`FLOW-RECONCILE`); also serves as cursor reconstruction after Redis loss. **No CLI subcommand form**.
+- `worker` (`POST /worker`): the same drain core, route-based; now an optional external safety net alongside the in-process loop. The shared consumer group plus per-message committed/in-flight leases make overlapping triggers no-ops, not double sends.
+- `reconcile` (`POST /reconcile`): external HTTPS trigger (`FLOW-RECONCILE`), also an optional safety net run by `scripts/cron-drain.sh` (covers pushes Stalwart never delivered, and mail piled up while an instance was down); also serves as cursor reconstruction after Redis loss. **No CLI subcommand form**.
 - sinceState writes external Redis (`state.rs`, `C-REDIS-ONLY-STATE`; not SQLite/file `NG-SQLITE-PERSIST`/`NG-LOCAL-VOLUME`).
 - Target acceptance: Send a test email to Stalwart. After Push callback + worker, Telegram receives the push in ~ seconds; it will not be resent after restarting; it will be reissued by reconciliation after clearing Redis. The current code gate has passed, but end-to-end verification still needs to be completed in real Stalwart, Redis and Telegram environments, and the goals here cannot be regarded as verified facts.
 
@@ -692,7 +693,7 @@ All values have now converged to the single authority table in `docs/reference.m
 - Affects: `Credentials::Basic` construct; no need for OAuth client/token auto-refresh module.
 
 ### 11.2 Real-time channel (decided, see docs/charter.md §8)
-- Channel = **JMAP Push HTTPS callback + external Cron reconciliation and draining** (`C-NO-LONG-CONN`/`C-HTTPS-INBOUND`); EventSource/SSE/WebSocket are all **non-target** (`NG-POLLING-SSE`). Push registration is triggered explicitly through protected `POST /api/push/register`; external Cron must call `/reconcile` (incremental enqueue) and `/worker` (drain concurrent notification) at the same time - **Only adjusting reconciliation will make the notification never sent** (README.md §3.10).
+- Channel = **JMAP Push HTTPS callback + in-process stream drain (normal drain path) + external Cron reconciliation safety net** (`C-NO-LONG-CONN`/`C-HTTPS-INBOUND`); EventSource/SSE/WebSocket are all **non-target** (`NG-POLLING-SSE`). Push registration is triggered explicitly through protected `POST /api/push/register`. Drain and notification are normally done by the **in-process stream drain** (`notify::run_stream_drain`, every 10 s plus once immediately after each reconcile) — the external Cron calling `/worker` is no longer required; `scripts/cron-drain.sh` (`--once` runs `/reconcile` and `/worker`) is demoted to an **optional safety net** covering pushes Stalwart never delivered and mail piled up while an instance was down. The shared consumer group plus per-message leases make overlapping external and in-process triggers no-ops, not double sends (README.md §3.10).
 - Stalwart side access has been verified in the real environment: the built-in role has the `PushSubscription` permission, and registration, verification and round-trip and real callback delivery are all passed. The match between the number of callback retries and the idempotent key TTL/reconciliation interval is an operation and maintenance parameter item, not an acceptance threshold: `docs/reference.md` §9.3 provides retry suggestions, README.md §3.10 provides a value for the reconciliation interval, and `docs/reference.md` §1 explains that the idempotent key TTL must cover the TG rate-limit fallback upper limit. `docs/opengaps.md` currently has no outstanding items.
 
 ### 11.3 Deployment form (confirmed, architecture related)

docs/design.zh-CN.md

@@ -234,8 +234,8 @@ Telegram 渠道用 `src/channel.rs` 的 `reqwest` 自研实现（`ARCH-DEPS-STAG
   1. 加载配置（`config::Config`）。
   2. 构造 `JmapService`（`Client::connect` 完成 session 解析、account_id 缓存、mailbox role→id 映射预热）；sinceState 从外部 Redis 恢复（`MOD-SINCESTATE`）。
   3. 启动 **HTTP 入口**（axum，单端口 `PORT`）：`/webhook/tg`、`/push/jmap`、`/reconcile`、`/healthz`、`/ready`；其中三条写路径先经 `SAF-AUTH-*` 入口鉴权（fail-closed，§7.3），`/healthz`、`/ready` 为公开探针（`SAF-PROBE-PUBLIC`）；`/ready` 已做端到端探测（配置 + Redis + 出站只读探测 JMAP session `GET` 与 TG `getMe`，各 3s、并行，最坏约 3s）。
-  4. 启动 **Redis Streams worker**（后台 task）消费 Push 事件 → `Email/changes` → 通知 → 发往 TG → 推进 sinceState → XACK。
+  4. 派生**进程内 stream drain**（`notify::run_stream_drain`，`src/main.rs` 中 `tokio::spawn` 循环）：每 `ttl::INPROC_DRAIN_INTERVAL_SECONDS`（10 s）排空一次 Redis Streams worker 队列（`stalwart:jmap`、`stalwart:telegram`）→ `Email/changes` → 通知 → 发往 TG → 推进 sinceState → XACK；每次 push 对账成功后立即再排空一次。drain 复用同一 consumer group，与 `/worker` 重叠触发时是 no-op 而非双发（`MOD-STREAMS-GROUP`、`MOD-DEDUP`）。
-  5. **不持有任何长连接、不自建定时器**（`C-NO-LONG-CONN`）；对账由**外部 HTTPS Cron** 触发 `/reconcile`（`FLOW-RECONCILE`）。
+  5. **不持有任何网络长连接**（`C-NO-LONG-CONN`）：每次 drain 迭代只是短 Redis 往返，形态与路由式 `POST /worker` 相同——无 WebSocket、SSE、长轮询或任何持久上游连接。drain 是进程内周期循环，不是网络长连接。对账还由**外部 HTTPS Cron** `/reconcile` 触发（`FLOW-RECONCILE`）；外部调度器现在是可选安全网（见 `scripts/cron-drain.sh`），不再是唯一排空路径。
 - 通知发送与命令处理共享 `Arc<JmapService>`，内部 `tokio::sync::RwLock` 保护可变缓存；跨请求状态一律落外部 Redis（`C-REDIS-ONLY-STATE`）。
 - **多实例与负载均衡（部署形态，`ARCH-LB-WORKER`）**：由于状态全在外部 Redis（`C-REDIS-ONLY-STATE`）且投递幂等（`MOD-DEDUP`），**同一镜像可跨多个 serverless 平台实例化**，前面用免费 Cloudflare Worker 做唯一入口与故障转移（`C-LB-SINGLE-REG-URL`）；`/reconcile` 用 Redis 锁单实例执行（`SAF-RECONCILE-LOCK`），Streams 用同一消费组自动分摊（`MOD-STREAMS-GROUP`）。详见 README.md §6.1。**领域/渠道逻辑无需改动。**
 - **生产红线（`C-NO-DB` / `C-NO-LOCAL-WRITE` / `C-LOG-STDOUT-ONLY` / `SAF-LOG-PURITY` / `C-NO-STATEFUL-RECOVERY`，详见 docs/charter.md §3 / docs/reference.md §9.1）**：
@@ -260,7 +260,7 @@ Stalwart JMAP Push → POST /push/jmap (StateChange{Email/EmailDelivery: new_sta
 ```
 > 关键：`Email/changes` 在 `created` 列出新建邮件 id，避免全量 `email_query`；`sinceState` 存外部 Redis（`C-REDIS-ONLY-STATE`），Redis 丢失时由对账从 JMAP 重建，**事实源在 JMAP**。
 > 通知只含发件人/主题/时间（+附件数），正文绝不出现在通知里（见 `docs/charter.md §3` 安全边界 `SAF-NOTIFY-META`）。
-> 无长连接：Push 为短请求回调，对账由**外部 HTTPS Cron**触发（`C-NO-LONG-CONN`），非容器内自持定时器。
+> 无长连接（`C-NO-LONG-CONN`）：Push 为短请求回调，进程内 stream drain 是短 Redis 往返的周期循环——无 WebSocket、SSE、长轮询或任何网络长连接。对账还可通过**外部 HTTPS Cron** 作为可选安全网触发。
 
 ### 5.5 数据流：`/read <seq>` 命令（长邮件分支）
 ```
@@ -624,12 +624,13 @@ pub enum BotError {
 ### 10.4 阶段 3.5：LLM 门面与回退（1.5 天，尚未执行）
 未实施，见 `docs/retired.md`。
 
-### 10.5 已实现能力：实时推送（推送回调 + Streams 工作进程 + 外部 Cron 定时协调）
+### 10.5 已实现能力：实时推送（推送回调 + 进程内 stream drain + 外部 Cron 安全网）
 `src/notify.rs` 是**单个近 2000 行的平铺文件**，没有子模块（下表中不存在 `notify::push_handler` / `notify::worker` / `notify::reconcile` / `mod_streams` / `mod_sincestate` 这些模块名，对应实现是文件内的自由函数）：
 - `register_push`（`POST /api/push/register`）：接受显式 HTTPS callback URL 并创建订阅；收到 Stalwart 推送时用 JMAP `PushSubscription/set` 回写 `verificationCode`。
 - `jmap_push`（`POST /push/jmap`）：校验 subscription ID + verificationCode → 去重 → 入 Redis Streams → 2xx。
-- `worker`（`POST /worker`）：Redis Streams 消费 → `Email/changes` 增量 → 通知 → 推进 sinceState → XACK。
+- 进程内 stream drain（`notify::run_stream_drain`，`src/main.rs` 中 `tokio::spawn` 循环）：每 `ttl::INPROC_DRAIN_INTERVAL_SECONDS`（10 s）一次，且每次 push 对账成功后立即一次，排空 Redis Streams worker 队列（`stalwart:jmap`、`stalwart:telegram`）→ `Email/changes` 增量 → 通知 → 推进 sinceState → XACK。只有短 Redis 往返（`C-NO-LONG-CONN`），不引入新的网络长连接。
-- `reconcile`（`POST /reconcile`）：外部 HTTPS Cron 触发（`FLOW-RECONCILE`）；兼做 Redis 丢失后的游标重建。**没有 CLI 子命令形态**。
+- `worker`（`POST /worker`）：与进程内循环共享同一 drain 核心，现为可选外部安全网。共享 consumer group 加 per-message committed/in-flight 租约使重叠触发成为 no-op 而非双发。
+- `reconcile`（`POST /reconcile`）：外部 HTTPS 触发（`FLOW-RECONCILE`），也是 `scripts/cron-drain.sh` 运行的可选安全网（覆盖 Stalwart 压根未发出的 push，以及实例宕机期间堆积的邮件）；兼做 Redis 丢失后的游标重建。**没有 CLI 子命令形态**。
 - sinceState 写外部 Redis（`state.rs`，`C-REDIS-ONLY-STATE`；不用 SQLite/文件 `NG-SQLITE-PERSIST`/`NG-LOCAL-VOLUME`）。
 - 目标验收：向 Stalwart 发测试邮件，经 Push 回调 + worker，Telegram 在 ~秒级收到推送；重启不重发；清空 Redis 后由对账补发。当前代码门禁已通过，但仍需在真实 Stalwart、Redis 和 Telegram 环境完成端到端验证，不能将此处目标当作已验证事实。
 
@@ -692,7 +693,7 @@ Push 事件经 Streams 消费并投递到 Telegram，其关键路径交付语义
 - 影响：`Credentials::Basic` 构造；无需 OAuth client / token 自动刷新模块。
 
 ### 11.2 实时通信渠道（已决定，参见 docs/charter.md §8）
-- 通道 = **JMAP Push HTTPS 回调 + 外部 Cron 对账与排空**（`C-NO-LONG-CONN`/`C-HTTPS-INBOUND`）；EventSource/SSE/WebSocket 均**非目标**（`NG-POLLING-SSE`）。Push 注册通过受保护的 `POST /api/push/register` 显式触发；外部 Cron 必须同时调 `/reconcile`（增量入队）与 `/worker`（排空并发通知）——**只调对账会让通知永远发不出去**（README.md §3.10）。
+- 通道 = **JMAP Push HTTPS 回调 + 进程内 stream drain（正常排空路径）+ 外部 Cron 对账安全网**（`C-NO-LONG-CONN`/`C-HTTPS-INBOUND`）；EventSource/SSE/WebSocket 均**非目标**（`NG-POLLING-SSE`）。Push 注册通过受保护的 `POST /api/push/register` 显式触发。排空与通知正常由**进程内 stream drain**（`notify::run_stream_drain`，每 10 s + 每次对账后立即一次）完成，不再依赖外部 Cron 调 `/worker`；`scripts/cron-drain.sh`（`--once` 跑 `/reconcile` 与 `/worker`）降级为**可选安全网**，覆盖 Stalwart 未发出的 push 与实例宕机堆积。共享 consumer group 与 per-message 租约使外部触发与进程内触发重叠时是 no-op 而非双发（README.md §3.10）。
 - Stalwart 侧接入已在真实环境验证：内置角色具备 `PushSubscription` 权限，注册、验证往返与真实回调投递全部走通。回调重试次数与幂等键 TTL / 对账间隔的匹配属于运维调参项，不是验收门槛：`docs/reference.md` §9.3 给重试建议、README.md §3.10 给对账间隔取值、`docs/reference.md` §1 说明幂等键 TTL 必须覆盖 TG rate-limit 回退上限。`docs/opengaps.md` 当前无未完成项。
 
 ### 11.3 部署形式（已确认，架构相关）

docs/opengaps.md

@@ -19,9 +19,9 @@ The TTL green light for real machines has already been passed using the Upstash
 | 1 | `newState` semantics for `Email/changes` | Closed (commit `269c8f6`) |
 | 2 | TTL actual measurement | Closed (code closing `0890eb1` + real machine `ttl` two assertions run through) |
 | 3 | Telegram inbound / Stalwart `PushSubscription` joint debugging | Closed (real traffic has been driven, please see `README.md` §3.7 for operation steps) |
-| 4 | `/worker` is not in the deployment document | Closed (`README.md` §3.10 has been added) |
+| 4 | `/worker` is not in the deployment document | Closed (`README.md` §3.10 was added; the in-process drain later made the external `/worker` trigger an optional safety net, so the normal path no longer depends on the external scheduler) |
 
-Access control: `cargo fmt --check` / `cargo check --locked` / `cargo clippy --locked --all-targets -- -D warnings` all green; `cargo test --locked` **106 passed / 0 failed / 4 ignored** (2 TTL assertions out of 4 `#[ignore]` have been run with real Upstash: `REDIS_TEST_URL=… cargo test --locked -- --ignored ttl` → **2 passed / 0 failed**; the remaining 2 require real JMAP servers, this capability has been verified in staging joint debugging); documentation gate is 6/6 PASS, including 49 line number anchor non-null verifications.
+Access control: `cargo fmt --check` / `cargo check --locked` / `cargo clippy --locked --all-targets -- -D warnings` all green; `cargo test --locked` **111 passed / 0 failed / 4 ignored** (2 TTL assertions out of 4 `#[ignore]` have been run with real Upstash: `REDIS_TEST_URL=… cargo test --locked -- --ignored ttl` → **2 passed / 0 failed**; the remaining 2 require real JMAP servers, this capability has been verified in staging joint debugging); documentation gate is 6/6 PASS, including 49 line number anchor non-null verifications.
 
 Item 5 is not within the stage goal: multi-account product decision-making, which has been determined not to be done by the user on 2026-09-28. The decision boundary and future changes will be discussed in §3.
 
@@ -113,6 +113,6 @@ This item is not skipped, but explicitly rejected: the deployment unit saved by
 
 ## Deployment document defects discovered and fixed in this round
 
-`README.md` §3.10 Originally, only the scheduling example of `/reconcile` was given, and `/worker` - the only queue consumption entry in the entire code base - was described as "manual triggering of operation and maintenance" and was deliberately not included in the Worker whitelist. **Copying the deployment according to the document = emails continue to be queued and notifications are never sent out**. This is the reason why the joint debugging environment is backlogged for several days before being emptied manually. `README.md` §3.10 The description of `/worker`’s scheduling steps, sequence requirements, batch limit and "204 cannot be used as a success signal" has been added.
+`README.md` §3.10 Originally, only the scheduling example of `/reconcile` was given, and `/worker` - the only queue consumption entry in the entire code base - was described as "manual triggering of operation and maintenance" and was deliberately not included in the Worker whitelist. **Copying the deployment according to the document = emails continue to be queued and notifications are never sent out**. This is the reason why the joint debugging environment is backlogged for several days before being emptied manually. `README.md` §3.10 The description of `/worker`’s scheduling steps, sequence requirements, batch limit and "204 cannot be used as a success signal" has been added. Later the in-process stream drain (`notify::run_stream_drain`, a spawned 10-second loop plus an immediate drain after each successful push reconcile) made the external `/worker` trigger an optional safety net instead of the only consumption path; `scripts/cron-drain.sh` was demoted accordingly.
 
-At the same time, the anchor points of `/worker` in `docs/reference.md` and `docs/retired.md` are corrected: the original value `notify.rs:340` falls in the `reconcile` function body, and the anchor point audit will not report an error (the line exists and is not empty), but the semantics are wrong; the correct value is `notify.rs:390`.
+At the same time, the anchor points of `/worker` in `docs/reference.md` and `docs/retired.md` are corrected: the original value `notify.rs:340` falls in the `reconcile` function body, and the anchor point audit will not report an error (the line exists and is not empty), but the semantics are wrong; the correct value is `notify.rs:390`. Per the repo's anchor policy these historical anchors were not backfilled further (see `AGENTS.md` §2.5).

docs/opengaps.zh-CN.md

@@ -19,9 +19,9 @@ TTL 那条真机绿灯已经用你给的 Upstash URL 跑过了（见 §1），
 | 1 | `Email/changes` 的 `newState` 语义 | 已关闭（提交 `269c8f6`） |
 | 2 | TTL 实测 | 已关闭（代码收口 `0890eb1` + 真机 `ttl` 两条断言跑通） |
 | 3 | Telegram 入站 / Stalwart `PushSubscription` 联调 | 已关闭（真实流量已驱动，操作步骤见 `README.md` §3.7） |
-| 4 | `/worker` 未在部署文档里 | 已关闭（`README.md` §3.10 已补） |
+| 4 | `/worker` 未在部署文档里 | 已关闭（`README.md` §3.10 已补；进程内 drain 后来把外部 `/worker` 触发降级为可选安全网，正常路径不再依赖外部调度器） |
 
-门禁：`cargo fmt --check` / `cargo check --locked` / `cargo clippy --locked --all-targets -- -D warnings` 全绿；`cargo test --locked` **106 passed / 0 failed / 4 ignored**（4 个 `#[ignore]` 里 2 个 TTL 断言已用真实 Upstash 跑过：`REDIS_TEST_URL=… cargo test --locked -- --ignored ttl` → **2 passed / 0 failed**；剩 2 个需要真实 JMAP 服务器，该能力已在 staging 联调验证）；文档门 6/6 全绿，含 49 处行号锚点非空校验。
+门禁：`cargo fmt --check` / `cargo check --locked` / `cargo clippy --locked --all-targets -- -D warnings` 全绿；`cargo test --locked` **111 passed / 0 failed / 4 ignored**（4 个 `#[ignore]` 里 2 个 TTL 断言已用真实 Upstash 跑过：`REDIS_TEST_URL=… cargo test --locked -- --ignored ttl` → **2 passed / 0 failed**；剩 2 个需要真实 JMAP 服务器，该能力已在 staging 联调验证）；文档门 6/6 全绿，含 49 处行号锚点非空校验。
 
 第 5 项不在阶段目标内：多账号产品决策，2026-09-28 已由用户定为不做，决定边界与将来要做的改动面见 §3。
 
@@ -113,6 +113,6 @@ TTL 那条真机绿灯已经用你给的 Upstash URL 跑过了（见 §1），
 
 ## 本轮发现并修复的部署文档问题
 
-`README.md` §3.10 原来只给了 `/reconcile` 的调度示例，而 `/worker`——全代码库唯一的队列消费入口——被描述成「运维手工触发」，且刻意不在 Worker 白名单里。**照文档照抄部署 = 邮件持续进队列、通知永远发不出去**，这正是联调环境积压了数天才被手工排空的成因。`README.md` §3.10 已补上 `/worker` 的调度步骤、顺序要求、批量上限与「204 不能当成功信号」的说明。
+`README.md` §3.10 原来只给了 `/reconcile` 的调度示例，而 `/worker`——全代码库唯一的队列消费入口——被描述成「运维手工触发」，且刻意不在 Worker 白名单里。**照文档照抄部署 = 邮件持续进队列、通知永远发不出去**，这正是联调环境积压了数天才被手工排空的成因。`README.md` §3.10 已补上 `/worker` 的调度步骤、顺序要求、批量上限与「204 不能当成功信号」的说明。后来进程内 stream drain（`notify::run_stream_drain`，一个被 spawn 的 10 秒循环 + 每次成功的 push 对账之后立即排空一次）把外部 `/worker` 触发从唯一消费路径降级成了**可选安全网**，`scripts/cron-drain.sh` 也相应降级。
 
-同时修正了 `docs/reference.md` 与 `docs/retired.md` 里 `/worker` 的锚点：原值 `notify.rs:340` 落在 `reconcile` 函数体内，锚点审计不会报错（该行存在且非空），但语义是错的；正确值是 `notify.rs:390`。
+同时修正了 `docs/reference.md` 与 `docs/retired.md` 里 `/worker` 的锚点：原值 `notify.rs:340` 落在 `reconcile` 函数体内，锚点审计不会报错（该行存在且非空），但语义是错的；正确值是 `notify.rs:390`。按仓库锚点策略，这些历史锚点未再回补（见 `AGENTS.md` §2.5）。

docs/reference.md

@@ -720,8 +720,16 @@ callback URL is idempotent. Registering with a different callback URL destroys t
 subscription first, then creates the new one and repoints the current-pointer mapping, so the old
 origin stops receiving pushes. Use `POST /api/push/disable` with that URL to remove it.
 
-The backend has no internal scheduler. [`scripts/cron-drain.sh`](../scripts/cron-drain.sh)
+The normal drain path is the **in-process stream drain** (`notify::run_stream_drain`): a spawned
-calls `/reconcile` (when `MW_RECONCILE_TOKEN` is set) and then `/worker`. Configure
+periodic loop that drains the Redis Streams worker queue every 10 seconds, plus an immediate
+one-shot drain after each successful push reconcile. It is composed only of short Redis
+round trips, so it does not violate `C-NO-LONG-CONN` (no network long connection). The shared
+consumer group plus per-message committed/in-flight leases make overlapping triggers no-ops
+rather than double sends.
+[`scripts/cron-drain.sh`](../scripts/cron-drain.sh) is the **optional external safety net**:
+it calls `/reconcile` (when `MW_RECONCILE_TOKEN` is set) and then `/worker`, covering the two
+cases the in-process loop cannot see — pushes Stalwart never delivered, and mail piled up
+while an instance was down. Configure
 `MW_APP_URL`, `MW_WORKER_TOKEN`, and optionally `MW_RECONCILE_TOKEN` in the external
 scheduler's secret store. `MW_WORKER_URL` can override the default when the operator needs
 to call a backend origin directly. `--once` is for cron; `--loop` is for a managed process.

docs/reference.zh-CN.md

@@ -695,8 +695,12 @@ Telegram、JMAP、allowlist、worker、reconcile 和 LLM 业务设置驻留在 R
 重复注册相同 URL 是幂等的。用一个不同的回调 URL 注册时，后端会先销毁上一个订阅，再创建新订阅并重指向当前指针映射，
 因此旧 origin 不再收到推送。调用 `POST /api/push/disable` 并传入该 URL 可删除订阅。
 
-后端没有内置调度器。外部调度器运行 [`scripts/cron-drain.sh`](../scripts/cron-drain.sh)：设置了 `MW_RECONCILE_TOKEN` 时，
+正常排空路径是**进程内 stream drain**（`notify::run_stream_drain`）：一个派生的周期循环，每 10 秒排空一次 Redis Streams 队列，
-脚本先调用 `/reconcile`，再调用 `/worker`。在调度器的密钥管理器中配置 `MW_APP_URL`、`MW_WORKER_TOKEN`，
+另在每次 push 对账成功后立即排空一次。它只由短 Redis 往返构成，不违反 `C-NO-LONG-CONN`（没有网络长连接）。
+共享 consumer group 加 per-message committed/in-flight 租约使重叠触发成为 no-op 而非双发。
+[`scripts/cron-drain.sh`](../scripts/cron-drain.sh) 是**可选的外部安全网**：
+它调用 `/reconcile`（设置了 `MW_RECONCILE_TOKEN` 时），再调用 `/worker`，覆盖进程内循环看不到的两种情况——
+Stalwart 压根未发出的 push，以及实例宕机期间堆积的邮件。在调度器的密钥管理器中配置 `MW_APP_URL`、`MW_WORKER_TOKEN`，
 以及可选的 `MW_RECONCILE_TOKEN`。需要绕过 Worker、直连后端源站时，可用 `MW_WORKER_URL` 覆盖默认地址。
 `--once` 用于 cron 单次运行，`--loop` 用于常驻进程；脚本还提供 `--diagnose` 和 `--test-notify` 排查问题。
 两个端点成功时均返回 `204` 和空响应正文。若另一轮对账仍持有锁，`/reconcile` 会返回 `409`。

docs/retired.md

@@ -78,7 +78,7 @@ These names never exist in the code. Some of them appear in the directory tree o
 | Directory scheme that splits Telegram access into four files "module entry/command parsing/session state/reply rendering" | Fictitious directory scheme | Drawn in the target directory tree, never created | Command parsing is in `parse_intent` (`src/worker.rs`); the session/authorization state is Redis TTL (the AI authorization state section of `docs/reference.md`); the rendering function is in `src/channel.rs` |
 | `delivery:pending:{stream}` | Fictional key | Confused with Redis Streams' pending-entries list (PEL) - PEL is maintained internally by Redis and is not a writable key | Real key: `delivery:inflight:{stream}:{id}` (EX 60) and `delivery:committed:{stream}:{id}` (EX 604_800), see Delivery pipeline keys section of `docs/reference.md` |
 | `check_config_reload` | Fictional function | Uncommitted name in draft, never made it into code | Hot updated to `refresh_business_config` (`src/notify.rs:1173`) |
-| `push:registration:{sha256(callback_url)}` was once written as "360s registration solo lock" | Fact mislabeling | The key is 7d callback → subscription ID mapping | The real 360s solo lock is `lock:push-register:{sha256(callback_url)}` (`src/notify.rs:1724-1727`) |
+| `push:registration:{sha256(callback_url)}` was once written as "360s registration solo lock" | Fact mislabeling | The key is 7d callback → subscription ID mapping | The real 360s solo lock is `lock:push-register:{sha256(callback_url)}` (`src/notify.rs:1805-1825`) |
 | `MESSAGWEAVE_DOMAIN` was once written as "Worker whitelist is invalid when production is not configured" | Fictional assertion (unsubmitted draft/appeared in discussion, not documented) | This environment variable has zero hits in all positions; Worker whitelist is **unconditional fail-closed**: unknown path 404, method does not match 405, backend is missing or parsing failure 503 (`cloudflare-worker/src/index.js:104-117`) | No need to configure switches, the whitelist will always take effect |
 | `read_batch` / `retry_or_dlq` "Redis may still return `Ok(())` when an error occurs, so the consumption loop will not exit due to a single failure" | Fictional assertion | The consumption entry is HTTP handler `worker` (`notify.rs:433`), **not a background loop**: full position `src/` zero hit `select!`, no signal processing, no resident worker process. `read_batch` (`state.rs:515`) throws the Redis error as it is (`state.rs:555`) via `?`, and the only discarded result is the `BUSYGROUP` idempotent protection of XGROUP `CREATE` (`state.rs:523-531`); `retry_or_dlq` is also thrown up via `?`. The caller returns `503 service_unavailable` + `retryable=true` for every `Err`, so the semantic premise of "the loop does not exit due to a single failure" itself is not true | None - the gap does not exist; `docs/opengaps.md` (original name `docs/roadmap.md`) "Code Gap" 4 → 3, `docs/design.md` "Known Boundary" same sentence has been deleted |
 

docs/retired.zh-CN.md

@@ -78,7 +78,7 @@
 | 把 Telegram 接入拆成「模块入口 / 命令解析 / 会话状态 / 回复渲染」四个文件的目录方案 | 虚构目录方案 | 目标目录树里画出，从未创建 | 命令解析在 `parse_intent`（`src/worker.rs`）；会话/授权态走 Redis TTL（`docs/reference.md` 的 AI 授权态一节）；渲染函数在 `src/channel.rs` 内 |
 | `delivery:pending:{stream}` | 虚构键 | 与 Redis Streams 的 pending-entries list（PEL）混淆——PEL 由 Redis 内部维护，不是可写键 | 真实键：`delivery:inflight:{stream}:{id}`（EX 60）与 `delivery:committed:{stream}:{id}`（EX 604_800），见 `docs/reference.md` 的投递流水线键一节 |
 | `check_config_reload` | 虚构函数 | 未提交草稿中的名字，从未进入代码 | 热更新为 `refresh_business_config`（`src/notify.rs:1173`） |
-| `push:registration:{sha256(callback_url)}` 曾被写成「360s 注册单飞锁」 | 事实误标 | 该键是 7d 回调→订阅 ID 映射 | 真正的 360s 单飞锁是 `lock:push-register:{sha256(callback_url)}`（`src/notify.rs:1724-1727`） |
+| `push:registration:{sha256(callback_url)}` 曾被写成「360s 注册单飞锁」 | 事实误标 | 该键是 7d 回调→订阅 ID 映射 | 真正的 360s 单飞锁是 `lock:push-register:{sha256(callback_url)}`（`src/notify.rs:1805-1825`） |
 | `MESSAGWEAVE_DOMAIN` 曾被写成「生产未配置时 Worker 白名单失效」 | 虚构断言（未提交草稿/讨论中出现，未进文档） | 该环境变量全仓零命中；Worker 白名单是**无条件 fail-closed**：未知路径 404、method 不符 405、后端缺失或解析失败 503（`cloudflare-worker/src/index.js:104-117`） | 无需配置开关，白名单恒生效 |
 | `read_batch` / `retry_or_dlq`「Redis 出错时仍可能返回 `Ok(())`，消费循环因此不会因单次失败退出」 | 虚构断言 | 消费入口是 HTTP handler `worker`（`notify.rs:433`），**不是后台循环**：全仓 `src/` 零命中 `select!`，无信号处理、无常驻 worker 进程。`read_batch`（`state.rs:515`）经 `?` 把 Redis 错误原样上抛（`state.rs:555`），唯一被丢弃的结果是 XGROUP `CREATE` 的 `BUSYGROUP` 幂等保护（`state.rs:523-531`）；`retry_or_dlq` 同样经 `?` 上抛。调用方对每一处 `Err` 都返回 `503 service_unavailable` + `retryable=true`，故「循环不因单次失败退出」这一语义前提本身不成立 | 无——不存在该缺口；`docs/opengaps.md`（原名 `docs/roadmap.md`）「代码缺口」4 → 3，`docs/design.md`「已知边界」同句已删 |
 

scripts/cron-drain.sh

@@ -1,18 +1,19 @@
 #!/usr/bin/env bash
 # ============================================================================
-# cron-drain.sh — 让 Telegram 通知真正发出去的外部调度器。
+# cron-drain.sh — 出站队列排空与增量对账的外部安全网。
 #
-# 后端**没有任何内部调度器**（C-NO-LONG-CONN：无 tokio-cron、无后台 poller）。
+# 后端现在有**进程内后台 drain**（`run_stream_drain`，每 10 秒一次；每次
-# 消费出站队列的唯一入口是 POST /worker，增量对账的唯一入口是 POST /reconcile，
+# Stalwart push reconcile 成功后还会立刻 drain 一次），正常路径不依赖本脚本。
-# 两者都必须由外部（cron / k8s CronJob / 定时器）触发。没有这一步，新邮件会一直
+# 本脚本是补充安全网，覆盖进程内循环看不到的两种情况：Stalwart 压根没发出的
-# 堆在 Redis 队列里，通知永远不发——这是「收不到 Telegram 提醒」最常见的原因。
+# push，以及实例宕机时堆积的邮件。多实例/多触发并发安全由共享 consumer group
+# 与 per-message committed/in-flight 租约保证：重复触发是 no-op，不会双发。
 #
 # 两个端点的成功码都是 **204 且响应体为空**，不是 200：
 #   POST /reconcile  → 204 成功 / 401 / 409（另一轮进行中）/ 503（带 Retry-After）
 #   POST /worker     → 204 成功 / 400（body 非法）/ 401 / 503（带 Retry-After）
 # 因此看到 204 是正常的，不要当成失败。
 #
-# 详设计说明见 README.md 第 3.10 节（调度与探活）。
+# 详设计说明见 README.md 第 3.10 节（drain 时序与外部安全网）。
 #
 # --- 凭据只从环境变量或命令行 flag 读取 ---
 # 不读 .env、不读配置文件、不写盘（唯一例外是可选的 --log-file / MW_LOG_FILE 运行日志）。
@@ -419,7 +420,7 @@ diagnose() {
     field="$(json_field "$body" ok)"
     if [ "$field" != "true" ]; then
       say "  [4] Telegram bot token 无效 detail=$(json_field "$body" detail)"
-      say "        ★ 通知永远发不出去，与 cron 无关"
+      say "        ★ 通知永远发不出去（与排空机制无关：token 无效时谁来排空都不行）"
       issues=$((issues + 1))
     else
       say "  [4] Telegram bot token 有效（getMe 成功）"
@@ -463,12 +464,12 @@ diagnose() {
 
   say ""
   if [ "$issues" -gt 0 ]; then
-    say "结论：上面标 ★ 的 $issues 项要先修；修完再确认 cron 确实在执行本脚本"
+    say "结论：上面标 ★ 的 $issues 项要先修；修完再跑一次本脚本 --once 复核"
-    say "      （crontab -l / kubectl get cronjob，或手动 bash 本脚本 --once 一次）。"
+    say "      （后台 drain 是进程内的，本脚本只是外部安全网，不是必需步骤）。"
     FAILURES=$((FAILURES + issues))
   else
-    say "结论：配置与依赖全部正常。收不到提醒多半是调度没跑——"
+    say "结论：配置与依赖全部正常。正常路径下进程内 drain 会自动排空出站队列；"
-    say "      确认 cron 确实在执行本脚本，并看它的退出码是否为 0。"
+    say "      收不到提醒时，本脚本 --once 是一次性外部检查手段（看退出码）。"
   fi
 }
 
@@ -598,7 +599,7 @@ main() {
     [ "$DO_TEST_NOTIFY" -eq 1 ] && [ -z "$DEBUG_TOKEN" ] && die "直发测试需要 --debug-token / MW_DEBUG_TOKEN"
   else
     [ -n "$WORKER_URL" ] || usage_die "缺少 --worker-domain / MW_WORKER_DOMAIN（默认取 --app-domain）"
-    [ -n "$WORKER_TOKEN" ] || usage_die "缺少 --worker-token / MW_WORKER_TOKEN（/worker 是排空的唯一入口）"
+    [ -n "$WORKER_TOKEN" ] || usage_die "缺少 --worker-token / MW_WORKER_TOKEN（POST /worker 需要 Bearer 鉴权）"
   fi
 
   BODY_FILE="$(mktemp -d 2>/dev/null)/body" || die "无法创建临时目录"

src/domain/jmap.rs

@@ -13,7 +13,7 @@ use thiserror::Error;
 /// while credentials are never included in these messages.
 #[derive(Debug, Error)]
 pub enum JmapError {
-    #[error("JMAP client error")]
+    #[error("JMAP client error: {0}")]
     Client(#[from] jmap_client::Error),
     #[error("JMAP object was not found")]
     NotFound,

src/main.rs

@@ -30,7 +30,7 @@ use crate::channel::telegram::TelegramClient;
 use crate::domain::jmap::{client::JmapClientBackend, JmapService};
 use config::{encryption_key_from_env, Config};
 use notify::{
-    router_configuration_setup, router_with_worker_state_runtime_bootstrap, MetadataWorker,
+    router_configuration_setup, router_with_worker_handle_runtime_bootstrap, MetadataWorker,
     WorkerHandle, WorkerHandler,
 };
 use state::{runtime_provider, RedisState};
@@ -119,6 +119,9 @@ async fn main() -> Result<(), error::BotError> {
         RedisState::connect_with_encryption(redis_url.expose_secret(), Some(encryption_key))
             .await?,
     );
+    // The in-process background drain keeps its own reference to the state
+    // handle: `worker_state` is moved into the metadata worker below.
+    let drain_state = worker_state.clone();
     let config = match worker_state
         .get_business_config()
         .await
@@ -153,6 +156,9 @@ async fn main() -> Result<(), error::BotError> {
     let allowlist: std::collections::HashSet<i64> =
         config.telegram.chat_allowlist.iter().copied().collect();
     let chat_id = config.telegram.chat_id;
+    // One shared `WorkerHandle` for both the router and the in-process
+    // background drain, so SPA-triggered worker reloads are observed by the
+    // drain without re-binding the handle.
     let worker: Arc<dyn WorkerHandler> = if config.telegram.bot_token.expose_secret().is_empty() {
         Arc::new(notify::NoopWorker)
     } else {
@@ -192,17 +198,24 @@ async fn main() -> Result<(), error::BotError> {
             }
         }
     };
-    let app = router_with_worker_state_runtime_bootstrap(
+    let worker_handle = Arc::new(WorkerHandle::new(worker));
+    let app = router_with_worker_handle_runtime_bootstrap(
         config.auth,
         config.worker_token,
         redis,
         allowlist,
-        Arc::new(WorkerHandle::new(worker)),
+        worker_handle.clone(),
         runtime,
         admin_token,
         debug_token,
     )
     .merge(web::router());
+    // Timely delivery without external cron: periodic drain plus the
+    // post-reconcile one-shot hook. The route-based `/worker` and any cron
+    // remain an optional safety net; shared committed/inflight leases make
+    // concurrent triggers a no-op rather than a double send.
+    let worker_drain: Arc<dyn WorkerHandler> = worker_handle.clone();
+    tokio::spawn(notify::run_stream_drain(worker_drain, drain_state));
     axum::serve(listener, app).await?;
     Ok(())
 }

src/notify.rs

@@ -458,6 +458,11 @@ pub(crate) fn error_response(status: StatusCode, code: &'static str, retry: bool
     }
 }
 
+/// Consumer identity for the in-process background drain. It shares the
+/// `WORKER_STREAM_GROUP` group with the `/worker` route, so committed/inflight
+/// leases decide delivery rather than consumer identity.
+pub(crate) const INPROC_CONSUMER: &str = "inproc-worker";
+
 async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) -> Response {
     refresh_business_config(&app).await;
     if !business_enabled(&app).await {
@@ -471,164 +476,214 @@ async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) ->
         .map(|token| constant_time_eq(token.as_bytes(), worker_token.expose_secret().as_bytes()))
         .unwrap_or(false);
     if !valid {
-        error_response(StatusCode::UNAUTHORIZED, "unauthorized", false)
+        return error_response(StatusCode::UNAUTHORIZED, "unauthorized", false);
+    }
+    let batch = if body.is_empty() {
+        WORKER_STREAM_BATCH
     } else {
-        const GROUP: &str = "stalwart-workers";
+        let Ok(request) = serde_json::from_slice::<WorkerRequest>(&body) else {
-        const CONSUMER: &str = "http-worker";
+            return error_response(StatusCode::BAD_REQUEST, "invalid_request", false);
-        const MAX: usize = 10;
-        let batch = if body.is_empty() {
-            MAX
-        } else {
-            let Ok(request) = serde_json::from_slice::<WorkerRequest>(&body) else {
-                return error_response(StatusCode::BAD_REQUEST, "invalid_request", false);
-            };
-            request
-                .batch
-                .filter(|count| *count > 0)
-                .map_or(MAX, |count| count.min(MAX))
         };
-        for stream in ["stalwart:jmap", "stalwart:telegram"] {
+        request
-            let messages = match app.state.read_batch(stream, GROUP, CONSUMER, batch).await {
+            .batch
-                Ok(messages) => messages,
+            .filter(|count| *count > 0)
-                Err(error) => {
+            .map_or(WORKER_STREAM_BATCH, |count| count.min(WORKER_STREAM_BATCH))
-                    tracing::warn!(stream, reason = %error, "worker: read_batch failed");
+    };
-                    return error_response(
+    if stream_drain(&app.worker, &app.state, WORKER_HTTP_CONSUMER, batch)
-                        StatusCode::SERVICE_UNAVAILABLE,
+        .await
-                        "service_unavailable",
+        .is_err()
-                        true,
+    {
-                    );
+        return error_response(StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", true);
-                }
+    }
-            };
+    StatusCode::NO_CONTENT.into_response()
-            if !messages.is_empty() {
+}
-                tracing::info!(stream, fetched = messages.len(), "worker: batch drained");
+
+/// Delivery consumer group. Both the `/worker` route (`http-worker`) and the
+/// in-process background drain (`inproc-worker`) read this same group, so the
+/// committed/inflight leases decide delivery rather than consumer identity.
+const WORKER_STREAM_GROUP: &str = "stalwart-workers";
+/// Stream entries a single drain pass handles at most.
+pub(crate) const WORKER_STREAM_BATCH: usize = 10;
+/// Streams a single drain pass walks, in order.
+const WORKER_STREAMS: [&str; 2] = ["stalwart:jmap", "stalwart:telegram"];
+/// Consumer identity used by the `/worker` HTTP route.
+pub(crate) const WORKER_HTTP_CONSUMER: &str = "http-worker";
+
+/// Pull one bounded batch from each delivery stream and deliver it.
+///
+/// Shared by the `/worker` route and the in-process background drain, so the
+/// two triggers cannot drift apart on delivery semantics. `Err(())` means a
+/// state-store call failed: the route maps that to a 503 with `Retry-After`,
+/// the background loop logs it and keeps running so one bad tick cannot stop
+/// the loop from running at all.
+///
+/// Per-message order is deliberate: the commit marker is written only after
+/// `process` succeeds, and the short in-flight lease is held only before it, so
+/// a crash cannot permanently suppress an XAUTOCLAIM retry. A message another
+/// consumer is already delivering is skipped, not failed.
+pub(crate) async fn stream_drain(
+    worker: &Arc<dyn WorkerHandler>,
+    state: &Arc<dyn ReliableState>,
+    consumer: &str,
+    batch: usize,
+) -> Result<(), ()> {
+    for stream in WORKER_STREAMS {
+        let messages = match state
+            .read_batch(stream, WORKER_STREAM_GROUP, consumer, batch)
+            .await
+        {
+            Ok(messages) => messages,
+            Err(error) => {
+                tracing::warn!(stream, reason = %error, "worker: read_batch failed");
+                return Err(());
             }
-            for message in messages.into_iter().take(batch) {
+        };
-                // Commit is written only after send succeeds. Before send,
+        if !messages.is_empty() {
-                // use a short in-flight lease so a crash cannot permanently
+            tracing::info!(stream, fetched = messages.len(), "worker: batch drained");
-                // suppress an XAUTOCLAIM retry.
+        }
-                let delivery_key = format!("delivery:committed:{stream}:{}", message.id);
+        for message in messages.into_iter().take(batch) {
-                match app.state.dedup_exists(&delivery_key).await {
+            // Commit is written only after send succeeds. Before send,
-                    Ok(true) => {
+            // use a short in-flight lease so a crash cannot permanently
-                        // Already committed: ack the stream entry and drop it.
+            // suppress an XAUTOCLAIM retry.
-                        // Logged so an acked-without-delivery is visible rather
+            let delivery_key = format!("delivery:committed:{stream}:{}", message.id);
-                        // than looking like a plain successful pass.
+            match state.dedup_exists(&delivery_key).await {
-                        tracing::info!(
+                Ok(true) => {
-                            stream,
+                    // Already committed: ack the stream entry and drop it.
-                            message_id = %message.id,
+                    // Logged so an acked-without-delivery is visible rather
-                            "worker: delivery already committed, acking stream entry"
+                    // than looking like a plain successful pass.
-                        );
-                        if app.state.ack(stream, GROUP, &message.id).await.is_err() {
-                            return error_response(
-                                StatusCode::SERVICE_UNAVAILABLE,
-                                "service_unavailable",
-                                true,
-                            );
-                        }
-                        continue;
-                    }
-                    Ok(false) => {}
-                    Err(error) => {
-                        tracing::warn!(stream, reason = %error, "worker: delivery lookup failed");
-                        return error_response(
-                            StatusCode::SERVICE_UNAVAILABLE,
-                            "service_unavailable",
-                            true,
-                        );
-                    }
-                }
-                let inflight_key = format!("delivery:inflight:{stream}:{}", message.id);
-                match app
-                    .state
-                    .claim_dedup(&inflight_key, ttl::DELIVERY_INFLIGHT_SECONDS)
-                    .await
-                {
-                    // Another instance is already delivering this message.
-                    Ok(false) => {
-                        tracing::info!(
-                            stream,
-                            message_id = %message.id,
-                            "worker: in-flight elsewhere, skipping"
-                        );
-                        continue;
-                    }
-                    Err(error) => {
-                        tracing::warn!(stream, reason = %error, "worker: inflight claim failed");
-                        return error_response(
-                            StatusCode::SERVICE_UNAVAILABLE,
-                            "service_unavailable",
-                            true,
-                        );
-                    }
-                    Ok(true) => {}
-                }
-                if app.worker.process(stream, &message.payload).await.is_ok() {
-                    if app
-                        .state
-                        .claim_dedup(&delivery_key, ttl::DELIVERY_COMMITTED_SECONDS)
-                        .await
-                        .is_err()
-                    {
-                        tracing::warn!(
-                            stream,
-                            message_id = %message.id,
-                            "worker: commit marker write failed"
-                        );
-                        let _ = app.state.release_dedup(&inflight_key).await;
-                        return error_response(
-                            StatusCode::SERVICE_UNAVAILABLE,
-                            "service_unavailable",
-                            true,
-                        );
-                    }
-                    let _ = app.state.release_dedup(&inflight_key).await;
-                    if app.state.ack(stream, GROUP, &message.id).await.is_err() {
-                        tracing::warn!(stream, message_id = %message.id, "worker: ack failed");
-                        return error_response(
-                            StatusCode::SERVICE_UNAVAILABLE,
-                            "service_unavailable",
-                            true,
-                        );
-                    }
                     tracing::info!(
                         stream,
                         message_id = %message.id,
-                        "worker: message delivered and acked"
+                        "worker: delivery already committed, acking stream entry"
                     );
-                } else {
+                    if state
-                    let _ = app.state.release_dedup(&inflight_key).await;
+                        .ack(stream, WORKER_STREAM_GROUP, &message.id)
-                    match app
-                        .state
-                        .retry_or_dlq(stream, &format!("{stream}:dlq"), GROUP, &message, 3)
                         .await
+                        .is_err()
                     {
-                        // `retry_or_dlq` returns whether the entry moved to the
+                        return Err(());
-                        // DLQ. The distinction is the whole point of the call:
+                    }
-                        // a moved entry will not be retried again, so an unlogged
+                    continue;
-                        // move is a silent, permanent loss.
+                }
-                        Ok(true) => tracing::error!(
+                Ok(false) => {}
-                            stream,
+                Err(error) => {
-                            message_id = %message.id,
+                    tracing::warn!(stream, reason = %error, "worker: delivery lookup failed");
-                            dlq = %format!("{stream}:dlq"),
+                    return Err(());
-                            "worker: delivery failed after retries, moved to DLQ"
+                }
-                        ),
+            }
-                        Ok(false) => tracing::warn!(
+            let inflight_key = format!("delivery:inflight:{stream}:{}", message.id);
-                            stream,
+            match state
-                            message_id = %message.id,
+                .claim_dedup(&inflight_key, ttl::DELIVERY_INFLIGHT_SECONDS)
-                            "worker: delivery failed, re-queued for retry"
+                .await
-                        ),
+            {
-                        Err(error) => {
+                // Another consumer is already delivering this message.
-                            tracing::error!(stream, reason = %error, "worker: retry/DLQ failed");
+                Ok(false) => {
-                            return error_response(
+                    tracing::info!(
-                                StatusCode::SERVICE_UNAVAILABLE,
+                        stream,
-                                "service_unavailable",
+                        message_id = %message.id,
-                                true,
+                        "worker: in-flight elsewhere, skipping"
-                            );
+                    );
-                        }
+                    continue;
+                }
+                Err(error) => {
+                    tracing::warn!(stream, reason = %error, "worker: inflight claim failed");
+                    return Err(());
+                }
+                Ok(true) => {}
+            }
+            if worker.process(stream, &message.payload).await.is_ok() {
+                if state
+                    .claim_dedup(&delivery_key, ttl::DELIVERY_COMMITTED_SECONDS)
+                    .await
+                    .is_err()
+                {
+                    tracing::warn!(
+                        stream,
+                        message_id = %message.id,
+                        "worker: commit marker write failed"
+                    );
+                    let _ = state.release_dedup(&inflight_key).await;
+                    return Err(());
+                }
+                let _ = state.release_dedup(&inflight_key).await;
+                if state
+                    .ack(stream, WORKER_STREAM_GROUP, &message.id)
+                    .await
+                    .is_err()
+                {
+                    tracing::warn!(stream, message_id = %message.id, "worker: ack failed");
+                    return Err(());
+                }
+                tracing::info!(
+                    stream,
+                    message_id = %message.id,
+                    "worker: message delivered and acked"
+                );
+            } else {
+                let _ = state.release_dedup(&inflight_key).await;
+                match state
+                    .retry_or_dlq(
+                        stream,
+                        &format!("{stream}:dlq"),
+                        WORKER_STREAM_GROUP,
+                        &message,
+                        3,
+                    )
+                    .await
+                {
+                    // `retry_or_dlq` returns whether the entry moved to the
+                    // DLQ. The distinction is the whole point of the call:
+                    // a moved entry will not be retried again, so an unlogged
+                    // move is a silent, permanent loss.
+                    Ok(true) => tracing::error!(
+                        stream,
+                        message_id = %message.id,
+                        dlq = %format!("{stream}:dlq"),
+                        "worker: delivery failed after retries, moved to DLQ"
+                    ),
+                    Ok(false) => tracing::warn!(
+                        stream,
+                        message_id = %message.id,
+                        "worker: delivery failed, re-queued for retry"
+                    ),
+                    Err(error) => {
+                        tracing::error!(stream, reason = %error, "worker: retry/DLQ failed");
+                        return Err(());
                     }
                 }
             }
         }
-        StatusCode::NO_CONTENT.into_response()
+    }
+    Ok(())
+}
+
+/// Background drain: pull the `stalwart:*` delivery streams without external
+/// cron.
+///
+/// C-NO-LONG-CONN forbids network long connections; this is an in-process task
+/// on the same short Redis round trips the `/worker` route already uses. The
+/// `delivery:committed:` / `delivery:inflight:` leases are shared with the
+/// route, so running this next to `/worker`, another instance, or cron never
+/// double-delivers. Each tick skips work while business mode is disabled; a
+/// failed tick logs and moves on, so the loop never ends.
+pub(crate) async fn run_stream_drain(
+    worker: Arc<dyn WorkerHandler>,
+    state: Arc<dyn ReliableState>,
+) {
+    loop {
+        tokio::time::sleep(std::time::Duration::from_secs(
+            ttl::INPROC_DRAIN_INTERVAL_SECONDS,
+        ))
+        .await;
+        if !state.is_enabled().await.unwrap_or(false) {
+            continue;
+        }
+        if stream_drain(&worker, &state, INPROC_CONSUMER, WORKER_STREAM_BATCH)
+            .await
+            .is_err()
+        {
+            tracing::warn!("worker: background drain tick failed; retrying next tick");
+        }
     }
 }
 
@@ -1490,8 +1545,8 @@ async fn push_reconcile_delta(worker: Arc<dyn WorkerHandler>, state: Arc<dyn Rel
         Ok(value) => value,
         Err(_) => {
             // Same shape as /reconcile: a state read failure means there is no
-            // safe `since` to reconcile from, so release and give up. The cron
+            // safe `since` to reconcile from, so release and give up. The
-            // reconcile covers the change on its next tick.
+            // next `/reconcile` or push call covers the change.
             tracing::warn!("push_reconcile_delta: reconcile state read failed");
             let _ = state.release_lock("lock:reconcile", &lock_owner).await;
             return;
@@ -1532,7 +1587,33 @@ async fn push_reconcile_delta(worker: Arc<dyn WorkerHandler>, state: Arc<dyn Rel
     } else {
         match result {
             Ok(new_state) => match state.set_reconcile_state(&new_state).await {
-                Ok(()) => tracing::info!("push_reconcile_delta: reconcile state advanced"),
+                // The reconcile just enqueued anything new into `stalwart:jmap`.
+                // Drain it here instead of waiting for the background loop's next
+                // tick, so a push-triggered email is delivered within seconds
+                // rather than up to INPROC_DRAIN_INTERVAL_SECONDS. Sharing
+                // stream_drain with the loop keeps the two triggers from drifting
+                // on delivery semantics; the shared committed/inflight leases make
+                // the overlap with the periodic tick a no-op, not a double send.
+                Ok(()) => {
+                    tracing::info!("push_reconcile_delta: reconcile state advanced");
+                    let drain_worker = Arc::clone(&worker);
+                    let drain_state = Arc::clone(&state);
+                    tokio::spawn(async move {
+                        if stream_drain(
+                            &drain_worker,
+                            &drain_state,
+                            INPROC_CONSUMER,
+                            WORKER_STREAM_BATCH,
+                        )
+                        .await
+                        .is_err()
+                        {
+                            tracing::warn!(
+                                "worker: post-reconcile drain failed; background drain will cover it"
+                            );
+                        }
+                    });
+                }
                 Err(_) => tracing::error!("push_reconcile_delta: reconcile state write failed"),
             },
             Err(reason) => {
@@ -2203,7 +2284,65 @@ fn router_with_worker_state_runtime_bootstrap_config<S: ReliableState + 'static>
     debug_token: Option<SecretString>,
     setup_missing: Vec<String>,
 ) -> Router {
-    let worker_handle = Arc::new(WorkerHandle::new(worker_handler));
+    router_with_worker_state_runtime_bootstrap_config_with_handle(
+        secrets,
+        worker_token,
+        state,
+        allowlist,
+        Arc::new(WorkerHandle::new(worker_handler)),
+        runtime,
+        admin_token,
+        debug_token,
+        setup_missing,
+    )
+}
+
+/// Same bootstrap router, but the caller's `WorkerHandle` is shared instead of
+/// a fresh one being built from the raw handler. Production wiring
+/// (`src/main.rs`) passes the same handle to `run_stream_drain`, so the
+/// background drain sees SPA-triggered worker reloads.
+#[expect(
+    clippy::too_many_arguments,
+    reason = "router wiring keeps state dependencies explicit"
+)]
+pub fn router_with_worker_handle_runtime_bootstrap<S: ReliableState + 'static>(
+    secrets: AuthSecrets,
+    worker_token: SecretString,
+    state: S,
+    allowlist: HashSet<i64>,
+    worker_handle: Arc<WorkerHandle>,
+    runtime: RuntimeConfigProvider,
+    admin_token: SecretString,
+    debug_token: Option<SecretString>,
+) -> Router {
+    router_with_worker_state_runtime_bootstrap_config_with_handle(
+        secrets,
+        worker_token,
+        state,
+        allowlist,
+        worker_handle,
+        runtime,
+        admin_token,
+        debug_token,
+        Vec::new(),
+    )
+}
+
+#[expect(
+    clippy::too_many_arguments,
+    reason = "router wiring keeps state dependencies explicit"
+)]
+fn router_with_worker_state_runtime_bootstrap_config_with_handle<S: ReliableState + 'static>(
+    secrets: AuthSecrets,
+    worker_token: SecretString,
+    state: S,
+    allowlist: HashSet<i64>,
+    worker_handle: Arc<WorkerHandle>,
+    runtime: RuntimeConfigProvider,
+    admin_token: SecretString,
+    debug_token: Option<SecretString>,
+    setup_missing: Vec<String>,
+) -> Router {
     let reload = Arc::new(ReloadCoordinator::new(worker_handle.clone(), None));
     let app_state = AppState {
         auth: AuthState::from(secrets),
@@ -4077,4 +4216,137 @@ mod tests {
             Some(url)
         );
     }
+
+    // ── stream_drain 单元测试（进程内 drain 共享的投递核心）──
+    //
+    // 与 `/worker` 路由走同一条路径：成功 → 提交标记 + ack；失败 → 重试计数，
+    // 第 3 次进 DLQ；已提交的条目短路 ack、不再调用 `process`。
+
+    /// 记录调用次数与 payload 的成功 Worker：`process` 恒成功。
+    struct DeliveringMockWorker {
+        calls: Arc<std::sync::atomic::AtomicUsize>,
+        payloads: Arc<std::sync::Mutex<Vec<String>>>,
+    }
+
+    #[async_trait::async_trait]
+    impl WorkerHandler for DeliveringMockWorker {
+        async fn process(&self, _stream: &str, payload: &str) -> Result<(), ()> {
+            self.calls.fetch_add(1, Ordering::SeqCst);
+            self.payloads.lock().unwrap().push(payload.to_owned());
+            Ok(())
+        }
+    }
+
+    #[tokio::test]
+    async fn stream_drain_delivers_acks_and_commits() {
+        let state: Arc<dyn ReliableState> = Arc::new(MemoryState::enabled_for_tests());
+        let id = state
+            .enqueue("stalwart:jmap", r#"{"event":"message/new"}"#)
+            .await
+            .unwrap();
+        let worker = DeliveringMockWorker {
+            calls: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
+            payloads: Arc::new(std::sync::Mutex::new(Vec::new())),
+        };
+        let calls = worker.calls.clone();
+        let payloads = worker.payloads.clone();
+        let handler: Arc<dyn WorkerHandler> = Arc::new(worker);
+
+        assert!(stream_drain(&handler, &state, "test-drain", 10)
+            .await
+            .is_ok());
+
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+        assert_eq!(
+            *payloads.lock().unwrap(),
+            vec![r#"{"event":"message/new"}"#.to_owned()]
+        );
+        let committed = format!("delivery:committed:stalwart:jmap:{id}");
+        let inflight = format!("delivery:inflight:stalwart:jmap:{id}");
+        assert!(state.dedup_exists(&committed).await.unwrap());
+        assert!(!state.dedup_exists(&inflight).await.unwrap());
+        // acked 条目不再可读：不会重复投递。
+        let drained = state
+            .read_batch("stalwart:jmap", WORKER_STREAM_GROUP, "test-drain", 10)
+            .await
+            .unwrap();
+        assert!(drained.is_empty());
+    }
+
+    #[tokio::test]
+    async fn stream_drain_moves_repeated_failures_to_dlq() {
+        let state: Arc<dyn ReliableState> = Arc::new(MemoryState::enabled_for_tests());
+        let id = state
+            .enqueue("stalwart:telegram", "payload-x")
+            .await
+            .unwrap();
+        let handler: Arc<dyn WorkerHandler> = Arc::new(NoopWorker);
+        let dlq = "stalwart:telegram:dlq";
+        for _ in 0..2 {
+            assert!(stream_drain(&handler, &state, "test-drain", 10)
+                .await
+                .is_ok());
+            // 前两次失败：重新入队等待重试，源流里仍在，DLQ 还是空的。
+            let pending = state
+                .read_batch("stalwart:telegram", WORKER_STREAM_GROUP, "test-drain", 10)
+                .await
+                .unwrap();
+            assert_eq!(pending.len(), 1);
+            assert_eq!(pending[0].id, id);
+            let dlq_batch = state
+                .read_batch(dlq, WORKER_STREAM_GROUP, "test-drain", 10)
+                .await
+                .unwrap();
+            assert!(dlq_batch.is_empty());
+        }
+        // 第 3 次失败：移入 DLQ，源流清空。
+        assert!(stream_drain(&handler, &state, "test-drain", 10)
+            .await
+            .is_ok());
+        assert!(state
+            .read_batch("stalwart:telegram", WORKER_STREAM_GROUP, "test-drain", 10)
+            .await
+            .unwrap()
+            .is_empty());
+        let dlq_batch = state
+            .read_batch(dlq, WORKER_STREAM_GROUP, "test-drain", 10)
+            .await
+            .unwrap();
+        assert_eq!(dlq_batch.len(), 1);
+        assert_eq!(dlq_batch[0].payload, "payload-x");
+    }
+
+    #[tokio::test]
+    async fn stream_drain_short_circuits_precommitted_messages() {
+        let state: Arc<dyn ReliableState> = Arc::new(MemoryState::enabled_for_tests());
+        let id = state
+            .enqueue("stalwart:jmap", "already-done")
+            .await
+            .unwrap();
+        assert!(state
+            .claim_dedup(
+                &format!("delivery:committed:stalwart:jmap:{id}"),
+                crate::state::ttl::DELIVERY_COMMITTED_SECONDS
+            )
+            .await
+            .unwrap());
+        let worker = DeliveringMockWorker {
+            calls: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
+            payloads: Arc::new(std::sync::Mutex::new(Vec::new())),
+        };
+        let calls = worker.calls.clone();
+        let handler: Arc<dyn WorkerHandler> = Arc::new(worker);
+
+        assert!(stream_drain(&handler, &state, "test-drain", 10)
+            .await
+            .is_ok());
+
+        // 已提交：`process` 一次都不调，条目 ack 掉。
+        assert_eq!(calls.load(Ordering::SeqCst), 0);
+        let drained = state
+            .read_batch("stalwart:jmap", WORKER_STREAM_GROUP, "test-drain", 10)
+            .await
+            .unwrap();
+        assert!(drained.is_empty());
+    }
 }

src/state.rs

@@ -59,6 +59,12 @@ pub(crate) mod ttl {
     pub const DELIVERY_INFLIGHT_SECONDS: u64 = MINUTE;
     /// Delivery idempotency window.
     pub const DELIVERY_COMMITTED_SECONDS: u64 = WEEK;
+    /// Interval between in-process background drains of `stalwart:*` delivery
+    /// streams. The background drain (consumer `inproc-worker`) is what makes
+    /// delivery timely without external cron; every `/worker` route call and
+    /// `push_reconcile_delta` completion also trigger a drain, so this only
+    /// bounds the latency of enqueues that no other trigger follows.
+    pub const INPROC_DRAIN_INTERVAL_SECONDS: u64 = 10;
     /// How long a `disabled` push subscription stays marked before re-check.
     pub const PUSH_DISABLED_SECONDS: u64 = DAY;
     /// How long a `pending` push subscription waits for verification.

src/worker.rs

@@ -36,15 +36,17 @@ pub trait WorkerHandler: Send + Sync {
     }
 
     /// Performs one bounded JMAP reconciliation pass. Implementations must not
-    /// persist the cursor; the HTTP coordinator commits it after enqueueing.
+    /// persist the cursor; the two commit sites — the `/reconcile` route
+    /// (`reconcile` in `src/notify.rs`) and `push_reconcile_delta` in
+    /// `src/notify.rs` — call `set_reconcile_state` after enqueuing.
     ///
     /// `initial_changes` sizes the first `/changes` call only. The pass may
     /// widen the window afterwards, which is why it is not called `max_changes`.
     ///
     /// On failure, returns a `&'static str` diagnostic code identifying which
-    /// step broke (e.g. `"jmap_session_failed"`, `"jmap_list_failed"`). The HTTP
+    /// step broke (e.g. `"jmap_session_failed"`, `"jmap_list_failed"`). The
-    /// layer surfaces this as a `reason` field in the 503 body so operators can
+    /// `/reconcile` route surfaces this as the `reason` field in the 503
-    /// see the cause without server-side logs.
+    /// body; `push_reconcile_delta` logs it and leaves the state unadvanced.
     async fn reconcile(
         &self,
         _since_state: Option<&str>,