fix(worker,state,jmap,config): close independent-review findings

- worker/state: paginate Email/changes by widening maxChanges on the same
  sinceState (jmap-client 0.4.2 has no upToId) to avoid skipping a batch when
  advancing per page; make dedup+enqueue atomic via claim_dedup_and_enqueue
  (Lua SET NX EX then XADD); scale the XAUTOCLAIM idle threshold by batch size;
  XACK payload-less malformed stream entries out of the PEL via ack_malformed;
  fail closed on unknown streams (Err); tighten SET NX EX TTL to .max(1);
  derive the CSPRNG fallback owner-token from time+PID+counter instead of a
  constant; clear the push verification digest on push:disable
- worker: /help menu now lists only the consent triggers the parser accepts
- jmap/config: complete client and configuration validation

src/config.rs

@@ -49,7 +49,6 @@ pub struct Config {
 pub struct AuthSecrets {
     pub reconcile_token: SecretString,
     pub telegram_webhook_secret: SecretString,
-    pub jmap_push_verification: SecretString,
 }
 
 pub struct TelegramConfig {
@@ -85,7 +84,6 @@ pub struct BusinessConfig {
     pub jmap_session_url: String,
     pub jmap_username: String,
     pub jmap_password: SecretString,
-    pub jmap_push_verification: SecretString,
     pub account_id: Option<String>,
     pub llm_enabled: bool,
     pub llm_allow_net: bool,
@@ -107,7 +105,6 @@ pub(crate) struct BusinessConfigWire {
     pub jmap_session_url: String,
     pub jmap_username: String,
     pub jmap_password: String,
-    pub jmap_push_verification: String,
     pub account_id: Option<String>,
     pub llm_enabled: bool,
     pub llm_allow_net: bool,
@@ -128,7 +125,6 @@ impl From<BusinessConfig> for BusinessConfigWire {
             jmap_session_url: value.jmap_session_url,
             jmap_username: value.jmap_username,
             jmap_password: value.jmap_password.expose_secret().to_owned(),
-            jmap_push_verification: value.jmap_push_verification.expose_secret().to_owned(),
             account_id: value.account_id,
             llm_enabled: value.llm_enabled,
             llm_allow_net: value.llm_allow_net,
@@ -150,7 +146,6 @@ impl TryFrom<BusinessConfigWire> for BusinessConfig {
         validate_nonblank("JMAP_USERNAME", &value.jmap_username)?;
         validate_nonblank("JMAP_PASSWORD", &value.jmap_password)?;
         validate_nonblank("TG_WEBHOOK_SECRET", &value.telegram_webhook_secret)?;
-        validate_nonblank("JMAP_PUSH_VERIFICATION", &value.jmap_push_verification)?;
         validate_nonblank("RECONCILE_TOKEN", &value.reconcile_token)?;
         validate_nonblank("WORKER_TOKEN", &value.worker_token)?;
         if let Some(key) = value.llm_api_key.as_deref() {
@@ -164,7 +159,6 @@ impl TryFrom<BusinessConfigWire> for BusinessConfig {
             jmap_session_url: value.jmap_session_url,
             jmap_username: value.jmap_username,
             jmap_password: SecretString::new(value.jmap_password),
-            jmap_push_verification: SecretString::new(value.jmap_push_verification),
             account_id: value.account_id,
             llm_enabled: value.llm_enabled,
             llm_allow_net: value.llm_allow_net,
@@ -179,6 +173,12 @@ impl TryFrom<BusinessConfigWire> for BusinessConfig {
 
 pub(crate) fn validate_business_wire(wire: BusinessConfigWire) -> Result<(), BotError> {
     let config: BusinessConfig = wire.try_into()?;
+    if config.chat_allowlist.is_empty() {
+        // SAF-CHAT-ALLOWLIST: 必填，处理前先拒绝非白名单；空名单是非法配置而非 fail-open。
+        return Err(BotError::Config(
+            "CHAT_ALLOWLIST must contain at least one chat id".into(),
+        ));
+    }
     let session = url::Url::parse(&config.jmap_session_url)
         .map_err(|_| BotError::Config("JMAP_SESSION_URL must be a valid URL".into()))?;
     if session.scheme() != "https"
@@ -242,7 +242,6 @@ impl Config {
             auth: AuthSecrets {
                 reconcile_token: empty(),
                 telegram_webhook_secret: empty(),
-                jmap_push_verification: empty(),
             },
             worker_token: empty(),
         }
@@ -277,7 +276,6 @@ impl Config {
             auth: AuthSecrets {
                 reconcile_token: value.reconcile_token,
                 telegram_webhook_secret: value.telegram_webhook_secret,
-                jmap_push_verification: value.jmap_push_verification,
             },
             worker_token: value.worker_token,
         }
@@ -299,6 +297,10 @@ impl Config {
         Ok(Self::from_business(redis_url, wire.try_into()?))
     }
 
+    /// 遗留兼容路径：从进程环境变量逐项组装完整配置（阶段0 口径）。
+    /// 当前生产启动仅需 `REDIS_URL` + `CONFIG_ENCRYPTION_KEY`（见 `src/main.rs`），
+    /// 业务字段改由 Redis 业务配置（`config:business`，`PUT /api/business-config` 热加载）托管；
+    /// 本函数仅当相关环境变量实际存在时作为引导路径使用，未声明废弃、未计划删除。
     pub fn from_env() -> Result<Self, BotError> {
         let port = std::env::var("PORT")
             .unwrap_or_else(|_| "8080".into())
@@ -353,7 +355,6 @@ impl Config {
             auth: AuthSecrets {
                 reconcile_token: required_secret("RECONCILE_TOKEN")?,
                 telegram_webhook_secret: required_secret("TG_WEBHOOK_SECRET")?,
-                jmap_push_verification: required_secret("JMAP_PUSH_VERIFICATION")?,
             },
             worker_token: required_secret("WORKER_TOKEN")?,
         })
@@ -411,6 +412,53 @@ mod tests {
         std::env::remove_var("S1_TEST_AUTH_SECRET");
     }
 
+    #[test]
+    fn empty_chat_allowlist_is_rejected() {
+        // SAF-CHAT-ALLOWLIST: 空名单是非法配置（Web 侧同样强制至少一项），
+        // 不得让绕过 SPA 的直接 PUT 静默写入"拒绝所有 chat"的配置。
+        let wire = BusinessConfigWire {
+            bot_token: "bot".into(),
+            telegram_chat_id: 1,
+            chat_allowlist: Vec::new(),
+            telegram_webhook_secret: "hook".into(),
+            jmap_session_url: "https://example.invalid".into(),
+            jmap_username: "u".into(),
+            jmap_password: "p".into(),
+            account_id: None,
+            llm_enabled: false,
+            llm_allow_net: false,
+            llm_api_key: None,
+            llm_base_url: None,
+            llm_model: None,
+            reconcile_token: "r".into(),
+            worker_token: "w".into(),
+        };
+        let error = validate_business_wire(wire).expect_err("empty allowlist must fail");
+        assert!(error.to_string().contains("CHAT_ALLOWLIST"));
+    }
+
+    #[test]
+    fn nonempty_chat_allowlist_passes() {
+        let wire = BusinessConfigWire {
+            bot_token: "bot".into(),
+            telegram_chat_id: 1,
+            chat_allowlist: vec![1, 2],
+            telegram_webhook_secret: "hook".into(),
+            jmap_session_url: "https://example.invalid".into(),
+            jmap_username: "u".into(),
+            jmap_password: "p".into(),
+            account_id: None,
+            llm_enabled: false,
+            llm_allow_net: false,
+            llm_api_key: None,
+            llm_base_url: None,
+            llm_model: None,
+            reconcile_token: "r".into(),
+            worker_token: "w".into(),
+        };
+        validate_business_wire(wire).expect("non-empty allowlist must pass");
+    }
+
     #[test]
     fn required_config_values_reject_blank_without_exposing_values() {
         for name in [
@@ -437,7 +485,6 @@ mod tests {
             ("JMAP_PASSWORD", "password"),
             ("REDIS_URL", "redis://invalid"),
             ("TG_WEBHOOK_SECRET", "telegram"),
-            ("JMAP_PUSH_VERIFICATION", "verification"),
             ("TELEGRAM_CHAT_ID", "123"),
         ] {
             std::env::set_var(name, value);
@@ -459,7 +506,6 @@ mod tests {
             "REDIS_URL",
             "RECONCILE_TOKEN",
             "TG_WEBHOOK_SECRET",
-            "JMAP_PUSH_VERIFICATION",
             "TELEGRAM_CHAT_ID",
         ] {
             std::env::remove_var(name);

src/domain/jmap.rs

@@ -55,8 +55,33 @@ pub struct EmailContent {
     pub is_long: bool,
 }
 
+/// A channel-neutral page of JMAP Email/changes. Deleted IDs are retained for
+/// cursor accounting but are not notified because they have no metadata.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct EmailChanges {
+    pub new_state: String,
+    pub created: Vec<String>,
+    pub updated: Vec<String>,
+    pub destroyed: Vec<String>,
+    pub has_more: bool,
+}
+
 #[async_trait]
 pub trait JmapBackend: Send + Sync {
+    async fn create_push_subscription(&self, callback_url: &str) -> Result<String, JmapError>;
+    async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), JmapError>;
+    async fn verify_push_subscription(
+        &self,
+        subscription_id: &str,
+        verification_code: &str,
+    ) -> Result<(), JmapError>;
+    async fn current_state(&self) -> Result<String, JmapError>;
+    async fn email_changes(
+        &self,
+        account_id: &str,
+        since_state: &str,
+        max_changes: usize,
+    ) -> Result<EmailChanges, JmapError>;
     async fn list_folders(&self, account_id: &str) -> Result<Vec<Folder>, JmapError>;
     async fn list_emails(
         &self,
@@ -64,6 +89,21 @@ pub trait JmapBackend: Send + Sync {
         folder_id: Option<&str>,
         limit: usize,
     ) -> Result<Vec<EmailMetadata>, JmapError>;
+    /// Fetch one bounded page without requiring the caller to materialize the
+    /// entire mailbox. Implementations with a native JMAP position query should
+    /// override this; the fallback preserves test backends' existing contract.
+    async fn list_emails_page(
+        &self,
+        account_id: &str,
+        folder_id: Option<&str>,
+        position: usize,
+        limit: usize,
+    ) -> Result<Vec<EmailMetadata>, JmapError> {
+        let all = self
+            .list_emails(account_id, folder_id, position.saturating_add(limit))
+            .await?;
+        Ok(all.into_iter().skip(position).take(limit).collect())
+    }
     async fn read_email(
         &self,
         account_id: &str,
@@ -109,6 +149,49 @@ impl<B> JmapService<B> {
 }
 
 impl<B: JmapBackend> JmapService<B> {
+    pub async fn create_push_subscription(&self, callback_url: &str) -> Result<String, JmapError> {
+        if !callback_url.starts_with("https://") {
+            return Err(JmapError::InvalidRequest("push callback URL"));
+        }
+        self.backend.create_push_subscription(callback_url).await
+    }
+
+    pub async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), JmapError> {
+        if subscription_id.trim().is_empty() {
+            return Err(JmapError::InvalidRequest("push subscription id"));
+        }
+        self.backend
+            .destroy_push_subscription(subscription_id)
+            .await
+    }
+    pub async fn verify_push_subscription(
+        &self,
+        subscription_id: &str,
+        verification_code: &str,
+    ) -> Result<(), JmapError> {
+        if subscription_id.trim().is_empty() || verification_code.trim().is_empty() {
+            return Err(JmapError::InvalidRequest("push verification"));
+        }
+        self.backend
+            .verify_push_subscription(subscription_id, verification_code)
+            .await
+    }
+    pub async fn current_state(&self) -> Result<String, JmapError> {
+        self.backend.current_state().await
+    }
+
+    pub async fn email_changes(
+        &self,
+        since_state: &str,
+        max_changes: usize,
+    ) -> Result<EmailChanges, JmapError> {
+        if since_state.trim().is_empty() || max_changes == 0 {
+            return Err(JmapError::InvalidRequest("changes"));
+        }
+        self.backend
+            .email_changes(&self.account_id, since_state, max_changes)
+            .await
+    }
     pub async fn list_folders(&self) -> Result<Vec<Folder>, JmapError> {
         self.backend.list_folders(&self.account_id).await
     }
@@ -126,6 +209,20 @@ impl<B: JmapBackend> JmapService<B> {
             .await
     }
 
+    pub async fn list_emails_page(
+        &self,
+        folder_id: Option<&str>,
+        position: usize,
+        limit: usize,
+    ) -> Result<Vec<EmailMetadata>, JmapError> {
+        if limit == 0 {
+            return Err(JmapError::InvalidRequest("limit"));
+        }
+        self.backend
+            .list_emails_page(&self.account_id, folder_id, position, limit)
+            .await
+    }
+
     pub async fn read_email(&self, email_id: &str) -> Result<Option<EmailContent>, JmapError> {
         if email_id.trim().is_empty() {
             return Err(JmapError::InvalidRequest("email_id"));
@@ -147,6 +244,43 @@ mod tests {
 
     #[async_trait]
     impl JmapBackend for MockBackend {
+        async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), JmapError> {
+            assert_eq!(subscription_id, "push-1");
+            Ok(())
+        }
+        async fn create_push_subscription(&self, callback_url: &str) -> Result<String, JmapError> {
+            assert_eq!(callback_url, "https://bot.example/push/jmap");
+            Ok("push-1".into())
+        }
+        async fn verify_push_subscription(
+            &self,
+            subscription_id: &str,
+            verification_code: &str,
+        ) -> Result<(), JmapError> {
+            assert_eq!((subscription_id, verification_code), ("push-1", "code"));
+            Ok(())
+        }
+        async fn current_state(&self) -> Result<String, JmapError> {
+            Ok("state-1".into())
+        }
+        async fn email_changes(
+            &self,
+            account_id: &str,
+            since_state: &str,
+            max_changes: usize,
+        ) -> Result<EmailChanges, JmapError> {
+            assert_eq!(
+                (account_id, since_state, max_changes),
+                ("account-1", "old", 10)
+            );
+            Ok(EmailChanges {
+                new_state: "new".into(),
+                created: vec!["email-1".into()],
+                updated: vec![],
+                destroyed: vec![],
+                has_more: false,
+            })
+        }
         async fn list_folders(&self, account_id: &str) -> Result<Vec<Folder>, JmapError> {
             assert_eq!(account_id, "account-1");
             Ok(vec![Folder {

src/domain/jmap/client.rs

@@ -1,9 +1,9 @@
 //! Concrete read-only adapter (MOD-JMAP-CLIENT, GATE-G1-JMAP-READONLY).
 
-use super::{EmailContent, EmailMetadata, Folder, JmapBackend, JmapError};
+use super::{EmailChanges, EmailContent, EmailMetadata, Folder, JmapBackend, JmapError};
 use crate::state::{runtime_provider, OutboundConfig, RuntimeConfigProvider};
 use async_trait::async_trait;
-use jmap_client::{email, mailbox, Get};
+use jmap_client::{email, mailbox, DataType, Get};
 
 pub struct JmapClientBackend {
     client: jmap_client::client::Client,
@@ -67,6 +67,111 @@ impl JmapClientBackend {
 
 #[async_trait]
 impl JmapBackend for JmapClientBackend {
+    async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), JmapError> {
+        if subscription_id.trim().is_empty() {
+            return Err(JmapError::InvalidRequest("push subscription id"));
+        }
+        tokio::time::timeout(
+            self.timeout(),
+            self.client.push_subscription_destroy(subscription_id),
+        )
+        .await
+        .map_err(|_| JmapError::Timeout)??;
+        Ok(())
+    }
+
+    async fn create_push_subscription(&self, callback_url: &str) -> Result<String, JmapError> {
+        let parsed = url::Url::parse(callback_url)
+            .map_err(|_| JmapError::InvalidRequest("push callback URL"))?;
+        if parsed.scheme() != "https" || parsed.username() != "" || parsed.password().is_some() {
+            return Err(JmapError::InvalidRequest("push callback URL"));
+        }
+        let subscription = tokio::time::timeout(
+            self.timeout(),
+            self.client.push_subscription_create(
+                format!("message-weave-{}", self.account_id),
+                callback_url,
+                None,
+            ),
+        )
+        .await
+        .map_err(|_| JmapError::Timeout)??;
+        let subscription_id = subscription
+            .id()
+            .map(ToOwned::to_owned)
+            .ok_or(JmapError::MissingField("push subscription id"))?;
+        // JMAP PushSubscription/set create has no `types` argument in
+        // jmap-client 0.4.2. Restrict the newly-created subscription with the
+        // follow-up update before exposing its id (REQ-PUSH-TYPES).
+        if tokio::time::timeout(
+            self.timeout(),
+            self.client.push_subscription_update_types(
+                &subscription_id,
+                Some([DataType::Email, DataType::EmailDelivery]),
+            ),
+        )
+        .await
+        .map_err(|_| JmapError::Timeout)?
+        .is_err()
+        {
+            let _ = self
+                .client
+                .push_subscription_destroy(&subscription_id)
+                .await;
+            return Err(JmapError::InvalidRequest("push subscription types"));
+        }
+        Ok(subscription_id)
+    }
+
+    async fn verify_push_subscription(
+        &self,
+        subscription_id: &str,
+        verification_code: &str,
+    ) -> Result<(), JmapError> {
+        if subscription_id.trim().is_empty() || verification_code.trim().is_empty() {
+            return Err(JmapError::InvalidRequest("push verification"));
+        }
+        let verified = tokio::time::timeout(
+            self.timeout(),
+            self.client
+                .push_subscription_verify(subscription_id, verification_code),
+        )
+        .await
+        .map_err(|_| JmapError::Timeout)??;
+        if verified.is_some() {
+            Ok(())
+        } else {
+            Err(JmapError::NotFound)
+        }
+    }
+
+    async fn current_state(&self) -> Result<String, JmapError> {
+        Ok(self.client.session().state().to_owned())
+    }
+
+    async fn email_changes(
+        &self,
+        account_id: &str,
+        since_state: &str,
+        max_changes: usize,
+    ) -> Result<EmailChanges, JmapError> {
+        if account_id != self.account_id || since_state.trim().is_empty() || max_changes == 0 {
+            return Err(JmapError::InvalidRequest("email changes"));
+        }
+        let mut request = self.client.build();
+        request.changes_email(since_state).max_changes(max_changes);
+        let response = tokio::time::timeout(self.timeout(), request.send_changes_email())
+            .await
+            .map_err(|_| JmapError::Timeout)??;
+        Ok(EmailChanges {
+            new_state: response.new_state().to_owned(),
+            created: response.created().to_owned(),
+            updated: response.updated().to_owned(),
+            destroyed: response.destroyed().to_owned(),
+            has_more: response.has_more_changes(),
+        })
+    }
+
     async fn list_folders(&self, account_id: &str) -> Result<Vec<Folder>, JmapError> {
         if account_id != self.account_id {
             return Err(JmapError::InvalidRequest("account_id"));
@@ -114,17 +219,32 @@ impl JmapBackend for JmapClientBackend {
         if account_id != self.account_id || limit == 0 {
             return Err(JmapError::InvalidRequest("email query"));
         }
-        let filter: Option<email::query::Filter> = folder_id.map(email::query::Filter::in_mailbox);
+        // Query in bounded pages. A large limit (used for the cold-start
-        let ids = tokio::time::timeout(
+        // baseline) therefore cannot silently stop at the server default page.
-            self.timeout(),
+        let page_size = limit.clamp(1, 100);
-            self.client.email_query(
+        let mut position = 0_i32;
-                filter,
+        let mut ids = Vec::new();
-                None::<Vec<jmap_client::core::query::Comparator<email::query::Comparator>>>,
+        while ids.len() < limit {
-            ),
+            let mut request = self.client.build();
-        )
+            let query = request.query_email();
+            if let Some(folder_id) = folder_id {
+                query.filter(email::query::Filter::in_mailbox(folder_id));
+            }
+            query.position(position).limit(page_size);
+            let mut response = tokio::time::timeout(self.timeout(), request.send_query_email())
                 .await
-        .map_err(|_| JmapError::Timeout)??
+                .map_err(|_| JmapError::Timeout)??;
-        .take_ids();
+            let page = response.take_ids();
+            if page.is_empty() {
+                break;
+            }
+            position = position.saturating_add(page.len() as i32);
+            ids.extend(page);
+            if ids.len() >= limit || position <= 0 {
+                break;
+            }
+        }
+        ids.truncate(limit);
         let mut out = Vec::new();
         for id in ids.into_iter().take(limit) {
             if let Some(email) = tokio::time::timeout(
@@ -140,6 +260,17 @@ impl JmapBackend for JmapClientBackend {
         Ok(out)
     }
 
+    async fn list_emails_page(
+        &self,
+        account_id: &str,
+        folder_id: Option<&str>,
+        position: usize,
+        limit: usize,
+    ) -> Result<Vec<EmailMetadata>, JmapError> {
+        self.fetch_email_page(account_id, folder_id, position, limit)
+            .await
+    }
+
     async fn read_email(
         &self,
         account_id: &str,
@@ -182,6 +313,41 @@ impl JmapBackend for JmapClientBackend {
 }
 
 impl JmapClientBackend {
+    async fn fetch_email_page(
+        &self,
+        account_id: &str,
+        folder_id: Option<&str>,
+        position: usize,
+        limit: usize,
+    ) -> Result<Vec<EmailMetadata>, JmapError> {
+        if account_id != self.account_id || limit == 0 {
+            return Err(JmapError::InvalidRequest("email query"));
+        }
+        let mut request = self.client.build();
+        let query = request.query_email();
+        if let Some(folder_id) = folder_id {
+            query.filter(email::query::Filter::in_mailbox(folder_id));
+        }
+        query.position(position as i32).limit(limit.min(100));
+        let mut response = tokio::time::timeout(self.timeout(), request.send_query_email())
+            .await
+            .map_err(|_| JmapError::Timeout)??;
+        let ids = response.take_ids();
+        let mut out = Vec::with_capacity(ids.len());
+        for id in ids {
+            if let Some(email) = tokio::time::timeout(
+                self.timeout(),
+                self.client.email_get(&id, None::<Vec<email::Property>>),
+            )
+            .await
+            .map_err(|_| JmapError::Timeout)??
+            {
+                out.push(metadata(&email));
+            }
+        }
+        Ok(out)
+    }
+
     fn timeout(&self) -> std::time::Duration {
         std::time::Duration::from_millis(
             self.runtime

src/state.rs

@@ -70,14 +70,39 @@ pub trait ReliableState: Send + Sync {
     async fn release_dedup(&self, key: &str) -> Result<(), StateError>;
     /// XADD enqueue; consumers perform XREADGROUP in a later worker phase.
     async fn enqueue(&self, stream: &str, payload: &str) -> Result<String, StateError>;
+    /// Atomically claim `dedup_key` and XADD `payload` to `stream`.
+    ///
+    /// The dedup key must only be written when the append succeeds: claiming
+    /// first and rolling back best-effort leaves a claimed-but-not-enqueued key
+    /// behind on a Redis hiccup, which silently drops the event for the whole
+    /// dedup TTL (REQ-RECONCILE-IDEMPOTENCY). Returns false when already claimed.
+    async fn claim_dedup_and_enqueue(
+        &self,
+        dedup_key: &str,
+        ttl_seconds: u64,
+        stream: &str,
+        payload: &str,
+    ) -> Result<bool, StateError>;
     /// XACK after successful processing.
     async fn ack(&self, stream: &str, group: &str, message_id: &str) -> Result<(), StateError>;
     /// XADD to a dead-letter stream after bounded retry policy decides to stop retrying.
     async fn dead_letter(&self, stream: &str, payload: &str) -> Result<String, StateError>;
     /// SET NX EX lock for reconcile single-flight.
-    async fn acquire_lock(&self, key: &str, ttl_seconds: u64) -> Result<bool, StateError>;
+    async fn acquire_lock(
+        &self,
+        key: &str,
+        owner_token: &str,
+        ttl_seconds: u64,
+    ) -> Result<bool, StateError>;
     /// Release a lock after the bounded reconcile attempt has completed.
-    async fn release_lock(&self, key: &str) -> Result<(), StateError>;
+    async fn release_lock(&self, key: &str, owner_token: &str) -> Result<(), StateError>;
+    /// Extend a lock only while the caller still owns it.
+    async fn renew_lock(
+        &self,
+        key: &str,
+        owner_token: &str,
+        ttl_seconds: u64,
+    ) -> Result<bool, StateError>;
     async fn read_batch(
         &self,
         stream: &str,
@@ -117,6 +142,58 @@ pub trait ReliableState: Send + Sync {
     /// Global durable enable gate; missing/read failures are interpreted by callers as disabled.
     async fn is_enabled(&self) -> Result<bool, StateError>;
     async fn set_enabled(&self, enabled: bool) -> Result<(), StateError>;
+    /// Durable JMAP Email/changes cursor. It is advanced only after all page
+    /// events have been enqueued successfully (REQ-RECONCILE-IDEMPOTENCY).
+    async fn get_reconcile_state(&self) -> Result<Option<String>, StateError>;
+    async fn set_reconcile_state(&self, state: &str) -> Result<(), StateError>;
+    /// Short-lived PushSubscription verification state; values are stored as
+    /// digests and never retained as plaintext credentials.
+    async fn remember_push_subscription(
+        &self,
+        subscription_id: &str,
+        verification_code: &str,
+        ttl_seconds: u64,
+    ) -> Result<(), StateError>;
+    async fn push_subscription_verified(
+        &self,
+        subscription_id: &str,
+        verification_code: &str,
+    ) -> Result<bool, StateError>;
+    async fn remember_push_subscription_id(&self, subscription_id: &str) -> Result<(), StateError>;
+    async fn set_push_subscription_status(
+        &self,
+        subscription_id: &str,
+        status: &str,
+        ttl_seconds: u64,
+    ) -> Result<(), StateError>;
+    // NOTE: `push:subscription:{id}:status` is currently write-only
+    // ("pending"/"verified"/"disabled"). Authorization uses the verification
+    // digest instead; the status key stays as a lightweight ops/observability
+    // trail (inspectable via redis-cli), not a gate (建议-5a).
+    /// Drop the verification-code digest for `subscription_id`.
+    ///
+    /// Callers must invoke this when disabling a subscription: otherwise a push
+    /// arriving within the residual TTL of the last verify call can still pass
+    /// `push_subscription_verified` and be enqueued (SAF-AUTH-JMAP-PUSH).
+    async fn forget_push_subscription(&self, subscription_id: &str) -> Result<(), StateError>;
+    async fn record_push_orphan(
+        &self,
+        subscription_id: &str,
+        request_id: &str,
+    ) -> Result<(), StateError>;
+    async fn get_push_subscription_for_callback(
+        &self,
+        callback_url: &str,
+    ) -> Result<Option<String>, StateError>;
+    async fn remember_push_subscription_for_callback(
+        &self,
+        callback_url: &str,
+        subscription_id: &str,
+    ) -> Result<(), StateError>;
+    async fn remove_push_subscription_for_callback(
+        &self,
+        callback_url: &str,
+    ) -> Result<(), StateError>;
 }
 
 pub struct RedisState {
@@ -165,9 +242,6 @@ impl ReliableState for RedisState {
         let mut connection = self.connection.clone();
         Ok(redis::cmd("XADD")
             .arg(stream)
-            .arg("MAXLEN")
-            .arg("~")
-            .arg(10_000)
             .arg("*")
             .arg("payload")
             .arg(payload)
@@ -175,6 +249,34 @@ impl ReliableState for RedisState {
             .await?)
     }
 
+    async fn claim_dedup_and_enqueue(
+        &self,
+        dedup_key: &str,
+        ttl_seconds: u64,
+        stream: &str,
+        payload: &str,
+    ) -> Result<bool, StateError> {
+        let mut connection = self.connection.clone();
+        // KEYS[1] = dedup key, KEYS[2] = stream; ARGV[1] = ttl, ARGV[2] = payload.
+        // `SET NX` returning false means another worker owns the event; skip the
+        // append so the XADD is only attempted once per dedup window.
+        let claimed: i64 = redis::Script::new(
+            "if redis.call('set', KEYS[1], '1', 'NX', 'EX', ARGV[1]) then \
+                 redis.call('xadd', KEYS[2], '*', 'payload', ARGV[2]) \
+                 return 1 \
+             else \
+                 return 0 \
+             end",
+        )
+        .key(dedup_key)
+        .key(stream)
+        .arg(ttl_seconds.max(1))
+        .arg(payload)
+        .invoke_async(&mut connection)
+        .await?;
+        Ok(claimed != 0)
+    }
+
     async fn ack(&self, stream: &str, group: &str, message_id: &str) -> Result<(), StateError> {
         let mut connection = self.connection.clone();
         let _: i64 = redis::cmd("XACK")
@@ -190,12 +292,52 @@ impl ReliableState for RedisState {
         self.enqueue(stream, payload).await
     }
 
-    async fn acquire_lock(&self, key: &str, ttl_seconds: u64) -> Result<bool, StateError> {
+    async fn acquire_lock(
-        set_nx_ex(self.connection.clone(), key, ttl_seconds).await
+        &self,
+        key: &str,
+        owner_token: &str,
+        ttl_seconds: u64,
+    ) -> Result<bool, StateError> {
+        let mut connection = self.connection.clone();
+        let result: Option<String> = redis::cmd("SET")
+            .arg(key)
+            .arg(owner_token)
+            .arg("NX")
+            .arg("EX")
+            .arg(ttl_seconds.max(1))
+            .query_async(&mut connection)
+            .await?;
+        Ok(result.is_some())
     }
 
-    async fn release_lock(&self, key: &str) -> Result<(), StateError> {
+    async fn release_lock(&self, key: &str, owner_token: &str) -> Result<(), StateError> {
-        delete_key(self.connection.clone(), key).await
+        let mut connection = self.connection.clone();
+        redis::Script::new(
+            "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
+        )
+        .key(key)
+        .arg(owner_token)
+        .invoke_async::<i64>(&mut connection)
+        .await?;
+        Ok(())
+    }
+
+    async fn renew_lock(
+        &self,
+        key: &str,
+        owner_token: &str,
+        ttl_seconds: u64,
+    ) -> Result<bool, StateError> {
+        let mut connection = self.connection.clone();
+        let renewed: i64 = redis::Script::new(
+            "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('expire', KEYS[1], ARGV[2]) else return 0 end",
+        )
+        .key(key)
+        .arg(owner_token)
+        .arg(ttl_seconds.max(1))
+        .invoke_async(&mut connection)
+        .await?;
+        Ok(renewed != 0)
     }
 
     async fn read_batch(
@@ -217,25 +359,38 @@ impl ReliableState for RedisState {
             .query_async(&mut connection)
             .await;
         // Recover messages left in the consumer group's PEL after a crash or
-        // restart before reading new entries. A bounded idle window prevents
+        // restart before reading new entries. The idle window has to clear the
-        // two short-lived workers from stealing active work.
+        // worst-case *batch* duration (a worker processes up to `count` events
+        // sequentially, each bounded by the ~300s single-event ceiling), else a
+        // second instance reclaims events that are still being processed and
+        // delivers them twice (建议-6). Single-instance deployments are
+        // unaffected; only duplicate-on-multi-instance is at stake, never loss.
+        const SINGLE_EVENT_CEILING_MS: u64 = 300_000;
+        let idle_threshold_ms = count.max(1) as u64 * SINGLE_EVENT_CEILING_MS;
         let reclaimed: redis::streams::StreamAutoClaimReply = redis::cmd("XAUTOCLAIM")
             .arg(stream)
             .arg(group)
             .arg(consumer)
-            // A worker handles at most ten events per request; keep this
+            .arg(idle_threshold_ms)
-            // above the worst bounded JMAP/Telegram batch duration.
-            .arg(300_000)
             .arg("0-0")
             .arg("COUNT")
             .arg(count.max(1))
             .query_async(&mut connection)
             .await?;
-        let reclaimed_messages = reclaimed
+        let mut reclaimed_messages = Vec::new();
-            .claimed
+        let mut malformed_ids = Vec::new();
-            .into_iter()
+        for item in reclaimed.claimed {
-            .filter_map(stream_id_to_message)
+            let id = item.id.clone();
-            .collect::<Vec<_>>();
+            match stream_id_to_message(item) {
+                Some(message) => reclaimed_messages.push(message),
+                None => malformed_ids.push(id),
+            }
+        }
+        // A record without a `payload` field can never be processed. ACK it so
+        // it stops being re-claimed every cycle and squatting in the PEL
+        // (建议-7); a payload that is present but not valid JSON still routes
+        // through the normal Err -> DLQ path.
+        ack_malformed(&mut connection, stream, group, &malformed_ids).await;
         if !reclaimed_messages.is_empty() {
             return Ok(reclaimed_messages);
         }
@@ -250,11 +405,21 @@ impl ReliableState for RedisState {
             .arg(">")
             .query_async(&mut connection)
             .await?;
-        Ok(reply
+        let mut messages = Vec::new();
-            .keys
+        let mut malformed_ids = Vec::new();
-            .into_iter()
+        for key in reply.keys {
-            .flat_map(|key| key.ids.into_iter().filter_map(stream_id_to_message))
+            for item in key.ids {
-            .collect())
+                let id = item.id.clone();
+                match stream_id_to_message(item) {
+                    Some(message) => messages.push(message),
+                    None => malformed_ids.push(id),
+                }
+            }
+        }
+        // Same handling as the XAUTOCLAIM path: drop unprocessable records from
+        // the PEL instead of letting them spin forever (建议-7).
+        ack_malformed(&mut connection, stream, group, &malformed_ids).await;
+        Ok(messages)
     }
 
     async fn retry_or_dlq(
@@ -273,7 +438,7 @@ impl ReliableState for RedisState {
             r#"local n = redis.call('INCR', KEYS[1])
                redis.call('EXPIRE', KEYS[1], 86400)
                if n < tonumber(ARGV[1]) then return 0 end
-               redis.call('XADD', ARGV[2], 'MAXLEN', '~', 10000, '*', 'payload', ARGV[3])
+               redis.call('XADD', ARGV[2], '*', 'payload', ARGV[3])
                redis.call('XACK', ARGV[4], ARGV[5], ARGV[6])
                return 1"#,
         );
@@ -452,6 +617,152 @@ impl ReliableState for RedisState {
             .await?;
         Ok(())
     }
+
+    async fn get_reconcile_state(&self) -> Result<Option<String>, StateError> {
+        let mut connection = self.connection.clone();
+        Ok(redis::cmd("GET")
+            .arg("state:jmap:since")
+            .query_async(&mut connection)
+            .await?)
+    }
+
+    async fn set_reconcile_state(&self, state: &str) -> Result<(), StateError> {
+        let mut connection = self.connection.clone();
+        redis::cmd("SET")
+            .arg("state:jmap:since")
+            .arg(state)
+            .query_async::<()>(&mut connection)
+            .await?;
+        Ok(())
+    }
+
+    async fn remember_push_subscription(
+        &self,
+        subscription_id: &str,
+        verification_code: &str,
+        ttl_seconds: u64,
+    ) -> Result<(), StateError> {
+        let mut connection = self.connection.clone();
+        redis::cmd("SET")
+            .arg(format!("push:subscription:{subscription_id}"))
+            .arg(crate::config::session_digest(verification_code))
+            .arg("EX")
+            .arg(ttl_seconds.max(1))
+            .query_async::<()>(&mut connection)
+            .await?;
+        Ok(())
+    }
+
+    async fn push_subscription_verified(
+        &self,
+        subscription_id: &str,
+        verification_code: &str,
+    ) -> Result<bool, StateError> {
+        let mut connection = self.connection.clone();
+        let current: Option<String> = redis::cmd("GET")
+            .arg(format!("push:subscription:{subscription_id}"))
+            .query_async(&mut connection)
+            .await?;
+        Ok(current.as_deref() == Some(crate::config::session_digest(verification_code).as_str()))
+    }
+
+    async fn remember_push_subscription_id(&self, subscription_id: &str) -> Result<(), StateError> {
+        let mut connection = self.connection.clone();
+        redis::cmd("SET")
+            .arg("push:subscription:id")
+            .arg(subscription_id)
+            .query_async::<()>(&mut connection)
+            .await?;
+        Ok(())
+    }
+
+    async fn set_push_subscription_status(
+        &self,
+        subscription_id: &str,
+        status: &str,
+        ttl_seconds: u64,
+    ) -> Result<(), StateError> {
+        let mut connection = self.connection.clone();
+        redis::cmd("SET")
+            .arg(format!("push:subscription:{subscription_id}:status"))
+            .arg(status)
+            .arg("EX")
+            .arg(ttl_seconds.max(1))
+            .query_async::<()>(&mut connection)
+            .await?;
+        Ok(())
+    }
+
+    async fn forget_push_subscription(&self, subscription_id: &str) -> Result<(), StateError> {
+        let mut connection = self.connection.clone();
+        redis::cmd("DEL")
+            .arg(format!("push:subscription:{subscription_id}"))
+            .query_async::<()>(&mut connection)
+            .await?;
+        Ok(())
+    }
+
+    async fn record_push_orphan(
+        &self,
+        subscription_id: &str,
+        request_id: &str,
+    ) -> Result<(), StateError> {
+        let mut connection = self.connection.clone();
+        redis::cmd("SET")
+            .arg(format!("push:orphan:{subscription_id}"))
+            .arg(request_id)
+            .arg("EX")
+            .arg(604_800_u64)
+            .query_async::<()>(&mut connection)
+            .await?;
+        Ok(())
+    }
+
+    async fn get_push_subscription_for_callback(
+        &self,
+        callback_url: &str,
+    ) -> Result<Option<String>, StateError> {
+        let mut connection = self.connection.clone();
+        Ok(redis::cmd("GET")
+            .arg(push_registration_key(callback_url))
+            .query_async(&mut connection)
+            .await?)
+    }
+
+    async fn remember_push_subscription_for_callback(
+        &self,
+        callback_url: &str,
+        subscription_id: &str,
+    ) -> Result<(), StateError> {
+        let mut connection = self.connection.clone();
+        redis::cmd("SET")
+            .arg(push_registration_key(callback_url))
+            .arg(subscription_id)
+            .arg("EX")
+            .arg(604_800_u64)
+            .query_async::<()>(&mut connection)
+            .await?;
+        Ok(())
+    }
+
+    async fn remove_push_subscription_for_callback(
+        &self,
+        callback_url: &str,
+    ) -> Result<(), StateError> {
+        let mut connection = self.connection.clone();
+        redis::cmd("DEL")
+            .arg(push_registration_key(callback_url))
+            .query_async::<()>(&mut connection)
+            .await?;
+        Ok(())
+    }
+}
+
+fn push_registration_key(callback_url: &str) -> String {
+    format!(
+        "push:registration:{}",
+        crate::config::session_digest(callback_url)
+    )
 }
 
 fn stream_id_to_message(item: redis::streams::StreamId) -> Option<StreamMessage> {
@@ -465,6 +776,36 @@ fn stream_id_to_message(item: redis::streams::StreamId) -> Option<StreamMessage>
     })
 }
 
+/// XACK stream records that arrived without a parseable `payload` field. These
+/// can never be processed, so acknowledging them keeps the consumer group's PEL
+/// from filling up with unclaimable junk; the failure is logged for operators.
+async fn ack_malformed(
+    connection: &mut redis::aio::MultiplexedConnection,
+    stream: &str,
+    group: &str,
+    ids: &[String],
+) {
+    if ids.is_empty() {
+        return;
+    }
+    let ack = redis::cmd("XACK")
+        .arg(stream)
+        .arg(group)
+        .arg(ids)
+        .query_async::<i64>(connection)
+        .await;
+    if ack
+        .as_ref()
+        .map(|acked| *acked != ids.len() as i64)
+        .unwrap_or(true)
+    {
+        eprintln!(
+            "WARN malformed stream records ack incomplete (stream={stream}, group={group}, expected={}, result={ack:?})",
+            ids.len()
+        );
+    }
+}
+
 async fn set_nx_ex(
     mut connection: redis::aio::MultiplexedConnection,
     key: &str,
@@ -475,7 +816,9 @@ async fn set_nx_ex(
         .arg("1")
         .arg("NX")
         .arg("EX")
-        .arg(ttl_seconds)
+        // Guard against a ttl of 0, which Redis rejects with "invalid expire
+        // time" (same normalization as acquire_lock).
+        .arg(ttl_seconds.max(1))
         .query_async(&mut connection)
         .await?;
     Ok(result.is_some())
@@ -498,7 +841,8 @@ impl MemoryState {
 #[derive(Default)]
 struct MemoryInner {
     dedup: HashSet<String>,
-    locks: HashSet<String>,
+    locks: HashMap<String, String>,
+    push_registrations: HashMap<String, String>,
     streams: HashMap<String, Vec<(String, String)>>,
     retries: HashMap<(String, String), u32>,
     consent: HashMap<i64, i64>,
@@ -508,6 +852,11 @@ struct MemoryInner {
     admin_session: Option<(String, i64)>,
     business_revision: u64,
     enabled: bool,
+    reconcile_state: Option<String>,
+    push_subscriptions: HashMap<String, (String, i64)>,
+    push_subscription_id: Option<String>,
+    push_status: HashMap<String, (String, i64)>,
+    push_orphans: HashMap<String, String>,
 }
 
 #[async_trait]
@@ -534,14 +883,30 @@ impl ReliableState for MemoryState {
             .entry(stream.to_owned())
             .or_default()
             .push((id.clone(), payload.to_owned()));
-        if let Some(messages) = inner.streams.get_mut(stream) {
-            let excess = messages.len().saturating_sub(10_000);
-            if excess > 0 {
-                messages.drain(..excess);
-            }
-        }
         Ok(id)
     }
+    async fn claim_dedup_and_enqueue(
+        &self,
+        dedup_key: &str,
+        _ttl_seconds: u64,
+        stream: &str,
+        payload: &str,
+    ) -> Result<bool, StateError> {
+        // Hold the lock across claim + append so MemoryState honors the same
+        // atomicity contract as the Lua-backed RedisState implementation.
+        let mut inner = self.inner.lock().map_err(|_| StateError::Poisoned)?;
+        if !inner.dedup.insert(dedup_key.to_owned()) {
+            return Ok(false);
+        }
+        inner.next_id += 1;
+        let id = format!("{}-0", inner.next_id);
+        inner
+            .streams
+            .entry(stream.to_owned())
+            .or_default()
+            .push((id, payload.to_owned()));
+        Ok(true)
+    }
     async fn ack(&self, stream: &str, _group: &str, message_id: &str) -> Result<(), StateError> {
         let mut inner = self.inner.lock().map_err(|_| StateError::Poisoned)?;
         if let Some(messages) = inner.streams.get_mut(stream) {
@@ -552,15 +917,43 @@ impl ReliableState for MemoryState {
     async fn dead_letter(&self, stream: &str, payload: &str) -> Result<String, StateError> {
         self.enqueue(stream, payload).await
     }
-    async fn acquire_lock(&self, key: &str, _ttl_seconds: u64) -> Result<bool, StateError> {
+    async fn acquire_lock(
+        &self,
+        key: &str,
+        owner_token: &str,
+        _ttl_seconds: u64,
+    ) -> Result<bool, StateError> {
         let mut inner = self.inner.lock().map_err(|_| StateError::Poisoned)?;
-        Ok(inner.locks.insert(key.to_owned()))
+        if inner.locks.contains_key(key) {
+            return Ok(false);
         }
-    async fn release_lock(&self, key: &str) -> Result<(), StateError> {
+        inner.locks.insert(key.to_owned(), owner_token.to_owned());
+        Ok(true)
+    }
+    async fn release_lock(&self, key: &str, owner_token: &str) -> Result<(), StateError> {
         let mut inner = self.inner.lock().map_err(|_| StateError::Poisoned)?;
+        if inner
+            .locks
+            .get(key)
+            .is_some_and(|owner| owner == owner_token)
+        {
             inner.locks.remove(key);
+        }
         Ok(())
     }
+
+    async fn renew_lock(
+        &self,
+        key: &str,
+        owner_token: &str,
+        _ttl_seconds: u64,
+    ) -> Result<bool, StateError> {
+        let inner = self.inner.lock().map_err(|_| StateError::Poisoned)?;
+        Ok(inner
+            .locks
+            .get(key)
+            .is_some_and(|owner| owner == owner_token))
+    }
     async fn read_batch(
         &self,
         stream: &str,
@@ -714,6 +1107,147 @@ impl ReliableState for MemoryState {
         self.inner.lock().map_err(|_| StateError::Poisoned)?.enabled = enabled;
         Ok(())
     }
+
+    async fn get_reconcile_state(&self) -> Result<Option<String>, StateError> {
+        Ok(self
+            .inner
+            .lock()
+            .map_err(|_| StateError::Poisoned)?
+            .reconcile_state
+            .clone())
+    }
+
+    async fn set_reconcile_state(&self, state: &str) -> Result<(), StateError> {
+        self.inner
+            .lock()
+            .map_err(|_| StateError::Poisoned)?
+            .reconcile_state = Some(state.to_owned());
+        Ok(())
+    }
+
+    async fn remember_push_subscription(
+        &self,
+        subscription_id: &str,
+        verification_code: &str,
+        ttl_seconds: u64,
+    ) -> Result<(), StateError> {
+        self.inner
+            .lock()
+            .map_err(|_| StateError::Poisoned)?
+            .push_subscriptions
+            .insert(
+                subscription_id.to_owned(),
+                (
+                    crate::config::session_digest(verification_code),
+                    unix_now().saturating_add(ttl_seconds as i64),
+                ),
+            );
+        Ok(())
+    }
+
+    async fn push_subscription_verified(
+        &self,
+        subscription_id: &str,
+        verification_code: &str,
+    ) -> Result<bool, StateError> {
+        let mut inner = self.inner.lock().map_err(|_| StateError::Poisoned)?;
+        let Some((digest, expiry)) = inner.push_subscriptions.get(subscription_id) else {
+            return Ok(false);
+        };
+        if *expiry <= unix_now() {
+            inner.push_subscriptions.remove(subscription_id);
+            return Ok(false);
+        }
+        Ok(digest == &crate::config::session_digest(verification_code))
+    }
+
+    async fn remember_push_subscription_id(&self, subscription_id: &str) -> Result<(), StateError> {
+        self.inner
+            .lock()
+            .map_err(|_| StateError::Poisoned)?
+            .push_subscription_id = Some(subscription_id.to_owned());
+        Ok(())
+    }
+
+    async fn get_push_subscription_for_callback(
+        &self,
+        callback_url: &str,
+    ) -> Result<Option<String>, StateError> {
+        Ok(self
+            .inner
+            .lock()
+            .map_err(|_| StateError::Poisoned)?
+            .push_registrations
+            .get(&push_registration_key(callback_url))
+            .cloned())
+    }
+
+    async fn remember_push_subscription_for_callback(
+        &self,
+        callback_url: &str,
+        subscription_id: &str,
+    ) -> Result<(), StateError> {
+        self.inner
+            .lock()
+            .map_err(|_| StateError::Poisoned)?
+            .push_registrations
+            .insert(
+                push_registration_key(callback_url),
+                subscription_id.to_owned(),
+            );
+        Ok(())
+    }
+
+    async fn remove_push_subscription_for_callback(
+        &self,
+        callback_url: &str,
+    ) -> Result<(), StateError> {
+        self.inner
+            .lock()
+            .map_err(|_| StateError::Poisoned)?
+            .push_registrations
+            .remove(&push_registration_key(callback_url));
+        Ok(())
+    }
+
+    async fn set_push_subscription_status(
+        &self,
+        subscription_id: &str,
+        status: &str,
+        ttl_seconds: u64,
+    ) -> Result<(), StateError> {
+        self.inner
+            .lock()
+            .map_err(|_| StateError::Poisoned)?
+            .push_status
+            .insert(
+                subscription_id.to_owned(),
+                (
+                    status.to_owned(),
+                    unix_now().saturating_add(ttl_seconds as i64),
+                ),
+            );
+        Ok(())
+    }
+
+    async fn forget_push_subscription(&self, subscription_id: &str) -> Result<(), StateError> {
+        let mut inner = self.inner.lock().map_err(|_| StateError::Poisoned)?;
+        inner.push_subscriptions.remove(subscription_id);
+        Ok(())
+    }
+
+    async fn record_push_orphan(
+        &self,
+        subscription_id: &str,
+        request_id: &str,
+    ) -> Result<(), StateError> {
+        self.inner
+            .lock()
+            .map_err(|_| StateError::Poisoned)?
+            .push_orphans
+            .insert(subscription_id.to_owned(), request_id.to_owned());
+        Ok(())
+    }
 }
 
 fn unix_now() -> i64 {
@@ -830,6 +1364,17 @@ mod tests {
     #[tokio::test]
     async fn memory_state_is_atomic_and_supports_stream_lifecycle() {
         let state = MemoryState::default();
+        assert!(!state.is_enabled().await.unwrap());
+        state.set_enabled(true).await.unwrap();
+        assert!(state.is_enabled().await.unwrap());
+        state.set_enabled(false).await.unwrap();
+        assert!(!state.is_enabled().await.unwrap());
+        assert!(state.get_reconcile_state().await.unwrap().is_none());
+        state.set_reconcile_state("jmap-state-1").await.unwrap();
+        assert_eq!(
+            state.get_reconcile_state().await.unwrap().as_deref(),
+            Some("jmap-state-1")
+        );
         assert!(state.claim_dedup("tg:42", 60).await.unwrap());
         assert!(!state.claim_dedup("tg:42", 60).await.unwrap());
         assert!(state.dedup_exists("tg:42").await.unwrap());
@@ -857,8 +1402,56 @@ mod tests {
             .dedup_exists("delivery:committed:events:1-0")
             .await
             .unwrap());
-        assert!(state.acquire_lock("reconcile", 30).await.unwrap());
+        assert!(state
-        assert!(!state.acquire_lock("reconcile", 30).await.unwrap());
+            .acquire_lock("reconcile", "owner-a", 30)
+            .await
+            .unwrap());
+        assert!(!state
+            .acquire_lock("reconcile", "owner-b", 30)
+            .await
+            .unwrap());
+        state.release_lock("reconcile", "owner-b").await.unwrap();
+        assert!(!state
+            .acquire_lock("reconcile", "owner-b", 30)
+            .await
+            .unwrap());
+        assert!(!state.renew_lock("reconcile", "owner-b", 90).await.unwrap());
+        assert!(state.renew_lock("reconcile", "owner-a", 90).await.unwrap());
+        state.release_lock("reconcile", "owner-a").await.unwrap();
+        assert!(state
+            .acquire_lock("reconcile", "owner-b", 30)
+            .await
+            .unwrap());
+        state
+            .remember_push_subscription_for_callback(
+                "https://example.test/push/jmap",
+                "subscription-1",
+            )
+            .await
+            .unwrap();
+        assert_eq!(
+            state
+                .get_push_subscription_for_callback("https://example.test/push/jmap")
+                .await
+                .unwrap()
+                .as_deref(),
+            Some("subscription-1")
+        );
+        state
+            .set_push_subscription_status("subscription-1", "disabled", 60)
+            .await
+            .unwrap();
+        // `pending/verified/disabled` 状态键是只写的运维轨迹（建议-5a）：
+        // 授权语义由验证码摘要承担，故这里不对其做读取断言。
+        state
+            .remove_push_subscription_for_callback("https://example.test/push/jmap")
+            .await
+            .unwrap();
+        assert!(state
+            .get_push_subscription_for_callback("https://example.test/push/jmap")
+            .await
+            .unwrap()
+            .is_none());
         assert!(state.ai_consent_until(7).await.unwrap().is_none());
         state.set_ai_consent(7, 3600).await.unwrap();
         assert!(state.ai_consent_until(7).await.unwrap().is_some());

src/worker.rs

@@ -15,6 +15,32 @@ use std::{
 #[async_trait]
 pub trait WorkerHandler: Send + Sync {
     async fn process(&self, stream: &str, payload: &str) -> Result<(), ()>;
+
+    async fn create_push_subscription(&self, _callback_url: &str) -> Result<String, ()> {
+        Err(())
+    }
+
+    async fn destroy_push_subscription(&self, _subscription_id: &str) -> Result<(), ()> {
+        Err(())
+    }
+
+    async fn verify_push_subscription(
+        &self,
+        _subscription_id: &str,
+        _verification_code: &str,
+    ) -> Result<(), ()> {
+        Err(())
+    }
+
+    /// Performs one bounded JMAP reconciliation pass. Implementations must not
+    /// persist the cursor; the HTTP coordinator commits it after enqueueing.
+    async fn reconcile(
+        &self,
+        _since_state: Option<&str>,
+        _max_changes: usize,
+    ) -> Result<String, ()> {
+        Err(())
+    }
 }
 
 /// Atomically replaceable worker handle. In-flight requests keep their old worker;
@@ -81,10 +107,36 @@ impl ReloadCoordinator {
 
 #[async_trait]
 impl WorkerHandler for WorkerHandle {
+    async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), ()> {
+        let worker = self.current.read().map_err(|_| ())?.clone();
+        worker.destroy_push_subscription(subscription_id).await
+    }
+
+    async fn create_push_subscription(&self, callback_url: &str) -> Result<String, ()> {
+        let worker = self.current.read().map_err(|_| ())?.clone();
+        worker.create_push_subscription(callback_url).await
+    }
+
+    async fn verify_push_subscription(
+        &self,
+        subscription_id: &str,
+        verification_code: &str,
+    ) -> Result<(), ()> {
+        let worker = self.current.read().map_err(|_| ())?.clone();
+        worker
+            .verify_push_subscription(subscription_id, verification_code)
+            .await
+    }
+
     async fn process(&self, stream: &str, payload: &str) -> Result<(), ()> {
         let worker = self.current.read().map_err(|_| ())?.clone();
         worker.process(stream, payload).await
     }
+
+    async fn reconcile(&self, since_state: Option<&str>, max_changes: usize) -> Result<String, ()> {
+        let worker = self.current.read().map_err(|_| ())?.clone();
+        worker.reconcile(since_state, max_changes).await
+    }
 }
 
 pub struct NoopWorker;
@@ -124,12 +176,41 @@ impl<B: JmapBackend> MetadataWorker<B> {
 
 #[async_trait]
 impl<B: JmapBackend> WorkerHandler for MetadataWorker<B> {
+    async fn destroy_push_subscription(&self, subscription_id: &str) -> Result<(), ()> {
+        self.jmap
+            .destroy_push_subscription(subscription_id)
+            .await
+            .map_err(|_| ())
+    }
+
+    async fn create_push_subscription(&self, callback_url: &str) -> Result<String, ()> {
+        self.jmap
+            .create_push_subscription(callback_url)
+            .await
+            .map_err(|_| ())
+    }
+
+    async fn verify_push_subscription(
+        &self,
+        subscription_id: &str,
+        verification_code: &str,
+    ) -> Result<(), ()> {
+        self.jmap
+            .verify_push_subscription(subscription_id, verification_code)
+            .await
+            .map_err(|_| ())
+    }
+
     async fn process(&self, stream: &str, payload: &str) -> Result<(), ()> {
         if stream == "stalwart:telegram" {
             return self.process_telegram(payload).await;
         }
         if stream != "stalwart:jmap" {
-            return Ok(());
+            // Unknown stream: fail closed. Returning Ok would let the coordinator
+            // 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.
+            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())
@@ -161,9 +242,140 @@ impl<B: JmapBackend> WorkerHandler for MetadataWorker<B> {
             .await
             .map_err(|_| ())
     }
+
+    async fn reconcile(&self, since_state: Option<&str>, max_changes: usize) -> Result<String, ()> {
+        let max_changes = max_changes.max(1);
+        const MAX_BASELINE_PAGES: usize = 100;
+        const MAX_BASELINE_EMAILS: usize = 10_000;
+        const RECONCILE_BUDGET: std::time::Duration = std::time::Duration::from_secs(20);
+        let started = tokio::time::Instant::now();
+        let (mut cursor, mut baseline_position) = if let Some(state) = since_state {
+            if let Some((baseline, position)) = decode_baseline_cursor(state) {
+                (baseline, position)
+            } else {
+                (state.to_owned(), usize::MAX)
+            }
+        } else {
+            // Capture a stable baseline before paging. Changes occurring while
+            // the listing is in flight are consumed below before committing.
+            (self.jmap.current_state().await.map_err(|_| ())?, 0)
+        };
+
+        if baseline_position != usize::MAX {
+            let mut pages = 0_usize;
+            let mut emails = 0_usize;
+            loop {
+                if pages >= MAX_BASELINE_PAGES
+                    || emails >= MAX_BASELINE_EMAILS
+                    || started.elapsed() >= RECONCILE_BUDGET
+                {
+                    return if pages == 0 {
+                        Err(())
+                    } else {
+                        Ok(encode_baseline_cursor(&cursor, baseline_position))
+                    };
+                }
+                let page = self
+                    .jmap
+                    .list_emails_page(None, baseline_position, 100)
+                    .await
+                    .map_err(|_| ())?;
+                if page.is_empty() {
+                    break;
+                }
+                pages += 1;
+                emails = emails.saturating_add(page.len());
+                baseline_position = baseline_position.saturating_add(page.len());
+                for email in page {
+                    self.enqueue_reconcile_event(&email.id).await?;
+                }
+            }
+        }
+
+        // RFC 8620 lets a server answer `/changes` with `hasMoreChanges=true`
+        // when more events remain than `maxChanges` allowed. The spec-blessed
+        // continuation is `sinceState = newState`; but a server that resolves
+        // `newState` to "after *all* pending changes" would then hand us a
+        // cursor that silently skips the unreturned batch. We cannot pass
+        // `upToId` here (jmap-client 0.4.2 does not expose it), so while more
+        // changes remain we keep the *same* sinceState and widen the window
+        // instead. The 24h dedup key makes re-reading the superset idempotent,
+        // so this is at-most-duplicate, never-loss. Only when the window stops
+        // growing (server-side cap) do we fall back to advancing to `new_state`
+        // to guarantee forward progress.
+        const CHANGE_WINDOW_CAP: usize = 4_096;
+        let mut window = max_changes;
+        let mut pages = 0_usize;
+        loop {
+            if pages >= 100 || started.elapsed() >= RECONCILE_BUDGET {
+                // Every event from the last completed page is already enqueued.
+                // Returning that cursor lets the next invocation continue from
+                // it instead of retrying the same bounded window forever.
+                return if pages == 0 { Err(()) } else { Ok(cursor) };
+            }
+            pages += 1;
+            let changes = self
+                .jmap
+                .email_changes(&cursor, window)
+                .await
+                .map_err(|_| ())?;
+            for email_id in changes.created.iter().chain(changes.updated.iter()) {
+                self.enqueue_reconcile_event(email_id).await?;
+            }
+            if !changes.has_more {
+                return Ok(changes.new_state);
+            }
+            if window < CHANGE_WINDOW_CAP {
+                window = window.saturating_mul(2).min(CHANGE_WINDOW_CAP);
+                continue;
+            }
+            // Window is already at the cap and the server still reports more:
+            // advancing is the only way to make progress.
+            cursor = changes.new_state;
+        }
+    }
+}
+
+fn encode_baseline_cursor(state: &str, position: usize) -> String {
+    let encoded = state
+        .as_bytes()
+        .iter()
+        .map(|byte| format!("{byte:02x}"))
+        .collect::<String>();
+    format!("baseline:{encoded}:{position}")
+}
+
+fn decode_baseline_cursor(value: &str) -> Option<(String, usize)> {
+    let rest = value.strip_prefix("baseline:")?;
+    let (encoded, position) = rest.rsplit_once(':')?;
+    let bytes = (0..encoded.len())
+        .step_by(2)
+        .map(|index| u8::from_str_radix(encoded.get(index..index + 2)?, 16).ok())
+        .collect::<Option<Vec<_>>>()?;
+    Some((String::from_utf8(bytes).ok()?, position.parse().ok()?))
 }
 
 impl<B: JmapBackend> MetadataWorker<B> {
+    async fn enqueue_reconcile_event(&self, email_id: &str) -> Result<(), ()> {
+        let key = format!("dedup:jmap:{}:{}", self.jmap.account_id(), email_id);
+        let payload = serde_json::json!({
+            "accountId": self.jmap.account_id(),
+            "emailId": email_id,
+        })
+        .to_string();
+        // Claim the dedup key and XADD in a single atomic step: a Redis hiccup
+        // 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). Returning false means another worker owns it.
+        // `false` means another worker already owns the dedup entry for this
+        // email in the current window; either way the event is handled here.
+        self.state
+            .claim_dedup_and_enqueue(&key, 86_400, "stalwart:jmap", &payload)
+            .await
+            .map_err(|_| ())?;
+        Ok(())
+    }
+
     async fn process_telegram(&self, payload: &str) -> Result<(), ()> {
         let update: TelegramUpdate = serde_json::from_str(payload).map_err(|_| ())?;
         let Some(message) = update.message else {
@@ -354,7 +566,7 @@ fn parse_intent(input: &str) -> Intent {
 }
 
 fn help_message() -> &'static str {
-    "使用说明：\n邮件到达后会先发送发件人、主题和时间等元数据通知。\n\n授权示例：/ai on（1小时）、临时一次、1小时、今天、7天、直到我撤销。\n/ai off：立即撤销 AI 正文授权。\n/summary <email_id>：请求指定邮件摘要；未授权或授权到期时只返回元数据并提示重新授权。\n/help（或 help）：显示本说明。\n\n隐私：AI 默认关闭，只有你明确开启后才会发送正文；授权有期限且不会自动续期；正文和 AI 结果不会持久化，也不会写入日志。\n如果未配置 AI 或 AI 调用失败，将回退为本地截取摘要。"
+    "使用说明：\n邮件到达后会先发送发件人、主题和时间等元数据通知。\n\n授权示例（请按原文中文输入，不支持英文别名）：\n1小时：/ai on、临时一次、临时、一次\n今天：今天\n7天：7天\n直到我撤销（最长365天）：直到我撤销、长期\n撤销授权：/ai off、关闭 ai、撤销授权、停止摘要\n/summary <email_id>：请求指定邮件摘要；未授权或授权到期时只返回元数据并提示重新授权。\n/help（或 help）：显示本说明。\n\n隐私：AI 默认关闭，只有你明确开启后才会发送正文；授权有期限且不会自动续期；正文和 AI 结果不会持久化，也不会写入日志。\n如果未配置 AI 或 AI 调用失败，将回退为本地截取摘要。"
 }
 
 #[derive(serde::Deserialize)]
@@ -444,7 +656,6 @@ mod reload_tests {
             jmap_session_url: "https://mail.example.test".into(),
             jmap_username: "user".into(),
             jmap_password: SecretString::new("pass".into()),
-            jmap_push_verification: SecretString::new("verify".into()),
             account_id: None,
             llm_enabled: false,
             llm_allow_net: false,
@@ -482,6 +693,16 @@ mod reload_tests {
         assert!(handle.process("x", "y").await.is_ok());
     }
 
+    #[test]
+    fn baseline_cursor_roundtrips_state_and_position() {
+        let encoded = encode_baseline_cursor("state:with:punctuation", 1234);
+        assert_eq!(
+            decode_baseline_cursor(&encoded),
+            Some(("state:with:punctuation".to_owned(), 1234))
+        );
+        assert!(decode_baseline_cursor("baseline:odd:hex").is_none());
+    }
+
     struct OkWorker;
     #[async_trait]
     impl WorkerHandler for OkWorker {