From 8eaf4d23f335861d652c652b3002e0f7c3c18123 Mon Sep 17 00:00:00 2001 From: john Date: Sat, 18 Jul 2026 18:15:22 +0800 Subject: [PATCH 1/2] fix: route WeChat media through shared vision pipeline --- server.mjs | 2 + tkmind-proxy.test.mjs | 61 +++++++++++++ wechat-mp.mjs | 121 +++++++++++++++++++++---- wechat-mp.test.mjs | 204 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 369 insertions(+), 19 deletions(-) diff --git a/server.mjs b/server.mjs index df10f16..99e1f9b 100644 --- a/server.mjs +++ b/server.mjs @@ -820,6 +820,8 @@ async function bootstrapUserAuth() { const target = await tkmindProxy.resolveTarget(sessionId); return tkmindProxy.apiFetchTo(target, pathname, init); }, + submitSessionReply: ({ userId, sessionId, requestId, userMessage }) => + tkmindProxy.submitSessionReplyForUser(userId, sessionId, requestId, userMessage), scheduleService: process.env.H5_SCHEDULE_ENABLED === '1' ? scheduleService : null, wechatScheduleLlmConfigService, llmProviderService, diff --git a/tkmind-proxy.test.mjs b/tkmind-proxy.test.mjs index d4e35fc..1370eeb 100644 --- a/tkmind-proxy.test.mjs +++ b/tkmind-proxy.test.mjs @@ -898,6 +898,67 @@ test('submitSessionReplyForUser adds goose metadata visibility flags before repl }); }); +test('submitSessionReplyForUser applies the shared Qwen vision preprocessing path', async () => { + await withFakeGoosedSession(async ({ apiTarget, workingDir, replyBodies }) => { + const proxy = createTkmindProxy({ + apiTarget, + apiSecret: 'test-secret', + userAuth: { + ...createMemoryTestUserAuth(workingDir), + async ownsSession() { + return true; + }, + async canUseChat() { + return { ok: true }; + }, + async getUserById() { + return { id: 'user-1' }; + }, + async getUserPublishLayout() { + return { publicUrl: 'https://example.com/MindSpace/user-1' }; + }, + async resolveUserPolicies() { + return { unrestricted: true, policies: {} }; + }, + }, + localFetchAsset: async () => ({ + buffer: Buffer.from('fake-image'), + mimeType: 'image/jpeg', + }), + llmProviderService: { + async applyBestProviderForSession() { + return { ok: true }; + }, + async hasVisionKey() { + return true; + }, + async analyzeImagesWithVision() { + return '一件蓝色产品，白色背景，竖版构图。'; + }, + }, + }); + + await proxy.submitSessionReplyForUser( + 'user-1', + 'session-1', + 'request-qwen-vision', + { + id: 'message-qwen-vision', + role: 'user', + content: [{ type: 'text', text: '请分析这张图片' }], + metadata: { + imageUrls: ['/api/mindspace/v1/assets/asset-1/download?inline=1'], + displayText: '请分析这张图片', + }, + }, + ); + + const forwardedText = replyBodies[0]?.user_message?.content?.[0]?.text ?? ''; + assert.match(forwardedText, /Qwen VL 图片描述/); + assert.match(forwardedText, /一件蓝色产品/); + }); +}); + test('submitSessionReplyForUser passes current prompt to Memory V2 resolve before existing reply path', async () => { let resolveInput = null; await withFakeGoosedSession(async ({ apiTarget, workingDir }) => { diff --git a/wechat-mp.mjs b/wechat-mp.mjs index 0ac0077..51ac35c 100644 --- a/wechat-mp.mjs +++ b/wechat-mp.mjs @@ -210,7 +210,14 @@ function pushMessage(messages, incoming) { return [...messages, incoming]; } -async function executeSessionReply(apiFetch, sessionId, requestId, prompt, metadata = {}) { +async function executeSessionReply( + apiFetch, + sessionId, + requestId, + prompt, + metadata = {}, + { submitReply = null } = {}, +) { const eventsResponse = await apiFetch(`/sessions/${sessionId}/events`, { method: 'GET', headers: { Accept: 'text/event-stream' }, @@ -220,18 +227,23 @@ async function executeSessionReply(apiFetch, sessionId, requestId, prompt, metad throw new Error(text || '无法建立公众号消息事件流'); } - const replyResponse = await apiFetch(`/sessions/${sessionId}/reply`, { - method: 'POST', - body: JSON.stringify({ - request_id: requestId, - user_message: createUserMessage(prompt, metadata), - }), - }); - if (!replyResponse.ok) { - const text = await replyResponse.text().catch(() => ''); - throw new Error(text || 'Agent reply 请求失败'); + const userMessage = createUserMessage(prompt, metadata); + if (submitReply) { + await submitReply({ sessionId, requestId, userMessage }); + } else { + const replyResponse = await apiFetch(`/sessions/${sessionId}/reply`, { + method: 'POST', + body: JSON.stringify({ + request_id: requestId, + user_message: userMessage, + }), + }); + if (!replyResponse.ok) { + const text = await replyResponse.text().catch(() => ''); + throw new Error(text || 'Agent reply 请求失败'); + } + replyResponse.body?.cancel().catch?.(() => {}); } - replyResponse.body?.cancel().catch?.(() => {}); const reader = eventsResponse.body.getReader(); const decoder = new TextDecoder(); @@ -1000,6 +1012,7 @@ export function isRecoverableWechatAgentSessionError(message) { if (/stale_session_poisoned_completion/i.test(normalized)) return true; if (/403|404|not found|无权访问/i.test(normalized)) return true; if (/tool_calls|tool_call_id|insufficient tool messages/i.test(normalized)) return true; + if (/wechat_agent_incomplete_reply/i.test(normalized)) return true; return false; } @@ -1030,6 +1043,13 @@ export function findRecoverableWechatAgentErrorInReply(reply) { export function assertWechatAgentReplyIsSendable(reply) { const recoverable = findRecoverableWechatAgentErrorInReply(reply); if (recoverable) throw new Error(recoverable); + const text = String(reply?.text ?? '').trim(); + if ( + /^(?:let me|i(?:'ll| will))\s+(?:first\s+)?(?:look|check|inspect|analy[sz]e)(?:\s+at)?\s+(?:the\s+)?image(?:\s+first)?[.!]?$/iu.test(text) || + /^(?:让我|我先)(?:先)?(?:看|查看|检查|分析)(?:一下)?(?:这张|该张|这个)?图片[。！!]?$/u.test(text) + ) { + throw new Error('wechat_agent_incomplete_reply'); + } } export function isWechatAgentApiErrorText(message) { @@ -1448,6 +1468,7 @@ export function createWechatMpService({ apiFetch, startAgentSession = null, sessionApiFetch = null, + submitSessionReply = null, scheduleService = null, wechatScheduleLlmConfigService = null, llmProviderService = null, @@ -1490,6 +1511,8 @@ export function createWechatMpService({ expiresAt: 0, }; const rememberedWechatContexts = new Map(); + const messageTasksByOpenid = new Map(); + const recentMediaByOpenid = new Map(); let jsapiTicketCache = { ticket: null, expiresAt: 0, @@ -1498,6 +1521,39 @@ export function createWechatMpService({ const fetchForSession = (sessionId, pathname, init) => sessionApiFetch ? sessionApiFetch(sessionId, pathname, init) : apiFetch(pathname, init); + const enqueueMessageTask = (openid, taskFactory) => { + const key = String(openid ?? '').trim(); + const previous = messageTasksByOpenid.get(key) ?? Promise.resolve(); + const task = previous.catch(() => undefined).then(taskFactory); + messageTasksByOpenid.set(key, task); + void task + .finally(() => { + if (messageTasksByOpenid.get(key) === task) messageTasksByOpenid.delete(key); + }) + .catch(() => {}); + return task; + }; + + const rememberRecentMedia = (openid, intent) => { + const publicUrl = String(intent?.media?.publicUrl ?? '').trim(); + if (!publicUrl) return; + recentMediaByOpenid.set(String(openid ?? '').trim(), { + media: { ...(intent.media ?? {}) }, + attachment: intent.attachment ? { ...intent.attachment } : null, + rememberedAt: Date.now(), + }); + }; + + const attachRecentMediaForFollowup = (openid, intent, mediaAnalysisEnabled) => { + if (!mediaAnalysisEnabled || intent?.msgType !== 'text' || intent?.media?.publicUrl) return; + const text = String(intent?.agentText ?? ''); + if (!/(?:刚才|之前|上一|这张|这份|图片|图像|照片|文件|文档|表格|excel|word)/iu.test(text)) return; + const recent = recentMediaByOpenid.get(String(openid ?? '').trim()); + if (!recent || Date.now() - recent.rememberedAt > 15 * 60 * 1000) return; + intent.media = { ...recent.media, source: 'wechat_recent_media' }; + if (recent.attachment) intent.attachment = { ...recent.attachment }; + }; + const resolveWechatBillingTokenState = (sessionId, tokenState) => resolveBillingTokenState(tokenState, { sessionId, @@ -1919,7 +1975,7 @@ export function createWechatMpService({ const buildIntentMetadata = (intent, { mediaAnalysisEnabled = false } = {}) => { const mediaPublicUrl = intent.media?.publicUrl || null; const fileAttachment = - intent.msgType === 'file' && mediaPublicUrl && intent.attachment?.filename + mediaPublicUrl && intent.attachment?.filename ? { assetId: '', downloadUrl: mediaPublicUrl, @@ -1933,7 +1989,7 @@ export function createWechatMpService({ originalMsgId: intent.msgId || null, displayText: intent.displayText || '', mediaPublicUrl, - ...(mediaAnalysisEnabled && intent.msgType === 'image' && mediaPublicUrl + ...(mediaAnalysisEnabled && mediaPublicUrl && !fileAttachment ? { imageUrls: [mediaPublicUrl] } : {}), ...(mediaAnalysisEnabled && fileAttachment ? { fileAttachments: [fileAttachment] } : {}), @@ -2032,6 +2088,17 @@ export function createWechatMpService({ requestId, agentPrompt, buildIntentMetadata(intent, { mediaAnalysisEnabled }), + { + submitReply: submitSessionReply + ? ({ requestId: replyRequestId, userMessage }) => + submitSessionReply({ + userId: user.userId, + sessionId, + requestId: replyRequestId, + userMessage, + }) + : null, + }, ); const workingDir = publishLayout?.publishDir ?? (await userAuth.resolveWorkingDir(user.userId)); const linkExistsForRequest = resolveLinkExistsForWorkingDir(workingDir, linkExists); @@ -2191,6 +2258,17 @@ export function createWechatMpService({ retryId, retryPrompt, buildIntentMetadata(intent, { mediaAnalysisEnabled }), + { + submitReply: submitSessionReply + ? ({ requestId: replyRequestId, userMessage }) => + submitSessionReply({ + userId: user.userId, + sessionId, + requestId: replyRequestId, + userMessage, + }) + : null, + }, ); const workingDir = publishLayout?.publishDir ?? (await userAuth.resolveWorkingDir(user.userId)); const linkExistsForRequest = resolveLinkExistsForWorkingDir(workingDir, linkExists); @@ -2416,6 +2494,7 @@ export function createWechatMpService({ boundUser, config.mediaAnalysisGrayUsers, ); + attachRecentMediaForFollowup(inbound.fromUserName, intent, mediaAnalysisEnabled); if (intent.msgType === 'file' && !mediaAnalysisEnabled) { await persistIntentDetail({ intent, userId: boundUser.userId, rawXmlHash }); @@ -2499,6 +2578,7 @@ export function createWechatMpService({ source: persisted.source, }; intent.agentText = `[图片1]: ${persisted.publicUrl}`; + rememberRecentMedia(inbound.fromUserName, intent); } catch (error) { await persistIntentDetail({ intent, userId: boundUser.userId, rawXmlHash }); return { @@ -2549,6 +2629,7 @@ export function createWechatMpService({ }; intent.agentText = `[文件1: ${persisted.filename}]: ${persisted.publicUrl}`; intent.displayText = `文件：${persisted.filename}`; + rememberRecentMedia(inbound.fromUserName, intent); } catch (error) { await persistIntentDetail({ intent, userId: boundUser.userId, rawXmlHash }); return { @@ -2686,11 +2767,13 @@ export function createWechatMpService({ }; } - const task = runIntentMessage({ - inbound, - intent, - user: boundUser, - }) + const task = enqueueMessageTask(inbound.fromUserName, () => + runIntentMessage({ + inbound, + intent, + user: boundUser, + }), + ) .then(async ({ sessionId } = {}) => { if (inbound.msgId && typeof userAuth.finishWechatMpMessage === 'function') { await userAuth.finishWechatMpMessage({ diff --git a/wechat-mp.test.mjs b/wechat-mp.test.mjs index 2518c6f..786caba 100644 --- a/wechat-mp.test.mjs +++ b/wechat-mp.test.mjs @@ -79,6 +79,7 @@ function createBoundWechatService({ config = {}, scheduleService = null, applySessionLlmProvider = null, + submitSessionReply = null, }) { return createWechatMpService({ config: { @@ -128,6 +129,7 @@ function createBoundWechatService({ }, startAgentSession, sessionApiFetch, + submitSessionReply, scheduleService, applySessionLlmProvider, wechatFetch, @@ -2752,6 +2754,17 @@ test('sanitizeWechatAgentOutboundText replaces raw api errors with friendly text ); }); +test('assertWechatAgentReplyIsSendable rejects image inspection placeholders', () => { + assert.throws( + () => assertWechatAgentReplyIsSendable({ text: 'Let me look at the image first.' }), + /wechat_agent_incomplete_reply/, + ); + assert.throws( + () => assertWechatAgentReplyIsSendable({ text: '我先看一下这张图片。' }), + /wechat_agent_incomplete_reply/, + ); +}); + test('wechat mp service recreates dedicated session when tool_calls error arrives via Finish assistant text', async () => { const token = 'token'; const timestamp = '1710000000'; @@ -3821,6 +3834,197 @@ test('wechat mp service persists image and routes image url into agent prompt', assert.match(detailCalls[0].mediaPublicUrl, /\/wechat-mp\//); }); +test('wechat mp image submission reuses the H5 prepared reply path', async () => { + const token = 'token'; + const timestamp = '1710000000'; + const nonce = 'nonce'; + const testUserId = 'test-user-image-prepared'; + const submitCalls = []; + const service = createBoundWechatService({ + token, + config: { + mediaAnalysisGrayUsers: [testUserId], + }, + userAuth: { + async findWechatUserByOpenid() { + return { userId: testUserId, status: 'active', nickname: '唐' }; + }, + }, + sessionApiFetch: async (_sessionId, pathname) => { + if (pathname === '/sessions/session-1/events') { + return new Response( + [ + 'data: {"type":"Message","message":{"id":"assistant-1","role":"assistant","metadata":{"userVisible":true},"content":[{"type":"text","text":"图片内容已识别。"}]}}\n\n', + 'data: {"type":"Finish"}\n\n', + ].join(''), + { status: 200, headers: { 'Content-Type': 'text/event-stream' } }, + ); + } + 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}`); + }, + submitSessionReply: async (input) => { + submitCalls.push(input); + return { ok: true }; + }, + 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/media/get')) { + return new Response(Buffer.from([0x89, 0x50, 0x4e, 0x47]), { + status: 200, + headers: { 'Content-Type': 'image/png' }, + }); + } + 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}`); + }, + }); + + try { + const result = await service.handleInboundMessage( + inboundXml({ + msgType: 'image', + content: '', + extraFields: { MediaId: 'media-prepared', PicUrl: 'https://wx.example.com/image.png' }, + }), + { timestamp, nonce, signature: signatureFor(token, timestamp, nonce) }, + ); + await result.task; + } finally { + fs.rmSync(path.join(process.cwd(), 'MindSpace', testUserId), { recursive: true, force: true }); + } + + assert.equal(submitCalls.length, 1); + assert.equal(submitCalls[0].userId, testUserId); + assert.equal(submitCalls[0].sessionId, 'session-1'); + assert.equal(submitCalls[0].userMessage.metadata.imageUrls.length, 1); + assert.match(submitCalls[0].userMessage.metadata.imageUrls[0], /\/public\/wechat-mp\//); +}); + +test('wechat mp serializes image and follow-up text and reattaches recent image', async () => { + const token = 'token'; + const timestamp = '1710000000'; + const nonce = 'nonce'; + const testUserId = 'test-user-image-followup'; + const submitCalls = []; + let eventCall = 0; + let releaseFirst = null; + const service = createBoundWechatService({ + token, + config: { + mediaAnalysisGrayUsers: [testUserId], + }, + userAuth: { + async findWechatUserByOpenid() { + return { userId: testUserId, status: 'active', nickname: '唐' }; + }, + }, + sessionApiFetch: async (_sessionId, pathname) => { + if (pathname === '/sessions/session-1/events') { + eventCall += 1; + if (eventCall === 1) { + return new Response( + new ReadableStream({ + start(controller) { + releaseFirst = () => { + controller.enqueue( + new TextEncoder().encode( + 'data: {"type":"Message","message":{"id":"assistant-image","role":"assistant","metadata":{"userVisible":true},"content":[{"type":"text","text":"图片已识别。"}]}}\n\n' + + 'data: {"type":"Finish"}\n\n', + ), + ); + controller.close(); + }; + }, + }), + { status: 200, headers: { 'Content-Type': 'text/event-stream' } }, + ); + } + return new Response( + [ + 'data: {"type":"Message","message":{"id":"assistant-followup","role":"assistant","metadata":{"userVisible":true},"content":[{"type":"text","text":"已结合刚才图片分析主题。"}]}}\n\n', + 'data: {"type":"Finish"}\n\n', + ].join(''), + { status: 200, headers: { 'Content-Type': 'text/event-stream' } }, + ); + } + 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}`); + }, + submitSessionReply: async (input) => { + submitCalls.push(input); + return { ok: true }; + }, + 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/media/get')) { + return new Response(Buffer.from([0x89, 0x50, 0x4e, 0x47]), { + status: 200, + headers: { 'Content-Type': 'image/png' }, + }); + } + 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}`); + }, + }); + + try { + const imageResult = await service.handleInboundMessage( + inboundXml({ + msgType: 'image', + content: '', + extraFields: { MediaId: 'media-followup', PicUrl: 'https://wx.example.com/image.png' }, + }), + { timestamp, nonce, signature: signatureFor(token, timestamp, nonce) }, + ); + while (submitCalls.length === 0) await new Promise((resolve) => setTimeout(resolve, 0)); + + const followupResult = await service.handleInboundMessage( + inboundXml({ msgType: 'text', content: '请根据刚才图片分析主题' }), + { timestamp, nonce, signature: signatureFor(token, timestamp, nonce) }, + ); + await new Promise((resolve) => setTimeout(resolve, 0)); + assert.equal(submitCalls.length, 1); + + releaseFirst(); + await imageResult.task; + await followupResult.task; + } finally { + fs.rmSync(path.join(process.cwd(), 'MindSpace', testUserId), { recursive: true, force: true }); + } + + assert.equal(submitCalls.length, 2); + assert.deepEqual( + submitCalls[1].userMessage.metadata.imageUrls, + submitCalls[0].userMessage.metadata.imageUrls, + ); + assert.equal(submitCalls[1].userMessage.metadata.msgType, 'text'); +}); + test('wechat mp service persists Word and Excel files in user public area and reuses H5 attachment metadata', async () => { const token = 'token'; const timestamp = '1710000000'; From ae3e1ba3fa036c9827c18d20bc504a5190d92dd3 Mon Sep 17 00:00:00 2001 From: john Date: Sat, 18 Jul 2026 19:05:33 +0800 Subject: [PATCH 2/2] fix: recover stale WeChat active requests --- wechat-mp.mjs | 1 + wechat-mp.test.mjs | 4 ++++ 2 files changed, 5 insertions(+) diff --git a/wechat-mp.mjs b/wechat-mp.mjs index 51ac35c..413027a 100644 --- a/wechat-mp.mjs +++ b/wechat-mp.mjs @@ -1012,6 +1012,7 @@ export function isRecoverableWechatAgentSessionError(message) { if (/stale_session_poisoned_completion/i.test(normalized)) return true; if (/403|404|not found|无权访问/i.test(normalized)) return true; if (/tool_calls|tool_call_id|insufficient tool messages/i.test(normalized)) return true; + if (/session already has an active request|active request.*cancel/i.test(normalized)) return true; if (/wechat_agent_incomplete_reply/i.test(normalized)) return true; return false; } diff --git a/wechat-mp.test.mjs b/wechat-mp.test.mjs index 786caba..cb96f50 100644 --- a/wechat-mp.test.mjs +++ b/wechat-mp.test.mjs @@ -2714,6 +2714,10 @@ test('isRecoverableWechatAgentSessionError detects poisoned tool_calls history', ); assert.equal(isRecoverableWechatAgentSessionError('stale_session_poisoned_completion'), true); assert.equal(isRecoverableWechatAgentSessionError('无权访问该会话'), true); + assert.equal( + isRecoverableWechatAgentSessionError('Session already has an active request. Cancel it first.'), + true, + ); assert.equal(isRecoverableWechatAgentSessionError('network timeout'), false); });