diff --git a/user-auth.mjs b/user-auth.mjs index 8fed207..ebef382 100644 --- a/user-auth.mjs +++ b/user-auth.mjs @@ -2259,6 +2259,21 @@ export function createUserAuth(pool, options = {}) { ); }; + const countWechatAgentSessionMessages = async ({ + appId, + openid, + agentSessionId, + }) => { + if (!appId || !openid || !agentSessionId) return 0; + const [rows] = await pool.query( + `SELECT COUNT(*) AS count + FROM h5_wechat_mp_messages + WHERE app_id = ? AND openid = ? AND agent_session_id = ?`, + [appId, openid, agentSessionId], + ); + return Number(rows?.[0]?.count ?? 0); + }; + const recordWechatMpMessage = async ({ appId, openid, @@ -2845,6 +2860,7 @@ export function createUserAuth(pool, options = {}) { upsertWechatAgentRoute, clearWechatAgentRoute, touchWechatAgentRoute, + countWechatAgentSessionMessages, recordWechatMpMessage, finishWechatMpMessage, insertWechatMpMessageDetail, diff --git a/wechat-mp.mjs b/wechat-mp.mjs index d723bca..6a05f10 100644 --- a/wechat-mp.mjs +++ b/wechat-mp.mjs @@ -26,6 +26,7 @@ const DEFAULT_UNSUPPORTED_TEXT = '当前先支持文字消息，你可以直接 const DEFAULT_UNBOUND_TEXT = '先点这里完成绑定，再继续和专属 Agent 对话：'; const DEFAULT_PROGRESS_DELAY_MS = 8000; const DEFAULT_SESSION_IDLE_ROTATE_MS = 30 * 60 * 1000; +const DEFAULT_SESSION_MESSAGE_ROTATE_COUNT = 200; const PUBLIC_HTML_LINK_PATTERN = /https?:\/\/[^\s<>"')\]]+\/MindSpace\/([0-9a-f-]{36}|[a-z0-9._-]+)\/public\/([^\s<>"')\]]+\.html)/gi; const SESSION_WORKSPACE_TOOL_NAMES = new Set([ @@ -1268,6 +1269,10 @@ export function loadWechatMpConfig(env = process.env) { 0, Number(env.H5_WECHAT_MP_SESSION_IDLE_ROTATE_MS ?? DEFAULT_SESSION_IDLE_ROTATE_MS), ), + sessionMessageRotateCount: Math.max( + 0, + Number(env.H5_WECHAT_MP_SESSION_MESSAGE_ROTATE_COUNT ?? DEFAULT_SESSION_MESSAGE_ROTATE_COUNT), + ), statusText: env.H5_WECHAT_MP_STATUS_TEXT?.trim() || DEFAULT_STATUS_TEXT, unsupportedText: env.H5_WECHAT_MP_UNSUPPORTED_TEXT?.trim() || DEFAULT_UNSUPPORTED_TEXT, unboundTextPrefix: env.H5_WECHAT_MP_UNBOUND_TEXT_PREFIX?.trim() || DEFAULT_UNBOUND_TEXT, @@ -1423,6 +1428,14 @@ export function createWechatMpService({ jsapiTicketUrl: config.jsapiTicketUrl || DEFAULT_WECHAT_JSAPI_TICKET_URL, progressText: config.progressText ?? DEFAULT_PROGRESS_TEXT, progressDelayMs: Math.max(0, Number(config.progressDelayMs ?? DEFAULT_PROGRESS_DELAY_MS)), + sessionIdleRotateMs: Math.max( + 0, + Number(config.sessionIdleRotateMs ?? DEFAULT_SESSION_IDLE_ROTATE_MS), + ), + sessionMessageRotateCount: Math.max( + 0, + Number(config.sessionMessageRotateCount ?? DEFAULT_SESSION_MESSAGE_ROTATE_COUNT), + ), statusText: config.statusText || DEFAULT_STATUS_TEXT, mediaPublicBaseUrl: config.mediaPublicBaseUrl || config.publicBaseUrl, maxImageBytes: Math.max(1, Number(config.maxImageBytes ?? 10 * 1024 * 1024)), @@ -1437,6 +1450,7 @@ export function createWechatMpService({ token: null, expiresAt: 0, }; + const rememberedWechatContexts = new Map(); let jsapiTicketCache = { ticket: null, expiresAt: 0, @@ -1663,33 +1677,67 @@ export function createWechatMpService({ const addressName = resolveWechatAddressName(userContext); const existingRoute = await userAuth.getWechatAgentRoute(config.appId, openid); if (existingRoute?.agentSessionId) { - await reconcileAgentSession( - (pathname, init) => fetchForSession(existingRoute.agentSessionId, pathname, init), - existingRoute.agentSessionId, - { - workingDir, + const now = Date.now(); + const routeUpdatedAt = Number(existingRoute.updatedAt ?? 0); + const routeIsIdle = + config.sessionIdleRotateMs > 0 && + routeUpdatedAt > 0 && + now - routeUpdatedAt > config.sessionIdleRotateMs; + const routeMessageCount = + config.sessionMessageRotateCount > 0 && + typeof userAuth.countWechatAgentSessionMessages === 'function' + ? await userAuth + .countWechatAgentSessionMessages({ + appId: config.appId, + openid, + agentSessionId: existingRoute.agentSessionId, + }) + .catch((err) => { + logger.warn?.('WeChat MP route message count lookup failed:', err); + return 0; + }) + : 0; + const routeIsTooLong = + config.sessionMessageRotateCount > 0 && + Number(routeMessageCount ?? 0) >= config.sessionMessageRotateCount; + if (routeIsIdle || routeIsTooLong) { + await userAuth.clearWechatAgentRoute(config.appId, openid); + rememberedWechatContexts.delete(existingRoute.agentSessionId); + } else { + await reconcileAgentSession( + (pathname, init) => fetchForSession(existingRoute.agentSessionId, pathname, init), + existingRoute.agentSessionId, + { + workingDir, + sessionPolicy, + sandboxConstraints: publishLayout?.constraints ?? null, + userContext: publishLayout + ? { + userId, + displayName: addressName || publishLayout.displayName, + username: addressName || null, + slug: null, + } + : null, + tolerateInvalidWorkingDir: true, + }, + ); + const routeHasTools = await sessionHasRequiredTools( + fetchForSession, + existingRoute.agentSessionId, sessionPolicy, - sandboxConstraints: publishLayout?.constraints ?? null, - userContext: publishLayout - ? { - userId, - displayName: addressName || publishLayout.displayName, - username: addressName || null, - slug: null, - } - : null, - tolerateInvalidWorkingDir: true, - }, - ); - const routeHasTools = await sessionHasRequiredTools( - fetchForSession, - existingRoute.agentSessionId, - sessionPolicy, - ).catch(() => false); - if (routeHasTools) { - return existingRoute.agentSessionId; + ).catch(() => false); + if (routeHasTools) { + if (typeof userAuth.touchWechatAgentRoute === 'function') { + await userAuth.touchWechatAgentRoute(config.appId, openid, now).catch((err) => { + logger.warn?.('WeChat MP route touch failed:', err); + }); + } + return { sessionId: existingRoute.agentSessionId, isNewSession: false }; + } + await userAuth.clearWechatAgentRoute(config.appId, openid); + rememberedWechatContexts.delete(existingRoute.agentSessionId); } - await userAuth.clearWechatAgentRoute(config.appId, openid); } else if (existingRoute?.status === 'disabled') { await userAuth.clearWechatAgentRoute(config.appId, openid); } @@ -1727,18 +1775,18 @@ export function createWechatMpService({ (pathname, init) => fetchForSession(sessionId, pathname, init), sessionId, { - workingDir, - sessionPolicy, - sandboxConstraints: publishLayout?.constraints ?? null, - userContext: publishLayout - ? { - userId, - displayName: addressName || publishLayout.displayName, - username: addressName || null, - slug: null, - } - : null, - tolerateInvalidWorkingDir: true, + workingDir, + sessionPolicy, + sandboxConstraints: publishLayout?.constraints ?? null, + userContext: publishLayout + ? { + userId, + displayName: addressName || publishLayout.displayName, + username: addressName || null, + slug: null, + } + : null, + tolerateInvalidWorkingDir: true, }, ); await userAuth.upsertWechatAgentRoute({ @@ -1747,7 +1795,7 @@ export function createWechatMpService({ openid, agentSessionId: sessionId, }); - return sessionId; + return { sessionId, isNewSession: true }; }; const refreshWechatSessionSnapshot = async (sessionId, userId) => { @@ -1759,9 +1807,12 @@ export function createWechatMpService({ } }; - const rememberWechatUserContext = async (sessionId, user) => { + const rememberWechatUserContext = async (sessionId, user, { forceBootstrap = false } = {}) => { const addressName = resolveWechatAddressName(user); if (!addressName) return; + const cacheKey = `${sessionId}:${addressName}`; + const alreadyRemembered = rememberedWechatContexts.get(sessionId) === cacheKey; + if (alreadyRemembered && !forceBootstrap) return; try { await readJsonResponse( await fetchForSession(sessionId, '/agent/harness_remember', { @@ -1780,9 +1831,10 @@ export function createWechatMpService({ await readJsonResponse( await fetchForSession(sessionId, '/agent/harness_bootstrap', { method: 'POST', - body: JSON.stringify({ sessionId, force: true }), + body: JSON.stringify({ sessionId, force: Boolean(forceBootstrap) }), }), ); + rememberedWechatContexts.set(sessionId, cacheKey); } catch (err) { logger.warn?.('WeChat MP user context remember failed:', err); } @@ -1839,15 +1891,16 @@ export function createWechatMpService({ const resetCandidate = intent.msgType === 'text' || intent.msgType === 'voice' ? intent.agentText : ''; const forceNew = isTopicResetIntent(resetCandidate); - let sessionId = await ensureWechatAgentSession({ + let route = await ensureWechatAgentSession({ userId: user.userId, openid: inbound.fromUserName, userContext: user, forceNew, }); + let sessionId = route.sessionId; const publishLayout = await userAuth.getUserPublishLayout(user.userId); await ensureSessionProvider(sessionId); - await rememberWechatUserContext(sessionId, user); + await rememberWechatUserContext(sessionId, user, { forceBootstrap: route.isNewSession }); let finished = false; let progressTimer = null; if (config.progressDelayMs > 0 && config.progressText) { @@ -1943,14 +1996,15 @@ export function createWechatMpService({ const message = err instanceof Error ? err.message : String(err); const mayBeStaleSession = sessionId && isRecoverableWechatAgentSessionError(message); if (mayBeStaleSession) { - sessionId = await ensureWechatAgentSession({ + route = await ensureWechatAgentSession({ userId: user.userId, openid: inbound.fromUserName, forceNew: true, userContext: user, }); + sessionId = route.sessionId; await ensureSessionProvider(sessionId); - await rememberWechatUserContext(sessionId, user); + await rememberWechatUserContext(sessionId, user, { forceBootstrap: route.isNewSession }); const retryId = crypto.randomUUID(); const retryStartedAt = Date.now(); const reply = await executeSessionReply( @@ -2484,7 +2538,7 @@ export function createWechatMpService({ message: '当前账号尚未绑定该服务号', }; } - const sessionId = await ensureWechatAgentSession({ + const { sessionId } = await ensureWechatAgentSession({ userId, openid, forceNew: true, diff --git a/wechat-mp.test.mjs b/wechat-mp.test.mjs index b89c2bf..4c88e3b 100644 --- a/wechat-mp.test.mjs +++ b/wechat-mp.test.mjs @@ -61,6 +61,7 @@ function createBoundWechatService({ token = 'token', wechatFetch, sessionApiFetch, + startAgentSession = null, userAuth = {}, config = {}, scheduleService = null, @@ -112,6 +113,7 @@ function createBoundWechatService({ async insertWechatMpMessageDetail() {}, ...userAuth, }, + startAgentSession, sessionApiFetch, scheduleService, applySessionLlmProvider, @@ -1016,7 +1018,7 @@ test('wechat mp service routes text to dedicated session and sends customer serv if (pathname === '/agent/harness_bootstrap') { const body = JSON.parse(init.body); assert.equal(body.sessionId, 'session-1'); - assert.equal(body.force, true); + assert.equal(body.force, false); return new Response('{}', { status: 200, headers: { 'Content-Type': 'application/json' } }); } throw new Error(`unexpected api path: ${pathname}`); @@ -1252,6 +1254,240 @@ test('wechat mp service applies provider before forwarding to dedicated session' assert.deepEqual(appliedSessions, ['session-1']); }); +test('wechat mp service rotates an idle dedicated session before replying', async () => { + const token = 'token'; + const timestamp = '1710000000'; + const nonce = 'nonce'; + const clearedRoutes = []; + const upsertedSessions = []; + const sessionApiCalls = []; + let routeCleared = false; + const service = createBoundWechatService({ + token, + config: { + sessionIdleRotateMs: 1000, + }, + startAgentSession: async () => ({ id: 'session-2' }), + sessionApiFetch: async (sessionId, pathname, init = {}) => { + sessionApiCalls.push([sessionId, pathname, init.body ?? null]); + assert.equal(sessionId, 'session-2'); + if (pathname === '/sessions/session-2/events') { + return new Response( + [ + 'data: {"type":"Message","request_id":"req-rotate","message":{"id":"assistant-1","role":"assistant","metadata":{"userVisible":true},"content":[{"type":"text","text":"收到，新的会话已接上。"}]}}\n\n', + 'data: {"type":"Finish","request_id":"req-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() - 60_000, + }; + }, + async clearWechatAgentRoute(appId, openid) { + routeCleared = true; + clearedRoutes.push([appId, openid]); + }, + async upsertWechatAgentRoute({ agentSessionId }) { + upsertedSessions.push(agentSessionId); + }, + async finishWechatMpMessage() {}, + }, + }); + const originalRandomUuid = crypto.randomUUID; + crypto.randomUUID = () => 'req-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.deepEqual(clearedRoutes, [['wx123', 'openid-1']]); + assert.deepEqual(upsertedSessions, ['session-2']); + assert.ok(sessionApiCalls.some(([sessionId, pathname]) => sessionId === 'session-2' && pathname === '/sessions/session-2/reply')); +}); + +test('wechat mp service rotates a dedicated session after the message-count threshold', async () => { + const token = 'token'; + const timestamp = '1710000000'; + const nonce = 'nonce'; + let routeCleared = false; + const countedSessions = []; + const service = createBoundWechatService({ + token, + config: { + sessionIdleRotateMs: 0, + sessionMessageRotateCount: 2, + }, + 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-count-rotate","message":{"id":"assistant-1","role":"assistant","metadata":{"userVisible":true},"content":[{"type":"text","text":"已换新会话。"}]}}\n\n', + 'data: {"type":"Finish","request_id":"req-count-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({ agentSessionId }) { + countedSessions.push(agentSessionId); + return 2; + }, + async clearWechatAgentRoute() { + routeCleared = true; + }, + async upsertWechatAgentRoute() {}, + async finishWechatMpMessage() {}, + }, + }); + const originalRandomUuid = crypto.randomUUID; + crypto.randomUUID = () => 'req-count-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.deepEqual(countedSessions, ['session-1']); + assert.equal(routeCleared, true); +}); + +test('wechat mp service skips repeated user-context bootstrap for the same session', async () => { + const token = 'token'; + const timestamp = '1710000000'; + const nonce = 'nonce'; + const bootstrapBodies = []; + const service = createBoundWechatService({ + token, + sessionApiFetch: async (sessionId, pathname, init = {}) => { + assert.equal(sessionId, 'session-1'); + if (pathname === '/sessions/session-1/events') { + return new Response( + [ + 'data: {"type":"Message","request_id":"req-bootstrap","message":{"id":"assistant-1","role":"assistant","metadata":{"userVisible":true},"content":[{"type":"text","text":"收到。"}]}}\n\n', + 'data: {"type":"Finish","request_id":"req-bootstrap","token_state":{"inputTokens":1,"outputTokens":2}}\n\n', + ].join(''), + { status: 200, headers: { 'Content-Type': 'text/event-stream' } }, + ); + } + if (pathname === '/sessions/session-1/reply') { + return new Response('{}', { status: 200, headers: { 'Content-Type': 'application/json' } }); + } + if (pathname === '/agent/harness_remember') { + return new Response('{}', { status: 200, headers: { 'Content-Type': 'application/json' } }); + } + if (pathname === '/agent/harness_bootstrap') { + bootstrapBodies.push(JSON.parse(String(init.body ?? '{}'))); + 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}`); + }, + }); + const originalRandomUuid = crypto.randomUUID; + crypto.randomUUID = () => 'req-bootstrap'; + try { + for (const content of ['第一条', '第二条']) { + 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.deepEqual(bootstrapBodies, [{ sessionId: 'session-1', force: false }]); +}); + test('wechat mp service injects static page publish instructions for html generation requests', async () => { const token = 'token'; const timestamp = '1710000000';