diff --git a/user-auth.mjs b/user-auth.mjs index ebef382..5bb7ee7 100644 --- a/user-auth.mjs +++ b/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, diff --git a/wechat-mp.mjs b/wechat-mp.mjs index 6a05f10..8037a79 100644 --- a/wechat-mp.mjs +++ b/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, diff --git a/wechat-mp.test.mjs b/wechat-mp.test.mjs index 4c88e3b..e4b5c23 100644 --- a/wechat-mp.test.mjs +++ b/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';