fix: harden agent and WeChat run completion

修复微信历史图片会话污染、换新会话失效及普通聊天误追加历史页面链接。

agent-run-gateway.mjs

@@ -488,7 +488,7 @@ export function createAgentRunGateway({
     return true;
   }
 
-  async function markRun(runId, status, fields = {}) {
+  async function markRun(runId, status, fields = {}, { expectedStatus = null } = {}) {
     const updates = ['status = ?', 'updated_at = ?'];
     const values = [status, nowMs()];
     for (const [key, value] of Object.entries(fields)) {
@@ -496,12 +496,18 @@ export function createAgentRunGateway({
       values.push(value);
     }
     values.push(runId);
-    await pool.query(
+    const where = expectedStatus
-      `UPDATE h5_agent_runs SET ${updates.join(', ')} WHERE id = ?`,
+      ? 'WHERE id = ? AND status = ?'
+      : 'WHERE id = ?';
+    if (expectedStatus) values.push(expectedStatus);
+    const [result] = await pool.query(
+      `UPDATE h5_agent_runs SET ${updates.join(', ')} ${where}`,
       values,
     );
+    if (Number(result?.affectedRows ?? 0) === 0) return false;
     await appendEvent(runId, status, fields);
     await appendRunSnapshot(runId);
+    return true;
   }
 
   function startRunHeartbeat(runId, { attempt }) {
@@ -940,11 +946,12 @@ export function createAgentRunGateway({
         throw error;
       }
     }
-    await markRun(runId, 'succeeded', {
+    const marked = await markRun(runId, 'succeeded', {
       agent_session_id: sessionId,
       completed_at: nowMs(),
       error_message: null,
-    });
+    }, { expectedStatus: 'running' });
+    if (!marked) return false;
     if (typeof observePersonalMemoryOnSuccess === 'function') {
       await observePersonalMemoryOnSuccess({
         userId: row.user_id,
@@ -999,7 +1006,7 @@ export function createAgentRunGateway({
       await markRun(runId, retryable ? 'retryable' : 'failed', {
         error_message: message,
         completed_at: retryable ? null : nowMs(),
-      });
+      }, { expectedStatus: 'running' });
       if (retryable && autoDispatch) {
         setTimeout(() => dispatchRun(runId), retryDelaysMs[nextAttempt - 1]);
       }
@@ -1129,6 +1136,7 @@ export function createAgentRunGateway({
     );
     const startedCutoff = nowMs() - normalizedStaleMs;
     const sessionFinishedCutoff = nowMs() - normalizedSessionFinishedGraceMs;
+    const heartbeatCutoff = nowMs() - normalizedStaleMs;
     const [rows] = await pool.query(
       `SELECT
          r.id,
@@ -1155,6 +1163,7 @@ export function createAgentRunGateway({
        ) sf ON sf.run_id = r.id
        WHERE r.status = 'running'
          AND r.started_at IS NOT NULL
+         AND (h.latest_heartbeat_at IS NULL OR h.latest_heartbeat_at <= ?)
          AND (
            r.started_at <= ?
            OR (
@@ -1164,7 +1173,7 @@ export function createAgentRunGateway({
          )
        ORDER BY COALESCE(sf.session_finished_at, r.started_at) ASC
        LIMIT ?`,
-      [startedCutoff, sessionFinishedCutoff, normalizedLimit],
+      [heartbeatCutoff, startedCutoff, sessionFinishedCutoff, normalizedLimit],
     );
     const recovered = [];
     for (const row of rows) {
@@ -1198,11 +1207,12 @@ export function createAgentRunGateway({
           requireRecoverableError: false,
         });
         if (deliverableRecovered) {
-          await markRun(row.id, 'succeeded', {
+          const marked = await markRun(row.id, 'succeeded', {
             agent_session_id: row.agent_session_id ?? null,
             completed_at: nowMs(),
             error_message: null,
-          });
+          }, { expectedStatus: 'running' });
+          if (!marked) continue;
           item.status = 'succeeded';
           item.recoveredAs = 'deliverables';
           recovered.push(item);
@@ -1223,15 +1233,21 @@ export function createAgentRunGateway({
              AND started_at IS NOT NULL
              AND (
                started_at <= ?
-               OR EXISTS (
+             OR EXISTS (
-                 SELECT 1
+               SELECT 1
-                 FROM h5_agent_run_events sf
+               FROM h5_agent_run_events sf
-                 WHERE sf.run_id = h5_agent_runs.id
+               WHERE sf.run_id = h5_agent_runs.id
-                   AND sf.event_type = 'session_finished'
+                 AND sf.event_type = 'session_finished'
-                   AND sf.created_at <= ?
+                 AND sf.created_at <= ?
                )
-             )`,
+             )
-          [message, completedAt, completedAt, row.id, startedCutoff, sessionFinishedCutoff],
+             AND (
+               SELECT COALESCE(MAX(hb.created_at), h5_agent_runs.started_at)
+               FROM h5_agent_run_events hb
+               WHERE hb.run_id = h5_agent_runs.id
+                 AND hb.event_type = 'worker_heartbeat'
+             ) <= ?`,
+          [message, completedAt, completedAt, row.id, startedCutoff, sessionFinishedCutoff, heartbeatCutoff],
         );
         if (Number(update?.affectedRows ?? 0) === 0) continue;
         await appendEvent(row.id, 'stale_recovered', {

agent-run-gateway.test.mjs

@@ -43,12 +43,16 @@ function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} }
     return timestamps.length ? Math.max(...timestamps) : null;
   };
 
-  const isStaleRunningRow = (row, startedCutoff, sessionFinishedCutoff) => {
+  const isStaleRunningRow = (row, heartbeatCutoff, startedCutoff, sessionFinishedCutoff) => {
     if (row.status !== 'running' || row.started_at == null) return false;
     const finishedAt = sessionFinishedAt(row.id);
+    const heartbeatAt = latestHeartbeatAt(row.id);
     return (
-      Number(row.started_at) <= Number(startedCutoff)
+      (heartbeatAt == null || Number(heartbeatAt) <= Number(heartbeatCutoff))
-      || (finishedAt != null && Number(finishedAt) <= Number(sessionFinishedCutoff))
+      && (
+        Number(row.started_at) <= Number(startedCutoff)
+        || (finishedAt != null && Number(finishedAt) <= Number(sessionFinishedCutoff))
+      )
     );
   };
 
@@ -112,9 +116,9 @@ function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} }
         }]];
       }
       if (sql.includes('SELECT') && sql.includes('session_finished_at') && sql.includes('r.started_at <= ?')) {
-        const [startedCutoff, sessionFinishedCutoff, limit = 1] = params;
+        const [heartbeatCutoff, startedCutoff, sessionFinishedCutoff, limit = 1] = params;
         return [[...runs.values()]
-          .filter((row) => isStaleRunningRow(row, startedCutoff, sessionFinishedCutoff))
+          .filter((row) => isStaleRunningRow(row, heartbeatCutoff, startedCutoff, sessionFinishedCutoff))
           .sort((a, b) => {
             const aKey = Number(sessionFinishedAt(a.id) ?? a.started_at ?? 0);
             const bKey = Number(sessionFinishedAt(b.id) ?? b.started_at ?? 0);
@@ -237,9 +241,9 @@ function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} }
         return [{ affectedRows: 1 }];
       }
       if (sql.includes("WHERE id = ?") && sql.includes("status = 'running'") && sql.includes('session_finished')) {
-        const [errorMessage, updatedAt, completedAt, id, startedCutoff, sessionFinishedCutoff] = params;
+        const [errorMessage, updatedAt, completedAt, id, startedCutoff, sessionFinishedCutoff, heartbeatCutoff] = params;
         const row = runs.get(id);
-        if (!row || !isStaleRunningRow(row, startedCutoff, sessionFinishedCutoff)) {
+        if (!row || !isStaleRunningRow(row, heartbeatCutoff, startedCutoff, sessionFinishedCutoff)) {
           return [{ affectedRows: 0 }];
         }
         Object.assign(row, {
@@ -283,7 +287,9 @@ function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} }
         return [rows];
       }
       if (sql.includes('UPDATE h5_agent_runs SET')) {
-        const id = params.at(-1);
+        const setSql = sql.split(' WHERE ')[0];
+        const columns = [...setSql.matchAll(/([a-z_]+) = \?/g)].map((match) => match[1]);
+        const id = params[columns.length];
         const row = runs.get(id);
         if (!row) return [{ affectedRows: 0 }];
         if (sql.includes("status = 'running'") && sql.includes('started_at = COALESCE(started_at, ?)')) {
@@ -297,7 +303,8 @@ function createFakePool({ sessionDeliverables = {}, workspaceDeliverables = {} }
           });
           return [{ affectedRows: 1 }];
         }
-        const columns = [...sql.matchAll(/([a-z_]+) = \?/g)].map((match) => match[1]);
+        const expectedStatus = sql.includes('AND status = ?') ? params[columns.length + 1] : null;
+        if (expectedStatus && row.status !== expectedStatus) return [{ affectedRows: 0 }];
         for (let i = 0; i < columns.length; i += 1) {
           row[columns[i]] = params[i];
         }
@@ -1817,7 +1824,7 @@ test('stale running recovery marks old running rows failed with an event', async
   );
 });
 
-test('stale running recovery still considers old runs even with fresh heartbeat', async () => {
+test('stale running recovery ignores old runs with a fresh heartbeat', async () => {
   const pool = createFakePool();
   const gateway = createAgentRunGateway({
     pool,
@@ -1847,9 +1854,9 @@ test('stale running recovery still considers old runs even with fresh heartbeat'
 
   const result = await gateway.recoverStaleRunningRuns({ staleMs: 1000, dryRun: false });
 
-  assert.equal(result.considered, 1);
+  assert.equal(result.considered, 0);
-  assert.equal(result.recovered, 1);
+  assert.equal(result.recovered, 0);
-  assert.equal(pool.runs.get(run.id).status, 'failed');
+  assert.equal(pool.runs.get(run.id).status, 'running');
 });
 
 test('stale running recovery succeeds when workspace pages exist after sync', async () => {

wechat-mp-config.mjs

@@ -87,6 +87,11 @@ export function loadWechatMpConfig(env = process.env) {
     acceptLocation: env.H5_WECHAT_MP_ACCEPT_LOCATION !== '0',
     acceptLink: env.H5_WECHAT_MP_ACCEPT_LINK !== '0',
     mediaAnalysisGrayUsers: parseCsvList(env.H5_WECHAT_MP_MEDIA_GRAY_USERS),
+    reliabilityGrayUsers: parseCsvList(env.H5_WECHAT_MP_RELIABILITY_GRAY_USERS),
+    agentReplyTimeoutMs: Math.max(
+      0,
+      Number(env.H5_WECHAT_MP_AGENT_REPLY_TIMEOUT_MS ?? 15 * 60 * 1000),
+    ),
     requireFreshPageThumbnail: env.H5_WECHAT_MP_FRESH_PAGE_THUMBNAILS !== '0',
     repairFreshPageThumbnail: env.H5_WECHAT_MP_FRESH_THUMBNAIL_REPAIR === '1',
     encodingAesKey: env.H5_WECHAT_MP_ENCODING_AES_KEY?.trim() ?? '',

wechat-mp.mjs

@@ -69,6 +69,7 @@ const DEFAULT_WECHAT_JSAPI_TICKET_URL = 'https://api.weixin.qq.com/cgi-bin/ticke
 const DEFAULT_ASR_TARGET = process.env.H5_ASR_TARGET ?? 'https://asr.tkmind.cn';
 const WECHAT_RECENT_MEDIA_TTL_MS = 15 * 60 * 1000;
 const WECHAT_RECENT_IMAGE_MAX_COUNT = 10;
+const DEFAULT_WECHAT_AGENT_REPLY_TIMEOUT_MS = 15 * 60 * 1000;
 export { loadWechatMpConfig };
 const PUBLIC_HTML_LINK_PATTERN =
   /https?:\/\/[^\s<>"')\]]+\/MindSpace\/([0-9a-f-]{36}|[a-z0-9._-]+)\/public\/([^\s<>"')\]]+\.html)/gi;
@@ -229,110 +230,137 @@ function pushMessage(messages, incoming) {
   return [...messages, incoming];
 }
 
-async function executeSessionReply(
+export async function executeSessionReply(
   apiFetch,
   sessionId,
   requestId,
   prompt,
   metadata = {},
-  { submitReply = null, prepareUserMessage = null } = {},
+  { submitReply = null, prepareUserMessage = null, timeoutMs = 0 } = {},
 ) {
-  const eventsResponse = await apiFetch(`/sessions/${sessionId}/events`, {
+  const normalizedTimeoutMs = Math.max(0, Number(timeoutMs) || 0);
-    method: 'GET',
+  const abortController = new AbortController();
-    headers: { Accept: 'text/event-stream' },
+  let timedOut = false;
-  });
+  let reader = null;
-  if (!eventsResponse.ok || !eventsResponse.body) {
+  const timeout = normalizedTimeoutMs > 0
-    const text = await eventsResponse.text().catch(() => '');
+    ? setTimeout(() => {
-    throw new Error(text || '无法建立公众号消息事件流');
+        timedOut = true;
-  }
+        abortController.abort();
+        void reader?.cancel?.().catch?.(() => {});
+      }, normalizedTimeoutMs)
+    : null;
 
-  let userMessage = createUserMessage(prompt, metadata);
+  try {
-  if (prepareUserMessage) {
+    const eventsResponse = await apiFetch(`/sessions/${sessionId}/events`, {
-    userMessage = (await prepareUserMessage(userMessage)) ?? userMessage;
+      method: 'GET',
-  }
+      headers: { Accept: 'text/event-stream' },
-  if (submitReply) {
+      signal: abortController.signal,
-    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) {
+    if (!eventsResponse.ok || !eventsResponse.body) {
-      const text = await replyResponse.text().catch(() => '');
+      const text = await eventsResponse.text().catch(() => '');
-      throw new Error(text || 'Agent reply 请求失败');
+      throw new Error(text || '无法建立公众号消息事件流');
     }
-    replyResponse.body?.cancel().catch?.(() => {});
-  }
 
-  const reader = eventsResponse.body.getReader();
+    let userMessage = createUserMessage(prompt, metadata);
-  const decoder = new TextDecoder();
+    if (prepareUserMessage) {
-  let buffer = '';
+      userMessage = (await prepareUserMessage(userMessage)) ?? userMessage;
-  let messages = [];
+    }
-  let requestMessages = [];
+    if (submitReply) {
-  let hasScopedAssistantUpdate = false;
+      await submitReply({ sessionId, requestId, userMessage, signal: abortController.signal });
-
+    } else {
-  while (true) {
+      const replyResponse = await apiFetch(`/sessions/${sessionId}/reply`, {
-    const { value, done } = await reader.read();
+        method: 'POST',
-    if (done) break;
+        body: JSON.stringify({
-    buffer += decoder.decode(value, { stream: true });
+          request_id: requestId,
-    const frames = buffer.split('\n\n');
+          user_message: userMessage,
-    buffer = frames.pop() ?? '';
+        }),
-    for (const frame of frames) {
+        signal: abortController.signal,
-      let data = '';
+      });
-      for (const line of frame.split('\n')) {
+      if (!replyResponse.ok) {
-        if (line.startsWith('data:')) data += line.slice(5).trim();
+        const text = await replyResponse.text().catch(() => '');
+        throw new Error(text || 'Agent reply 请求失败');
       }
-      if (!data) continue;
+      replyResponse.body?.cancel().catch?.(() => {});
-      let event;
+    }
-      try {
-        event = JSON.parse(data);
-      } catch {
-        continue;
-      }
-      const routingId = event.chat_request_id ?? event.request_id;
-      if (routingId && routingId !== requestId) continue;
 
-      if (event.type === 'Message' && event.message?.metadata?.userVisible) {
+    reader = eventsResponse.body.getReader();
-        const hasActionRequired = event.message.content?.some((item) => item.type === 'actionRequired');
+    const decoder = new TextDecoder();
-        if (hasActionRequired) {
+    let buffer = '';
-          throw new Error('当前回复需要人工确认，公众号通道暂不支持');
+    let messages = [];
+    let requestMessages = [];
+    let hasScopedAssistantUpdate = false;
+
+    while (true) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      buffer += decoder.decode(value, { stream: true });
+      const frames = buffer.split('\n\n');
+      buffer = frames.pop() ?? '';
+      for (const frame of frames) {
+        let data = '';
+        for (const line of frame.split('\n')) {
+          if (line.startsWith('data:')) data += line.slice(5).trim();
         }
-        if (event.message.role === 'assistant') hasScopedAssistantUpdate = true;
+        if (!data) continue;
-        messages = pushMessage(messages, event.message);
+        let event;
-        requestMessages = pushMessage(requestMessages, event.message);
+        try {
-      } else if (event.type === 'UpdateConversation') {
+          event = JSON.parse(data);
-        // Ignore unscoped snapshots until this request has yielded an assistant update.
+        } catch {
-        // Otherwise a stale session snapshot can overwrite the current reply with a
+          continue;
-        // previous page/link from the same WeChat-dedicated session.
-        if (hasScopedAssistantUpdate) {
-          messages = (event.conversation ?? []).filter((item) => item.metadata?.userVisible);
         }
-      } else if (event.type === 'Error') {
+        const routingId = event.chat_request_id ?? event.request_id;
-        throw new Error(event.error || '任务执行失败');
+        if (routingId && routingId !== requestId) continue;
-      } else if (event.type === 'Finish') {
+
-        const assistant = [...messages].reverse().find((item) => item.role === 'assistant');
+        if (event.type === 'Message' && event.message?.metadata?.userVisible) {
-        if (!hasScopedAssistantUpdate || !assistant) {
+          const hasActionRequired = event.message.content?.some((item) => item.type === 'actionRequired');
-          throw new Error('本轮未收到可发送的新回复，请稍后重试');
+          if (hasActionRequired) {
+            throw new Error('当前回复需要人工确认，公众号通道暂不支持');
+          }
+          if (event.message.role === 'assistant') hasScopedAssistantUpdate = true;
+          messages = pushMessage(messages, event.message);
+          requestMessages = pushMessage(requestMessages, event.message);
+        } else if (event.type === 'UpdateConversation') {
+          // Ignore unscoped snapshots until this request has yielded an assistant update.
+          // Otherwise a stale session snapshot can overwrite the current reply with a
+          // previous page/link from the same WeChat-dedicated session.
+          if (hasScopedAssistantUpdate) {
+            messages = (event.conversation ?? []).filter((item) => item.metadata?.userVisible);
+          }
+        } else if (event.type === 'Error') {
+          throw new Error(event.error || '任务执行失败');
+        } else if (event.type === 'Finish') {
+          const assistant = [...messages].reverse().find((item) => item.role === 'assistant');
+          if (!hasScopedAssistantUpdate || !assistant) {
+            throw new Error('本轮未收到可发送的新回复，请稍后重试');
+          }
+          const reply = {
+            text: messageVisibleText(assistant),
+            tokenState: event.token_state ?? null,
+            messages,
+            // Keep request-scoped stream messages separate from a later full
+            // UpdateConversation snapshot. Artifact delivery must never inspect
+            // historical tool calls from the whole dedicated session.
+            requestMessages,
+          };
+          assertWechatAgentReplyIsSendable(reply);
+          return reply;
         }
-        const reply = {
-          text: messageVisibleText(assistant),
-          tokenState: event.token_state ?? null,
-          messages,
-          // Keep request-scoped stream messages separate from a later full
-          // UpdateConversation snapshot. Artifact delivery must never inspect
-          // historical tool calls from the whole dedicated session.
-          requestMessages,
-        };
-        assertWechatAgentReplyIsSendable(reply);
-        return reply;
       }
     }
-  }
 
-  throw new Error('公众号消息事件流提前结束');
+    throw new Error('公众号消息事件流提前结束');
+  } catch (err) {
+    if (timedOut) {
+      const timeoutError = new Error(`公众号消息处理超时（${normalizedTimeoutMs}ms），请稍后重试`);
+      timeoutError.code = 'WECHAT_AGENT_REPLY_TIMEOUT';
+      timeoutError.retryable = true;
+      throw timeoutError;
+    }
+    throw err;
+  } finally {
+    if (timeout) clearTimeout(timeout);
+    await reader?.cancel?.().catch?.(() => {});
+  }
 }
 
 const WECHAT_CUSTOMER_TEXT_MAX_BYTES = 2048;
@@ -1580,6 +1608,13 @@ export function createWechatMpService({
     mediaAnalysisGrayUsers: Array.isArray(config.mediaAnalysisGrayUsers)
       ? config.mediaAnalysisGrayUsers
       : [],
+    reliabilityGrayUsers: Array.isArray(config.reliabilityGrayUsers)
+      ? config.reliabilityGrayUsers
+      : [],
+    agentReplyTimeoutMs: Math.max(
+      0,
+      Number(config.agentReplyTimeoutMs ?? DEFAULT_WECHAT_AGENT_REPLY_TIMEOUT_MS),
+    ),
     requireFreshPageThumbnail,
     repairFreshPageThumbnail,
     asrTarget: config.asrTarget || DEFAULT_ASR_TARGET,
@@ -2321,6 +2356,8 @@ export function createWechatMpService({
   const runIntentMessage = async ({ inbound, intent, user }) => {
     const wechatIntent = classifyWechatIntent(intent);
     const mediaAnalysisEnabled = isWechatMediaGrayUser(user, config.mediaAnalysisGrayUsers);
+    const reliabilityEnabled = isWechatMediaGrayUser(user, config.reliabilityGrayUsers);
+    const agentReplyTimeoutMs = reliabilityEnabled ? config.agentReplyTimeoutMs : 0;
     const resetCandidate =
       intent.msgType === 'text' || intent.msgType === 'voice' ? intent.agentText : '';
     const imagePolicy = resolveWechatImageGenerationPolicy({
@@ -2396,6 +2433,7 @@ export function createWechatMpService({
                   options: { requireHistoricalImageIsolation: true },
                 })
             : null,
+          timeoutMs: agentReplyTimeoutMs,
         },
       );
       const generatedImages = collectWechatGeneratedImages(replyRequestMessages(reply));
@@ -2549,6 +2587,11 @@ export function createWechatMpService({
       return { sessionId };
     } catch (err) {
       const message = err instanceof Error ? err.message : String(err);
+      if (err?.code === 'WECHAT_AGENT_REPLY_TIMEOUT') {
+        await userAuth.clearWechatAgentRoute(config.appId, inbound.fromUserName).catch((clearErr) => {
+          logger.warn?.('WeChat MP timed-out session route clear failed:', clearErr);
+        });
+      }
       // A Page Data request must fail closed. Retrying a poisoned completion in
       // another session while the finish guard is also active can turn one
       // request into repairs against historical pages. Drop only this user's
@@ -2611,6 +2654,7 @@ export function createWechatMpService({
                     options: { requireHistoricalImageIsolation: true },
                   })
               : null,
+            timeoutMs: agentReplyTimeoutMs,
           },
         );
         const generatedImages = collectWechatGeneratedImages(replyRequestMessages(reply));

wechat-mp.test.mjs

@@ -22,6 +22,7 @@ import {
   verifyWechatMpSignature,
   verifyWechatMpUrlChallenge,
   decryptWechatMpPayload,
+  executeSessionReply,
   WECHAT_CUSTOMER_TEXT_MAX_BYTES,
 } from './wechat-mp.mjs';
 import { createChatIntentRouter } from './chat-intent-router.mjs';
@@ -69,6 +70,33 @@ function previewReadyPageHtml({ title = 'Page', subtitle = '测试页面', cover
   return `<!doctype html><html><head><meta name="description" content="${subtitle}"><meta name="mindspace-cover" content='{"tag":"页面","accent":"#3366cc","accent2":"#112233","subtitle":"${subtitle}"${coverField}}'><title>${title}</title></head><body><main>${'x'.repeat(600)}</main><p data-mindspace-page-tag="platform-brand">TKMind · 智趣</p></body></html>`;
 }
 
+test('executeSessionReply converts a hanging event stream into a bounded timeout', async () => {
+  let cancelled = false;
+  const body = new ReadableStream({
+    start(controller) {
+      controller.enqueue(new TextEncoder().encode(': connected\n\n'));
+    },
+    cancel() {
+      cancelled = true;
+    },
+  });
+  await assert.rejects(
+    executeSessionReply(
+      async (_pathname, init) => {
+        assert.equal(init.method, 'GET');
+        return { ok: true, body };
+      },
+      'session-timeout',
+      'request-timeout',
+      'hello',
+      {},
+      { timeoutMs: 20, submitReply: async () => {} },
+    ),
+    (error) => error?.code === 'WECHAT_AGENT_REPLY_TIMEOUT',
+  );
+  assert.equal(cancelled, true);
+});
+
 function jsonEscapedPreviewReadyPageHtml(options = {}) {
   return previewReadyPageHtml(options).replace(/\\/g, '\\\\').replace(/"/g, '\\"');
 }