merge: fix WeChat media gray pipeline

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,

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 }) => {

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`, {
+  const userMessage = createUserMessage(prompt, metadata);
-    method: 'POST',
+  if (submitReply) {
-    body: JSON.stringify({
+    await submitReply({ sessionId, requestId, userMessage });
-      request_id: requestId,
+  } else {
-      user_message: createUserMessage(prompt, metadata),
+    const replyResponse = await apiFetch(`/sessions/${sessionId}/reply`, {
-    }),
+      method: 'POST',
-  });
+      body: JSON.stringify({
-  if (!replyResponse.ok) {
+        request_id: requestId,
-    const text = await replyResponse.text().catch(() => '');
+        user_message: userMessage,
-    throw new Error(text || 'Agent reply 请求失败');
+      }),
+    });
+    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,8 @@ 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;
 }
 
@@ -1030,6 +1044,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 +1469,7 @@ export function createWechatMpService({
   apiFetch,
   startAgentSession = null,
   sessionApiFetch = null,
+  submitSessionReply = null,
   scheduleService = null,
   wechatScheduleLlmConfigService = null,
   llmProviderService = null,
@@ -1490,6 +1512,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 +1522,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 +1976,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 +1990,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 +2089,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 +2259,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 +2495,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 +2579,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 +2630,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 +2768,13 @@ export function createWechatMpService({
       };
     }
 
-    const task = runIntentMessage({
+    const task = enqueueMessageTask(inbound.fromUserName, () =>
-      inbound,
+      runIntentMessage({
-      intent,
+        inbound,
-      user: boundUser,
+        intent,
-    })
+        user: boundUser,
+      }),
+    )
       .then(async ({ sessionId } = {}) => {
         if (inbound.msgId && typeof userAuth.finishWechatMpMessage === 'function') {
           await userAuth.finishWechatMpMessage({

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,
@@ -2712,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);
 });
 
@@ -2752,6 +2758,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 +3838,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';