diag(worker): instrument silent delivery paths; fix %/ ? error formatters

worker.process now logs each of the seven failure reasons it already had
(unknown stream, missing id, account mismatch, read_email err/None,
telegram send failure) without echoing credentials, so the delivery tail
can be diagnosed from stdout alone. notify.rs /worker now logs the
retry_or_dlq outcome it previously discarded, and logs each
commit/ack step. Fixes two formatting slips in the new code: a read
error logged with ?error (Debug) instead of %error (Display), and a
reconcile error line that misattributed old_state to the position walk
it actually discarded.

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

docs/charter.md

@@ -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: 106 passed / 0 failed / 4 ignored** (4 cursor/walk tests + 2 `/api/status` Host-header tests (label echo + null fallback)), retested 2026-10-14; 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: 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`).
 
 **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/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:1565-1566`) |
+| `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`) |
 | `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:1565-1566`） |
+| `push:registration:{sha256(callback_url)}` 曾被写成「360s 注册单飞锁」 | 事实误标 | 该键是 7d 回调→订阅 ID 映射 | 真正的 360s 单飞锁是 `lock:push-register:{sha256(callback_url)}`（`src/notify.rs:1724-1727`） |
 | `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`「已知边界」同句已删 |
 

src/notify.rs

@@ -490,14 +490,18 @@ async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) ->
         for stream in ["stalwart:jmap", "stalwart:telegram"] {
             let messages = match app.state.read_batch(stream, GROUP, CONSUMER, batch).await {
                 Ok(messages) => messages,
-                Err(_) => {
+                Err(error) => {
+                    tracing::warn!(stream, reason = %error, "worker: read_batch failed");
                     return error_response(
                         StatusCode::SERVICE_UNAVAILABLE,
                         "service_unavailable",
                         true,
-                    )
+                    );
                 }
             };
+            if !messages.is_empty() {
+                tracing::info!(stream, fetched = messages.len(), "worker: batch drained");
+            }
             for message in messages.into_iter().take(batch) {
                 // Commit is written only after send succeeds. Before send,
                 // use a short in-flight lease so a crash cannot permanently
@@ -505,6 +509,14 @@ async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) ->
                 let delivery_key = format!("delivery:committed:{stream}:{}", message.id);
                 match app.state.dedup_exists(&delivery_key).await {
                     Ok(true) => {
+                        // Already committed: ack the stream entry and drop it.
+                        // Logged so an acked-without-delivery is visible rather
+                        // than looking like a plain successful pass.
+                        tracing::info!(
+                            stream,
+                            message_id = %message.id,
+                            "worker: delivery already committed, acking stream entry"
+                        );
                         if app.state.ack(stream, GROUP, &message.id).await.is_err() {
                             return error_response(
                                 StatusCode::SERVICE_UNAVAILABLE,
@@ -515,12 +527,13 @@ async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) ->
                         continue;
                     }
                     Ok(false) => {}
-                    Err(_) => {
+                    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);
@@ -529,13 +542,22 @@ async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) ->
                     .claim_dedup(&inflight_key, ttl::DELIVERY_INFLIGHT_SECONDS)
                     .await
                 {
-                    Ok(false) => continue,
+                    // Another instance is already delivering this message.
-                    Err(_) => {
+                    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) => {}
                 }
@@ -546,6 +568,11 @@ async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) ->
                         .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,
@@ -555,20 +582,42 @@ async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) ->
                     }
                     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"
+                    );
                 } else {
                     let _ = app.state.release_dedup(&inflight_key).await;
-                    if app
+                    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
+                        // 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 error_response(
                                 StatusCode::SERVICE_UNAVAILABLE,
                                 "service_unavailable",
@@ -578,6 +627,7 @@ async fn worker(State(app): State<AppState>, headers: HeaderMap, body: Bytes) ->
                     }
                 }
             }
+        }
         StatusCode::NO_CONTENT.into_response()
     }
 }

src/worker.rs

@@ -321,22 +321,46 @@ impl<B: JmapBackend> WorkerHandler for MetadataWorker<B> {
             // write `delivery:committed` + XACK and silently drop an event we do
             // not understand. Falling through to Err routes it to retry/DLQ so it
             // stays observable instead of vanishing.
+            tracing::warn!(
+                stream,
+                "worker.process: unknown stream, routing to retry/DLQ"
+            );
             return Err(());
         }
         let event: WorkerEvent = serde_json::from_str(payload).map_err(|_| ())?;
         let (Some(account), Some(email)) = (event.account_id.as_deref(), event.email_id.as_deref())
         else {
+            tracing::warn!(
+                has_account = event.account_id.is_some(),
+                has_email = event.email_id.is_some(),
+                "worker.process: payload missing accountId or emailId"
+            );
             return Err(());
         };
         if account != self.jmap.account_id() {
+            // The id we are bound to versus the id the event names: logging both
+            // distinguishes a stale event from a misconfigured account binding.
+            tracing::warn!(
+                event_account = %account,
+                bound_account = %self.jmap.account_id(),
+                "worker.process: event account does not match the bound account"
+            );
             return Err(());
         }
