diff --git a/adapters/wechat/__tests__/protocol.test.ts b/adapters/wechat/__tests__/protocol.test.ts index ab377de1..602fe779 100644 --- 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[0], init?: Parameters[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') + }) }) diff --git a/adapters/wechat/__tests__/typing.test.ts b/adapters/wechat/__tests__/typing.test.ts new file mode 100644 index 00000000..45cca8c6 --- /dev/null +++ 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']) + }) +}) diff --git a/adapters/wechat/index.ts b/adapters/wechat/index.ts index b1a577fb..273e71e3 100644 --- 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() const contextTokens = new Map() const typingTickets = new Map() const pendingPermissions = new Map>() +const typingController = new WechatTypingController(sendTypingIndicator) let getUpdatesBuf = '' let stopped = false @@ -84,13 +86,24 @@ async function sendText(chatId: string, text: string): Promise { 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 { 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 { + 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 { + 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 { // 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 { 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 { 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 { 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) diff --git a/adapters/wechat/protocol.ts b/adapters/wechat/protocol.ts index e7e0a751..4bf3c00f 100644 --- 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 { - 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 { @@ -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 + 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 } diff --git a/adapters/wechat/typing.ts b/adapters/wechat/typing.ts new file mode 100644 index 00000000..d907c5af --- /dev/null +++ b/adapters/wechat/typing.ts @@ -0,0 +1,42 @@ +export type WechatTypingStatus = 'typing' | 'cancel' + +type SendTyping = (chatId: string, status: WechatTypingStatus) => Promise + +/** + * 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>() + + 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() + } +} diff --git a/desktop/src/api/adapters.ts b/desktop/src/api/adapters.ts index 8855c48d..8bac289b 100644 --- a/desktop/src/api/adapters.ts +++ b/desktop/src/api/adapters.ts @@ -41,6 +41,10 @@ export const adaptersApi = { return api.post('/api/adapters/wechat/unbind', {}) }, + unbindDingtalk() { + return api.post('/api/adapters/dingtalk/unbind', {}) + }, + beginDingtalkRegistration() { return api.post('/api/adapters/dingtalk/registration/begin', {}) }, diff --git a/desktop/src/stores/adapterStore.test.ts b/desktop/src/stores/adapterStore.test.ts index 88cdf057..73ab270c 100644 --- a/desktop/src/stores/adapterStore.test.ts +++ b/desktop/src/stores/adapterStore.test.ts @@ -7,6 +7,7 @@ describe('adapterStore IM pairing behavior', () => { startWechatLogin: vi.fn(), pollWechatLogin: vi.fn(), unbindWechat: vi.fn(), + unbindDingtalk: vi.fn(), beginDingtalkRegistration: vi.fn(), pollDingtalkRegistration: vi.fn(), } @@ -59,4 +60,17 @@ describe('adapterStore IM pairing behavior', () => { expect(adaptersApi.unbindWechat).toHaveBeenCalledTimes(1) expect(useAdapterStore.getState().config).toBe(nextConfig) }) + + it('unbinds the DingTalk bot through the explicit bot action', async () => { + const nextConfig = { dingtalk: { pairedUsers: [], allowedUsers: [] } } + adaptersApi.unbindDingtalk.mockResolvedValue(nextConfig) + + const { useAdapterStore } = await import('./adapterStore') + + await useAdapterStore.getState().unbindDingtalkBot() + + expect(adaptersApi.updateConfig).not.toHaveBeenCalled() + expect(adaptersApi.unbindDingtalk).toHaveBeenCalledTimes(1) + expect(useAdapterStore.getState().config).toBe(nextConfig) + }) }) diff --git a/desktop/src/stores/adapterStore.ts b/desktop/src/stores/adapterStore.ts index 47ff82b5..69dbd799 100644 --- a/desktop/src/stores/adapterStore.ts +++ b/desktop/src/stores/adapterStore.ts @@ -132,13 +132,9 @@ export const useAdapterStore = create((set, get) => ({ }, unbindDingtalkBot: async () => { - await get().updateConfig({ - dingtalk: { - clientId: undefined, - clientSecret: undefined, - pairedUsers: [], - }, - }) + const config = await adaptersApi.unbindDingtalk() + set({ config }) + void notifyTauriRestartAdapters() }, removePairedUser: async (platform, userId) => { diff --git a/src/server/__tests__/adapters.test.ts b/src/server/__tests__/adapters.test.ts index 7229a7bb..fd3c7d11 100644 --- a/src/server/__tests__/adapters.test.ts +++ b/src/server/__tests__/adapters.test.ts @@ -125,4 +125,27 @@ describe('Adapters API', () => { expect(json.wechat.allowedUsers).toEqual([]) expect(json.wechat.pairedUsers).toEqual([]) }) + + it('clears DingTalk credentials on unbind', async () => { + const put = makeRequest('PUT', '/api/adapters', { + dingtalk: { + clientId: 'ding-client-1', + clientSecret: 'dingtalk-client-secret', + allowedUsers: ['ding-allowed-user'], + permissionCardTemplateId: 'permission-template', + pairedUsers: [{ userId: 'ding-user', displayName: 'DingTalk User', pairedAt: 1 }], + }, + }) + await handleAdaptersApi(put.req, put.url, put.segments) + + const unbind = makeRequest('POST', '/api/adapters/dingtalk/unbind') + const res = await handleAdaptersApi(unbind.req, unbind.url, unbind.segments) + expect(res.status).toBe(200) + const json = await res.json() as any + expect(json.dingtalk.clientId).toBeUndefined() + expect(json.dingtalk.clientSecret).toBeUndefined() + expect(json.dingtalk.allowedUsers).toEqual([]) + expect(json.dingtalk.permissionCardTemplateId).toBeUndefined() + expect(json.dingtalk.pairedUsers).toEqual([]) + }) }) diff --git a/src/server/api/adapters.ts b/src/server/api/adapters.ts index 5e3a6096..c4e67cae 100644 --- a/src/server/api/adapters.ts +++ b/src/server/api/adapters.ts @@ -143,6 +143,18 @@ export async function handleAdaptersApi( if (tail[0] === 'wechat') { return handleWechatAdaptersApi(req, tail.slice(1)) } + if (tail[0] === 'dingtalk' && req.method === 'POST' && tail[1] === 'unbind') { + await adapterService.updateConfig({ + dingtalk: { + clientId: undefined, + clientSecret: undefined, + allowedUsers: [], + pairedUsers: [], + permissionCardTemplateId: undefined, + }, + }) + return Response.json(await adapterService.getConfig()) + } if (tail[0] === 'dingtalk' && tail[1] === 'registration') { if (req.method === 'POST' && tail[2] === 'begin') { return Response.json(await beginDingtalkRegistration())