tkmind/memind
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix wechat mp session rotation and bootstrap reuse

Browse Source
This commit is contained in:
john
2026-07-04 12:10:22 +08:00
parent 3139395b58
commit bef72b2768
3 changed files with 352 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
user-auth.mjs
+16
View File
@@ -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,
wechat-mp.mjs
+99 -45
View File
@@ -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,
wechat-mp.test.mjs
+237 -1
View File
@@ -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';