fix: keep WeChat agent replies visible

WeChat development turns were reaching the local CLI, but the adapter treated iLink business failures as successful sends and only emitted a short-lived typing indicator. The adapter now validates sendmessage/sendtyping ret codes, retries text replies without stale context tokens, keeps typing alive during long-running tool work, and surfaces queue failures back to the chat. Constraint: WeChat uses separate sendmessage and sendtyping APIs with business-level ret codes inside HTTP 200 responses. Rejected: Only fixing the typing indicator | the transcript showed the agent completed the task, so the reply delivery path also needed hardening. Confidence: high Scope-risk: narrow Directive: Do not treat WeChat HTTP 200 responses as successful until the iLink ret/errcode body has been checked. Tested: bun test adapters/wechat/__tests__/protocol.test.ts adapters/wechat/__tests__/typing.test.ts Tested: cd adapters && bunx tsc --noEmit Tested: bun run check:adapters Tested: real bound WeChat getconfig, typing, text send, cancel typing smoke Not-tested: Full inbound WeChat user-message loop without a fresh user-triggered message
2026-07-18 13:23:33 +08:00 · 2026-05-05 19:58:53 +08:00 · 2026-05-05 19:58:53 +08:00 · 622b94011f
commit 622b94011f
parent 71f851d60a
5 changed files with 284 additions and 28 deletions
--- a/adapters/wechat/tests/protocol.test.ts
+++ b/adapters/wechat/tests/protocol.test.ts
@ -1,7 +1,13 @@
-import { describe, expect, it } from 'bun:test'
-import { buildClientVersion, extractWechatText } from '../protocol.js'
+import { afterEach, describe, expect, it } from 'bun:test'
+import { buildClientVersion, extractWechatText, sendWechatText, sendWechatTyping } from '../protocol.js'
 import { collectWechatMediaCandidates } from '../media.js'

+const originalFetch = globalThis.fetch
+
+afterEach(() => {
+  globalThis.fetch = originalFetch
+})
+
 describe('WeChat protocol helpers', () => {
  it('encodes iLink client versions like the OpenClaw Weixin plugin', () => {
    expect(buildClientVersion('2.1.7')).toBe((2 << 16) | (1 << 8) | 7)
@ -71,4 +77,47 @@ describe('WeChat protocol helpers', () => {
      },
    ])
  })
+
+  it('throws when sendmessage returns a non-zero WeChat ret code', async () => {
+    globalThis.fetch = (async () => new Response(JSON.stringify({ ret: 40001, errmsg: 'bad context_token' }), { status: 200 })) as unknown as typeof fetch
+
+    await expect(sendWechatText({
+      baseUrl: 'https://api.example.com',
+      token: 'token',
+      to: 'user',
+      text: 'hello',
+      contextToken: 'stale-context',
+    })).rejects.toThrow('wechatSendMessage returned 40001: bad context_token')
+  })
+
+  it('allows successful sendmessage responses', async () => {
+    const requests: string[] = []
+    globalThis.fetch = (async (_url: Parameters<typeof fetch>[0], init?: Parameters<typeof fetch>[1]) => {
+      requests.push(String(init?.body ?? ''))
+      return new Response(JSON.stringify({ ret: 0 }), { status: 200 })
+    }) as unknown as typeof fetch
+
+    await sendWechatText({
+      baseUrl: 'https://api.example.com',
+      token: 'token',
+      to: 'user',
+      text: 'hello',
+      contextToken: 'ctx',
+    })
+
+    expect(requests).toHaveLength(1)
+    expect(JSON.parse(requests[0]!).msg.context_token).toBe('ctx')
+  })
+
+  it('throws when sendtyping returns a non-zero WeChat ret code', async () => {
+    globalThis.fetch = (async () => new Response(JSON.stringify({ ret: 42001, errmsg: 'typing ticket expired' }), { status: 200 })) as unknown as typeof fetch
+
+    await expect(sendWechatTyping({
+      baseUrl: 'https://api.example.com',
+      token: 'token',
+      ilinkUserId: 'user',
+      typingTicket: 'ticket',
+      status: 'typing',
+    })).rejects.toThrow('wechatSendTyping returned 42001: typing ticket expired')
+  })
 })
