diff --git a/.changeset/amsg-review-hardening.md b/.changeset/amsg-review-hardening.md
new file mode 100644
index 0000000..39f5512
--- /dev/null
+++ b/.changeset/amsg-review-hardening.md
@@ -0,0 +1,23 @@
+---
+"@rei-standard/amsg-server": minor
+"@rei-standard/amsg-shared": patch
+---
+
+代码评审加固：存量任务订阅兜底、串行分组写偏斜收口、重试状态重置、错误分类与门禁去重
+
+**@rei-standard/amsg-server**
+
+- **升级前创建的任务不再必然投递失败。** 投递时解析订阅新增兜底：用户级 `push_subscriptions` 存储里没有订阅时，回退到旧任务 payload 里内嵌的 `pushSubscription`（存储里有则永远优先用存储的那份）。普通投递路径和 agentic 路径都生效——存量部署升级后，用户来不及打开新客户端登记订阅，旧任务照样发得出去。
+- **pg / neon 串行分组占位补上写偏斜收口。** READ COMMITTED 下两个并发 tick 各领同组「不同」行时，`NOT EXISTS` 互相看不见对方未提交的租约，同组两条任务可能并发执行。现在占位提交后再复查一次同组活租约，撞上就放掉自己刚写的租约、这一跳不跑（两边都让路也安全：行保持 pending，下一跳重试）。`claimTask` 与 `push_subscriptions` 三方法同时收拢到 `adapters/pg-shared.js`，pg / neon 共用一份 SQL，语义不再可能分歧。
+- **tick 内串行分组预占用按用户隔离。** 内存侧的占坑键带上 `user_id`，与落库侧 per-user HMAC 的隔离语义对齐——多用户部署下两个用户恰好返回同一个分组 key（如共用的默认角色名）不再互相顶掉对方的任务。
+- **`PUT /update-message` 重置重试状态。** 更新任务时 `retry_count` 归零、`retry_after` 清空（后者仅在支持 `claimTask` 的适配器上写）——刚修好 apiKey / 改好排期的任务不再背着耗尽的重试预算，下一次瞬时故障不会直接把它打成永久 failed。
+- **`POST /schedule-message` 的订阅预检改为存在性检查。** 不再解密（解出来的值本来也用不上）；查询本身失败时报可重试的 503 `PUSH_SUBSCRIPTION_LOOKUP_FAILED`，不再把瞬时 DB 故障伪装成 409 `PUSH_SUBSCRIPTION_MISSING` 引导客户端去走多余的重订阅流程。
+- **订阅类错误带稳定 `code`，投递失败按类别处置。** `resolvePushSubscription` 抛出的错误带 `err.code`（`PUSH_SUBSCRIPTION_MISSING` / `PUSH_SUBSCRIPTION_STORE_UNSUPPORTED`），消费方按 code 分支即可、不必匹配 message 文案；tick 的失败处置对这两类「重试也好不了」的错误短路退避阶梯——一次性任务直接进终审处置，循环任务直接作废本次 occurrence，不再每次白跑 3 轮重试。
+- **过期守卫两处收紧 / 放开。** 重试链上的任务（`retry_count > 0`）在排定的重试时刻（`retry_after`）本身也被拖过阈值时同样按过期处理——停摆恰好落在重试窗口里的任务不再于恢复后把几天前的旧内容推出去（`getPendingTasks` 随之在返回行里带上 `retry_after`）。阈值本身可用 `ctx.staleAfterMs` 覆盖（单用户 worker 从 config 的 `staleAfterMs` 透传），依赖「再晚也送达」语义的宿主有了官方出口。
+- **单用户 worker 的两处错误边界补齐。** `cors.origin` 回调抛错按「不放行这个 origin」处理，不再逃出 `fetch()` 变成 Cloudflare 1101 错误页；`scheduled()` 的配置构建失败改为记日志跳过这一跳，不再以未捕获异常崩掉 cron 调用。
+- **存量多租户租户自动补列。** 多租户侧每个进程首次取得适配器时补跑一遍幂等的 `initSchema`（建表 / `ADD COLUMN IF NOT EXISTS`），升级加列后第一个请求就把 schema 补齐——不再依赖 CHANGELOG 里的手工 DDL 步骤（同 tenantId 重放 `/init-tenant` 到不了 `initSchema` 就 409，此前存量租户没有任何自动迁移路径）。
+- **门禁与工具函数去重。** X-User-Id 门禁（8 个 handler 里的复制粘贴，文案已分裂成两种）收拢为 `lib/request.js` 的 `requireUserId()`，对同一错误码的 message 统一为「缺少用户标识符」；`UPDATABLE_COLUMNS` 白名单三个适配器共用 `schema.js` 一份；`isValidUrl` 改为 re-export shared 的实现；tenant/blob-store 的 base64url 改用 shared 实现；tick 的预解密 payload 直通投递侧（`processSingleMessage` 新增 `predecrypted` 参数），同一份密文不再解两遍，相关失实注释一并修正；过期跳过的循环/一次性两个近似复制的分支收拢为单一尾部。
+
+**@rei-standard/amsg-shared**
+
+- `verifyVapidJwt` 的 JWT payload 解码改用 `webcrypto-utils` 的 `utf8Decode`，兑现本模块「编码辅助只住在 webcrypto-utils」的约定（行为不变）。
diff --git a/packages/rei-standard-amsg/server/src/server/adapters/d1.js b/packages/rei-standard-amsg/server/src/server/adapters/d1.js
index ab73596..fd5eee3 100644
--- a/packages/rei-standard-amsg/server/src/server/adapters/d1.js
+++ b/packages/rei-standard-amsg/server/src/server/adapters/d1.js
@@ -15,15 +15,8 @@ import {
   CLIENT_STATE_TABLE_SQL,
   PUSH_SUBSCRIPTION_TABLE_SQL
 } from './schema.sqlite.js';
