diff --git a/README.md b/README.md index 958d340..70f0fb3 100644 --- a/README.md +++ b/README.md @@ -148,7 +148,7 @@ WeChat ⇄ pi TUI session ⇄ AI model + tools - WeChat messages are fetched via iLink Bot API long polling - Incoming messages are injected into the active pi session via `pi.sendUserMessage()` - When you type in TUI, a preview is sent to WeChat -- AI replies are delivered incrementally (per `message_end`) and finalized on `agent_end` +- AI replies are delivered incrementally (per `message_end`); the `agent_end` catch-up replay was removed to prevent duplicate historical messages after session restore - Only the TUI session that runs `/wechat start` holds the connection ## FAQ @@ -304,7 +304,7 @@ pi install git:github.com/shenjiecode/pi-wechat-assistant - 微信消息通过 iLink Bot API 长轮询获取 - 收到的消息通过 `pi.sendUserMessage()` 注入当前 pi 会话 - TUI 输入时微信端会收到预览 -- AI 回复增量发送（每条 `message_end` 即发），`agent_end` 时补发遗漏 +- AI 回复增量发送（每条 `message_end` 即发），不依赖 `agent_end` 补发（防止恢复会话后历史回复重发） - 只有执行 `/wechat start` 的 TUI 会话持有连接 ## 常见问题 @@ -460,7 +460,7 @@ WeChat ⇄ pi TUIセッション ⇄ AIモデル + ツール - WeChatメッセージはiLink Bot APIのロングポーリングで取得 - 受信メッセージは `pi.sendUserMessage()` で現在のpiセッションに注入 - TUIで入力するとWeChat側にプレビューが送信 -- AI返信は増分的に送信（`message_end` ごと）、`agent_end` で残りを補完 +- AI返信は増分的に送信（`message_end` ごと）、`agent_end` での補完は行わない（セッション復元後の過去返信重複送信を防止） - `/wechat start` を実行したTUIセッションのみが接続を保持 ## FAQ @@ -616,7 +616,7 @@ WeChat ⇄ pi TUI 세션 ⇄ AI 모델 + 도구 - WeChat 메시지는 iLink Bot API 롱 폴링으로 가져옴 - 수신 메시지는 `pi.sendUserMessage()` 로 현재 pi 세션에 주입 - TUI에서 입력하면 WeChat 측에 미리보기 전송 -- AI 응답은 증분 전송 (`message_end` 마다), `agent_end` 에서 나머지 보완 +- AI 응답은 증분 전송 (`message_end` 마다), `agent_end` 보완 없음 (세션 복원 후 과거 응답 중복 방지) - `/wechat start` 를 실행한 TUI 세션만 연결 유지 ## FAQ diff --git a/src/auth.ts b/src/auth.ts index 4ae8d91..6fbb953 100644 --- a/src/auth.ts +++ b/src/auth.ts @@ -15,6 +15,8 @@ const CREDS_FILE = path.join(STATE_DIR, 'credentials.json') const CONFIG_FILE = path.join(STATE_DIR, 'config.json') const LOCK_FILE = path.join(STATE_DIR, 'session.lock') const CONTEXT_TOKENS_FILE = path.join(STATE_DIR, 'context-tokens.json') +const CURSOR_FILE = path.join(STATE_DIR, 'cursor.json') +const SEEN_IDS_FILE = path.join(STATE_DIR, 'seen-ids.json') export function getStateDir(): string { return STATE_DIR @@ -52,6 +54,35 @@ async function deleteFile(filePath: string): Promise { } } +// --- 消息游标（跨重启持久化，防止重放历史消息导致重复回复） --- + +export async function loadCursor(): Promise { + const data = await readJsonFile<{ cursor: string }>(CURSOR_FILE) + return data?.cursor ?? '' +} + +export async function saveCursor(cursor: string): Promise { + if (!cursor) return + await writeJsonFile(CURSOR_FILE, { cursor, updatedAt: new Date().toISOString() }) +} + +// --- 已见消息 id（持久化去重，游标丢失/过期时兜底防重发） --- + +const MAX_SEEN_IDS = 5_000 + +export async function loadSeenIds(): Promise> { + const data = await readJsonFile<{ ids: string[] }>(SEEN_IDS_FILE) + return new Set(data?.ids ?? []) +} + +export async function saveSeenIds(ids: Set): Promise { + const arr = Array.from(ids) + if (arr.length === 0) return + // 只保留最近 N 个，防止文件无限增长 + const trimmed = arr.length > MAX_SEEN_IDS ? arr.slice(arr.length - MAX_SEEN_IDS) : arr + await writeJsonFile(SEEN_IDS_FILE, { ids: trimmed, updatedAt: new Date().toISOString() }) +} + // --- 凭证 --- export async function loadCredentials(): Promise { diff --git a/src/client.ts b/src/client.ts index db2e383..c4d57bd 100644 --- a/src/client.ts +++ b/src/client.ts @@ -22,6 +22,10 @@ import { loadContextTokens, saveContextTokensThrottled, flushContextTokens, + loadCursor, + saveCursor, + loadSeenIds, + saveSeenIds, } from './auth.js' import { CDN_BASE, @@ -130,6 +134,7 @@ export class WeixinClient { private cursor = '' private readonly typingTickets = new Map() private readonly contextTokens = new Map() + private _seenMessageIds = new Set() private _lastActiveUserId: string | null = null private _contextTokensDirty = false private _disposed = false @@ -152,6 +157,10 @@ export class WeixinClient { for (const [userId, token] of Object.entries(persisted.tokens)) { this.contextTokens.set(userId, token) } + // 恢复上次游标，避免重启后服务端重放历史消息导致重复回复 + this.cursor = await loadCursor() + // 恢复已见消息 id（游标丢失/过期时兜底去重） + this._seenMessageIds = await loadSeenIds() } get accountId(): string { return this.credentials.accountId } @@ -187,12 +196,35 @@ export class WeixinClient { } this.cursor = response.get_updates_buf || this.cursor + // 持久化游标（写失败不致命，下次可能重放；由 _seenMessageIds 兜底去重） + if (this.cursor) { + saveCursor(this.cursor).catch(() => {}) + } + const incoming: IncomingMessage[] = [] for (const raw of response.msgs ?? []) { this.rememberContext(raw) const normalized = this.normalizeIncomingMessage(raw) - if (normalized) incoming.push(normalized) + if (!normalized) continue + const id = normalized.messageId + if (id && this._seenMessageIds.has(id)) { + debugLog(`[DEDUP] skip already-seen message ${id}`) + continue + } + if (id) { + this._seenMessageIds.add(id) + // 防止 Set 无限增长：超限重建（极端情况下去重失效，游标仍是主防线） + if (this._seenMessageIds.size > 10_000) { + this._seenMessageIds.clear() + this._seenMessageIds.add(id) + } + } + incoming.push(normalized) + } + // 持久化已见 id（fire-and-forget，节流由文件大小控制） + if (incoming.length > 0) { + saveSeenIds(this._seenMessageIds).catch(() => {}) } return incoming } diff --git a/src/index.ts b/src/index.ts index 169b360..8f9f13b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -15,7 +15,7 @@ import { splitAndFilterMarkdown } from './message.js' import { MessageQueue } from './queue.js' import { handleRemoteCommand, type RemoteCommandDeps } from './remote-commands.js' import { registerCommands, type CommandDeps } from './commands.js' -import { ok, fail, formatError, isAbortError, extractAllAssistantReplies, extractTextFromMessageContent } from './utils.js' +import { ok, fail, formatError, isAbortError, extractTextFromMessageContent } from './utils.js' import { POLL_RETRY_BASE_MS, POLL_RETRY_MAX_MS, @@ -480,7 +480,9 @@ export default function wechatAssistant(pi: ExtensionAPI) { } }) - // agent 结束 → 补发遗漏 + 收尾 + // agent 结束 → 收尾（补发已移除：message_end 已逐条增量发送， + // 补发依赖 event.messages 为“本次 run”的范围假设，omp -c 恢复会话后 + // messages 含全量历史，slice(sentCount) 会把历史回复全部重发到微信） pi.on('agent_end', async (event, ctx) => { latestCtx = ctx agentIdle = true @@ -491,24 +493,6 @@ export default function wechatAssistant(pi: ExtensionAPI) { const assistantMsgs = turn.messages.filter(m => m?.role === 'assistant').length log(`[AGENT-END] turn#${turn.seq} source=${turn.wechatConversationActive ? 'WECHAT' : 'TUI'} targetUser=${turn.targetUser} messages=${msgCount} assistant=${assistantMsgs} sentCount=${turn.sentCount}`) - const allReplies = extractAllAssistantReplies(turn.messages) - const newReplies = allReplies.slice(turn.sentCount) - log(`[AGENT-END-REPLIES] all=${allReplies.length} sent=${turn.sentCount} new=${newReplies.length}`) - - if (turn.wechatConversationActive && newReplies.length > 0 && client && turn.targetUser) { - try { - await queue.sendRepliesToWechat(newReplies, turn.targetUser) - log(`[AGENT-END-DONE] sent ${newReplies.length} remaining replies`) - } catch (err) { - log(`[AGENT-END-ERROR] ${formatError(err)}`) - notify(`发送微信回复失败: ${formatError(err)}`, 'error') - } - } else if (allReplies.length === 0) { - log(`[AGENT-END-NOREPLY] no assistant text`) - } else { - log(`[AGENT-END-SAFE] all replies already sent incrementally`) - } - if (queue.activeRequest) { await client?.stopTyping(queue.activeRequest.userId).catch(() => {}) queue.activeRequest = null diff --git a/src/queue.ts b/src/queue.ts index 0f700b5..fbef963 100644 --- a/src/queue.ts +++ b/src/queue.ts @@ -6,7 +6,6 @@ import { randomUUID } from 'node:crypto' import type { ExtensionContext, ExtensionCommandContext } from '@mariozechner/pi-coding-agent' import { debugLog } from './logger.js' import { WeixinClient } from './client.js' -import { splitAndFilterMarkdown } from './message.js' import { fetchImageAsBase64, fetchFile, saveFileToDisk, type ImageData } from './media.js' import { ACK_TEXT, @@ -374,19 +373,6 @@ export class MessageQueue { } } - // --- 统一发送到微信 --- - - async sendRepliesToWechat(replies: string[], targetUserId: string): Promise { - const client = this.getClient() - if (!client) return - for (const reply of replies) { - const chunks = splitAndFilterMarkdown(reply) - for (const chunk of chunks) { - await client.sendText(targetUserId, chunk) - } - } - } - // --- 重置 --- reset(): void { diff --git a/src/utils.ts b/src/utils.ts index d3f0414..dcba6f7 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -19,33 +19,6 @@ export async function renderQrCode(url: string): Promise { }) } -export function extractAllAssistantReplies( - messages: Array<{ role?: string; content?: unknown }>, -): string[] { - const replies: string[] = [] - for (let i = 0; i < messages.length; i++) { - const message = messages[i] - if (message?.role !== 'assistant') continue - if (typeof message.content === 'string') { - const text = message.content.trim() - if (text) replies.push(text) - continue - } - if (!Array.isArray(message.content)) continue - const text = message.content - .filter( - (part): part is { type: 'text'; text: string } => - typeof part === 'object' && part !== null && (part as { type?: string }).type === 'text', - ) - .map((part) => part.text.trim()) - .filter(Boolean) - .join('\n') - .trim() - if (text) replies.push(text) - } - return replies -} - export function extractTextFromMessageContent(content: unknown): string | null { if (typeof content === 'string') return content.trim() || null if (!Array.isArray(content)) return null diff --git a/tests/utils.test.ts b/tests/utils.test.ts index 4674099..e8aa0e9 100644 --- a/tests/utils.test.ts +++ b/tests/utils.test.ts @@ -6,7 +6,6 @@ import { describe, it, expect } from 'vitest' import { ok, fail, - extractAllAssistantReplies, extractTextFromMessageContent, summarizePreview, formatError, @@ -26,50 +25,6 @@ describe('ok / fail', () => { }) }) -describe('extractAllAssistantReplies', () => { - it('extracts string content from assistant messages', () => { - const messages = [ - { role: 'user', content: 'hello' }, - { role: 'assistant', content: 'Hi there' }, - { role: 'assistant', content: ' How can I help? ' }, - ] - expect(extractAllAssistantReplies(messages)).toEqual(['Hi there', 'How can I help?']) - }) - - it('extracts array content with text parts', () => { - const messages = [ - { - role: 'assistant', - content: [ - { type: 'text', text: 'Part 1' }, - { type: 'toolCall', name: 'send_file' }, - { type: 'text', text: 'Part 2' }, - ], - }, - ] - expect(extractAllAssistantReplies(messages)).toEqual(['Part 1\nPart 2']) - }) - - it('skips non-assistant messages', () => { - const messages = [ - { role: 'user', content: 'hello' }, - { role: 'toolResult', content: 'result' }, - ] - expect(extractAllAssistantReplies(messages)).toEqual([]) - }) - - it('skips empty assistant messages', () => { - const messages = [ - { role: 'assistant', content: '' }, - { role: 'assistant', content: ' ' }, - ] - expect(extractAllAssistantReplies(messages)).toEqual([]) - }) - - it('handles empty array', () => { - expect(extractAllAssistantReplies([])).toEqual([]) - }) -}) describe('extractTextFromMessageContent', () => { it('extracts from string', () => {