--- a/adapters/wechat/tests/typing.test.ts
+++ b/adapters/wechat/tests/typing.test.ts
@ -0,0 +1,38 @@
+import { describe, expect, it } from 'bun:test'
+import { WechatTypingController } from '../typing.js'
+
+describe('WechatTypingController', () => {
+  it('sends typing immediately and keeps it alive until stopped', async () => {
+    const calls: Array<[string, string]> = []
+    const controller = new WechatTypingController(async (chatId, status) => {
+      calls.push([chatId, status])
+    }, 10)
+
+    controller.start('chat-1')
+    await Bun.sleep(25)
+    controller.stop('chat-1')
+    controller.destroy()
+
+    expect(calls[0]).toEqual(['chat-1', 'typing'])
+    expect(calls.filter(([, status]) => status === 'typing').length).toBeGreaterThanOrEqual(2)
+    expect(calls.at(-1)).toEqual(['chat-1', 'cancel'])
+  })
+
+  it('does not create duplicate keepalive timers for the same chat', async () => {
+    const calls: Array<[string, string]> = []
+    const controller = new WechatTypingController(async (chatId, status) => {
+      calls.push([chatId, status])
+    }, 10)
+
+    controller.start('chat-1')
+    controller.start('chat-1')
+    await Bun.sleep(25)
+    controller.stop('chat-1')
+    controller.destroy()
+
+    const typingCalls = calls.filter(([, status]) => status === 'typing')
+    expect(typingCalls.length).toBeGreaterThanOrEqual(3)
+    expect(typingCalls.length).toBeLessThanOrEqual(5)
+    expect(calls.at(-1)).toEqual(['chat-1', 'cancel'])
+  })
+})
--- a/adapters/wechat/index.ts
+++ b/adapters/wechat/index.ts
@ -20,6 +20,7 @@ import { AdapterHttpClient } from '../common/http-client.js'
 import { isAllowedUser, tryPair } from '../common/pairing.js'
 import { AttachmentStore } from '../common/attachment/attachment-store.js'
 import { checkAttachmentLimit } from '../common/attachment/attachment-limits.js'
