merge: fix/wechat-mp-perf

user-auth.mjs

@@ -2274,6 +2274,18 @@ export function createUserAuth(pool, options = {}) {
     return Number(rows?.[0]?.count ?? 0);
   };
 
+  const getWechatAgentSessionSnapshotMessageCount = async (agentSessionId) => {
+    if (!agentSessionId) return 0;
+    const [rows] = await pool.query(
+      `SELECT synced_msg_count
+       FROM h5_session_snapshots
+       WHERE agent_session_id = ?
+       LIMIT 1`,
+      [agentSessionId],
+    );
+    return Number(rows?.[0]?.synced_msg_count ?? 0);
+  };
+
   const recordWechatMpMessage = async ({
     appId,
     openid,
@@ -2861,6 +2873,7 @@ export function createUserAuth(pool, options = {}) {
     clearWechatAgentRoute,
     touchWechatAgentRoute,
     countWechatAgentSessionMessages,
+    getWechatAgentSessionSnapshotMessageCount,
     recordWechatMpMessage,
     finishWechatMpMessage,
     insertWechatMpMessageDetail,

wechat-mp.mjs

@@ -1683,10 +1683,12 @@ export function createWechatMpService({
         config.sessionIdleRotateMs > 0 &&
         routeUpdatedAt > 0 &&
         now - routeUpdatedAt > config.sessionIdleRotateMs;
-      const routeMessageCount =
-        config.sessionMessageRotateCount > 0 &&
-        typeof userAuth.countWechatAgentSessionMessages === 'function'
-          ? await userAuth
+      let routeMessageCount = 0;
+      if (config.sessionMessageRotateCount > 0) {
+        const counts = [];
+        if (typeof userAuth.countWechatAgentSessionMessages === 'function') {
+          counts.push(
+            userAuth
               .countWechatAgentSessionMessages({
                 appId: config.appId,
                 openid,
@@ -1695,8 +1697,24 @@ export function createWechatMpService({
               .catch((err) => {
                 logger.warn?.('WeChat MP route message count lookup failed:', err);
                 return 0;
-              })
-          : 0;
+              }),
+          );
+        }
+        if (typeof userAuth.getWechatAgentSessionSnapshotMessageCount === 'function') {
+          counts.push(
+            userAuth
+              .getWechatAgentSessionSnapshotMessageCount(existingRoute.agentSessionId)
+              .catch((err) => {
+                logger.warn?.('WeChat MP snapshot message count lookup failed:', err);
+                return 0;
+              }),
+          );
+        }
+        if (counts.length > 0) {
+          const resolved = await Promise.all(counts);
+          routeMessageCount = Math.max(0, ...resolved.map((value) => Number(value) || 0));
+        }
+      }
       const routeIsTooLong =
         config.sessionMessageRotateCount > 0 &&
         Number(routeMessageCount ?? 0) >= config.sessionMessageRotateCount;
@@ -1807,6 +1825,10 @@ export function createWechatMpService({
     }
   };
 
+  const scheduleWechatSessionSnapshotRefresh = (sessionId, userId) => {
+    void refreshWechatSessionSnapshot(sessionId, userId);
+  };
+
   const rememberWechatUserContext = async (sessionId, user, { forceBootstrap = false } = {}) => {
     const addressName = resolveWechatAddressName(user);
     if (!addressName) return;
@@ -1984,7 +2006,7 @@ export function createWechatMpService({
         publicBaseUrl: config.publicBaseUrl,
         artifacts: verifiedArtifacts.length > 0 ? verifiedArtifacts : confirmedArtifacts,
       });
-      await refreshWechatSessionSnapshot(sessionId, user.userId);
+      scheduleWechatSessionSnapshotRefresh(sessionId, user.userId);
       await sendCustomerServiceText(inbound.fromUserName, await guardScheduleReply(finalizedReply), user, {
         verifiedHtmlUrls: (verifiedArtifacts.length > 0 ? verifiedArtifacts : confirmedArtifacts).map(
           (artifact) => artifact.url,
@@ -2053,7 +2075,7 @@ export function createWechatMpService({
           publicBaseUrl: config.publicBaseUrl,
           artifacts: verifiedArtifacts.length > 0 ? verifiedArtifacts : confirmedArtifacts,
         });
-        await refreshWechatSessionSnapshot(sessionId, user.userId);
+        scheduleWechatSessionSnapshotRefresh(sessionId, user.userId);
         await sendCustomerServiceText(inbound.fromUserName, await guardScheduleReply(finalizedReply), user, {
           verifiedHtmlUrls: (verifiedArtifacts.length > 0 ? verifiedArtifacts : confirmedArtifacts).map(
             (artifact) => artifact.url,

wechat-mp.test.mjs

@@ -1424,6 +1424,90 @@ test('wechat mp service rotates a dedicated session after the message-count thre
   assert.equal(routeCleared, true);
 });
 
+test('wechat mp service rotates a dedicated session when snapshot message count exceeds threshold', async () => {
+  const token = 'token';
+  const timestamp = '1710000000';
+  const nonce = 'nonce';
+  let routeCleared = false;
+  const service = createBoundWechatService({
+    token,
+    config: {
+      sessionIdleRotateMs: 0,
+      sessionMessageRotateCount: 200,
+    },
+    startAgentSession: async () => ({ id: 'session-2' }),
+    sessionApiFetch: async (sessionId, pathname) => {
+      assert.equal(sessionId, 'session-2');
+      if (pathname === '/sessions/session-2/events') {
+        return new Response(
+          [
+            'data: {"type":"Message","request_id":"req-snapshot-rotate","message":{"id":"assistant-1","role":"assistant","metadata":{"userVisible":true},"content":[{"type":"text","text":"已换新会话。"}]}}\n\n',
+            'data: {"type":"Finish","request_id":"req-snapshot-rotate","token_state":{"inputTokens":1,"outputTokens":2}}\n\n',
+          ].join(''),
+          { status: 200, headers: { 'Content-Type': 'text/event-stream' } },
+        );
+      }
+      if (pathname === '/sessions/session-2/reply') {
+        return new Response('{}', { status: 200, headers: { 'Content-Type': 'application/json' } });
+      }
+      if (pathname === '/agent/harness_remember' || pathname === '/agent/harness_bootstrap') {
+        return new Response('{}', { status: 200, headers: { 'Content-Type': 'application/json' } });
+      }
+      throw new Error(`unexpected api path: ${pathname}`);
+    },
+    wechatFetch: async (url) => {
+      if (String(url).includes('/cgi-bin/stable_token')) {
+        return new Response(JSON.stringify({ access_token: 'access-1', expires_in: 7200 }), {
+          status: 200,
+          headers: { 'Content-Type': 'application/json' },
+        });
+      }
+      if (String(url).includes('/cgi-bin/message/custom/send')) {
+        return new Response(JSON.stringify({ errcode: 0, errmsg: 'ok' }), {
+          status: 200,
+          headers: { 'Content-Type': 'application/json' },
+        });
+      }
+      throw new Error(`unexpected wechat url: ${url}`);
+    },
+    userAuth: {
+      async getWechatAgentRoute() {
+        if (routeCleared) return null;
+        return {
+          agentSessionId: 'session-1',
+          status: 'active',
+          updatedAt: Date.now(),
+        };
+      },
+      async countWechatAgentSessionMessages() {
+        return 3;
+      },
+      async getWechatAgentSessionSnapshotMessageCount() {
+        return 616;
+      },
+      async clearWechatAgentRoute() {
+        routeCleared = true;
+      },
+      async upsertWechatAgentRoute() {},
+      async finishWechatMpMessage() {},
+    },
+  });
+  const originalRandomUuid = crypto.randomUUID;
+  crypto.randomUUID = () => 'req-snapshot-rotate';
+  try {
+    const result = await service.handleInboundMessage(inboundXml({ content: '继续' }), {
+      timestamp,
+      nonce,
+      signature: signatureFor(token, timestamp, nonce),
+    });
+    assert.equal(result.status, 200);
+    await result.task;
+  } finally {
+    crypto.randomUUID = originalRandomUuid;
+  }
+  assert.equal(routeCleared, true);
+});
+
 test('wechat mp service skips repeated user-context bootstrap for the same session', async () => {
   const token = 'token';
   const timestamp = '1710000000';