fix(push): accept RFC 8620 StateChange pushes and reconcile on them

Stalwart's event pushes are StateChange objects (RFC 8620 §7.1.1), which
carry only @type plus a changed map of account->type->state-string. They
carry no pushSubscriptionId, no accountId and no emailId, so the old
handler deserialized every field to None and returned 401 from the
subscription_id check. The event-push delivery path was dead from the
first deploy; only the /reconcile cron could deliver new mail.

Confirmed against Stalwart's own log: the diagnostic entry log showed
subscription_id=None account_id=None email_id=None has_verification_code
=false for request_ids fe126c05cee9d7493f4619b2727be3a5 and
4869dee4e7360803d8f9136ec1020cd5, matching Stalwart's `Push subscription
error ... code = 401` lines at 01:53:24Z and 01:54:24Z.

jmap_push now discriminates by content, not by @type, because Stalwart's
raw body had never been observed and @type may be absent:
- verificationCode present -> PushVerification (RFC 8620 §7.2), the
  verification branch, unchanged, still returns the code as a JSON body.
- changed present -> StateChange: ack 204, then spawn
  push_reconcile_delta, which pulls the delta via the same /changes path
  /reconcile uses, under the same lock:reconcile so a push-triggered pull
  and the cron pull never touch reconcile:state concurrently. If the lock
  is held it defers: the running pull observes the same change.
- neither -> 400 instead of a silent 401, so a malformed push is visible.

StateChange cannot supply an emailId, so the dead dedup_exists / inline
enqueue branch predicated on it is deleted, along with the four temporary
diagnostic logs from ffa53b5. JmapPush keeps only its two live
PushVerification fields. The stalwart:jmap queue contract
{"accountId","emailId"}, worker.reconcile and the queue keys are untouched.

Tests: jmap_push_requires_verification_code replaced by three focused
tests -- a body with neither discriminator 400s (was: silent 401), a
StateChange body 204s (was: 401 for the route's whole lifetime), and a
verification body returns the code as JSON. GATE-P0 108 passed / 0 failed
/ 4 ignored; GATE-DOCS 6/6 clean. The one remaining `jmap_push request
received` entry log is kept permanently for correlation against
Stalwart's Push subscription error lines.

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

src/notify.rs

@@ -1290,26 +1290,45 @@ async fn jmap_push(State(app): State<AppState>, body: Bytes) -> Response {
     if !business_enabled(&app).await {
         return error_response(StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", true);
     }
-    let Ok(push) = serde_json::from_slice::<JmapPush>(&body) else {
+    // Discern the body kind by *content*, not by `@type`: Stalwart's raw body
+    // has never been observed, so `@type` may be absent and keying on it alone
+    // would misroute a real event push to the 400 arm below.
+    // - `verificationCode` present → PushVerification (RFC 8620 §7.2), which
+    //   DOES carry `pushSubscriptionId`.
+    // - `changed` present         → StateChange (RFC 8620 §7.1.1), which
+    //   carries NEITHER `pushSubscriptionId` NOR any emailId — only
+    //   account→type→state-string. The emailIds must be fetched via `/changes`,
+    //   which is what the reconcile path does.
+    // - otherwise → fail closed with 400. A body that is neither used to
+    //   401 silently as "unauthorized", hiding a malformed push behind an
+    //   auth error; 400 makes it observable.
+    let Ok(value): serde_json::Result<Value> = serde_json::from_slice(&body) else {
         return error_response(StatusCode::BAD_REQUEST, "invalid_request", false);
     };
-    // Diagnostic request_id minted once here so every push 401 response and
+    // Correlation id minted once per push so the entry log, the branch logs and
-    // branch log line in this handler carries the same correlation id as the
+    // the 401 response all share it — that is what makes a push 401 in this log
-    // entry log. The verification_code value is never logged (AGENTS.md §3.2).
+    // matchable against Stalwart's `Push subscription error` line. The
+    // verification_code value is never logged (AGENTS.md §3.2).
     let request_id = reconcile_lock_owner();
     tracing::info!(
         request_id = %request_id,
-        subscription_id = ?push.subscription_id,
+        type_tag = ?value.get("@type").and_then(|v| v.as_str()),
-        account_id = ?push.account_id,
+        has_verification_code = value.get("verificationCode").is_some(),
-        email_id = ?push.email_id,
+        has_changed = value.get("changed").is_some(),
-        has_verification_code = push.verification_code.is_some(),
         "jmap_push request received"
     );
-    // subscription_id is required on every request (RFC 8620 §7.2/§7.3).
+    if value.get("verificationCode").is_some() {
+        let Ok(push) = serde_json::from_value::<JmapPush>(value).map_err(|_| ()) else {
+            return error_response(StatusCode::BAD_REQUEST, "invalid_request", false);
+        };
+        // subscription_id is required on the verification request (RFC 8620 §7.2).
         let Some(subscription) = push.subscription_id.as_deref() else {
             return push_401(request_id);
         };
-    if let Some(code) = push.verification_code.as_deref() {
+        let Some(code) = push.verification_code.as_deref() else {
+            // Unreachable: the branch was selected on verification_code being Some.
+            unreachable!("verificationCode branch without a verification code")
+        };
         // Verification request (RFC 8620 §7.2): carries verificationCode.
         if !matches!(
             app.state
@@ -1364,103 +1383,114 @@ async fn jmap_push(State(app): State<AppState>, body: Bytes) -> Response {
                         "push verify API call failed; continuing (RFC 8620 §7.2 verification is via callback response)");
                 }
             }
-            // Diagnostic: a verification request rewrites push:subscription:{id}.
-            // If a concurrent event push 401s, this line tells us a verification
-            // raced it (key rewritten mid-window). The code value is not logged.
-            tracing::info!(
-                request_id = %request_id,
-                subscription = %subscription,
-                "jmap_push verification branch wrote key"
-            );
         }
-    } else {
+        // RFC 8620 §7.2: the verification response MUST be a JSON body
-        // Event push (RFC 8620 §7.3): no verificationCode; the subscription
-        // must have been verified during the initial handshake.
-        let push_key = format!("push:subscription:{subscription}");
-        let exists = matches!(app.state.dedup_exists(&push_key).await, Ok(true));
-        // Diagnostic: this is the line that decides the 401. If subscription is
-        // "b" and exists is false here while a CLI probe shows the key present,
-        // the key was deleted/rewritten in the window (see the verify-branch
-        // log). If subscription is not "b", Stalwart sent a different id than
-        // our stored push:subscription:id pointer — the contradiction's root.
-        tracing::info!(
-            request_id = %request_id,
-            key = %push_key,
-            exists,
-            "jmap_push event dedup_exists check"
-        );
-        if !exists {
-            return push_401(request_id);
-        }
-    }
-    if push.account_id.is_none() && push.email_id.is_none() {
-        // RFC 8620 §7.2: verification request response MUST be a JSON body
         // containing exactly the verificationCode property (not 204).
-        if let Some(code) = push.verification_code.as_deref() {
         return axum::Json(serde_json::json!({
             "verificationCode": code
         }))
         .into_response();
     }
-        return StatusCode::NO_CONTENT.into_response();
+    if value.get("changed").is_some() {
-    }
+        // Event push (StateChange): acknowledge so the server stops retrying,
-    let (Some(account), Some(email)) = (push.account_id, push.email_id) else {
+        // then pull the delta. The spawned task takes the same `lock:reconcile`
-        return error_response(StatusCode::BAD_REQUEST, "invalid_request", false);
+        // the /reconcile endpoint holds, so a push-triggered pull and a
-    };
+        // cron-triggered pull never run concurrently (they would race on
-    let key = format!("dedup:jmap:{account}:{email}");
+        // `reconcile:state`). If the lock is already held, do nothing — the
-    match app.state.claim_dedup(&key, ttl::DEDUP_JMAP_SECONDS).await {
+        // running reconcile will pick this change up on its own.
-        Ok(false) => StatusCode::NO_CONTENT.into_response(),
+        let worker = Arc::clone(&app.worker);
-        Ok(true) => match app
+        let state = Arc::clone(&app.state);
-            .state
-            .enqueue("stalwart:jmap", &String::from_utf8_lossy(&body))
-            .await
-        {
-            Ok(message_id) => {
-                // Diagnostic: the event push was accepted and queued for
-                // delivery. If Telegram still doesn't arrive after this line,
-                // the failure is downstream (worker.process / Telegram), not
-                // in the push 401 path.
         tracing::info!(
             request_id = %request_id,
-                    %message_id,
+            accounts_changed = value
-                    "jmap_push event enqueued"
+                .get("changed")
+                .and_then(|v| v.as_object())
+                .map_or(0usize, |m| m.len()),
+            "jmap_push state change acked; reconciling"
         );
-                // Claim in-flight lease so the cron-triggered worker() drain
+        tokio::spawn(push_reconcile_delta(worker, state));
-                // skips this message while we process it immediately.
+        return StatusCode::NO_CONTENT.into_response();
-                let inflight_key = format!("delivery:inflight:stalwart:jmap:{message_id}");
+    }
-                if matches!(
+    error_response(StatusCode::BAD_REQUEST, "invalid_request", false)
-                    app.state
+}
-                        .claim_dedup(&inflight_key, ttl::DELIVERY_INFLIGHT_SECONDS)
+
+/// Pull the JMAP delta a StateChange push triggered, on the `/reconcile` lock.
+///
+/// A StateChange body carries no emailId (RFC 8620 §7.1.1 gives only
+/// account→type→state-string), so the only way to learn which messages changed
+/// is the `/changes` call this runs — the same path `/reconcile` uses. It
+/// acquires `lock:reconcile` so a push-triggered pull and the cron pull never
+/// touch `reconcile:state` concurrently. If the lock is held, this does
+/// nothing: the running pull will observe the same state change.
+async fn push_reconcile_delta(worker: Arc<dyn WorkerHandler>, state: Arc<dyn ReliableState>) {
+    const PUSH_RECONCILE_INITIAL_CHANGES: usize = 100;
+    let lock_owner = reconcile_lock_owner();
+    if !matches!(
+        state
+            .acquire_lock("lock:reconcile", &lock_owner, ttl::RECONCILE_LOCK_SECONDS)
             .await,
         Ok(true)
     ) {
-                    let worker = Arc::clone(&app.worker);
+        tracing::info!(
-                    let state = Arc::clone(&app.state);
+            "push_reconcile_delta: lock:reconcile busy; deferring to the running reconcile"
-                    let payload = String::from_utf8_lossy(&body).into_owned();
+        );
-                    tokio::spawn(async move {
+        return;
-                        if worker.process("stalwart:jmap", &payload).await.is_ok() {
-                            let committed =
-                                format!("delivery:committed:stalwart:jmap:{message_id}");
-                            let _ = state
-                                .claim_dedup(&committed, ttl::DELIVERY_COMMITTED_SECONDS)
-                                .await;
-                            let _ = state
-                                .ack("stalwart:jmap", "stalwart-workers", &message_id)
-                                .await;
-                        }
-                        let inflight = format!("delivery:inflight:stalwart:jmap:{message_id}");
-                        let _ = state.release_dedup(&inflight).await;
-                    });
-                }
-                StatusCode::NO_CONTENT.into_response()
     }
+    let since = match state.get_reconcile_state().await {
+        Ok(value) => value,
         Err(_) => {
-                let _ = app.state.release_dedup(&key).await;
+            // Same shape as /reconcile: a state read failure means there is no
-                error_response(StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", true)
+            // safe `since` to reconcile from, so release and give up. The cron
+            // reconcile covers the change on its next tick.
+            tracing::warn!("push_reconcile_delta: reconcile state read failed");
+            let _ = state.release_lock("lock:reconcile", &lock_owner).await;
+            return;
         }
+    };
+    let heartbeat_state = Arc::clone(&state);
+    let heartbeat_owner = lock_owner.clone();
+    let lease_lost = Arc::new(AtomicBool::new(false));
+    let heartbeat_lease_lost = Arc::clone(&lease_lost);
+    let heartbeat = tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(std::time::Duration::from_secs(
+                ttl::RECONCILE_HEARTBEAT_INTERVAL_SECONDS,
+            ))
+            .await;
+            match heartbeat_state
+                .renew_lock(
+                    "lock:reconcile",
+                    &heartbeat_owner,
+                    ttl::RECONCILE_HEARTBEAT_SECONDS,
+                )
+                .await
+            {
+                Ok(true) => {}
+                _ => {
+                    heartbeat_lease_lost.store(true, Ordering::Release);
+                    break;
+                }
+            }
+        }
+    });
+    let result = worker
+        .reconcile(since.as_deref(), PUSH_RECONCILE_INITIAL_CHANGES)
+        .await;
+    heartbeat.abort();
+    if lease_lost.load(Ordering::Acquire) {
+        tracing::warn!("push_reconcile_delta: lock lease lost mid-pull; state not advanced");
+    } else {
+        match result {
+            Ok(new_state) => match state.set_reconcile_state(&new_state).await {
+                Ok(()) => tracing::info!("push_reconcile_delta: reconcile state advanced"),
+                Err(_) => tracing::error!("push_reconcile_delta: reconcile state write failed"),
             },
-        Err(_) => error_response(StatusCode::SERVICE_UNAVAILABLE, "service_unavailable", true),
+            Err(reason) => {
+                tracing::warn!(reason = %reason, "push_reconcile_delta: reconcile failed")
             }
+        }
+    }
+    let _ = state.release_lock("lock:reconcile", &lock_owner).await;
 }
 
 #[derive(serde::Deserialize)]
@@ -1933,16 +1963,16 @@ impl TelegramUpdate {
         self.message.as_ref().map(|m| m.chat.id)
     }
 }
+/// PushVerification body (RFC 8620 §7.2.3). A StateChange body does not fit
+/// this shape at all — it carries neither `pushSubscriptionId` nor
+/// `verificationCode` — so `jmap_push` discriminates on the parsed `Value`
+/// before reaching for this type.
 #[derive(serde::Deserialize)]
 struct JmapPush {
     #[serde(rename = "pushSubscriptionId")]
     subscription_id: Option<String>,
     #[serde(rename = "verificationCode")]
     verification_code: Option<String>,
-    #[serde(rename = "accountId")]
-    account_id: Option<String>,
-    #[serde(rename = "emailId")]
-    email_id: Option<String>,
 }
 
 fn header_value_matches(headers: &HeaderMap, name: &str, expected: &str) -> bool {
@@ -2538,7 +2568,10 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn jmap_push_requires_verification_code() {
+    async fn jmap_push_rejects_bodies_without_verification_or_change() {
+        // Neither a verificationCode nor a `changed` map: this is not a push
+        // body at all (RFC 8620 §7). It used to 401 as "unauthorized", which
+        // hid a malformed push behind an auth error; 400 makes it observable.
         assert_eq!(
             request(
                 Request::builder()
@@ -2548,20 +2581,10 @@ mod tests {
                     .unwrap(),
             )
             .await,
-            StatusCode::UNAUTHORIZED
+            StatusCode::BAD_REQUEST
-        );
-        assert_eq!(
-            request(
-                Request::builder()
-                    .method("POST")
-                    .uri("/push/jmap")
-                    .header("content-type", "application/json")
-                    .body(Body::from(r#"{"verificationCode":""}"#))
-                    .unwrap(),
-            )
-            .await,
-            StatusCode::UNAUTHORIZED
         );
+        // A verificationCode without a pushSubscriptionId cannot be attributed
+        // to a subscription, so it is refused before any state is written.
         assert_eq!(
             request(
                 Request::builder()
@@ -2574,6 +2597,14 @@ mod tests {
             .await,
             StatusCode::UNAUTHORIZED
         );
+    }
+
+    #[tokio::test]
+    async fn jmap_push_acks_state_change() {
+        // RFC 8620 §7.1.1 StateChange: no pushSubscriptionId, no accountId, no
+        // emailId — only account→type→state-string. This is what Stalwart
+        // actually sends for a new message, and it is the body that used to
+        // 401 for the lifetime of this route.
         assert_eq!(
             request(
                 Request::builder()
@@ -2581,20 +2612,42 @@ mod tests {
                     .uri("/push/jmap")
                     .header("content-type", "application/json")
                     .body(Body::from(
-                        r#"{"pushSubscriptionId":"push-1","verificationCode":"code","accountId":"a","emailId":"e"}"#
+                        r#"{"@type":"StateChange","changed":{"a3123":{"Email":"d35ecb040aab","EmailDelivery":"428d565f2440"}}}"#
                     ))
                     .unwrap(),
             )
             .await,
-            // Hybrid payload (verificationCode + accountId + emailId): the
-            // verification is recorded, then the request falls through to event
-            // enqueue (RFC 8620 §7.3). remember_push_subscription succeeds
-            // (PUSH_SUBSCRIPTION_SECONDS == 0 → no expiry), so the response is
-            // 204 — not 503. The pre-TTL-0 expectation of 503 was stale.
             StatusCode::NO_CONTENT
         );
     }
 
+    #[tokio::test]
+    async fn jmap_push_verification_returns_the_code() {
+        // RFC 8620 §7.2: the verification response MUST be a JSON body holding
+        // the verificationCode property, not an empty 204.
+        let response = test_router()
+            .oneshot(
+                Request::builder()
+                    .method("POST")
+                    .uri("/push/jmap")
+                    .header("content-type", "application/json")
+                    .body(Body::from(
+                        r#"{"@type":"PushVerification","pushSubscriptionId":"push-1","verificationCode":"code"}"#
+                    ))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::OK);
+        let body = axum::body::to_bytes(response.into_body(), usize::MAX)
+            .await
+            .unwrap();
+        assert_eq!(
+            serde_json::from_slice::<Value>(&body).unwrap(),
+            json!({"verificationCode": "code"})
+        );
+    }
+
     #[tokio::test]
     async fn worker_requires_bearer_token() {
         let status = request(