+import { WechatTypingController } from './typing.js'
 import {
  extractWechatText,
  getWechatConfig,
@ -56,6 +57,7 @@ const blockBuffers = new Map<string, MessageBuffer>()
 const contextTokens = new Map<string, string>()
 const typingTickets = new Map<string, string>()
 const pendingPermissions = new Map<string, Set<string>>()
+const typingController = new WechatTypingController(sendTypingIndicator)

 let getUpdatesBuf = ''
 let stopped = false
@ -84,13 +86,24 @@ async function sendText(chatId: string, text: string): Promise<void> {
  const chunks = splitMessage(text, WECHAT_TEXT_LIMIT)
  const contextToken = contextTokens.get(chatId)
  for (const chunk of chunks) {
-    await sendWechatText({
-      baseUrl,
-      token: botToken,
-      to: chatId,
-      text: chunk,
-      contextToken,
-    })
+    try {
+      await sendWechatText({
+        baseUrl,
+        token: botToken,
+        to: chatId,
+        text: chunk,
+        contextToken,
+      })
+    } catch (err) {
+      if (!contextToken) throw err
+      console.warn('[WeChat] sendText with context token failed, retrying without context:', err instanceof Error ? err.message : err)
+      await sendWechatText({
+        baseUrl,
+        token: botToken,
+        to: chatId,
+        text: chunk,
+      })
+    }
  }
  console.log(`[WeChat] Sent ${chunks.length} message chunk(s) to ${redactChatId(chatId)}`)
 }
@ -112,17 +125,7 @@ function getBlockBuffer(chatId: string): MessageBuffer {

 async function sendTypingIndicator(chatId: string, status: 'typing' | 'cancel'): Promise<void> {
  try {
-    let typingTicket = typingTickets.get(chatId)
-    if (!typingTicket && status === 'typing') {
-      const configResp = await getWechatConfig({
-        baseUrl,
-        token: botToken,
-        ilinkUserId: chatId,
-        contextToken: contextTokens.get(chatId),
-      })
-      typingTicket = configResp.typing_ticket
-      if (typingTicket) typingTickets.set(chatId, typingTicket)
-    }
+    const typingTicket = await getTypingTicket(chatId, status)
    if (!typingTicket) return
    await sendWechatTyping({
      baseUrl,
@ -132,20 +135,73 @@ async function sendTypingIndicator(chatId: string, status: 'typing' | 'cancel'):
      status,
    })
  } catch (err) {
+    typingTickets.delete(chatId)
+    if (status === 'typing') {
+      try {
+        const typingTicket = await getTypingTicket(chatId, status)
+        if (!typingTicket) return
+        await sendWechatTyping({
+          baseUrl,
+          token: botToken,
+          ilinkUserId: chatId,
+          typingTicket,
+          status,
+        })
+        return
+      } catch {
+        // Fall through to the warning below with the original error.
+      }
+    }
    console.warn('[WeChat] sendTyping failed:', err instanceof Error ? err.message : err)
  }
 }

+async function getTypingTicket(chatId: string, status: 'typing' | 'cancel'): Promise<string | null> {
+  let typingTicket = typingTickets.get(chatId)
+  if (!typingTicket && status === 'typing') {
+    const configResp = await getWechatConfig({
+      baseUrl,
+      token: botToken,
+      ilinkUserId: chatId,
+      contextToken: contextTokens.get(chatId),
+    })
+    if (typeof configResp.ret === 'number' && configResp.ret !== 0) {
+      throw new Error(`getconfig returned ${configResp.ret}: ${configResp.errmsg ?? ''}`)
+    }
+    typingTicket = configResp.typing_ticket
+    if (typingTicket) typingTickets.set(chatId, typingTicket)
+  }
+  return typingTicket || null
+}
+
 function clearTransientChatState(chatId: string): void {
  blockBuffers.get(chatId)?.reset()
  blockBuffers.delete(chatId)
  pendingPermissions.delete(chatId)
+  typingController.stop(chatId)
  const runtime = getRuntimeState(chatId)
  runtime.state = 'idle'
  runtime.verb = undefined
  runtime.pendingPermissionCount = 0
 }

+function enqueueWechat(chatId: string, task: () => Promise<void>): void {
+  enqueue(chatId, async () => {
+    try {
+      await task()
+    } catch (err) {
+      typingController.stop(chatId)
+      const message = err instanceof Error ? err.message : String(err)
+      console.error(`[WeChat] Failed to handle message for ${redactChatId(chatId)}:`, err)
+      try {
+        await sendText(chatId, `处理消息失败：${message}`)
+      } catch (sendErr) {
+        console.error(`[WeChat] Failed to report message handling error for ${redactChatId(chatId)}:`, sendErr)
+      }
+    }
+  })
+}
+
 async function ensureExistingSession(chatId: string): Promise<{ sessionId: string; workDir: string } | null> {
  const stored = sessionStore.get(chatId)
  if (!stored) return null
@ -176,6 +232,29 @@ async function buildStatusText(chatId: string): Promise<string> {
    // Keep IM status best-effort.
  }

+  let taskCounts:
+    | {
+        total: number
+        pending: number
+        inProgress: number
+        completed: number
+      }
+    | undefined
+
+  try {
+    const tasks = await httpClient.getTasksForSession(stored.sessionId)
+    if (tasks.length > 0) {
+      taskCounts = {
+        total: tasks.length,
+        pending: tasks.filter((task) => task.status === 'pending').length,
+        inProgress: tasks.filter((task) => task.status === 'in_progress').length,
+        completed: tasks.filter((task) => task.status === 'completed').length,
+      }
+    }
+  } catch {
+    // Keep IM status best-effort.
+  }
+
  return formatImStatus({
    sessionId: stored.sessionId,
    projectName,
@ -184,6 +263,7 @@ async function buildStatusText(chatId: string): Promise<string> {
    state: runtime.state,
    verb: runtime.verb,
    pendingPermissionCount: runtime.pendingPermissionCount,
+    taskCounts,
  })
 }

@ -287,9 +367,18 @@ async function handleServerMessage(chatId: string, msg: ServerMessage): Promise<
      runtime.state = msg.state
      runtime.verb = typeof msg.verb === 'string' ? msg.verb : undefined
      if (msg.state === 'thinking' || msg.state === 'tool_executing') {
-        void sendTypingIndicator(chatId, 'typing')
+        typingController.start(chatId)
      } else if (msg.state === 'idle') {
-        void sendTypingIndicator(chatId, 'cancel')
+        typingController.stop(chatId)
+      }
+      break
+    case 'content_start':
+      if (msg.blockType === 'text') {
+        runtime.state = 'streaming'
+      } else if (msg.blockType === 'tool_use') {
+        runtime.state = 'tool_executing'
+        runtime.verb = typeof msg.toolName === 'string' ? msg.toolName : runtime.verb
+        typingController.start(chatId)
      }
      break
    case 'content_delta':
@ -297,6 +386,16 @@ async function handleServerMessage(chatId: string, msg: ServerMessage): Promise<
        getBlockBuffer(chatId).append(msg.text)
      }
      break
+    case 'tool_use_complete':
+      runtime.state = 'tool_executing'
+      runtime.verb = typeof msg.toolName === 'string' ? msg.toolName : runtime.verb
+      typingController.start(chatId)
+      break
+    case 'tool_result':
+      runtime.state = 'thinking'
+      runtime.verb = undefined
+      typingController.start(chatId)
+      break
    case 'permission_request': {
      runtime.pendingPermissionCount += 1
      runtime.state = 'permission_pending'
@ -306,6 +405,7 @@ async function handleServerMessage(chatId: string, msg: ServerMessage): Promise<
        pendingPermissions.set(chatId, pending)
      }
      pending.add(msg.requestId)
+      typingController.stop(chatId)
      await sendText(
        chatId,
        `${formatPermissionRequest(msg.toolName, msg.input, msg.requestId)}\n\n${formatPermissionInstructions(msg.requestId)}`,
@ -315,7 +415,7 @@ async function handleServerMessage(chatId: string, msg: ServerMessage): Promise<
    case 'message_complete': {
      runtime.state = 'idle'
      runtime.verb = undefined
-      void sendTypingIndicator(chatId, 'cancel')
+      typingController.stop(chatId)
      await blockBuffers.get(chatId)?.complete()
      blockBuffers.delete(chatId)
      break
@ -323,7 +423,7 @@ async function handleServerMessage(chatId: string, msg: ServerMessage): Promise<
    case 'error':
      runtime.state = 'idle'
      runtime.verb = undefined
-      void sendTypingIndicator(chatId, 'cancel')
+      typingController.stop(chatId)
      blockBuffers.get(chatId)?.reset()
      blockBuffers.delete(chatId)
      await sendText(chatId, `错误: ${msg.message}`)
@ -362,7 +462,7 @@ async function routeUserMessage(message: WechatMessage): Promise<void> {
    return
  }

-  enqueue(chatId, async () => {
+  enqueueWechat(chatId, async () => {
    const hasAttachments = mediaCandidates.length > 0
    if (!hasAttachments && (text === '/help' || text === '帮助')) {
      await sendText(chatId, formatImHelp())
@ -429,7 +529,7 @@ async function routeUserMessage(message: WechatMessage): Promise<void> {
    const attachments = await collectAttachments(chatId, mediaCandidates)
    const effectiveText = text || (attachments.length > 0 ? '(用户发送了附件)' : '')
    if (!effectiveText && attachments.length === 0) return
-    void sendTypingIndicator(chatId, 'typing')
+    typingController.start(chatId)
    const sent = bridge.sendUserMessage(chatId, effectiveText, attachments.length ? attachments : undefined)
    if (!sent) await sendText(chatId, '消息发送失败，连接可能已断开。请发送 /new 重新开始。')
  })
@ -525,6 +625,7 @@ void pollLoop()
 process.on('SIGINT', () => {
  console.log('[WeChat] Shutting down...')
  stopped = true
+  typingController.destroy()
  bridge.destroy()
  dedup.destroy()
  process.exit(0)
--- a/adapters/wechat/protocol.ts
+++ b/adapters/wechat/protocol.ts
@ -275,7 +275,7 @@ export async function sendWechatText(params: {
    base_info: buildBaseInfo(),
  }

-  await apiPostFetch({
+  const rawText = await apiPostFetch({
    baseUrl: params.baseUrl,
    endpoint: 'ilink/bot/sendmessage',
    body: JSON.stringify(body),
@ -283,6 +283,7 @@ export async function sendWechatText(params: {
    timeoutMs: params.timeoutMs ?? API_TIMEOUT_MS,
    label: 'wechatSendMessage',
  })
+  assertWechatApiOk(rawText, 'wechatSendMessage')
 }

 export async function getWechatConfig(params: {
@ -315,7 +316,7 @@ export async function sendWechatTyping(params: {
  status: 'typing' | 'cancel'
  timeoutMs?: number
 }): Promise<void> {
-  await apiPostFetch({
+  const rawText = await apiPostFetch({
    baseUrl: params.baseUrl,
    endpoint: 'ilink/bot/sendtyping',
    body: JSON.stringify({
@ -328,6 +329,7 @@ export async function sendWechatTyping(params: {
    timeoutMs: params.timeoutMs ?? 10_000,
    label: 'wechatSendTyping',
  })
+  assertWechatApiOk(rawText, 'wechatSendTyping')
 }

 async function pollQrStatus(apiBaseUrl: string, qrcode: string): Promise<QrStatusResponse> {
@ -428,6 +430,30 @@ function randomWechatUin(): string {
  return Buffer.from(String(uint32), 'utf-8').toString('base64')
 }

+function assertWechatApiOk(rawText: string, label: string): void {
+  if (!rawText.trim()) return
+
+  let body: unknown
+  try {
+    body = JSON.parse(rawText)
+  } catch {
+    return
+  }
+
+  if (!body || typeof body !== 'object') return
+
+  const record = body as Record<string, unknown>
+  const code = typeof record.ret === 'number'
+    ? record.ret
+    : typeof record.errcode === 'number'
+      ? record.errcode
+      : 0
+  if (code === 0) return
+
+  const message = typeof record.errmsg === 'string' ? record.errmsg : rawText
+  throw new Error(`${label} returned ${code}: ${message}`)
+}
+
 function isLoginFresh(login: ActiveLogin): boolean {
  return Date.now() - login.startedAt < QR_LOGIN_TTL_MS
 }
--- a/adapters/wechat/typing.ts
+++ b/adapters/wechat/typing.ts
@ -0,0 +1,42 @@
+export type WechatTypingStatus = 'typing' | 'cancel'
+
+type SendTyping = (chatId: string, status: WechatTypingStatus) => Promise<void>
+
+/**
+ * WeChat typing indicators expire quickly. Keep sending typing while the
+ * desktop session is active, then cancel when the turn completes.
+ */
+export class WechatTypingController {
+  private readonly active = new Map<string, ReturnType<typeof setInterval>>()
+
+  constructor(
+    private readonly sendTyping: SendTyping,
+    private readonly keepaliveIntervalMs = 5000,
+  ) {}
+
+  start(chatId: string): void {
+    void this.sendTyping(chatId, 'typing')
+    if (this.active.has(chatId)) return
+
+    const timer = setInterval(() => {
+      void this.sendTyping(chatId, 'typing')
+    }, this.keepaliveIntervalMs)
+    this.active.set(chatId, timer)
+  }
+
+  stop(chatId: string): void {
+    const timer = this.active.get(chatId)
+    if (timer) {
+      clearInterval(timer)
+      this.active.delete(chatId)
+    }
+    void this.sendTyping(chatId, 'cancel')
+  }
+
+  destroy(): void {
+    for (const timer of this.active.values()) {
+      clearInterval(timer)
+    }
+    this.active.clear()
+  }
+}