-        let content = self
+        let content = match self.jmap.read_email(email).await {
-            .jmap
+            Ok(content) => content,
-            .read_email(email)
+            Err(error) => {
-            .await
+                tracing::warn!(
-            .map_err(|_| ())?
+                    reason = %error,
-            .ok_or(())?;
+                    email_id = %email,
+                    "worker.process: read_email failed"
+                );
+                return Err(());
+            }
+        }
+        .ok_or_else(
+            || tracing::warn!(email_id = %email, "worker.process: read_email returned no email"),
+        )?;
         let metadata = content.metadata;
         self.telegram
             .send_notification(
@@ -351,7 +375,16 @@ impl<B: JmapBackend> WorkerHandler for MetadataWorker<B> {
                 },
             )
             .await
-            .map_err(|_| ())
+            .map_err(|reason| {
+                // The client's render target is a chat id, so the failure can be
+                // logged as such without exposing the token (AGENTS.md §3.2).
+                tracing::warn!(
+                    chat_id = %self.chat_id,
+                    reason = %reason,
+                    email_id = %email,
+                    "worker.process: telegram send failed"
+                )
+            })
     }
 
     async fn reconcile(
@@ -484,7 +517,7 @@ impl<B: JmapBackend> MetadataWorker<B> {
             pages += 1;
             let changes = match self.jmap.email_changes(&state, window).await {
                 Ok(changes) => changes,
-                Err(_) => {
+                Err(error) => {
                     // Re-baseline instead of failing the invocation. jmap-client
                     // 0.4.2 exposes no reset signal: `ChangesResponse` only
                     // carries `oldState`/`newState`/`hasMoreChanges`/`created`/
@@ -499,19 +532,36 @@ impl<B: JmapBackend> MetadataWorker<B> {
                     // state at position 0 recovers on the next run. The 24h
                     // dedup key makes the resulting re-read at-most-duplicate,
                     // never-loss.
-                    let rebaselined = ReconcileCursor::walk_from(
+                    let recovered = ReconcileCursor::walk_from(
                         self.jmap
                             .session_state()
                             .await
                             .map_err(|_| "jmap_session_failed")?,
                     );
-                    return Ok(rebaselined.to_persisted());
+                    // Log at error, not warn: the cursor advances here, so from
+                    // the operator's view the pass succeeded while not one email
+                    // was enqueued. Without this line the state-change log and a
+                    // genuinely successful pull are indistinguishable.
+                    tracing::error!(
+                        reason = %error,
+                        pages,
+                        discarded_since_state = %state,
+                        "reconcile: /changes failed, cursor reset to the position walk at position 0 with zero emails enqueued"
+                    );
+                    return Ok(recovered.to_persisted());
                 }
             };
             for email_id in changes.created.iter().chain(changes.updated.iter()) {
                 self.enqueue_reconcile_event(email_id).await?;
             }
             if !changes.has_more_changes {
+                tracing::info!(
+                    pages,
+                    window,
+                    created = changes.created.len(),
+                    updated = changes.updated.len(),
+                    "reconcile: /changes replay complete"
+                );
                 return Ok(changes.new_state);
             }
             if window < CHANGE_WINDOW_CAP {
@@ -556,12 +606,39 @@ impl<B: JmapBackend> MetadataWorker<B> {
         // mid-way must not leave the dedup key claimed while the payload is
         // absent from the stream (which would drop the event for the whole
         // 24h dedup window).
-        self.state
+        match self
+            .state
             .claim_dedup_and_enqueue(&key, ttl::DEDUP_JMAP_SECONDS, "stalwart:jmap", &payload)
             .await
-            .map_err(|_| "enqueue_failed")?;
+        {
+            // A `false` here means the dedup key was already claimed within the
+            // last 24h, so no event was enqueued at all. That is correct
+            // deduplication, but it is otherwise the one silent path in which a
+            // `/changes` response naming this email produces no Telegram message.
+            Ok(true) => {
+                tracing::info!(
+                    email_id = %email_id,
+                    "reconcile: email enqueued to stalwart:jmap"
+                );
                 Ok(())
             }
+            Ok(false) => {
+                tracing::info!(
+                    email_id = %email_id,
+                    "reconcile: email suppressed by the 24h dedup key"
+                );
+                Ok(())
+            }
+            Err(error) => {
+                tracing::warn!(
+                    reason = %error,
+                    email_id = %email_id,
+                    "reconcile: enqueue failed"
+                );
+                Err("enqueue_failed")
+            }
+        }
+    }
 
     async fn process_telegram(&self, payload: &str) -> Result<(), ()> {
         let update: TelegramUpdate = serde_json::from_str(payload).map_err(|_| ())?;