-
-// Update methods build a dynamic SET clause from object keys. Callers pass only
-// hardcoded column names today, but enforcing a whitelist keeps a future caller
-// from ever turning a caller-supplied key into interpolated SQL.
-const UPDATABLE_COLUMNS = new Set([
-  'user_id', 'uuid', 'encrypted_payload', 'message_type',
-  'next_send_at', 'lease_until', 'retry_after', 'serialize_group',
-  'status', 'retry_count', 'created_at', 'updated_at'
-]);
+// 列名不分方言：三个适配器共用 schema.js 里的这一份白名单，加列只改一处。
+import { UPDATABLE_COLUMNS } from './schema.js';
 
 // LIKE 前缀转义：用户 key 里的 % _ \ 不能变成通配符/转义符。
 function escapeLikePrefix(prefix) {
@@ -197,7 +190,7 @@ export class D1Adapter {
   async getPendingTasks(limit = 50) {
     const now = this._now();
     const res = await this._db.prepare(
-      `SELECT id, user_id, uuid, encrypted_payload, message_type, next_send_at, status, retry_count
+      `SELECT id, user_id, uuid, encrypted_payload, message_type, next_send_at, retry_after, status, retry_count
        FROM scheduled_messages
        WHERE status = 'pending' AND next_send_at <= ?
          AND (lease_until IS NULL OR lease_until <= ?)
diff --git a/packages/rei-standard-amsg/server/src/server/adapters/neon.js b/packages/rei-standard-amsg/server/src/server/adapters/neon.js
index 194df79..bf33442 100644
--- a/packages/rei-standard-amsg/server/src/server/adapters/neon.js
+++ b/packages/rei-standard-amsg/server/src/server/adapters/neon.js
@@ -14,6 +14,7 @@ import {
   COLUMNS_SQL,
   UPDATABLE_COLUMNS
 } from './schema.js';
+import * as pgShared from './pg-shared.js';
 
 export class NeonAdapter {
   /** @param {string} connectionString */
@@ -198,7 +199,7 @@ export class NeonAdapter {
   async getPendingTasks(limit = 50) {
     const sql = this._getSql();
     return sql.query(
-      `SELECT id, user_id, uuid, encrypted_payload, message_type, next_send_at, status, retry_count
+      `SELECT id, user_id, uuid, encrypted_payload, message_type, next_send_at, retry_after, status, retry_count
        FROM scheduled_messages
        WHERE status = 'pending' AND next_send_at <= NOW()
          AND (lease_until IS NULL OR lease_until <= NOW())
@@ -209,65 +210,11 @@ export class NeonAdapter {
     );
   }
 
-  /**
-   * 领取一条到点的任务：在 lease_until 上写下「这条归我管到什么时候」，
-   * 本次投递期间别的 tick 领不走它。
-   *
-   * 租约写在自己的列上，next_send_at 全程不动——那一列是用户设的触发时刻，
-   * 任务列表要读它、循环任务推进下一次也要拿它当基准。
-   *
-   * 两个 tick 抢同一行时只有一个改得动，另一个拿不到 RETURNING 行，据此跳
-   * 过。WHERE 里的两个条件各管一件事：
-   *   - lease_until 为空或已过期：没人正在跑这条。领了任务的 tick 中途没了
-   *     也不会把行焊死，租约到期后自然可以被接手。
-   *   - next_send_at 等于读这行时看到的值：读出来之后用户又改了排期的话，
-   *     这一跳就不该再按旧时刻发。
-   *
-   * 不加一个 'sending' 状态来表达「正在跑」：status 上有 CHECK 约束，加值
-   * 要改表。
-   *
-   * 比 next_send_at 时两边都截到毫秒：列是 timestamptz（微秒精度），驱动读
-   * 出来是 JS Date（毫秒精度），原值送回去可能因为亚毫秒差对不上。
-   *
-   * 带 serializeGroup 时多一道分组门：同一分组里已经有别的行拿着未到期的租
-   * 约，这条就领不走（同一分组同时只跑一条）。判定和写租约在同一条 UPDATE
-   * 里完成，「先查再占」的空档天然不存在。分组门只看租约，不看
-   * `retry_after`：等着重试的任务其实闲着，不该把同分组的其他任务一起堵住。
-   *
-   * @param {number} taskId
-   * @param {string|Date} expectedNextSendAt - 读这行时拿到的 next_send_at 原值
-   * @param {string|Date} leaseUntil - 租期末尾
-   * @param {string|null} [serializeGroup] - 串行分组标识；空表示不参与分组串行
-   * @returns {Promise<boolean>} true = 领到了；false = 别人正拿着租约、同分组
-   *   有任务正在跑、排期被改过、或行已不是 pending
-   */
+  // 领取一条到点的任务。SQL 与并发语义在 pg-shared.js（pg / neon 共用一份，
+  // 语义说明也在那里）。
   async claimTask(taskId, expectedNextSendAt, leaseUntil, serializeGroup = null) {
     const sql = this._getSql();
-    const grouped = typeof serializeGroup === 'string' && serializeGroup.length > 0;
-    const params = [leaseUntil, taskId, expectedNextSendAt];
-    let setClause = 'lease_until = $1, updated_at = NOW()';
-    let groupGuard = '';
-    if (grouped) {
-      params.push(serializeGroup); // $4
-      setClause = 'lease_until = $1, serialize_group = $4, updated_at = NOW()';
-      groupGuard = `
-          AND NOT EXISTS (
-            SELECT 1 FROM scheduled_messages busy
-             WHERE busy.serialize_group = $4 AND busy.id <> $2
-               AND busy.status = 'pending' AND busy.lease_until > NOW()
-          )`;
-    }
-    const rows = await sql.query(
-      `UPDATE scheduled_messages
-          SET ${setClause}
-        WHERE id = $2 AND status = 'pending'
-          AND date_trunc('milliseconds', next_send_at)
-            = date_trunc('milliseconds', $3::timestamptz)
-          AND (lease_until IS NULL OR lease_until <= NOW())${groupGuard}
-       RETURNING id`,
-      params
-    );
-    return rows.length > 0;
+    return pgShared.claimTask((text, params) => sql.query(text, params), taskId, expectedNextSendAt, leaseUntil, serializeGroup);
   }
 
   async listTasks(userId, opts = {}) {
@@ -327,60 +274,20 @@ export class NeonAdapter {
     return rows.length > 0 ? rows[0].status : null;
   }
 
-  // ── push_subscriptions (user-level Web Push subscription) ──────────────
+  // ── push_subscriptions（实现见 pg-shared.js，pg / neon 共用一份）─────────
 
-  /**
-   * 这个用户当前登记的推送订阅（密文原样返回，解密在上层）。
-   *
-   * @param {string} userId
-   * @returns {Promise<{ subscription: string, updated_at: number }|null>}
-   */
   async getPushSubscription(userId) {
     const sql = this._getSql();
-    const rows = await sql.query(
-      'SELECT subscription, updated_at FROM push_subscriptions WHERE user_id = $1 LIMIT 1',
-      [userId]
-    );
-    if (rows.length === 0) return null;
-    // BIGINT 在 pg 驱动里读出来是字符串，统一成 number 再往上给。
-    return { subscription: rows[0].subscription, updated_at: Number(rows[0].updated_at) };
+    return pgShared.getPushSubscription((text, params) => sql.query(text, params), userId);
   }
 
-  /**
-   * 覆盖写这个用户的订阅。一个用户一行，没有 last-write-wins 之类的比较——
-   * 客户端拿到的新订阅永远比旧的有效，旧的那份只会 410。
-   *
-   * @param {string} userId
-   * @param {string} encryptedSubscription
-   * @param {number} updatedAt - epoch 毫秒
-   * @returns {Promise<boolean>}
-   */
   async upsertPushSubscription(userId, encryptedSubscription, updatedAt) {
     const sql = this._getSql();
-    const rows = await sql.query(
-      `INSERT INTO push_subscriptions (user_id, subscription, updated_at)
-       VALUES ($1, $2, $3)
-       ON CONFLICT (user_id) DO UPDATE SET
-         subscription = EXCLUDED.subscription,
-         updated_at = EXCLUDED.updated_at
-       RETURNING user_id`,
-      [userId, encryptedSubscription, updatedAt]
-    );
-    return rows.length > 0;
+    return pgShared.upsertPushSubscription((text, params) => sql.query(text, params), userId, encryptedSubscription, updatedAt);
   }
 
-  /**
-   * 删掉这个用户的订阅（设置页「停止接收推送」）。
-   *
-   * @param {string} userId
-   * @returns {Promise<boolean>} true = 确实删掉了一行
-   */
   async deletePushSubscription(userId) {
     const sql = this._getSql();
-    const rows = await sql.query(
-      'DELETE FROM push_subscriptions WHERE user_id = $1 RETURNING user_id',
-      [userId]
-    );
-    return rows.length > 0;
+    return pgShared.deletePushSubscription((text, params) => sql.query(text, params), userId);
   }
 }
diff --git a/packages/rei-standard-amsg/server/src/server/adapters/pg-shared.js b/packages/rei-standard-amsg/server/src/server/adapters/pg-shared.js
new file mode 100644
index 0000000..5fece7f
--- /dev/null
+++ b/packages/rei-standard-amsg/server/src/server/adapters/pg-shared.js
@@ -0,0 +1,151 @@
+/**
+ * pg / neon 共用的 Postgres 查询实现。
+ *
+ * 两个适配器只在「怎么把 SQL 发出去」上不同（pg 的连接池 vs neon 的 HTTP
+ * 驱动），SQL 与并发语义必须逐字一致——各自复制一份的话，修一边漏一边就会让
+ * 两种 Postgres 部署的串行化行为静默分歧，而各自的测试还都是绿的。所以这里
+ * 按执行器参数化：适配器只递一个 `query(text, params) → rows`。
+ *
+ * D1（SQLite 方言、单写者、ISO TEXT 时间戳）不走这份实现。
+ *
+ * @typedef {(text: string, params?: any[]) => Promise<any[]>} PgQuery
+ */
+
+/**
+ * 领取一条到点的任务：在 lease_until 上写下「这条归我管到什么时候」，
+ * 本次投递期间别的 tick 领不走它。
+ *
+ * 租约写在自己的列上，next_send_at 全程不动——那一列是用户设的触发时刻，
+ * 任务列表要读它、循环任务推进下一次也要拿它当基准。
+ *
+ * 两个 tick 抢同一行时只有一个改得动，另一个拿不到 RETURNING 行，据此跳
+ * 过。WHERE 里的两个条件各管一件事：
+ *   - lease_until 为空或已过期：没人正在跑这条。领了任务的 tick 中途没了
+ *     也不会把行焊死，租约到期后自然可以被接手。
+ *   - next_send_at 等于读这行时看到的值：读出来之后用户又改了排期的话，
+ *     这一跳就不该再按旧时刻发。
+ *
+ * 不加一个 'sending' 状态来表达「正在跑」：status 上有 CHECK 约束，加值
+ * 要改表。
+ *
+ * 比 next_send_at 时两边都截到毫秒：列是 timestamptz（微秒精度），驱动读
+ * 出来是 JS Date（毫秒精度），原值送回去可能因为亚毫秒差对不上。
+ *
+ * 带 serializeGroup 时多一道分组门：同一分组里已经有别的行拿着未到期的租
+ * 约，这条就领不走（同一分组同时只跑一条）。判定和写租约在同一条 UPDATE
+ * 里完成——但那只对「抢同一行」成立；READ COMMITTED 下两个并发 tick 各领
+ * 同组的**不同**行时，各自的 NOT EXISTS 子查询都看不到对方尚未提交的租约，
+ * 也没有行锁冲突逼它重查（写偏斜）。所以占位成功后再回头查一次：真撞上了
+ * 就把自己刚写的租约放掉、这一跳不跑——两边都让也没事，行保持 pending，
+ * 下一跳重试。分组门只看租约，不看 `retry_after`：等着重试的任务其实闲着，
+ * 不该把同分组的其他任务一起堵住。
+ *
+ * @param {PgQuery} query
+ * @param {number} taskId
+ * @param {string|Date} expectedNextSendAt - 读这行时拿到的 next_send_at 原值
+ * @param {string|Date} leaseUntil - 租期末尾
+ * @param {string|null} [serializeGroup] - 串行分组标识；空表示不参与分组串行
+ * @returns {Promise<boolean>} true = 领到了；false = 已被别人领走、同分组有
+ *   任务正在跑、排期被改过、或行已不是 pending
+ */
+export async function claimTask(query, taskId, expectedNextSendAt, leaseUntil, serializeGroup = null) {
+  const grouped = typeof serializeGroup === 'string' && serializeGroup.length > 0;
+  const params = [leaseUntil, taskId, expectedNextSendAt];
+  let setClause = 'lease_until = $1, updated_at = NOW()';
+  let groupGuard = '';
+  if (grouped) {
+    params.push(serializeGroup); // $4
+    setClause = 'lease_until = $1, serialize_group = $4, updated_at = NOW()';
+    groupGuard = `
+        AND NOT EXISTS (
+          SELECT 1 FROM scheduled_messages busy
+           WHERE busy.serialize_group = $4 AND busy.id <> $2
+             AND busy.status = 'pending' AND busy.lease_until > NOW()
+        )`;
+  }
+  const rows = await query(
+    `UPDATE scheduled_messages
+        SET ${setClause}
+      WHERE id = $2 AND status = 'pending'
+        AND date_trunc('milliseconds', next_send_at)
+          = date_trunc('milliseconds', $3::timestamptz)
+        AND (lease_until IS NULL OR lease_until <= NOW())${groupGuard}
+     RETURNING id`,
+    params
+  );
+  if (rows.length === 0) return false;
+  if (!grouped) return true;
+
+  // 写偏斜的收口（见函数头注释）：自己的租约已提交，此刻再查，对方若也领
+  // 到了同组的另一行，双方至少有一方看得见冲突并让路。
+  const conflict = await query(
+    `SELECT 1 FROM scheduled_messages busy
+      WHERE busy.serialize_group = $1 AND busy.id <> $2
+        AND busy.status = 'pending' AND busy.lease_until > NOW()
+      LIMIT 1`,
+    [serializeGroup, taskId]
+  );
+  if (conflict.length === 0) return true;
+  await query(
+    'UPDATE scheduled_messages SET lease_until = NULL, updated_at = NOW() WHERE id = $1',
+    [taskId]
+  );
+  return false;
+}
+
+// ── push_subscriptions (user-level Web Push subscription) ──────────────
+
+/**
+ * 这个用户当前登记的推送订阅（密文原样返回，解密在上层）。
+ *
+ * @param {PgQuery} query
+ * @param {string} userId
+ * @returns {Promise<{ subscription: string, updated_at: number }|null>}
+ */
+export async function getPushSubscription(query, userId) {
+  const rows = await query(
+    'SELECT subscription, updated_at FROM push_subscriptions WHERE user_id = $1 LIMIT 1',
+    [userId]
+  );
+  if (rows.length === 0) return null;
+  // BIGINT 在 pg 驱动里读出来是字符串，统一成 number 再往上给。
+  return { subscription: rows[0].subscription, updated_at: Number(rows[0].updated_at) };
+}
+
+/**
+ * 覆盖写这个用户的订阅。一个用户一行，没有 last-write-wins 之类的比较——
+ * 客户端拿到的新订阅永远比旧的有效，旧的那份只会 410。
+ *
+ * @param {PgQuery} query
+ * @param {string} userId
+ * @param {string} encryptedSubscription
+ * @param {number} updatedAt - epoch 毫秒
+ * @returns {Promise<boolean>}
+ */
+export async function upsertPushSubscription(query, userId, encryptedSubscription, updatedAt) {
+  const rows = await query(
+    `INSERT INTO push_subscriptions (user_id, subscription, updated_at)
+     VALUES ($1, $2, $3)
+     ON CONFLICT (user_id) DO UPDATE SET
+       subscription = EXCLUDED.subscription,
+       updated_at = EXCLUDED.updated_at
+     RETURNING user_id`,
+    [userId, encryptedSubscription, updatedAt]
+  );
+  return rows.length > 0;
+}
+
+/**
+ * 删掉这个用户的订阅（设置页「停止接收推送」）。
+ *
+ * @param {PgQuery} query
+ * @param {string} userId
+ * @returns {Promise<boolean>} true = 确实删掉了一行
+ */
+export async function deletePushSubscription(query, userId) {
+  const rows = await query(
+    'DELETE FROM push_subscriptions WHERE user_id = $1 RETURNING user_id',
+    [userId]
+  );
+  return rows.length > 0;
+}
diff --git a/packages/rei-standard-amsg/server/src/server/adapters/pg.js b/packages/rei-standard-amsg/server/src/server/adapters/pg.js
index 3b0117d..012eb91 100644
--- a/packages/rei-standard-amsg/server/src/server/adapters/pg.js
+++ b/packages/rei-standard-amsg/server/src/server/adapters/pg.js
@@ -16,6 +16,7 @@ import {
   COLUMNS_SQL,
   UPDATABLE_COLUMNS
 } from './schema.js';
+import * as pgShared from './pg-shared.js';
 
 export class PgAdapter {
   /** @param {string} connectionString */
@@ -196,7 +197,7 @@ export class PgAdapter {
 
   async getPendingTasks(limit = 50) {
     return this._query(
-      `SELECT id, user_id, uuid, encrypted_payload, message_type, next_send_at, status, retry_count
+      `SELECT id, user_id, uuid, encrypted_payload, message_type, next_send_at, retry_after, status, retry_count
        FROM scheduled_messages
        WHERE status = 'pending' AND next_send_at <= NOW()
          AND (lease_until IS NULL OR lease_until <= NOW())
@@ -207,64 +208,10 @@ export class PgAdapter {
     );
   }
 
-  /**
-   * 领取一条到点的任务：在 lease_until 上写下「这条归我管到什么时候」，
-   * 本次投递期间别的 tick 领不走它。
-   *
-   * 租约写在自己的列上，next_send_at 全程不动——那一列是用户设的触发时刻，
-   * 任务列表要读它、循环任务推进下一次也要拿它当基准。
-   *
-   * 两个 tick 抢同一行时只有一个改得动，另一个拿不到 RETURNING 行，据此跳
-   * 过。WHERE 里的两个条件各管一件事：
-   *   - lease_until 为空或已过期：没人正在跑这条。领了任务的 tick 中途没了
-   *     也不会把行焊死，租约到期后自然可以被接手。
-   *   - next_send_at 等于读这行时看到的值：读出来之后用户又改了排期的话，
-   *     这一跳就不该再按旧时刻发。
-   *
-   * 不加一个 'sending' 状态来表达「正在跑」：status 上有 CHECK 约束，加值
-   * 要改表。
-   *
-   * 比 next_send_at 时两边都截到毫秒：列是 timestamptz（微秒精度），驱动读
-   * 出来是 JS Date（毫秒精度），原值送回去可能因为亚毫秒差对不上。
-   *
-   * 带 serializeGroup 时多一道分组门：同一分组里已经有别的行拿着未到期的租
-   * 约，这条就领不走（同一分组同时只跑一条）。判定和写租约在同一条 UPDATE
-   * 里完成，「先查再占」的空档天然不存在。分组门只看租约，不看
-   * `retry_after`：等着重试的任务其实闲着，不该把同分组的其他任务一起堵住。
-   *
-   * @param {number} taskId
-   * @param {string|Date} expectedNextSendAt - 读这行时拿到的 next_send_at 原值
-   * @param {string|Date} leaseUntil - 租期末尾
-   * @param {string|null} [serializeGroup] - 串行分组标识；空表示不参与分组串行
-   * @returns {Promise<boolean>} true = 领到了；false = 已被别人领走、同分组有
-   *   任务正在跑、或行已不是 pending
-   */
+  // 领取一条到点的任务。SQL 与并发语义在 pg-shared.js（pg / neon 共用一份，
+  // 语义说明也在那里）。
   async claimTask(taskId, expectedNextSendAt, leaseUntil, serializeGroup = null) {
-    const grouped = typeof serializeGroup === 'string' && serializeGroup.length > 0;
-    const params = [leaseUntil, taskId, expectedNextSendAt];
-    let setClause = 'lease_until = $1, updated_at = NOW()';
-    let groupGuard = '';
-    if (grouped) {
-      params.push(serializeGroup); // $4
-      setClause = 'lease_until = $1, serialize_group = $4, updated_at = NOW()';
-      groupGuard = `
-          AND NOT EXISTS (
-            SELECT 1 FROM scheduled_messages busy
-             WHERE busy.serialize_group = $4 AND busy.id <> $2
-               AND busy.status = 'pending' AND busy.lease_until > NOW()
-          )`;
-    }
-    const rows = await this._query(
-      `UPDATE scheduled_messages
-          SET ${setClause}
-        WHERE id = $2 AND status = 'pending'
-          AND date_trunc('milliseconds', next_send_at)
-            = date_trunc('milliseconds', $3::timestamptz)
-          AND (lease_until IS NULL OR lease_until <= NOW())${groupGuard}
-       RETURNING id`,
-      params
-    );
-    return rows.length > 0;
+    return pgShared.claimTask((text, params) => this._query(text, params), taskId, expectedNextSendAt, leaseUntil, serializeGroup);
   }
 
   async listTasks(userId, opts = {}) {
@@ -321,57 +268,17 @@ export class PgAdapter {
     return rows.length > 0 ? rows[0].status : null;
   }
 
-  // ── push_subscriptions (user-level Web Push subscription) ──────────────
+  // ── push_subscriptions（实现见 pg-shared.js，pg / neon 共用一份）─────────
 
-  /**
-   * 这个用户当前登记的推送订阅（密文原样返回，解密在上层）。
-   *
-   * @param {string} userId
-   * @returns {Promise<{ subscription: string, updated_at: number }|null>}
-   */
   async getPushSubscription(userId) {
-    const rows = await this._query(
-      'SELECT subscription, updated_at FROM push_subscriptions WHERE user_id = $1 LIMIT 1',
-      [userId]
-    );
-    if (rows.length === 0) return null;
-    // BIGINT 在 pg 驱动里读出来是字符串，统一成 number 再往上给。
-    return { subscription: rows[0].subscription, updated_at: Number(rows[0].updated_at) };
+    return pgShared.getPushSubscription((text, params) => this._query(text, params), userId);
   }
 
-  /**
-   * 覆盖写这个用户的订阅。一个用户一行，没有 last-write-wins 之类的比较——
-   * 客户端拿到的新订阅永远比旧的有效，旧的那份只会 410。
-   *
-   * @param {string} userId
-   * @param {string} encryptedSubscription
-   * @param {number} updatedAt - epoch 毫秒
-   * @returns {Promise<boolean>}
-   */
   async upsertPushSubscription(userId, encryptedSubscription, updatedAt) {
-    const rows = await this._query(
-      `INSERT INTO push_subscriptions (user_id, subscription, updated_at)
-       VALUES ($1, $2, $3)
-       ON CONFLICT (user_id) DO UPDATE SET
-         subscription = EXCLUDED.subscription,
-         updated_at = EXCLUDED.updated_at
-       RETURNING user_id`,
-      [userId, encryptedSubscription, updatedAt]
-    );
-    return rows.length > 0;
+    return pgShared.upsertPushSubscription((text, params) => this._query(text, params), userId, encryptedSubscription, updatedAt);
   }
 
-  /**
-   * 删掉这个用户的订阅（设置页「停止接收推送」）。
-   *
-   * @param {string} userId
-   * @returns {Promise<boolean>} true = 确实删掉了一行
-   */
   async deletePushSubscription(userId) {
-    const rows = await this._query(
-      'DELETE FROM push_subscriptions WHERE user_id = $1 RETURNING user_id',
-      [userId]
-    );
-    return rows.length > 0;
+    return pgShared.deletePushSubscription((text, params) => this._query(text, params), userId);
   }
 }
diff --git a/packages/rei-standard-amsg/server/src/server/adapters/schema.js b/packages/rei-standard-amsg/server/src/server/adapters/schema.js
index 41c8132..4b86e88 100644
--- a/packages/rei-standard-amsg/server/src/server/adapters/schema.js
+++ b/packages/rei-standard-amsg/server/src/server/adapters/schema.js
@@ -130,7 +130,7 @@ export const COLUMNS_SQL = `
 // Update methods build a dynamic SET clause from object keys. Callers pass only
 // hardcoded column names today, but enforcing a whitelist keeps a future caller
 // from ever turning a caller-supplied key into interpolated SQL.
-// (The D1 adapter enforces the same list with its own in-file copy.)
+// 列名不分方言：pg / neon / D1 三个适配器 import 的都是这一份，加列只改这里。
 export const UPDATABLE_COLUMNS = new Set([
   'user_id', 'uuid', 'encrypted_payload', 'message_type',
   'next_send_at', 'lease_until', 'retry_after', 'serialize_group',
diff --git a/packages/rei-standard-amsg/server/src/server/cloudflare/single-user-worker.js b/packages/rei-standard-amsg/server/src/server/cloudflare/single-user-worker.js
index 465bf94..af68776 100644
--- a/packages/rei-standard-amsg/server/src/server/cloudflare/single-user-worker.js
+++ b/packages/rei-standard-amsg/server/src/server/cloudflare/single-user-worker.js
@@ -80,9 +80,19 @@ const CORS_ALLOW_METHODS = 'GET, POST, PUT, DELETE, OPTIONS';
  */
 function corsHeadersFor(cors, requestOrigin) {
   if (!cors || cors.origin == null) return null;
-  const allowOrigin = typeof cors.origin === 'function'
-    ? cors.origin(requestOrigin) || null
-    : cors.origin; // e.g. '*' or a fixed origin like 'https://app.example.com'
+  let allowOrigin;
+  if (typeof cors.origin === 'function') {
+    // 宿主的回调抛错不能让整个请求逃出错误边界（这个函数在 fetch() 的
+    // try 之外也会被调）：按「不放行这个 origin」处理，响应照常走 JSON 包体。
+    try {
+      allowOrigin = cors.origin(requestOrigin) || null;
+    } catch (error) {
+      console.warn('[amsg single-user] cors.origin 回调抛错，按不放行处理:', error && error.message);
+      allowOrigin = null;
+    }
+  } else {
+    allowOrigin = cors.origin; // e.g. '*' or a fixed origin like 'https://app.example.com'
+  }
   if (!allowOrigin) return null;
 
   const headers = {
@@ -216,7 +226,15 @@ export function createSingleUserCloudflareWorker(buildConfig) {
   }
 
   async function scheduled(event, env /* , ctx */) {
-    const cfg = await resolveConfig(env);
+    // fetch() 对 buildConfig 失败有降级路径，cron 这边同样不该以未捕获异常
+    // 崩掉：记日志跳过这一跳，行保持 pending，配置修好后下一跳照常。
+    let cfg;
+    try {
+      cfg = await resolveConfig(env);
+    } catch (error) {
+      console.error('[amsg single-user] scheduled(): config build failed; skipping tick:', error && error.message);
+      return;
+    }
     const vapid = cfg.vapid || {};
     if (!cfg.webpush || !vapid.email || !vapid.publicKey || !vapid.privateKey) {
       console.error('[amsg single-user] scheduled(): VAPID/webpush not configured; skipping tick');
@@ -257,7 +275,9 @@ export function createSingleUserCloudflareWorker(buildConfig) {
         // 跨跳也算（见 lib/run-tick.js）。不配 = 全并发，与以前一致。
         serializeBy: cfg.serializeBy,
         // 任务占位租期（默认 10 分钟，随 totalTimeoutMs 抬高）。
-        claimLeaseMs: cfg.claimLeaseMs
+        claimLeaseMs: cfg.claimLeaseMs,
+        // 补发新鲜度阈值（默认 60 分钟；见 lib/run-tick.js 的 STALE_AFTER_MS）。
+        staleAfterMs: cfg.staleAfterMs
       });
     } catch (error) {
       console.error('[amsg single-user] scheduled(): tick failed:', error && error.message);
diff --git a/packages/rei-standard-amsg/server/src/server/handlers/cancel-message.js b/packages/rei-standard-amsg/server/src/server/handlers/cancel-message.js
index 9a2a174..aa72ecf 100644
--- a/packages/rei-standard-amsg/server/src/server/handlers/cancel-message.js
+++ b/packages/rei-standard-amsg/server/src/server/handlers/cancel-message.js
@@ -5,8 +5,7 @@
  * @returns {{ DELETE: function }}
  */
 
-import { isValidUUIDv4 } from '../lib/validation.js';
-import { getHeader } from '../lib/request.js';
+import { requireUserId } from '../lib/request.js';
 
 export function createCancelMessageHandler(ctx) {
   async function DELETE(url, headers) {
@@ -23,13 +22,9 @@ export function createCancelMessageHandler(ctx) {
       return { status: 400, body: { success: false, error: { code: 'TASK_ID_REQUIRED', message: '缺少任务ID' } } };
     }
 
-    const userId = getHeader(headers, 'x-user-id');
-    if (!userId) {
-      return { status: 400, body: { success: false, error: { code: 'USER_ID_REQUIRED', message: '缺少用户标识符' } } };
-    }
-    if (!isValidUUIDv4(userId)) {
-      return { status: 400, body: { success: false, error: { code: 'INVALID_USER_ID_FORMAT', message: 'X-User-Id 必须是 UUID v4 格式' } } };
-    }
+    const gate = requireUserId(headers);
+    if (gate.error) return gate.error;
+    const { userId } = gate;
 
     const deleted = await db.deleteTaskByUuid(taskUuid, userId);
 
diff --git a/packages/rei-standard-amsg/server/src/server/handlers/client-state.js b/packages/rei-standard-amsg/server/src/server/handlers/client-state.js
index 9699844..2c0d07c 100644
--- a/packages/rei-standard-amsg/server/src/server/handlers/client-state.js
+++ b/packages/rei-standard-amsg/server/src/server/handlers/client-state.js
@@ -29,8 +29,7 @@
  */
 
 import { deriveUserEncryptionKey, decryptPayload, encryptPayload, decryptFromStorage } from '../lib/encryption.js';
-import { getHeader, isPlainObject, parseEncryptedBody } from '../lib/request.js';
-import { isValidUUIDv4 } from '../lib/validation.js';
+import { getHeader, isPlainObject, parseEncryptedBody, requireUserId } from '../lib/request.js';
 import {
   STATE_CHUNK_SLICE_BYTES,
   DEFAULT_MAX_STATE_VALUE_BYTES,
@@ -58,13 +57,6 @@ function err(status, code, message, details) {
   return { status, body: { success: false, error } };
 }
 
-function requireUserId(headers) {
-  const userId = getHeader(headers, 'x-user-id');
-  if (!userId) return { error: err(400, 'USER_ID_REQUIRED', '缺少用户标识符') };
-  if (!isValidUUIDv4(userId)) return { error: err(400, 'INVALID_USER_ID_FORMAT', 'X-User-Id 必须是 UUID v4 格式') };
-  return { userId };
-}
-
 function rejectEntry(entry, index, code, message, extra) {
   const rejection = { index, code, message, ...(extra || {}) };
   if (entry && typeof entry === 'object') {
diff --git a/packages/rei-standard-amsg/server/src/server/handlers/get-message.js b/packages/rei-standard-amsg/server/src/server/handlers/get-message.js
index 0178903..fecce5d 100644
--- a/packages/rei-standard-amsg/server/src/server/handlers/get-message.js
+++ b/packages/rei-standard-amsg/server/src/server/handlers/get-message.js
@@ -20,8 +20,7 @@
  */
 
 import { deriveUserEncryptionKey, decryptFromStorage, encryptPayload } from '../lib/encryption.js';
-import { getHeader } from '../lib/request.js';
-import { isValidUUIDv4 } from '../lib/validation.js';
+import { requireUserId } from '../lib/request.js';
 import { projectTask } from '../lib/task-projection.js';
 
 export function createGetMessageHandler(ctx) {
@@ -34,20 +33,9 @@ export function createGetMessageHandler(ctx) {
     const tenantCtx = tenantResult.context;
     const db = tenantCtx.db;
     const masterKey = tenantCtx.masterKey;
-    const userId = getHeader(headers, 'x-user-id');
-
-    if (!userId) {
-      return {
-        status: 400,
-        body: { success: false, error: { code: 'USER_ID_REQUIRED', message: '必须提供 X-User-Id 请求头' } }
-      };
-    }
-    if (!isValidUUIDv4(userId)) {
-      return {
-        status: 400,
-        body: { success: false, error: { code: 'INVALID_USER_ID_FORMAT', message: 'X-User-Id 必须是 UUID v4 格式' } }
-      };
-    }
+    const gate = requireUserId(headers);
+    if (gate.error) return gate.error;
+    const { userId } = gate;
 
     const taskUuid = new URL(url, 'https://dummy').searchParams.get('id');
     if (!taskUuid) {
diff --git a/packages/rei-standard-amsg/server/src/server/handlers/get-user-key.js b/packages/rei-standard-amsg/server/src/server/handlers/get-user-key.js
index 7ca6b4e..8933af8 100644
--- a/packages/rei-standard-amsg/server/src/server/handlers/get-user-key.js
+++ b/packages/rei-standard-amsg/server/src/server/handlers/get-user-key.js
@@ -6,8 +6,7 @@
  */
 
 import { deriveUserEncryptionKey } from '../lib/encryption.js';
-import { getHeader } from '../lib/request.js';
-import { isValidUUIDv4 } from '../lib/validation.js';
+import { requireUserId } from '../lib/request.js';
 
 export function createGetUserKeyHandler(ctx) {
   async function GET(url, headers) {
@@ -18,21 +17,9 @@ export function createGetUserKeyHandler(ctx) {
     }
 
     const { masterKey } = tenantResult.context;
-    const userId = getHeader(effectiveHeaders, 'x-user-id');
-
-    if (!userId) {
-      return {
-        status: 400,
-        body: { success: false, error: { code: 'USER_ID_REQUIRED', message: '缺少用户标识符' } }
-      };
-    }
-
-    if (!isValidUUIDv4(userId)) {
-      return {
-        status: 400,
-        body: { success: false, error: { code: 'INVALID_USER_ID_FORMAT', message: 'X-User-Id 必须是 UUID v4 格式' } }
-      };
-    }
+    const gate = requireUserId(effectiveHeaders);
+    if (gate.error) return gate.error;
+    const { userId } = gate;
 
     return {
       status: 200,
diff --git a/packages/rei-standard-amsg/server/src/server/handlers/messages.js b/packages/rei-standard-amsg/server/src/server/handlers/messages.js
index 8092963..09ab407 100644
--- a/packages/rei-standard-amsg/server/src/server/handlers/messages.js
+++ b/packages/rei-standard-amsg/server/src/server/handlers/messages.js
@@ -6,8 +6,7 @@
  */
 
 import { deriveUserEncryptionKey, decryptFromStorage, encryptPayload } from '../lib/encryption.js';
-import { getHeader } from '../lib/request.js';
-import { isValidUUIDv4 } from '../lib/validation.js';
+import { requireUserId } from '../lib/request.js';
 import { projectTask } from '../lib/task-projection.js';
 
 export function createMessagesHandler(ctx) {
@@ -20,20 +19,9 @@ export function createMessagesHandler(ctx) {
     const tenantCtx = tenantResult.context;
     const db = tenantCtx.db;
     const masterKey = tenantCtx.masterKey;
-    const userId = getHeader(headers, 'x-user-id');
-
-    if (!userId) {
-      return {
-        status: 400,
-        body: { success: false, error: { code: 'USER_ID_REQUIRED', message: '必须提供 X-User-Id 请求头' } }
-      };
-    }
-    if (!isValidUUIDv4(userId)) {
-      return {
-        status: 400,
-        body: { success: false, error: { code: 'INVALID_USER_ID_FORMAT', message: 'X-User-Id 必须是 UUID v4 格式' } }
-      };
-    }
+    const gate = requireUserId(headers);
+    if (gate.error) return gate.error;
+    const { userId } = gate;
 
     const u = new URL(url, 'https://dummy');
     const status = u.searchParams.get('status') || 'all';
diff --git a/packages/rei-standard-amsg/server/src/server/handlers/push-subscription.js b/packages/rei-standard-amsg/server/src/server/handlers/push-subscription.js
index c805103..bad2410 100644
--- a/packages/rei-standard-amsg/server/src/server/handlers/push-subscription.js
+++ b/packages/rei-standard-amsg/server/src/server/handlers/push-subscription.js
@@ -23,8 +23,7 @@
  */
 
 import { deriveUserEncryptionKey, decryptPayload } from '../lib/encryption.js';
-import { getHeader, isPlainObject, parseEncryptedBody } from '../lib/request.js';
-import { isValidUUIDv4 } from '../lib/validation.js';
+import { getHeader, isPlainObject, parseEncryptedBody, requireUserId } from '../lib/request.js';
 import {
   isPushSubscriptionShape,
   loadPushSubscription,
@@ -37,13 +36,6 @@ function err(status, code, message, details) {
   return { status, body: { success: false, error } };
 }
 
-function requireUserId(headers) {
-  const userId = getHeader(headers, 'x-user-id');
-  if (!userId) return { error: err(400, 'USER_ID_REQUIRED', '缺少用户标识符') };
-  if (!isValidUUIDv4(userId)) return { error: err(400, 'INVALID_USER_ID_FORMAT', 'X-User-Id 必须是 UUID v4 格式') };
-  return { userId };
-}
-
 const UNSUPPORTED = err(
   501,
   'PUSH_SUBSCRIPTION_NOT_SUPPORTED',
diff --git a/packages/rei-standard-amsg/server/src/server/handlers/schedule-message.js b/packages/rei-standard-amsg/server/src/server/handlers/schedule-message.js
index 8d92639..e505d36 100644
--- a/packages/rei-standard-amsg/server/src/server/handlers/schedule-message.js
+++ b/packages/rei-standard-amsg/server/src/server/handlers/schedule-message.js
@@ -8,9 +8,9 @@
 import { randomUUID } from '../lib/webcrypto-utils.js';
 import { deriveUserEncryptionKey, decryptPayload, encryptForStorage } from '../lib/encryption.js';
 import { isUniqueViolation } from '../lib/db-errors.js';
-import { getHeader, isPlainObject, parseEncryptedBody } from '../lib/request.js';
-import { validateScheduleMessagePayload, isValidUUIDv4 } from '../lib/validation.js';
-import { loadPushSubscription, supportsPushSubscriptionStore } from '../lib/push-subscription-store.js';
+import { getHeader, isPlainObject, parseEncryptedBody, requireUserId } from '../lib/request.js';
+import { validateScheduleMessagePayload } from '../lib/validation.js';
+import { supportsPushSubscriptionStore } from '../lib/push-subscription-store.js';
 import { processMessagesByUuid } from '../lib/message-processor.js';
 
 export function createScheduleMessageHandler(ctx) {
@@ -25,17 +25,13 @@ export function createScheduleMessageHandler(ctx) {
     const masterKey = tenantCtx.masterKey;
     const isEncrypted = getHeader(headers, 'x-payload-encrypted') === 'true';
     const encryptionVersion = getHeader(headers, 'x-encryption-version');
-    const userId = getHeader(headers, 'x-user-id');
 
     if (!isEncrypted) {
       return { status: 400, body: { success: false, error: { code: 'ENCRYPTION_REQUIRED', message: '请求体必须加密' } } };
     }
-    if (!userId) {
-      return { status: 400, body: { success: false, error: { code: 'USER_ID_REQUIRED', message: '缺少用户标识符' } } };
-    }
-    if (!isValidUUIDv4(userId)) {
-      return { status: 400, body: { success: false, error: { code: 'INVALID_USER_ID_FORMAT', message: 'X-User-Id 必须是 UUID v4 格式' } } };
-    }
+    const gate = requireUserId(headers);
+    if (gate.error) return gate.error;
+    const { userId } = gate;
     if (encryptionVersion !== '1') {
       return { status: 400, body: { success: false, error: { code: 'UNSUPPORTED_ENCRYPTION_VERSION', message: '加密版本不支持' } } };
     }
@@ -75,13 +71,25 @@ export function createScheduleMessageHandler(ctx) {
     if (!supportsPushSubscriptionStore(db)) {
       return { status: 501, body: { success: false, error: { code: 'PUSH_SUBSCRIPTION_NOT_SUPPORTED', message: '当前数据库适配器不支持用户级推送订阅存储' } } };
     }
-    let storedSubscription = null;
+    // 只做存在性检查（不解密——解出来的订阅这里也用不上）。查询本身失败是
+    // 基础设施问题，按可重试的 503 报出去，别伪装成「没登记」把客户端引去
+    // 走一遍多余的重订阅流程。
+    let subscriptionRow;
     try {
-      storedSubscription = await loadPushSubscription({ db, userId, userKey });
+      subscriptionRow = await db.getPushSubscription(userId);
     } catch (_error) {
-      storedSubscription = null;
+      return {
+        status: 503,
+        body: {
+          success: false,
+          error: {
+            code: 'PUSH_SUBSCRIPTION_LOOKUP_FAILED',
+            message: '推送订阅读取失败，请稍后重试'
+          }
+        }
+      };
     }
-    if (!storedSubscription) {
+    if (!subscriptionRow || typeof subscriptionRow.subscription !== 'string' || !subscriptionRow.subscription) {
       return {
         status: 409,
         body: {
diff --git a/packages/rei-standard-amsg/server/src/server/handlers/update-message.js b/packages/rei-standard-amsg/server/src/server/handlers/update-message.js
index 42da953..20769da 100644
--- a/packages/rei-standard-amsg/server/src/server/handlers/update-message.js
+++ b/packages/rei-standard-amsg/server/src/server/handlers/update-message.js
@@ -6,8 +6,8 @@
  */
 
 import { deriveUserEncryptionKey, decryptPayload, encryptForStorage, decryptFromStorage } from '../lib/encryption.js';
-import { getHeader, isPlainObject, parseEncryptedBody } from '../lib/request.js';
-import { isValidISO8601, isValidUUIDv4, isValidTimeZoneId, validateLlmMessagesArray, validateSplitPattern, validateAvatarUrl } from '../lib/validation.js';
+import { getHeader, isPlainObject, parseEncryptedBody, requireUserId } from '../lib/request.js';
+import { isValidISO8601, isValidTimeZoneId, validateLlmMessagesArray, validateSplitPattern, validateAvatarUrl } from '../lib/validation.js';
 
 export function createUpdateMessageHandler(ctx) {
   async function PUT(url, headers, body) {
@@ -26,13 +26,9 @@ export function createUpdateMessageHandler(ctx) {
       return { status: 400, body: { success: false, error: { code: 'TASK_ID_REQUIRED', message: '缺少任务ID' } } };
     }
 
-    const userId = getHeader(headers, 'x-user-id');
-    if (!userId) {
-      return { status: 400, body: { success: false, error: { code: 'USER_ID_REQUIRED', message: '缺少用户标识符' } } };
-    }
-    if (!isValidUUIDv4(userId)) {
-      return { status: 400, body: { success: false, error: { code: 'INVALID_USER_ID_FORMAT', message: 'X-User-Id 必须是 UUID v4 格式' } } };
-    }
+    const gate = requireUserId(headers);
+    if (gate.error) return gate.error;
+    const { userId } = gate;
 
     const isEncrypted = getHeader(headers, 'x-payload-encrypted') === 'true';
     const encryptionVersion = getHeader(headers, 'x-encryption-version');
@@ -204,7 +200,16 @@ export function createUpdateMessageHandler(ctx) {
     };
 
     const encryptedPayload = await encryptForStorage(JSON.stringify(updatedData), userKey);
-    const extraFields = updates.nextSendAt ? { next_send_at: updates.nextSendAt } : undefined;
+    // 更新即视为「重新出发」：把重试计数清零、把退避放掉。不清的话，刚修好
+    // apiKey / 改好排期的任务还背着之前攒下的 retry_count，下一次哪怕是瞬时
+    // 故障也可能直接触发终审处置。retry_after 只在支持占位的适配器上写
+    //（与 run-tick 的 updateAndRelease 同一判据——没实现 claimTask 的适配器
+    // 未必有这一列）。
+    const extraFields = {
+      retry_count: 0,
+      ...(typeof db.claimTask === 'function' ? { retry_after: null } : {}),
+      ...(updates.nextSendAt ? { next_send_at: updates.nextSendAt } : {})
+    };
 
     const result = await db.updateTaskByUuid(taskUuid, userId, encryptedPayload, extraFields);
 
diff --git a/packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js b/packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js
index 096a70e..d9e2f66 100644
--- a/packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js
+++ b/packages/rei-standard-amsg/server/src/server/lib/agentic-fire.js
@@ -848,11 +848,13 @@ async function sendHookPushPayloads({
     if (!ctx.vapid || !ctx.vapid.email || !ctx.vapid.publicKey || !ctx.vapid.privateKey) {
       throw new Error('VAPID configuration missing - push notifications cannot be sent');
     }
-    // 用户级订阅，投递时现读（任务行不携带它）。
+    // 用户级订阅，投递时现读（任务行不携带它）。升级前创建的任务把订阅冻结
+    // 在 payload 里，用户级存储没有时兜底用那一份。
     const pushSubscription = await resolvePushSubscription({
       db: ctx.db,
       userId: task.user_id,
       userKey,
+      legacyFallback: (decryptedPayload && decryptedPayload.pushSubscription) ?? null,
     });
     const messageIdBase = task.id != null ? `msg_task_${task.id}${occurrenceSuffix(task)}` : `msg_${randomUUID()}`;
 
diff --git a/packages/rei-standard-amsg/server/src/server/lib/message-processor.js b/packages/rei-standard-amsg/server/src/server/lib/message-processor.js
index 2626156..19fa701 100644
--- a/packages/rei-standard-amsg/server/src/server/lib/message-processor.js
+++ b/packages/rei-standard-amsg/server/src/server/lib/message-processor.js
@@ -97,17 +97,21 @@ function splitMessageIntoSentences(messageContent, splitPattern = null) {
  * @param {import('../adapters/interface.js').TaskRow} task
  * @param {ProcessorContext} ctx
  * @param {string} [providedMasterKey]
- * @returns {Promise<{ success: boolean, messagesSent: number, error?: string }>}
+ * @param {{ userKey: string, payload: Object } | null} [predecrypted] - 调用方
+ *   （run-tick 的预扫描）已经解好的 payload；传了就不再解第二遍。
+ * @returns {Promise<{ success: boolean, messagesSent: number, error?: string, errorCode?: string|null }>}
  */
-export async function processSingleMessage(task, ctx, providedMasterKey) {
+export async function processSingleMessage(task, ctx, providedMasterKey, predecrypted = null) {
   try {
     const masterKey = providedMasterKey || ctx.masterKey;
     if (!masterKey) {
       return { success: false, messagesSent: 0, error: 'TENANT_MASTER_KEY_MISSING' };
     }
 
-    const userKey = await deriveUserEncryptionKey(task.user_id, masterKey);
-    const decryptedPayload = JSON.parse(await decryptFromStorage(task.encrypted_payload, userKey));
+    const userKey = (predecrypted && predecrypted.userKey)
+      || await deriveUserEncryptionKey(task.user_id, masterKey);
+    const decryptedPayload = (predecrypted && predecrypted.payload)
+      || JSON.parse(await decryptFromStorage(task.encrypted_payload, userKey));
 
     // Fire-time hooks: when the host configured onBeforeFire and the task
     // needs the LLM, offer the agentic path first. onBeforeFire → null
@@ -170,8 +174,15 @@ export async function processSingleMessage(task, ctx, providedMasterKey) {
     }
 
     // 订阅是用户级的一份，投递时现读（任务行不携带它）。取不到就抛，走既有
-    // 的失败/重试逻辑——静默不发会让任务「成功」地什么都没做。
-    const pushSubscription = await resolvePushSubscription({ db: ctx.db, userId: task.user_id, userKey });
+    // 的失败/重试逻辑——静默不发会让任务「成功」地什么都没做。升级前创建的
+    // 任务把订阅冻结在 payload 里，用户级存储没有时兜底用那一份，存量任务
+    // 不必等用户重新登记。
+    const pushSubscription = await resolvePushSubscription({
+      db: ctx.db,
+      userId: task.user_id,
+      userKey,
+      legacyFallback: decryptedPayload.pushSubscription ?? null
+    });
     // sessionId is shared across the optional ReasoningPush and every
     // ContentPush from this LLM round. Pin it to (task id + 名义触发时刻)
     // when available (scheduled tasks) so retries of the same occurrence
@@ -248,7 +259,9 @@ export async function processSingleMessage(task, ctx, providedMasterKey) {
     return { success: true, messagesSent: messages.length };
 
   } catch (error) {
-    return { success: false, messagesSent: 0, error: error.message };
+    // errorCode 透传底层错误的稳定 `code`（如 PUSH_SUBSCRIPTION_MISSING），
+    // run-tick 按它区分「重试也好不了」的永久性失败。
+    return { success: false, messagesSent: 0, error: error.message, errorCode: error.code || null };
   }
 }
 
diff --git a/packages/rei-standard-amsg/server/src/server/lib/push-subscription-store.js b/packages/rei-standard-amsg/server/src/server/lib/push-subscription-store.js
index bdc2a65..7a99034 100644
--- a/packages/rei-standard-amsg/server/src/server/lib/push-subscription-store.js
+++ b/packages/rei-standard-amsg/server/src/server/lib/push-subscription-store.js
@@ -69,24 +69,41 @@ export async function loadPushSubscription({ db, userId, userKey }) {
   return { subscription, updatedAt: row.updated_at ?? null };
 }
 
+/** 带稳定 `code` 属性的错误：消费方按 error.code 分支，不用字符串匹配 message。 */
+function codedError(code, message) {
+  const error = new Error(`${code}: ${message}`);
+  error.code = code;
+  return error;
+}
+
 /**
  * 投递前取订阅。取不到就抛——静默不发会让任务「成功」地什么都没做，用户
  * 只看到消息凭空消失。抛出去走既有的重试 / 标记逻辑，原因也会记进 payload
- * 的 lastError，`GET /messages` 上看得见。
+ * 的 lastError，`GET /messages` 上看得见。抛出的错误带稳定的 `code` 属性
+ * （'PUSH_SUBSCRIPTION_MISSING' / 'PUSH_SUBSCRIPTION_STORE_UNSUPPORTED'），
+ * 按类别分支请用它，别匹配 message 文案。
+ *
+ * `legacyFallback`：用户级存储里没有订阅时的兜底（升级前创建的任务把订阅
+ * 冻结在自己的 payload 里，这份订阅仍然有效）。存储里有订阅时永远用存储的
+ * 那份——它是用户最近一次登记的。
  *
  * @param {Object} args
  * @param {import('../adapters/interface.js').DbAdapter} args.db
  * @param {string} args.userId
  * @param {string} args.userKey
+ * @param {unknown} [args.legacyFallback] - 旧任务 payload 里内嵌的订阅（可选）
  * @returns {Promise<Object>} 明文订阅对象
  */
-export async function resolvePushSubscription({ db, userId, userKey }) {
+export async function resolvePushSubscription({ db, userId, userKey, legacyFallback = null }) {
+  const fallback = isPushSubscriptionShape(legacyFallback) ? legacyFallback : null;
   if (!supportsPushSubscriptionStore(db)) {
-    throw new Error('PUSH_SUBSCRIPTION_STORE_UNSUPPORTED: 当前数据库适配器不支持用户级推送订阅存储');
+    if (fallback) return fallback;
+    throw codedError('PUSH_SUBSCRIPTION_STORE_UNSUPPORTED', '当前数据库适配器不支持用户级推送订阅存储');
   }
   const stored = await loadPushSubscription({ db, userId, userKey });
   if (!stored) {
-    throw new Error('PUSH_SUBSCRIPTION_MISSING: 该用户还没有登记推送订阅（PUT /push-subscription）');
+    if (fallback) return fallback;
+    throw codedError('PUSH_SUBSCRIPTION_MISSING', '该用户还没有登记推送订阅（PUT /push-subscription）');
   }
   return stored.subscription;
 }
diff --git a/packages/rei-standard-amsg/server/src/server/lib/request.js b/packages/rei-standard-amsg/server/src/server/lib/request.js
index f225d8b..7222e8c 100644
--- a/packages/rei-standard-amsg/server/src/server/lib/request.js
+++ b/packages/rei-standard-amsg/server/src/server/lib/request.js
@@ -3,6 +3,8 @@
  * Keeps body parsing and shape validation consistent across handlers.
  */
 
+import { isValidUUIDv4 } from './validation.js';
+
 export const REQUEST_ERRORS = {
   INVALID_JSON: { code: 'INVALID_JSON', message: '请求体不是有效的 JSON' },
   INVALID_REQUEST_BODY: { code: 'INVALID_REQUEST_BODY', message: '请求体格式无效' },
@@ -119,6 +121,33 @@ export function parseEncryptedBody(body) {
   return parsedBody;
 }
 
+/**
+ * 标准错误信封：{ status, body: { success: false, error: { code, message, details? } } }。
+ *
+ * @param {number} status
+ * @param {string} code
+ * @param {string} message
+ * @param {Object} [details]
+ */
+export function errorResponse(status, code, message, details) {
+  const error = details === undefined ? { code, message } : { code, message, details };
+  return { status, body: { success: false, error } };
+}
+
+/**
+ * X-User-Id 门禁。所有按用户读写的端点共用这一份：规则（必填 + UUID v4）和
+ * 文案只此一处，改口径不用挨个 handler 找复制粘贴的副本。
+ *
+ * @param {Record<string, any>} headers
+ * @returns {{ userId: string, error?: undefined } | { error: ReturnType<typeof errorResponse> }}
+ */
+export function requireUserId(headers) {
+  const userId = getHeader(headers, 'x-user-id');
+  if (!userId) return { error: errorResponse(400, 'USER_ID_REQUIRED', '缺少用户标识符') };
+  if (!isValidUUIDv4(userId)) return { error: errorResponse(400, 'INVALID_USER_ID_FORMAT', 'X-User-Id 必须是 UUID v4 格式') };
+  return { userId };
+}
+
 /**
  * Read a header value case-insensitively.
  *
diff --git a/packages/rei-standard-amsg/server/src/server/lib/run-tick.js b/packages/rei-standard-amsg/server/src/server/lib/run-tick.js
index 3456213..d1539be 100644
--- a/packages/rei-standard-amsg/server/src/server/lib/run-tick.js
+++ b/packages/rei-standard-amsg/server/src/server/lib/run-tick.js
@@ -21,7 +21,7 @@
  * 等租约到期才会被接手——这是租约本身的代价，把租期设得比最慢的一次投递长
  * 一点就行。
  *
- * @param {Object} ctx - { db, masterKey, vapid, webpush, claimLeaseMs?, serializeBy?, onStaleSkip? }
+ * @param {Object} ctx - { db, masterKey, vapid, webpush, claimLeaseMs?, staleAfterMs?, serializeBy?, onStaleSkip? }
  *   ctx.serializeBy?.(task)：可选的分组串行。返回一个分组标识（同一个角色、
  *   同一份台账……宿主自己定义），同一分组的任务同时只放行一条；返回 null /
  *   空串 / 不配这个函数 = 这条任务不参与串行，行为与以前完全一致。参数是与
@@ -56,9 +56,15 @@ const CLAIM_LEASE_MARGIN_MS = 2 * 60 * 1000;
 // 补发新鲜度：错过名义触发时刻超过这个时长的任务不再照常补发。服务停摆几天
 // 恢复后，一次性任务不该把攒了几天的旧话一口气倒出来，循环任务更不该每分钟
 // 补发一天。正在重试链上的任务（retry_count > 0）不算过期——它的 next_send_at
-// 一直是名义时刻，重试拖过一小时不等于用户错过了它。
+// 一直是名义时刻，重试拖过一小时不等于用户错过了它——但重试时刻（retry_after）
+// 本身也被拖过了这个时长的除外：那说明停摆发生在重试窗口里，内容一样旧。
+// 宿主可用 ctx.staleAfterMs 覆盖（与 claimLeaseMs 同一模式）。
 export const STALE_AFTER_MS = 60 * 60 * 1000;
 
+function resolveStaleAfterMs(ctx) {
+  return positiveNumber(ctx.staleAfterMs) || STALE_AFTER_MS;
+}
+
 /**
  * @typedef {Object} StaleSkipInfo
  * @property {'stale'} reason
@@ -112,6 +118,7 @@ export async function runScheduledTick(ctx) {
   const db = ctx.db;
   const masterKey = ctx.masterKey;
   const claimLeaseMs = resolveClaimLeaseMs(ctx);
+  const staleAfterMs = resolveStaleAfterMs(ctx);
   const serializeBy = typeof ctx.serializeBy === 'function' ? ctx.serializeBy : null;
 
   const startTime = Date.now();
@@ -192,8 +199,12 @@ export async function runScheduledTick(ctx) {
       return { taken: true, rawKey: null };
     }
     if (typeof rawKey !== 'string' || !rawKey) return { taken: false, rawKey: null };
-    if (groupsTakenThisTick.has(rawKey)) return { taken: true, rawKey };
-    groupsTakenThisTick.add(rawKey);
+    // 占坑的键带上 user_id：分组是按用户隔离的（落库那侧靠 per-user HMAC 天然
+    // 隔开，见 deriveSerializeGroup），内存这侧不带用户维度的话，两个用户恰好
+    // 返回同一个 rawKey（例如共用的默认角色名）会互相顶掉对方的任务。
+    const scopedKey = `${task.user_id}\u0000${rawKey}`;
+    if (groupsTakenThisTick.has(scopedKey)) return { taken: true, rawKey };
+    groupsTakenThisTick.add(scopedKey);
     return { taken: false, rawKey };
   }
 
@@ -250,11 +261,16 @@ export async function runScheduledTick(ctx) {
    * 久漂了。没有这两列的适配器（没实现 claimTask）退回老行为，把重试时刻写
    * 进 next_send_at。
    */
-  async function handleDeliveryFailure(task, reason, recurrenceType, decryptedPayload, userKey) {
+  async function handleDeliveryFailure(task, reason, recurrenceType, decryptedPayload, userKey, errorCode = null) {
     results.failedCount++;
     const tzId = decryptedPayload ? (decryptedPayload.tzId ?? null) : null;
+    // 重试也好不了的失败（订阅没登记 / 适配器不支持订阅存储）不进退避阶梯：
+    // 一次性任务直接进终审处置，循环任务直接作废本次 occurrence——在用户重新
+    // 登记订阅之前，隔两分钟再试三次只是白跑。
+    const permanent = errorCode === 'PUSH_SUBSCRIPTION_MISSING'
+      || errorCode === 'PUSH_SUBSCRIPTION_STORE_UNSUPPORTED';
     try {
-      if (task.retry_count >= 3) {
+      if (permanent || task.retry_count >= 3) {
         const encrypted = await encryptPayloadWithLastError(task, decryptedPayload, userKey, reason);
         if (isRecurringType(recurrenceType)) {
           const nextSendAt = nextFutureOccurrence(Date.parse(task.next_send_at), recurrenceType, Date.now(), tzId);
@@ -362,9 +378,8 @@ export async function runScheduledTick(ctx) {
       return;
     }
 
-    // recurrenceType 是过期判定、终审失败处置的依据。解密不了的话投递也必然
-    // 失败（processSingleMessage 解的是同一份），按投递失败走既有的重试/终态
-    // 逻辑。
+    // recurrenceType 是过期判定、终审失败处置的依据。解密不了的话投递也无从
+    // 谈起（投递用的就是这份解密结果），按投递失败走既有的重试/终态逻辑。
     if (!decrypted.ok) {
       await handleDeliveryFailure(task, (decrypted.error && decrypted.error.message) || '任务载荷解密失败', null, null, null);
       return;
@@ -377,12 +392,17 @@ export async function runScheduledTick(ctx) {
     // 服务停摆 N 天恢复后，攒下的旧任务不照常补发：一次性任务标 'failed'；
     // 循环任务把排期快进到未来第一个名义时刻。两边都写 lastError、都调
     // onStaleSkip——「昨天那次没响」对用户是一样的事实，循环任务不该无声无息
-    // 地把它抹掉。正在重试链上的（retry_count > 0）不算过期，见 STALE_AFTER_MS
-    // 的注释。
+    // 地把它抹掉。正在重试链上的（retry_count > 0）不算过期——除非排定的重试
+    // 时刻（retry_after）本身也被拖过了 staleAfterMs：正常退避只有几分钟，
+    // 拖过一小时说明停摆恰好落在重试窗口里，这时内容跟没进重试链的一样旧。
+    // 见 STALE_AFTER_MS 的注释。
     const occurrenceMs = Date.parse(task.next_send_at);
+    const retryAfterMs = task.retry_after ? Date.parse(task.retry_after) : NaN;
+    const notOnFreshRetryChain = (task.retry_count || 0) === 0
+      || (Number.isFinite(retryAfterMs) && Date.now() - retryAfterMs > staleAfterMs);
     if (Number.isFinite(occurrenceMs)
-        && Date.now() - occurrenceMs > STALE_AFTER_MS
-        && (task.retry_count || 0) === 0) {
+        && Date.now() - occurrenceMs > staleAfterMs
+        && notOnFreshRetryChain) {
       try {
         // hook 的 client_state 读写口：过期跳过往往正是宿主要留一条痕迹的时
         // 候（服务停摆恢复后的第一跳，此前这个 tick 里可能一次 fire 都没跑
@@ -394,62 +414,47 @@ export async function runScheduledTick(ctx) {
           maxStateValueBytes: ctx.maxStateValueBytes
         });
 
-        if (isRecurringType(recurrenceType)) {
-          const plan = planNextOccurrence(occurrenceMs, recurrenceType, Date.now(), tzId);
-          const nextSendAt = new Date(plan.nextMs).toISOString();
-          const encrypted = await encryptPayloadWithLastError(task, decryptedPayload, userKey, 'stale', {
-            skippedCount: plan.skippedCount,
-            nextSendAt
-          });
-          await updateAndRelease(task.id, {
-            next_send_at: nextSendAt,
-            retry_count: 0,
-            ...(encrypted ? { encrypted_payload: encrypted } : {})
-          });
-          results.staleTasks.push({
-            taskId: task.id,
-            reason: 'stale',
-            action: 'fast_forwarded',
-            nextSendAt,
-            skippedCount: plan.skippedCount
-          });
-          await notifyStaleSkip(task, {
-            reason: 'stale',
-            action: 'fast_forwarded',
-            metadata: decryptedPayload.metadata ?? null,
-            recurrenceType,
-            occurrenceMs,
-            skippedCount: plan.skippedCount,
-            skippedOccurrences: plan.skippedOccurrences,
-            skippedTruncated: plan.skippedTruncated,
-            nextSendAt,
-            ...stateAccessors
-          });
-        } else {
-          const encrypted = await encryptPayloadWithLastError(task, decryptedPayload, userKey, 'stale');
-          await updateAndRelease(task.id, {
-            status: 'failed',
-            ...(encrypted ? { encrypted_payload: encrypted } : {})
-          });
-          results.staleTasks.push({ taskId: task.id, reason: 'stale', action: 'expired', skippedCount: 1 });
-          // 消费方用它写「错过了」回执；best-effort，hook 抛错不影响主流程。
-          // task 是数据库行原样（metadata 锁在 encrypted_payload 密文里），所以
-          // 把解密出的 metadata 单独递过去，hook 才能对上是哪个角色的任务。
-          // 只透传 metadata 这一个子字段——解密 payload 里还有 apiKey /
-          // pushSubscription 等凭据，不能整个递出去。
-          await notifyStaleSkip(task, {
-            reason: 'stale',
-            action: 'expired',
-            metadata: decryptedPayload.metadata ?? null,
-            recurrenceType: recurrenceType || 'none',
-            occurrenceMs,
-            skippedCount: 1,
-            skippedOccurrences: [occurrenceMs],
-            skippedTruncated: false,
-            nextSendAt: null,
-            ...stateAccessors
-          });
-        }
+        // 循环任务快进（fast_forwarded）与一次性任务作废（expired）的收尾
+        // 完全同构：写 lastError → 写库 → 记 staleTasks → 调 onStaleSkip。
+        // 差异全部收进 plan 里，尾部只写一遍——两个近似复制的分支各改各的，
+        // 漏改一边是迟早的事。
+        const recurring = isRecurringType(recurrenceType);
+        const plan = recurring ? planNextOccurrence(occurrenceMs, recurrenceType, Date.now(), tzId) : null;
+        const nextSendAt = recurring ? new Date(plan.nextMs).toISOString() : null;
+        const action = recurring ? 'fast_forwarded' : 'expired';
+        const skippedCount = recurring ? plan.skippedCount : 1;
+        const encrypted = await encryptPayloadWithLastError(
+          task, decryptedPayload, userKey, 'stale',
+          recurring ? { skippedCount, nextSendAt } : undefined
+        );
+        await updateAndRelease(task.id, {
+          ...(recurring ? { next_send_at: nextSendAt, retry_count: 0 } : { status: 'failed' }),
+          ...(encrypted ? { encrypted_payload: encrypted } : {})
+        });
+        results.staleTasks.push({
+          taskId: task.id,
+          reason: 'stale',
+          action,
+          ...(recurring ? { nextSendAt } : {}),
+          skippedCount
+        });
+        // 消费方用它写「错过了」回执；best-effort，hook 抛错不影响主流程。
+        // task 是数据库行原样（metadata 锁在 encrypted_payload 密文里），所以
+        // 把解密出的 metadata 单独递过去，hook 才能对上是哪个角色的任务。
+        // 只透传 metadata 这一个子字段——解密 payload 里还有 apiKey /
+        // pushSubscription 等凭据，不能整个递出去。
+        await notifyStaleSkip(task, {
+          reason: 'stale',
+          action,
+          metadata: decryptedPayload.metadata ?? null,
+          recurrenceType: recurrenceType || 'none',
+          occurrenceMs,
+          skippedCount,
+          skippedOccurrences: recurring ? plan.skippedOccurrences : [occurrenceMs],
+          skippedTruncated: recurring ? plan.skippedTruncated : false,
+          nextSendAt,
+          ...stateAccessors
+        });
       } catch (error) {
         results.failedCount++;
         results.failedTasks.push({ taskId: task.id, reason: error.message || '过期任务处理失败', status: 'stale_update_failed' });
@@ -459,14 +464,18 @@ export async function runScheduledTick(ctx) {
 
     let sendResult;
     try {
-      sendResult = await processSingleMessage(task, { ...ctx, db, masterKey }, masterKey);
+      // 预扫描解好的 payload 一并递过去，投递侧不再解第二遍。
+      sendResult = await processSingleMessage(
+        task, { ...ctx, db, masterKey }, masterKey,
+        { userKey, payload: decryptedPayload }
+      );
     } catch (error) {
-      await handleDeliveryFailure(task, error.message || '消息发送失败', recurrenceType, decryptedPayload, userKey);
+      await handleDeliveryFailure(task, error.message || '消息发送失败', recurrenceType, decryptedPayload, userKey, error.code || null);
       return;
     }
 
     if (!sendResult.success) {
-      await handleDeliveryFailure(task, sendResult.error || '消息发送失败', recurrenceType, decryptedPayload, userKey);
+      await handleDeliveryFailure(task, sendResult.error || '消息发送失败', recurrenceType, decryptedPayload, userKey, sendResult.errorCode || null);
       return;
     }
 
@@ -493,8 +502,9 @@ export async function runScheduledTick(ctx) {
   // 一条来」：并发解密谁先跑完是没准的，让它来决定同组谁先放行的话，两条都到
   // 点时晚的那条可能抢在早的前面发出去——这正是分组串行想避免的事。
   //
-  // 没配 serializeBy 时这一步只是顺手把 payload 解出来（下面不会再解第二遍），
-  // 谁都不拦。
+  // 没配 serializeBy 时这一步只是顺手把 payload 解出来，谁都不拦。解好的
+  // payload 会随任务一路递到投递侧（processSingleMessage 的 predecrypted
+  // 参数），全程只解这一遍。
   const taskQueue = [];
   for (const task of tasks) {
     const decrypted = await decryptTask(task);
diff --git a/packages/rei-standard-amsg/server/src/server/lib/validation.js b/packages/rei-standard-amsg/server/src/server/lib/validation.js
index 23c185a..2214968 100644
--- a/packages/rei-standard-amsg/server/src/server/lib/validation.js
+++ b/packages/rei-standard-amsg/server/src/server/lib/validation.js
@@ -2,11 +2,16 @@
  * Validation utility library (SDK version)
  */
 
-import { validateAvatarUrl, validateLlmMessagesShape } from '@rei-standard/amsg-shared';
+import { isValidUrl, validateAvatarUrl, validateLlmMessagesShape } from '@rei-standard/amsg-shared';
 import { isValidTimeZoneId } from './recurrence.js';
 
 export { isValidTimeZoneId };
 
+// URL 校验统一走 shared 的实现（validateAvatarUrl 内部用的也是同一份），
+// 两个包对「什么算 URL」不再各持一版。此处重导出，保持本模块及
+// `createReiServer` 的公开导出不变。
+export { isValidUrl };
+
 /**
  * Validate ISO 8601 date string.
  * @param {string} dateString
@@ -17,20 +22,6 @@ export function isValidISO8601(dateString) {
   return date instanceof Date && !isNaN(date.getTime());
 }
 
-/**
- * Validate URL format.
- * @param {string} urlString
- * @returns {boolean}
- */
-export function isValidUrl(urlString) {
-  try {
-    new URL(urlString);
-    return true;
-  } catch {
-    return false;
-  }
-}
-
 /**
  * Validate UUID format.
  * @param {string} uuid
diff --git a/packages/rei-standard-amsg/server/src/server/tenant/blob-store.js b/packages/rei-standard-amsg/server/src/server/tenant/blob-store.js
index 21e672d..5db7ca5 100644
--- a/packages/rei-standard-amsg/server/src/server/tenant/blob-store.js
+++ b/packages/rei-standard-amsg/server/src/server/tenant/blob-store.js
@@ -1,23 +1,13 @@
 import { createCipheriv, createDecipheriv, createHash, randomBytes } from 'crypto';
+// base64url 编解码统一用 shared 的实现（经 lib/webcrypto-utils.js 重导出），
+// 与同目录 token.js 同一份——本文件其余部分仍是刻意的 Node-only（node:crypto）。
+import {
+  bytesToBase64Url as base64UrlEncode,
+  base64UrlToBytes as base64UrlDecode,
+} from '../lib/webcrypto-utils.js';
 
 const inMemoryNamespaces = new Map();
 
-function base64UrlEncode(input) {
-  return Buffer.from(input)
-    .toString('base64')
-    .replace(/\+/g, '-')
-    .replace(/\//g, '_')
-    .replace(/=+$/g, '');
-}
-
-function base64UrlDecode(input) {
-  const normalized = input
-    .replace(/-/g, '+')
-    .replace(/_/g, '/');
-  const padLength = (4 - (normalized.length % 4)) % 4;
-  return Buffer.from(normalized + '='.repeat(padLength), 'base64');
-}
-
 function getKekBuffer(kek) {
   const value = String(kek || '').trim();
   if (!value) {
diff --git a/packages/rei-standard-amsg/server/src/server/tenant/context.js b/packages/rei-standard-amsg/server/src/server/tenant/context.js
index 9456a59..3b1c818 100644
--- a/packages/rei-standard-amsg/server/src/server/tenant/context.js
+++ b/packages/rei-standard-amsg/server/src/server/tenant/context.js
@@ -57,8 +57,22 @@ export function createTenantContextManager(options) {
   async function getOrCreateAdapter(dbConfig) {
     const cacheKey = `${dbConfig.driver}:${dbConfig.connectionString}`;
     if (!adapterCache.has(cacheKey)) {
-      const adapter = await adapterFactory(dbConfig);
-      adapterCache.set(cacheKey, adapter);
+      // 缓存的是 promise：并发首次请求只建一次；失败则清掉缓存，下次重试。
+      const created = (async () => {
+        const adapter = await adapterFactory(dbConfig);
+        // 每个进程首次拿到适配器时把 schema 补齐到最新（建表 / ADD COLUMN IF
+        // NOT EXISTS，全部幂等）。存量租户没有别的路径重跑 initSchema——同
+        // tenantId 重放 /init-tenant 在到达 initSchema 前就 409——不在这里补，
+        // 升级加列后每个 tick 都会因缺列报错，直到运维手工执行 DDL。
+        if (typeof adapter.initSchema === 'function') {
+          await adapter.initSchema();
+        }
+        return adapter;
+      })().catch((error) => {
+        adapterCache.delete(cacheKey);
+        throw error;
+      });
+      adapterCache.set(cacheKey, created);
     }
     return adapterCache.get(cacheKey);
   }
diff --git a/packages/rei-standard-amsg/server/test/pg-neon-adapter.test.mjs b/packages/rei-standard-amsg/server/test/pg-neon-adapter.test.mjs
index 5b759c7..bb81759 100644
--- a/packages/rei-standard-amsg/server/test/pg-neon-adapter.test.mjs
+++ b/packages/rei-standard-amsg/server/test/pg-neon-adapter.test.mjs
@@ -83,14 +83,19 @@ for (const backend of BACKENDS) {
 
   // 分组串行：判定和占位必须在同一条语句里。拆成「先查这组忙不忙、再占位」的
   // 话，两个 tick 的查询都会在对方占位之前返回「不忙」，双双进同一个分组。
+  // 占位提交之后另有一次复查（READ COMMITTED 写偏斜的收口，见 pg-shared.js）：
+  // 两个 tick 各领同组「不同」行时 NOT EXISTS 互相看不见对方未提交的租约，
+  // 复查发生在自己的租约提交之后，至少一方看得见冲突并让路。
   test(`${backend.name}: 带 serializeGroup 时，分组判定和占位是同一条 UPDATE`, async () => {
-    const { adapter, calls } = backend.make(() => [{ id: 1 }]);
+    const { adapter, calls } = backend.make((text) =>
+      /^\s*SELECT 1 FROM scheduled_messages busy/i.test(text) ? [] : [{ id: 1 }]
+    );
     const lease = '2026-01-01T00:10:00.000Z';
 
     assert.equal(await adapter.claimTask(7, '2026-01-01T00:00:00.000Z', lease, 'grp-abc'), true);
 
-    assert.equal(calls.length, 1, '只发一条语句，不能先查一次再占一次');
-    const { text, params } = calls.at(-1);
+    assert.equal(calls.length, 2, '判定+占位合一的 UPDATE，加一次占位提交后的复查');
+    const { text, params } = calls[0];
     // 子查询自带 WHERE，这里就不切段了，直接对整条语句断言。
     const sql = flat(text);
     // 同一分组里有别的行拿着未到期的租约就领不走。
@@ -102,6 +107,29 @@ for (const backend of BACKENDS) {
     // 领到的同时把分组写在行上，下一跳靠它判断这组忙不忙。
     assert.match(setClauseOf(text), /serialize_group\s*=\s*\$4/);
     assert.deepEqual(params, [lease, 7, '2026-01-01T00:00:00.000Z', 'grp-abc']);
+    // 复查只看同组其他行的活租约。
+    const verify = flat(calls[1].text);
+    assert.match(verify, /^SELECT 1 FROM scheduled_messages busy/i);
+    assert.match(verify, /busy\.lease_until > NOW\(\)/i);
+    assert.deepEqual(calls[1].params, ['grp-abc', 7]);
+  });
+
+  test(`${backend.name}: 占位后复查撞上同组另一条活租约 → 放掉租约、返回 false`, async () => {
+    const { adapter, calls } = backend.make((text) =>
+      /^\s*SELECT 1 FROM scheduled_messages busy/i.test(text) ? [{ ok: 1 }] : [{ id: 1 }]
+    );
+
+    assert.equal(
+      await adapter.claimTask(7, '2026-01-01T00:00:00.000Z', '2026-01-01T00:10:00.000Z', 'grp-abc'),
+      false
+    );
+
+    // 最后一条语句把自己刚写的租约放掉（两边都让路也没事：行保持 pending，
+    // 下一跳重试）。
+    const release = flat(calls.at(-1).text);
+    assert.match(release, /SET lease_until = NULL/i);
+    assert.match(release, /WHERE id = \$1/i);
+    assert.deepEqual(calls.at(-1).params, [7]);
   });
 
   test(`${backend.name}: 不传 serializeGroup 时语句里没有分组门`, async () => {
diff --git a/packages/rei-standard-amsg/shared/src/webpush.js b/packages/rei-standard-amsg/shared/src/webpush.js
index d7e4a35..c937152 100644
--- a/packages/rei-standard-amsg/shared/src/webpush.js
+++ b/packages/rei-standard-amsg/shared/src/webpush.js
@@ -26,6 +26,7 @@ import {
   concatBytes,
   base64UrlToBytes,
   utf8,
+  utf8Decode,
   bytesToBase64Url,
   jsonToBase64Url,
   randomBytes,
@@ -329,7 +330,7 @@ export async function verifyVapidJwt(jwt, publicKey) {
   );
   if (!ok) throw new Error('VAPID JWT: signature mismatch');
 
-  const payload = JSON.parse(new TextDecoder().decode(base64UrlToBytes(p)));
+  const payload = JSON.parse(utf8Decode(base64UrlToBytes(p)));
   if (!payload.exp || payload.exp <= Math.floor(Date.now() / 1000)) {
     throw new Error('VAPID JWT: expired');
   }