From 69765a2b2bf240dd12de3350b9cb1f139b7ab097 Mon Sep 17 00:00:00 2001 From: scotthuang Date: Wed, 2 Sep 2026 02:08:18 +0800 Subject: [PATCH 1/5] feat: resolve ID-only Weixin quotes from local cache --- CHANGELOG.md | 7 + CHANGELOG.zh_CN.md | 7 + README.md | 32 ++ README.zh_CN.md | 28 ++ src/api/api.test.ts | 56 ++- src/api/api.ts | 70 ++- src/api/types.ts | 17 +- src/auth/accounts.ts | 2 + src/channel.ts | 32 +- src/config/config-schema.test.ts | 14 + src/config/config-schema.ts | 17 + src/messaging/error-notice.ts | 2 + src/messaging/inbound.test.ts | 224 +++++++++- src/messaging/inbound.ts | 146 ++++++- src/messaging/partial-quote.test.ts | 50 +++ src/messaging/partial-quote.ts | 65 +++ src/messaging/process-message.ts | 46 +- src/messaging/quote-store.test.ts | 432 +++++++++++++++++++ src/messaging/quote-store.ts | 570 +++++++++++++++++++++++++ src/messaging/reply-progress-sender.ts | 3 +- src/messaging/send-media.test.ts | 2 + src/messaging/send-media.ts | 8 +- src/messaging/send.test.ts | 88 +++- src/messaging/send.ts | 134 +++++- src/messaging/slash-commands.ts | 3 +- 25 files changed, 2004 insertions(+), 51 deletions(-) create mode 100644 src/messaging/partial-quote.test.ts create mode 100644 src/messaging/partial-quote.ts create mode 100644 src/messaging/quote-store.test.ts create mode 100644 src/messaging/quote-store.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 67718f9..718c6f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,13 @@ This project follows the [Keep a Changelog](https://keepachangelog.com/) format. +## [Unreleased] + +### Added + +- **Quote reconstruction for newer WeChat clients:** Losslessly parse message IDs and resolve ID-only `svr_id` text and partial quotes through a per-account, per-conversation SQLite side store. Images, video, voice, and attachments are copied into managed storage with time, count, byte-budget, and single-file eviction limits. +- **Compatible degradation:** Disable quote caching when `node:sqlite` is unavailable or `quoteCache.enabled=false`, with no in-memory fallback. Cache failures never interrupt normal message delivery. + ## [2.4.7] - 2026-08-31 ### Fixed diff --git a/CHANGELOG.zh_CN.md b/CHANGELOG.zh_CN.md index c052898..668f5e1 100644 --- a/CHANGELOG.zh_CN.md +++ b/CHANGELOG.zh_CN.md @@ -4,6 +4,13 @@ 本项目遵循 [Keep a Changelog](https://keepachangelog.com/) 格式。 +## [未发布] + +### 新增 + +- **新版微信引用消息还原：** 对消息 ID 做无损解析，并用按账号、会话隔离的 SQLite 旁路存储还原只携带 `svr_id` 的文本引用和局部引用。图片、视频、语音及附件会复制到受管目录，支持按时间、条数、空间和单文件大小淘汰。 +- **兼容降级：** `node:sqlite` 不可用或 `quoteCache.enabled=false` 时关闭引用缓存，不启用内存替代方案；缓存异常不会影响正常消息收发。 + ## [2.4.7] - 2026-08-31 ### 修复 diff --git a/README.md b/README.md index 05ad188..f853a0b 100644 --- a/README.md +++ b/README.md @@ -111,6 +111,38 @@ authentication or routing. All registered agents on this plugin instance currently share the same `botAgent` declaration; per-agent overrides may be added in a future version if needed. +## Local quote cache + +Newer WeChat clients may send only a server message ID for a quoted message. The +plugin therefore stores text and media metadata in SQLite and copies images, +video, voice, and attachments into managed storage. Records are isolated by +account and conversation. By default, text is retained for 30 days with a limit +of 10,000 messages per account; media is retained for 7 days with a 256 MiB +per-account budget and a 25 MiB single-file limit. Cleanup runs at startup, +hourly, every 100 writes, whenever the media budget is exceeded, and when an +account is deleted. + +If `node:sqlite` is unavailable, the plugin logs a warning and disables this +feature; it does not fall back to an in-memory cache. You can also disable it or +change the limits explicitly: + +```json +{ + "channels": { + "openclaw-weixin": { + "quoteCache": { + "enabled": false, + "retentionDays": 30, + "maxMessagesPerAccount": 10000, + "mediaRetentionDays": 7, + "maxMediaBytesPerAccount": 268435456, + "maxSingleMediaBytes": 26214400 + } + } + } +} +``` + ## Backend API Protocol This plugin communicates with the backend gateway via HTTP JSON API. Developers integrating with their own backend need to implement the following interfaces. diff --git a/README.zh_CN.md b/README.zh_CN.md index 9322541..6a60af5 100644 --- a/README.zh_CN.md +++ b/README.zh_CN.md @@ -107,6 +107,34 @@ openclaw config set session.dmScope per-account-channel-peer 已注册的 agent 共享同一个 `botAgent` 声明；如有需要按 agent 单独标识的场景， 可在后续版本扩展配置。 +## 引用消息本地缓存 + +新版微信的引用消息可能只携带服务端消息 ID。插件默认使用 SQLite 保存文本和媒体元数据， +并将图片、视频、语音及附件复制到受管目录，从而在后续引用时还原原消息。缓存按账号和会话 +隔离；文本默认保留 30 天且每账号最多 10,000 条，媒体默认保留 7 天、每账号最多 256 MiB、 +单文件最多 25 MiB。淘汰在启动时、每小时、每写入 100 条及媒体超出空间上限时触发；删除 +账号时会同步删除其引用缓存。 + +如当前 Node.js 不提供 `node:sqlite`，插件会记录警告并自动关闭此功能，不使用内存缓存降级。 +也可以显式关闭或调整限制： + +```json +{ + "channels": { + "openclaw-weixin": { + "quoteCache": { + "enabled": false, + "retentionDays": 30, + "maxMessagesPerAccount": 10000, + "mediaRetentionDays": 7, + "maxMediaBytesPerAccount": 268435456, + "maxSingleMediaBytes": 26214400 + } + } + } +} +``` + ## 后端 API 协议 本插件通过 HTTP JSON API 与后端网关通信。二次开发者若需对接自有后端，需实现以下接口。 diff --git a/src/api/api.test.ts b/src/api/api.test.ts index 831289f..7f446ca 100644 --- a/src/api/api.test.ts +++ b/src/api/api.test.ts @@ -37,6 +37,7 @@ import { sanitizeBotAgent, readPackageJsonFromDir, classifyFetchError, + parseWeixinApiJson, } from "./api.js"; function mockResponse(body: object | string, status = 200, ok = true): Response { @@ -179,7 +180,14 @@ describe("sendMessage", () => { mockFetch.mockResolvedValueOnce({ ok: true, status: 200, text: () => Promise.resolve("{}") } as Response); await expect( sendMessage({ baseUrl: "https://api.example.com/", body: { msg: { to_user_id: "u" } } }), - ).resolves.toBeUndefined(); + ).resolves.toEqual({}); + }); + + it("returns a lossless server message ID", async () => { + mockFetch.mockResolvedValueOnce(mockResponse('{"ret":0,"message_id":18446744073709551615}')); + await expect( + sendMessage({ baseUrl: "https://api.example.com/", body: { msg: { to_user_id: "u" } } }), + ).resolves.toMatchObject({ message_id: "18446744073709551615" }); }); it("throws on non-ok response", async () => { @@ -188,6 +196,52 @@ describe("sendMessage", () => { sendMessage({ baseUrl: "https://api.example.com/", body: { msg: {} } }), ).rejects.toThrow("sendMessage 403"); }); + + it("throws on a successful HTTP response with a non-zero API ret", async () => { + mockFetch.mockResolvedValueOnce(mockResponse({ ret: 1 })); + await expect( + sendMessage({ baseUrl: "https://api.example.com/", body: { msg: {} } }), + ).rejects.toThrow("sendMessage ret=1 errmsg=(none)"); + }); +}); + +describe("parseWeixinApiJson", () => { + it("preserves all known uint64 message ID fields as strings", () => { + const parsed = parseWeixinApiJson<{ + message_id: string; + nested: { msg_id: string; ref: { svr_id: string } }; + }>( + '{"message_id":18446744073709551615,"nested":{"msg_id" : 9007199254740993,"ref":{"svr_id": 123}}}', + ); + expect(parsed).toEqual({ + message_id: "18446744073709551615", + nested: { msg_id: "9007199254740993", ref: { svr_id: "123" } }, + }); + }); + + it("does not rewrite matching text inside JSON strings", () => { + const raw = '{"text":"\\\"message_id\\\":18446744073709551615","message_id":"7"}'; + expect(parseWeixinApiJson<{ text: string; message_id: string }>(raw)).toEqual({ + text: '"message_id":18446744073709551615', + message_id: "7", + }); + }); + + it("leaves non-ID keys and non-numeric ID values unchanged", () => { + expect(parseWeixinApiJson( + '{ "other": 9007199254740993, "message_id": null, "msg_id": "already-string" }', + )).toEqual({ + other: 9007199254740992, + message_id: null, + msg_id: "already-string", + }); + }); + + it("handles negative numeric IDs without losing precision", () => { + expect(parseWeixinApiJson<{ svr_id: string }>('{"svr_id" : -9007199254740993}')).toEqual({ + svr_id: "-9007199254740993", + }); + }); }); describe("getConfig", () => { diff --git a/src/api/api.ts b/src/api/api.ts index 6a152b4..19e5fa4 100644 --- a/src/api/api.ts +++ b/src/api/api.ts @@ -453,7 +453,7 @@ export async function getUpdates( label: "getUpdates", abortSignal: params.abortSignal, }); - const resp: GetUpdatesResp = JSON.parse(rawText); + const resp = parseWeixinApiJson(rawText); return resp; } catch (err) { // Long-poll timeout or external abort are both normal control-flow exits. @@ -499,10 +499,71 @@ export async function getUploadUrl( return resp; } -/** Send a single message downstream. */ +const LOSSLESS_ID_FIELDS = new Set(["message_id", "msg_id", "svr_id"]); + +/** + * Quote uint64 message identifiers before JSON.parse sees them. This scanner + * only rewrites actual object properties, never matching text inside JSON strings. + */ +export function parseWeixinApiJson(rawText: string): T { + let output = ""; + let index = 0; + while (index < rawText.length) { + if (rawText[index] !== '"') { + output += rawText[index++]; + continue; + } + + const stringStart = index; + index++; + let escaped = false; + while (index < rawText.length) { + const char = rawText[index++]; + if (escaped) { + escaped = false; + } else if (char === "\\") { + escaped = true; + } else if (char === '"') { + break; + } + } + const stringToken = rawText.slice(stringStart, index); + output += stringToken; + + let cursor = index; + while (/\s/.test(rawText[cursor] ?? "")) cursor++; + if (rawText[cursor] !== ":") continue; + + let key: unknown; + try { + key = JSON.parse(stringToken); + } catch { + continue; + } + if (typeof key !== "string" || !LOSSLESS_ID_FIELDS.has(key)) continue; + + output += rawText.slice(index, cursor + 1); + cursor++; + while (/\s/.test(rawText[cursor] ?? "")) { + output += rawText[cursor++]; + } + const numberStart = cursor; + if (rawText[cursor] === "-") cursor++; + while (/\d/.test(rawText[cursor] ?? "")) cursor++; + if (cursor > numberStart && !(cursor === numberStart + 1 && rawText[numberStart] === "-")) { + output += `"${rawText.slice(numberStart, cursor)}"`; + index = cursor; + } else { + index = numberStart; + } + } + return JSON.parse(output) as T; +} + +/** Send a single message downstream and return the server-assigned message ID. */ export async function sendMessage( params: WeixinApiOptions & { body: SendMessageReq }, -): Promise { +): Promise { const rawText = await apiPostFetch({ baseUrl: params.baseUrl, endpoint: "ilink/bot/sendmessage", @@ -511,12 +572,13 @@ export async function sendMessage( timeoutMs: params.timeoutMs ?? DEFAULT_API_TIMEOUT_MS, label: "sendMessage", }); - const resp: SendMessageResp = JSON.parse(rawText); + const resp = parseWeixinApiJson(rawText); if (resp.ret && resp.ret !== 0) { throw new Error( `sendMessage ret=${resp.ret} errmsg=${resp.errmsg ?? "(none)"}`, ); } + return resp; } /** Fetch bot config (includes typing_ticket) for a given user. */ diff --git a/src/api/types.ts b/src/api/types.ts index f63656d..276e845 100644 --- a/src/api/types.ts +++ b/src/api/types.ts @@ -147,6 +147,18 @@ export interface VideoItem { export interface RefMessage { message_item?: MessageItem; title?: string; // 摘要 + /** Server message ID used when newer clients omit the quoted content. */ + svr_id?: string; + /** Optional metadata describing a selected substring of the quoted message. */ + partial_text?: PartialText; +} + +export interface PartialText { + start: string; + end: string; + startindex: number; + endindex: number; + quotemd5: string; } export interface ToolCallStartItem { @@ -179,7 +191,8 @@ export interface MessageItem { /** Unified message (proto: WeixinMessage). Replaces the old split Message + MessageContent + FullMessage. */ export interface WeixinMessage { seq?: number; - message_id?: number; + /** uint64 on the wire; parsed losslessly as a string. */ + message_id?: string; from_user_id?: string; to_user_id?: string; client_id?: string; @@ -224,6 +237,8 @@ export interface SendMessageReq { } export interface SendMessageResp { + /** uint64 on the wire; parsed losslessly as a string. */ + message_id?: string; ret?: number; errmsg?: string; } diff --git a/src/auth/accounts.ts b/src/auth/accounts.ts index b9d2f01..ff2bb1c 100644 --- a/src/auth/accounts.ts +++ b/src/auth/accounts.ts @@ -5,6 +5,7 @@ import { normalizeAccountId } from "openclaw/plugin-sdk/account-id"; import type { OpenClawConfig } from "openclaw/plugin-sdk/core"; import { resolveStateDir } from "../storage/state-dir.js"; +import { deleteQuoteCacheForAccount } from "../messaging/quote-store.js"; import { resolveFrameworkAllowFromPath } from "./pairing.js"; import { logger } from "../util/logger.js"; @@ -218,6 +219,7 @@ export function saveWeixinAccount( * - credentials/openclaw-weixin-{accountId}-allowFrom.json (authorized users) */ export function clearWeixinAccount(accountId: string): void { + deleteQuoteCacheForAccount(accountId); const dir = resolveAccountsDir(); const accountFiles = [ `${accountId}.json`, diff --git a/src/channel.ts b/src/channel.ts index 2a084df..6d2fa87 100644 --- a/src/channel.ts +++ b/src/channel.ts @@ -18,6 +18,10 @@ import type { ResolvedWeixinAccount } from "./auth/accounts.js"; import { notifyStop, notifyStart } from "./api/api.js"; import { assertSessionActive } from "./api/session-guard.js"; import { getContextToken, findAccountIdsByContextToken, restoreContextTokens, clearContextTokensForAccount } from "./messaging/inbound.js"; +import { + deactivateQuoteStoreAccount, + initializeQuoteStore, +} from "./messaging/quote-store.js"; import { logger } from "./util/logger.js"; import { DEFAULT_ILINK_BOT_TYPE, @@ -143,6 +147,7 @@ async function sendWeixinOutbound(params: { baseUrl: account.baseUrl, token: account.token, contextToken: params.contextToken, + accountId: account.accountId, }}); emitWeixinMessageSent({ to: params.to, content: filteredText, success: true, accountId: account.accountId }); return { channel: "openclaw-weixin", messageId: result.messageId }; @@ -173,6 +178,27 @@ export const weixinPlugin: ChannelPlugin = { default: true, description: "Send structured tool-call progress messages.", }, + quoteCache: { + type: "object", + additionalProperties: false, + description: "Persist message content locally so ID-only Weixin quotes can be resolved.", + properties: { + enabled: { type: "boolean", default: true }, + retentionDays: { type: "number", default: 30, minimum: 0.01 }, + maxMessagesPerAccount: { type: "integer", default: 10000, minimum: 1 }, + mediaRetentionDays: { type: "number", default: 7, minimum: 0.01 }, + maxMediaBytesPerAccount: { + type: "integer", + default: 268435456, + minimum: 1, + }, + maxSingleMediaBytes: { + type: "integer", + default: 26214400, + minimum: 1, + }, + }, + }, }, }, }, @@ -271,7 +297,7 @@ export const weixinPlugin: ChannelPlugin = { filePath, to: ctx.to, text, - opts: { baseUrl: account.baseUrl, token: account.token, contextToken }, + opts: { baseUrl: account.baseUrl, token: account.token, contextToken, accountId: account.accountId }, cdnBaseUrl: account.cdnBaseUrl, }); emitWeixinMessageSent({ to: ctx.to, content: text, success: true, accountId: account.accountId }); @@ -288,6 +314,7 @@ export const weixinPlugin: ChannelPlugin = { baseUrl: account.baseUrl, token: account.token, contextToken, + accountId: account.accountId, }}); emitWeixinMessageSent({ to: ctx.to, content: text, success: true, accountId: account.accountId }); return { channel: "openclaw-weixin", messageId: result.messageId }; @@ -426,6 +453,8 @@ export const weixinPlugin: ChannelPlugin = { throw new Error("weixin not configured: missing token"); } + await initializeQuoteStore(ctx.cfg, account.accountId); + ctx.log?.info?.(`[${account.accountId}] starting weixin provider (${DEFAULT_BASE_URL})`); try { @@ -470,6 +499,7 @@ export const weixinPlugin: ChannelPlugin = { stopAccount: async (ctx) => { const account = ctx.account; const aLog = logger.withAccount(account.accountId); + deactivateQuoteStoreAccount(account.accountId); if (!account.configured || !account.token?.trim()) { aLog.debug(`gateway.stopAccount: skip notifyStop (not configured or no token)`); return; diff --git a/src/config/config-schema.test.ts b/src/config/config-schema.test.ts index df04d27..9332381 100644 --- a/src/config/config-schema.test.ts +++ b/src/config/config-schema.test.ts @@ -7,6 +7,14 @@ describe("WeixinConfigSchema", () => { expect(result.baseUrl).toBe("https://ilinkai.weixin.qq.com"); expect(result.cdnBaseUrl).toBe("https://novac2c.cdn.weixin.qq.com/c2c"); expect(result.replyProgressMessages).toBe(true); + expect(result.quoteCache).toEqual({ + enabled: true, + retentionDays: 30, + maxMessagesPerAccount: 10_000, + mediaRetentionDays: 7, + maxMediaBytesPerAccount: 256 * 1024 * 1024, + maxSingleMediaBytes: 25 * 1024 * 1024, + }); }); it("accepts custom baseUrl and cdnBaseUrl", () => { @@ -34,6 +42,12 @@ describe("WeixinConfigSchema", () => { expect(result.replyProgressMessages).toBe(false); }); + it("accepts disabling quote caching for hosts without SQLite", () => { + const result = WeixinConfigSchema.parse({ quoteCache: { enabled: false } }); + expect(result.quoteCache.enabled).toBe(false); + expect(result.quoteCache.retentionDays).toBe(30); + }); + it("accepts accounts map", () => { const result = WeixinConfigSchema.parse({ accounts: { diff --git a/src/config/config-schema.ts b/src/config/config-schema.ts index 4d152cc..46d20a4 100644 --- a/src/config/config-schema.ts +++ b/src/config/config-schema.ts @@ -14,10 +14,27 @@ const weixinAccountSchema = z.object({ routeTag: z.number().optional(), }); +const quoteCacheSchema = z.object({ + enabled: z.boolean().default(true), + retentionDays: z.number().positive().default(30), + maxMessagesPerAccount: z.number().int().positive().default(10_000), + mediaRetentionDays: z.number().positive().default(7), + maxMediaBytesPerAccount: z.number().int().positive().default(256 * 1024 * 1024), + maxSingleMediaBytes: z.number().int().positive().default(25 * 1024 * 1024), +}); + /** Top-level weixin config schema (token is stored in credentials file, not config). */ export const WeixinConfigSchema = weixinAccountSchema.extend({ accounts: z.record(z.string(), weixinAccountSchema).optional(), replyProgressMessages: z.boolean().default(true), + quoteCache: quoteCacheSchema.default({ + enabled: true, + retentionDays: 30, + maxMessagesPerAccount: 10_000, + mediaRetentionDays: 7, + maxMediaBytesPerAccount: 256 * 1024 * 1024, + maxSingleMediaBytes: 25 * 1024 * 1024, + }), /** ISO 8601; bumped on each successful login to refresh gateway config from disk. */ channelConfigUpdatedAt: z.string().optional(), }); diff --git a/src/messaging/error-notice.ts b/src/messaging/error-notice.ts index 7b1db6e..ae75360 100644 --- a/src/messaging/error-notice.ts +++ b/src/messaging/error-notice.ts @@ -13,6 +13,7 @@ export async function sendWeixinErrorNotice(params: { baseUrl: string; token?: string; runId?: string; + accountId?: string; errLog: (m: string) => void; }): Promise { if (!params.contextToken) { @@ -23,6 +24,7 @@ export async function sendWeixinErrorNotice(params: { baseUrl: params.baseUrl, token: params.token, contextToken: params.contextToken, + accountId: params.accountId, ...(params.runId ? { runId: params.runId } : {}), }}); logger.debug(`sendWeixinErrorNotice: sent to=${params.to}`); diff --git a/src/messaging/inbound.test.ts b/src/messaging/inbound.test.ts index 8c507f7..0f6ce18 100644 --- a/src/messaging/inbound.test.ts +++ b/src/messaging/inbound.test.ts @@ -1,5 +1,15 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + import { describe, it, expect, vi, beforeEach } from "vitest"; -import { isMediaItem, weixinMessageToMsgContext, getContextTokenFromMsgContext } from "./inbound.js"; +import { + getContextTokenFromMsgContext, + getWeixinMessageId, + isMediaItem, + resolveStoredQuoteContext, + weixinMessageToMsgContext, +} from "./inbound.js"; import type { WeixinMsgContext } from "./inbound.js"; import { MessageItemType } from "../api/types.js"; import type { WeixinMessage, MessageItem } from "../api/types.js"; @@ -47,6 +57,17 @@ describe("isMediaItem", () => { }); }); +describe("getMediaLabel", () => { + it("labels every supported media kind", async () => { + const { getMediaLabel } = await import("./inbound.js"); + expect(getMediaLabel(MessageItemType.IMAGE)).toBe("[图片]"); + expect(getMediaLabel(MessageItemType.VIDEO)).toBe("[视频]"); + expect(getMediaLabel(MessageItemType.FILE)).toBe("[文件]"); + expect(getMediaLabel(MessageItemType.VOICE)).toBe("[语音]"); + expect(getMediaLabel(MessageItemType.TEXT)).toBe(""); + }); +}); + describe("weixinMessageToMsgContext", () => { beforeEach(() => { vi.spyOn(Date, "now").mockReturnValue(1700000000000); @@ -142,6 +163,14 @@ describe("weixinMessageToMsgContext", () => { expect(ctx.MediaType).toBe("audio/wav"); }); + it("uses transcribed voice text as the body", () => { + const ctx = weixinMessageToMsgContext({ + from_user_id: "u", + item_list: [{ type: MessageItemType.VOICE, voice_item: { text: "voice transcript" } }], + }, "acc"); + expect(ctx.Body).toBe("voice transcript"); + }); + it("prioritizes pic > video > file > voice", () => { const ctx = weixinMessageToMsgContext(baseMsg, "acc", { decryptedPicPath: "/tmp/pic.png", @@ -153,6 +182,17 @@ describe("weixinMessageToMsgContext", () => { expect(ctx.MediaType).toBe("image/*"); }); + it("keeps downloaded inline-quote media out of the current message attachment", () => { + const ctx = weixinMessageToMsgContext(baseMsg, "acc", { + decryptedPicPath: "/tmp/quoted.png", + referencedMedia: true, + }); + expect(ctx.MediaPath).toBeUndefined(); + expect(ctx.MediaType).toBeUndefined(); + expect(ctx.MediaPaths).toEqual(["/tmp/quoted.png"]); + expect(ctx.MediaTypes).toEqual(["image/*"]); + }); + it("builds quoted context from ref_msg title", () => { const msg: WeixinMessage = { from_user_id: "u", @@ -165,7 +205,9 @@ describe("weixinMessageToMsgContext", () => { ], }; const ctx = weixinMessageToMsgContext(msg, "acc"); - expect(ctx.Body).toBe("[引用: original title]\nreply"); + expect(ctx.Body).toBe("reply"); + expect(ctx.ReplyToBody).toBe("original title"); + expect(ctx.ReplyToIsQuote).toBe(true); }); it("skips quoted context when ref_msg is a media item", () => { @@ -203,7 +245,9 @@ describe("weixinMessageToMsgContext", () => { ], }; const ctx = weixinMessageToMsgContext(msg, "acc"); - expect(ctx.Body).toBe("[引用: Author | original text]\nmy reply"); + expect(ctx.Body).toBe("my reply"); + expect(ctx.ReplyToBody).toBe("Author | original text"); + expect(ctx.ReplyToIsQuote).toBe(true); }); it("builds quoted context with only message_item (no title)", () => { @@ -223,7 +267,8 @@ describe("weixinMessageToMsgContext", () => { ], }; const ctx = weixinMessageToMsgContext(msg, "acc"); - expect(ctx.Body).toBe("[引用: quoted]\nreply"); + expect(ctx.Body).toBe("reply"); + expect(ctx.ReplyToBody).toBe("quoted"); }); it("returns text when ref_msg has no extractable content", () => { @@ -241,7 +286,7 @@ describe("weixinMessageToMsgContext", () => { expect(ctx.Body).toBe("reply"); }); - it("returns empty body when item_list has only non-text items", () => { + it("uses a stable label when item_list has only a media item", () => { const msg: WeixinMessage = { from_user_id: "u", item_list: [ @@ -249,7 +294,174 @@ describe("weixinMessageToMsgContext", () => { ], }; const ctx = weixinMessageToMsgContext(msg, "acc"); - expect(ctx.Body).toBe(""); + expect(ctx.Body).toBe("[图片]"); + }); + + it("keeps the provider uint64 ID separately from the generated MessageSid", () => { + const ctx = weixinMessageToMsgContext({ ...baseMsg, message_id: "18446744073709551615" }, "acc"); + expect(ctx.MessageSidFull).toBe("18446744073709551615"); + expect(ctx.MessageSid).not.toBe(ctx.MessageSidFull); + }); +}); + +describe("stored quote resolution", () => { + const quotedMessage = (partial_text?: NonNullable["partial_text"]>): WeixinMessage => ({ + from_user_id: "user1", + item_list: [{ + type: MessageItemType.TEXT, + text_item: { text: "reply" }, + ref_msg: { svr_id: "9007199254740993123", ...(partial_text ? { partial_text } : {}) }, + }], + }); + + it("looks up an ID-only quote in the account and conversation scope", () => { + const msg = quotedMessage(); + const ctx = weixinMessageToMsgContext(msg, "acc"); + const find = vi.fn(() => ({ + accountId: "acc", + conversationId: "user1", + messageId: "9007199254740993123", + direction: "inbound" as const, + body: "original text", + createdAt: Date.now(), + })); + resolveStoredQuoteContext(ctx, msg, "acc", { find }); + expect(find).toHaveBeenCalledWith("acc", "user1", "9007199254740993123"); + expect(ctx).toMatchObject({ + Body: "reply", + ReplyToId: "9007199254740993123", + ReplyToBody: "original text", + ReplyToIsQuote: true, + }); + }); + + it("marks a cache miss without modifying the current message body", () => { + const msg = quotedMessage(); + const ctx = weixinMessageToMsgContext(msg, "acc"); + resolveStoredQuoteContext(ctx, msg, "acc", { find: () => null }); + expect(ctx.Body).toBe("reply"); + expect(ctx.ReplyToBody).toBe("[引用消息内容未缓存]"); + }); + + it("resolves the legacy nested message ID", () => { + const msg: WeixinMessage = { + from_user_id: "user1", + item_list: [{ + type: MessageItemType.TEXT, + text_item: { text: "reply" }, + ref_msg: { message_item: { type: MessageItemType.TEXT, msg_id: "legacy-id" } }, + }], + }; + const ctx = weixinMessageToMsgContext(msg, "acc"); + resolveStoredQuoteContext(ctx, msg, "acc", { + find: () => ({ + accountId: "acc", conversationId: "user1", messageId: "legacy-id", + direction: "inbound", body: "legacy body", createdAt: Date.now(), + }), + }); + expect(ctx.ReplyToId).toBe("legacy-id"); + expect(ctx.ReplyToBody).toBe("legacy body"); + }); + + it("keeps complete inline quote content instead of replacing it from storage", () => { + const msg: WeixinMessage = { + from_user_id: "user1", + item_list: [{ + type: MessageItemType.TEXT, + text_item: { text: "reply" }, + ref_msg: { + svr_id: "id", + message_item: { type: MessageItemType.TEXT, text_item: { text: "inline body" } }, + }, + }], + }; + const ctx = weixinMessageToMsgContext(msg, "acc"); + const find = vi.fn(); + resolveStoredQuoteContext(ctx, msg, "acc", { find }); + expect(find).not.toHaveBeenCalled(); + expect(ctx.ReplyToBody).toBe("inline body"); + }); + + it("sets ReplyToQuoteText for a partial quote", () => { + const selected = "second abc"; + const msg = quotedMessage({ + start: "s", + end: "c", + startindex: 1, + endindex: 2, + quotemd5: "", + }); + const ctx = weixinMessageToMsgContext(msg, "acc"); + resolveStoredQuoteContext(ctx, msg, "acc", { + find: () => ({ + accountId: "acc", + conversationId: "user1", + messageId: "9007199254740993123", + direction: "inbound", + body: "start abc, second abc", + createdAt: Date.now(), + }), + }); + expect(ctx.ReplyToQuoteText).toBe(selected); + }); + + it.each([ + ["image/png", "picture.png", "[引用的图片已过期: picture.png]"], + ["video/mp4", undefined, "[引用的视频已过期]"], + ["audio/mpeg", undefined, "[引用的语音已过期]"], + ["application/pdf", "doc.pdf", "[引用的附件已过期: doc.pdf]"], + ])("describes expired managed media (%s)", (mediaMime, mediaName, expected) => { + const msg = quotedMessage(); + const ctx = weixinMessageToMsgContext(msg, "acc"); + resolveStoredQuoteContext(ctx, msg, "acc", { + find: () => ({ + accountId: "acc", conversationId: "user1", messageId: "9007199254740993123", + direction: "inbound", body: "[媒体]", mediaMime, ...(mediaName ? { mediaName } : {}), + createdAt: Date.now(), + }), + }); + expect(ctx.ReplyToBody).toBe(expected); + }); + + it("adds an existing quoted media file alongside the current attachment", () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "weixin-inbound-quote-")); + const quotedPath = path.join(tempDir, "quoted.png"); + fs.writeFileSync(quotedPath, "image"); + try { + const msg = quotedMessage(); + const ctx = weixinMessageToMsgContext(msg, "acc", { decryptedFilePath: "/tmp/current.pdf" }); + resolveStoredQuoteContext(ctx, msg, "acc", { + find: () => ({ + accountId: "acc", conversationId: "user1", messageId: "9007199254740993123", + direction: "inbound", body: "[图片]", mediaPath: quotedPath, mediaMime: "image/png", + createdAt: Date.now(), + }), + }); + expect(ctx.MediaPaths).toEqual(["/tmp/current.pdf", quotedPath]); + expect(ctx.MediaTypes).toEqual(["application/octet-stream", "image/png"]); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it("marks a referenced media path that disappeared from disk as expired", () => { + const msg = quotedMessage(); + const ctx = weixinMessageToMsgContext(msg, "acc"); + resolveStoredQuoteContext(ctx, msg, "acc", { + find: () => ({ + accountId: "acc", conversationId: "user1", messageId: "9007199254740993123", + direction: "inbound", body: "[视频]", mediaPath: "/definitely/missing/video.mp4", + mediaMime: "video/mp4", createdAt: Date.now(), + }), + }); + expect(ctx.ReplyToBody).toBe("[引用的视频已过期]"); + }); +}); + +describe("getWeixinMessageId", () => { + it("prefers the top-level ID and falls back to an item ID", () => { + expect(getWeixinMessageId({ message_id: "10", item_list: [{ msg_id: "11" }] })).toBe("10"); + expect(getWeixinMessageId({ item_list: [{ msg_id: "11" }] })).toBe("11"); }); }); diff --git a/src/messaging/inbound.ts b/src/messaging/inbound.ts index fc25ab8..f1d8033 100644 --- a/src/messaging/inbound.ts +++ b/src/messaging/inbound.ts @@ -6,6 +6,9 @@ import { generateId } from "../util/random.js"; import type { WeixinMessage, MessageItem } from "../api/types.js"; import { MessageItemType } from "../api/types.js"; import { resolveStateDir } from "../storage/state-dir.js"; +import { resolvePartialQuote } from "./partial-quote.js"; +import { getQuoteStore } from "./quote-store.js"; +import type { QuoteStore } from "./quote-store.js"; // --------------------------------------------------------------------------- // Context token store (in-process cache + disk persistence) @@ -144,6 +147,8 @@ export type WeixinMsgContext = { OriginatingChannel: "openclaw-weixin"; OriginatingTo: string; MessageSid: string; + /** Provider message ID retained losslessly for transcript metadata. */ + MessageSidFull?: string; Timestamp?: number; Provider: "openclaw-weixin"; ChatType: "direct"; @@ -153,6 +158,12 @@ export type WeixinMsgContext = { MediaUrl?: string; MediaPath?: string; MediaType?: string; + MediaPaths?: string[]; + MediaTypes?: string[]; + ReplyToId?: string; + ReplyToBody?: string; + ReplyToQuoteText?: string; + ReplyToIsQuote?: boolean; /** Raw message body for framework command authorization. */ CommandBody?: string; /** Whether the sender is authorized to execute slash commands. */ @@ -169,29 +180,42 @@ export function isMediaItem(item: MessageItem): boolean { ); } +export function getMediaLabel(type: number | undefined): string { + switch (type) { + case MessageItemType.IMAGE: + return "[图片]"; + case MessageItemType.VIDEO: + return "[视频]"; + case MessageItemType.FILE: + return "[文件]"; + case MessageItemType.VOICE: + return "[语音]"; + default: + return ""; + } +} + +export function getWeixinMessageId(msg: WeixinMessage): string | undefined { + const topLevel = msg.message_id?.trim(); + if (topLevel) return topLevel; + for (const item of msg.item_list ?? []) { + const itemId = item.msg_id?.trim(); + if (itemId) return itemId; + } + return undefined; +} + function bodyFromItemList(itemList?: MessageItem[]): string { if (!itemList?.length) return ""; for (const item of itemList) { if (item.type === MessageItemType.TEXT && item.text_item?.text != null) { - const text = String(item.text_item.text); - const ref = item.ref_msg; - if (!ref) return text; - // Quoted media is passed as MediaPath; only include the current text as body. - if (ref.message_item && isMediaItem(ref.message_item)) return text; - // Build quoted context from both title and message_item content. - const parts: string[] = []; - if (ref.title) parts.push(ref.title); - if (ref.message_item) { - const refBody = bodyFromItemList([ref.message_item]); - if (refBody) parts.push(refBody); - } - if (!parts.length) return text; - return `[引用: ${parts.join(" | ")}]\n${text}`; + return String(item.text_item.text); } // 语音转文字：如果语音消息有 text 字段，直接使用文字内容 if (item.type === MessageItemType.VOICE && item.voice_item?.text) { return item.voice_item.text; } + if (isMediaItem(item)) return getMediaLabel(item.type); } return ""; } @@ -209,6 +233,8 @@ export type WeixinInboundMediaOpts = { fileMediaType?: string; /** Local path to decrypted video file. */ decryptedVideoPath?: string; + /** The downloaded media belongs to an inline quoted message, not the current message. */ + referencedMedia?: boolean; }; /** @@ -238,6 +264,8 @@ export function weixinMessageToMsgContext( if (msg.context_token) { ctx.context_token = msg.context_token; } + const providerMessageId = getWeixinMessageId(msg); + if (providerMessageId) ctx.MessageSidFull = providerMessageId; if (opts?.decryptedPicPath) { ctx.MediaPath = opts.decryptedPicPath; @@ -253,9 +281,99 @@ export function weixinMessageToMsgContext( ctx.MediaType = opts.voiceMediaType ?? "audio/wav"; } + if (opts?.referencedMedia && ctx.MediaPath) { + ctx.MediaPaths = [ctx.MediaPath]; + ctx.MediaTypes = [ctx.MediaType ?? "application/octet-stream"]; + delete ctx.MediaPath; + delete ctx.MediaType; + } + + applyInlineQuoteContext(ctx, msg); + return ctx; } +function findReferenceItem(msg: WeixinMessage): MessageItem | undefined { + return msg.item_list?.find((item) => item.ref_msg); +} + +function inlineQuoteBody(item: MessageItem): string | undefined { + const ref = item.ref_msg; + if (!ref) return undefined; + const parts: string[] = []; + if (ref.title?.trim()) parts.push(ref.title.trim()); + if (ref.message_item) { + const body = bodyFromItemList([ref.message_item]); + if (body) parts.push(body); + } + return parts.length ? parts.join(" | ") : undefined; +} + +function applyInlineQuoteContext(ctx: WeixinMsgContext, msg: WeixinMessage): void { + const item = findReferenceItem(msg); + if (!item?.ref_msg) return; + const replyToId = item.ref_msg.svr_id?.trim() || item.ref_msg.message_item?.msg_id?.trim(); + const body = inlineQuoteBody(item); + ctx.ReplyToIsQuote = true; + if (replyToId) ctx.ReplyToId = replyToId; + if (body) ctx.ReplyToBody = body; +} + +function expiredMediaLabel(mime?: string, name?: string): string { + const kind = mime?.startsWith("image/") + ? "图片" + : mime?.startsWith("video/") + ? "视频" + : mime?.startsWith("audio/") + ? "语音" + : "附件"; + return name ? `[引用的${kind}已过期: ${name}]` : `[引用的${kind}已过期]`; +} + +/** Resolve an ID-only quote after sender authorization has succeeded. */ +export function resolveStoredQuoteContext( + ctx: WeixinMsgContext, + msg: WeixinMessage, + accountId: string, + store: Pick | null = getQuoteStore(), +): void { + const item = findReferenceItem(msg); + const ref = item?.ref_msg; + const referenceId = ref?.svr_id?.trim() || ref?.message_item?.msg_id?.trim(); + if (!item || !ref || !referenceId) return; + + ctx.ReplyToId = referenceId; + ctx.ReplyToIsQuote = true; + if (ctx.ReplyToBody && (ref.title?.trim() || ref.message_item)) return; + const record = store?.find(accountId, ctx.From, referenceId); + if (!record) { + if (!ctx.ReplyToBody) ctx.ReplyToBody = "[引用消息内容未缓存]"; + return; + } + + ctx.ReplyToBody = record.body || expiredMediaLabel(record.mediaMime, record.mediaName); + if (ref.partial_text && record.body) { + const partial = resolvePartialQuote(record.body, ref.partial_text); + if (!partial.fallback && partial.resolved) ctx.ReplyToQuoteText = partial.resolved; + } + + if (!record.mediaPath) { + if (record.mediaMime || record.mediaName) { + ctx.ReplyToBody = expiredMediaLabel(record.mediaMime, record.mediaName); + } + return; + } + if (!fs.existsSync(record.mediaPath)) { + ctx.ReplyToBody = expiredMediaLabel(record.mediaMime, record.mediaName); + return; + } + const currentPaths = ctx.MediaPaths ?? (ctx.MediaPath ? [ctx.MediaPath] : []); + const currentTypes = + ctx.MediaTypes ?? (ctx.MediaPath ? [ctx.MediaType ?? "application/octet-stream"] : []); + ctx.MediaPaths = [...currentPaths, record.mediaPath]; + ctx.MediaTypes = [...currentTypes, record.mediaMime ?? "application/octet-stream"]; +} + /** Extract the context_token from an inbound WeixinMsgContext. */ export function getContextTokenFromMsgContext(ctx: WeixinMsgContext): string | undefined { return ctx.context_token; diff --git a/src/messaging/partial-quote.test.ts b/src/messaging/partial-quote.test.ts new file mode 100644 index 0000000..b4d3796 --- /dev/null +++ b/src/messaging/partial-quote.test.ts @@ -0,0 +1,50 @@ +import crypto from "node:crypto"; + +import { describe, expect, it } from "vitest"; + +import { resolvePartialQuote } from "./partial-quote.js"; + +function md5(value: string): string { + return crypto.createHash("md5").update(value, "utf8").digest("hex"); +} + +describe("resolvePartialQuote", () => { + it("resolves the protocol example using global occurrence indexes", () => { + expect(resolvePartialQuote("abcedfabcgh", { + start: "a", + end: "c", + startindex: 1, + endindex: 1, + quotemd5: md5("abc"), + })).toEqual({ resolved: "abc", fallback: false }); + }); + + it("uses the hash to support end indexes relative to the selected start", () => { + const full = "x-end start first-end second-end"; + const selected = "start first-end second-end"; + expect(resolvePartialQuote(full, { + start: "start", + end: "end", + startindex: 0, + endindex: 1, + quotemd5: md5(selected), + })).toEqual({ resolved: selected, fallback: false }); + }); + + it("falls back when indexes are invalid or the hash does not match", () => { + expect(resolvePartialQuote("hello", { + start: "h", + end: "o", + startindex: -1, + endindex: 0, + quotemd5: "", + })).toEqual({ resolved: null, fallback: true }); + expect(resolvePartialQuote("hello", { + start: "h", + end: "o", + startindex: 0, + endindex: 0, + quotemd5: "not-the-md5", + })).toEqual({ resolved: null, fallback: true }); + }); +}); diff --git a/src/messaging/partial-quote.ts b/src/messaging/partial-quote.ts new file mode 100644 index 0000000..8f323ad --- /dev/null +++ b/src/messaging/partial-quote.ts @@ -0,0 +1,65 @@ +import crypto from "node:crypto"; + +import type { PartialText } from "../api/types.js"; + +export type PartialQuoteResult = { + resolved: string | null; + fallback: boolean; +}; + +function nthIndexOf(text: string, value: string, occurrence: number, fromIndex = 0): number { + if (!value || !Number.isInteger(occurrence) || occurrence < 0) return -1; + let position = fromIndex; + for (let current = 0; current <= occurrence; current++) { + position = text.indexOf(value, position); + if (position < 0) return -1; + if (current < occurrence) position += value.length; + } + return position; +} + +function hashQuote(text: string): string { + return crypto.createHash("md5").update(text, "utf8").digest("hex"); +} + +function candidate( + fullText: string, + partial: PartialText, + endSearchMode: "global" | "relative", +): string | null { + const start = nthIndexOf(fullText, partial.start, partial.startindex); + if (start < 0) return null; + const end = + endSearchMode === "global" + ? nthIndexOf(fullText, partial.end, partial.endindex) + : nthIndexOf(fullText, partial.end, partial.endindex, start + partial.start.length); + if (end < start) return null; + return fullText.slice(start, end + partial.end.length); +} + +/** + * Resolve both observed interpretations of endindex. When quotemd5 is present + * it disambiguates the protocol variants; without a hash, global indexes match + * the examples supplied with the newer Weixin payload definition. + */ +export function resolvePartialQuote( + fullText: string, + partial: PartialText, +): PartialQuoteResult { + if (!fullText || !partial.start || !partial.end) { + return { resolved: null, fallback: true }; + } + const candidates = [ + candidate(fullText, partial, "global"), + candidate(fullText, partial, "relative"), + ].filter((value, index, all): value is string => Boolean(value) && all.indexOf(value) === index); + + if (!partial.quotemd5) { + return candidates[0] + ? { resolved: candidates[0], fallback: false } + : { resolved: null, fallback: true }; + } + const expected = partial.quotemd5.toLowerCase(); + const resolved = candidates.find((value) => hashQuote(value) === expected) ?? null; + return { resolved, fallback: resolved === null }; +} diff --git a/src/messaging/process-message.ts b/src/messaging/process-message.ts index 3336a70..e7ddf9a 100644 --- a/src/messaging/process-message.ts +++ b/src/messaging/process-message.ts @@ -27,13 +27,16 @@ import { setContextToken, weixinMessageToMsgContext, getContextTokenFromMsgContext, + getWeixinMessageId, isMediaItem, + resolveStoredQuoteContext, } from "./inbound.js"; import type { WeixinInboundMediaOpts } from "./inbound.js"; import { sendWeixinMediaFile } from "./send-media.js"; import { StreamingMarkdownFilter } from "./markdown-filter.js"; import { sendMessageWeixin } from "./send.js"; import { WeixinReplyProgressSender } from "./reply-progress-sender.js"; +import { getQuoteStore } from "./quote-store.js"; import { handleSlashCommand } from "./slash-commands.js"; const MEDIA_OUTBOUND_TEMP_DIR = path.join(resolvePreferredOpenClawTmpDir(), "weixin/media/outbound-temp"); @@ -153,6 +156,7 @@ export async function processOneMessage( label, }); Object.assign(mediaOpts, downloaded); + if (refMediaItem) mediaOpts.referencedMedia = true; } const mediaDownloadMs = Date.now() - mediaDownloadStart; @@ -216,6 +220,28 @@ export async function processOneMessage( ); } + resolveStoredQuoteContext(ctx, full, deps.accountId); + const inboundMessageId = getWeixinMessageId(full); + if (inboundMessageId) { + try { + await getQuoteStore()?.put({ + accountId: deps.accountId, + conversationId: senderId, + messageId: inboundMessageId, + direction: "inbound", + body: ctx.Body, + ...(ctx.MediaPath ? { sourceMediaPath: ctx.MediaPath } : {}), + ...(ctx.MediaType ? { mediaMime: ctx.MediaType } : {}), + ...(mainMediaItem?.file_item?.file_name + ? { mediaName: mainMediaItem.file_item.file_name } + : {}), + createdAt: full.create_time_ms ?? Date.now(), + }); + } catch (err) { + logger.warn(`quote cache: failed to save inbound message id=${inboundMessageId}: ${String(err)}`); + } + } + const route = deps.channelRuntime.routing.resolveAgentRoute({ cfg: deps.config, channel: "openclaw-weixin", @@ -283,6 +309,7 @@ export async function processOneMessage( baseUrl: deps.baseUrl, token: deps.token, contextToken, + accountId: deps.accountId, }, }) : undefined; @@ -386,6 +413,7 @@ export async function processOneMessage( token: deps.token, contextToken, runId, + accountId: deps.accountId, }}); emitWeixinMessageSent({ to: ctx.To, content: text, success: true, accountId: deps.accountId, runId }); logger.info(`outbound: text sent to=${ctx.To}`); @@ -395,7 +423,13 @@ export async function processOneMessage( filePath, to: ctx.To, text, - opts: { baseUrl: deps.baseUrl, token: deps.token, contextToken, runId }, + opts: { + baseUrl: deps.baseUrl, + token: deps.token, + contextToken, + runId, + accountId: deps.accountId, + }, cdnBaseUrl: deps.cdnBaseUrl, }); emitWeixinMessageSent({ to: ctx.To, content: text, success: true, accountId: deps.accountId, runId }); @@ -407,6 +441,7 @@ export async function processOneMessage( token: deps.token, contextToken, runId, + accountId: deps.accountId, }}); emitWeixinMessageSent({ to: ctx.To, content: text, success: true, accountId: deps.accountId, runId }); logger.info(`outbound: text sent OK to=${ctx.To}`); @@ -441,6 +476,7 @@ export async function processOneMessage( baseUrl: deps.baseUrl, token: deps.token, runId, + accountId: deps.accountId, errLog: deps.errLog, }); }, @@ -514,7 +550,13 @@ export async function processOneMessage( await sendMessageWeixin({ to: ctx.To, text: timingText, - opts: { baseUrl: deps.baseUrl, token: deps.token, contextToken, runId }, + opts: { + baseUrl: deps.baseUrl, + token: deps.token, + contextToken, + runId, + accountId: deps.accountId, + }, }); logger.info(`debug-timing: sent OK`); } catch (debugErr) { diff --git a/src/messaging/quote-store.test.ts b/src/messaging/quote-store.test.ts new file mode 100644 index 0000000..9bec2ab --- /dev/null +++ b/src/messaging/quote-store.test.ts @@ -0,0 +1,432 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import type { QuoteCachePolicy } from "./quote-store.js"; +import { + QuoteStore, + closeQuoteStore, + deactivateQuoteStoreAccount, + deleteQuoteCacheForAccount, + getQuoteStore, + initializeQuoteStore, + resolveQuoteCachePolicy, +} from "./quote-store.js"; + +vi.mock("../util/logger.js", () => ({ + logger: { + info: vi.fn(), + debug: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, +})); + +let rootDir: string; +let stores: QuoteStore[]; + +function policy(overrides: Partial = {}): QuoteCachePolicy { + return { + enabled: true, + retentionMs: 30 * 24 * 60 * 60 * 1000, + maxMessagesPerAccount: 10_000, + mediaRetentionMs: 7 * 24 * 60 * 60 * 1000, + maxMediaBytesPerAccount: 256 * 1024 * 1024, + maxSingleMediaBytes: 25 * 1024 * 1024, + ...overrides, + }; +} + +async function open(overrides: Partial = {}): Promise { + const store = await QuoteStore.open({ rootDir, policy: policy(overrides) }); + expect(store).not.toBeNull(); + stores.push(store!); + return store!; +} + +beforeEach(() => { + rootDir = fs.mkdtempSync(path.join(os.tmpdir(), "weixin-quote-store-")); + stores = []; +}); + +afterEach(() => { + closeQuoteStore(); + delete process.env.OPENCLAW_STATE_DIR; + for (const store of stores) store.close(); + fs.rmSync(rootDir, { recursive: true, force: true }); +}); + +describe("QuoteStore", () => { + it("persists lossless IDs and scopes records by account and conversation", async () => { + const store = await open(); + const createdAt = Date.now(); + await store.put({ + accountId: "account-a", + conversationId: "user-a", + messageId: "18446744073709551615", + direction: "inbound", + body: "hello", + createdAt, + }); + + expect(store.find("account-a", "user-a", "18446744073709551615")?.body).toBe("hello"); + expect(store.find("account-b", "user-a", "18446744073709551615")).toBeNull(); + expect(store.find("account-a", "user-b", "18446744073709551615")).toBeNull(); + + store.close(); + const reopened = await open(); + expect(reopened.find("account-a", "user-a", "18446744073709551615")).toMatchObject({ + messageId: "18446744073709551615", + body: "hello", + createdAt, + }); + }); + + it("evicts expired and over-count messages during GC", async () => { + const now = Date.now(); + const store = await open({ retentionMs: 100, maxMessagesPerAccount: 2 }); + for (let index = 0; index < 3; index++) { + await store.put({ + accountId: "account", + conversationId: "user", + messageId: String(index), + direction: "inbound", + body: `message-${index}`, + createdAt: now + index, + }); + } + store.runGc(now + 3); + expect(store.find("account", "user", "0")).toBeNull(); + expect(store.find("account", "user", "1")?.body).toBe("message-1"); + expect(store.find("account", "user", "2")?.body).toBe("message-2"); + + store.runGc(now + 1000); + expect(store.find("account", "user", "1")).toBeNull(); + expect(store.find("account", "user", "2")).toBeNull(); + }); + + it("copies media into managed storage, deduplicates it, and deletes it with the account", async () => { + const source = path.join(rootDir, "source.png"); + fs.writeFileSync(source, "same-content"); + const store = await open(); + for (const messageId of ["1", "2"]) { + await store.put({ + accountId: "account", + conversationId: "user", + messageId, + direction: "inbound", + body: "[图片]", + sourceMediaPath: source, + mediaMime: "image/png", + createdAt: Date.now(), + }); + } + const first = store.find("account", "user", "1"); + const second = store.find("account", "user", "2"); + expect(first?.mediaPath).toBe(second?.mediaPath); + expect(first?.mediaPath).not.toBe(source); + expect(fs.readFileSync(first!.mediaPath!, "utf8")).toBe("same-content"); + + const managedPath = first!.mediaPath!; + store.deleteAccount("account"); + expect(store.find("account", "user", "1")).toBeNull(); + expect(fs.existsSync(managedPath)).toBe(false); + }); + + it("does not let sanitized account names share a media directory", async () => { + const source = path.join(rootDir, "source.png"); + fs.writeFileSync(source, "same-content"); + const store = await open(); + for (const accountId of ["a/b", "a_b"]) { + await store.put({ + accountId, + conversationId: "user", + messageId: "1", + direction: "inbound", + body: "[图片]", + sourceMediaPath: source, + createdAt: Date.now(), + }); + } + const firstPath = store.find("a/b", "user", "1")!.mediaPath!; + const secondPath = store.find("a_b", "user", "1")!.mediaPath!; + expect(path.dirname(firstPath)).not.toBe(path.dirname(secondPath)); + store.deleteAccount("a/b"); + expect(fs.existsSync(firstPath)).toBe(false); + expect(fs.existsSync(secondPath)).toBe(true); + }); + + it("keeps message metadata but skips oversized media", async () => { + const source = path.join(rootDir, "large.bin"); + fs.writeFileSync(source, "12345"); + const store = await open({ maxSingleMediaBytes: 4 }); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "large", + direction: "inbound", + body: "[文件]", + sourceMediaPath: source, + mediaMime: "application/octet-stream", + mediaName: "large.bin", + createdAt: Date.now(), + }); + expect(store.find("account", "user", "large")).toMatchObject({ + body: "[文件]", + mediaMime: "application/octet-stream", + mediaName: "large.bin", + }); + expect(store.find("account", "user", "large")?.mediaPath).toBeUndefined(); + }); + + it("enforces the per-account media byte budget oldest-first", async () => { + const firstSource = path.join(rootDir, "first.bin"); + const secondSource = path.join(rootDir, "second.bin"); + fs.writeFileSync(firstSource, "1111"); + fs.writeFileSync(secondSource, "2222"); + const store = await open({ maxMediaBytesPerAccount: 4 }); + const now = Date.now(); + await store.put({ + accountId: "account", conversationId: "user", messageId: "1", + direction: "inbound", body: "[文件]", sourceMediaPath: firstSource, createdAt: now, + }); + await store.put({ + accountId: "account", conversationId: "user", messageId: "2", + direction: "inbound", body: "[文件]", sourceMediaPath: secondSource, createdAt: now + 1, + }); + expect(store.find("account", "user", "1")?.mediaPath).toBeUndefined(); + expect(store.find("account", "user", "2")?.mediaPath).toBeDefined(); + }); + + it("expires managed media without discarding the message body", async () => { + const source = path.join(rootDir, "old.mp3"); + fs.writeFileSync(source, "voice"); + const now = Date.now(); + const store = await open({ mediaRetentionMs: 10 }); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "old-media", + direction: "inbound", + body: "[语音]", + sourceMediaPath: source, + mediaMime: "audio/mpeg", + createdAt: now, + }); + const managedPath = store.find("account", "user", "old-media")!.mediaPath!; + store.runGc(now + 20); + expect(store.find("account", "user", "old-media")).toMatchObject({ + body: "[语音]", + mediaMime: "audio/mpeg", + }); + expect(store.find("account", "user", "old-media")?.mediaPath).toBeUndefined(); + expect(fs.existsSync(managedPath)).toBe(false); + }); + + it("removes an expired message but preserves media shared by a newer record", async () => { + const source = path.join(rootDir, "shared.png"); + fs.writeFileSync(source, "shared"); + const now = Date.now(); + const store = await open({ retentionMs: 100 }); + await store.put({ + accountId: "account", conversationId: "user", messageId: "old", + direction: "inbound", body: "old", sourceMediaPath: source, createdAt: now - 1000, + }); + await store.put({ + accountId: "account", conversationId: "user", messageId: "new", + direction: "inbound", body: "new", sourceMediaPath: source, createdAt: now, + }); + const managedPath = store.find("account", "user", "new")!.mediaPath!; + expect(store.find("account", "user", "old")).toBeNull(); + expect(fs.existsSync(managedPath)).toBe(true); + }); + + it("prunes orphan files and tolerates a missing media source", async () => { + const store = await open(); + await store.put({ + accountId: "account", + conversationId: "user", + messageId: "missing", + direction: "inbound", + body: "[文件]", + sourceMediaPath: path.join(rootDir, "does-not-exist.bin"), + createdAt: Date.now(), + }); + expect(store.find("account", "user", "missing")?.mediaPath).toBeUndefined(); + + const orphanDir = path.join(rootDir, "ref-media", "nested"); + fs.mkdirSync(orphanDir, { recursive: true }); + const orphan = path.join(orphanDir, "orphan.bin"); + fs.writeFileSync(orphan, "orphan"); + store.runGc(); + expect(fs.existsSync(orphan)).toBe(false); + }); + + it("applies policy updates asynchronously", async () => { + const store = await open(); + const now = Date.now(); + for (const messageId of ["1", "2"]) { + await store.put({ + accountId: "account", conversationId: "user", messageId, + direction: "inbound", body: messageId, createdAt: now, + }); + } + store.updatePolicy(policy({ maxMessagesPerAccount: 1 })); + await new Promise((resolve) => queueMicrotask(resolve)); + const remaining = ["1", "2"].filter((id) => store.find("account", "user", id)); + expect(remaining).toHaveLength(1); + }); + + it("does not open when explicitly disabled", async () => { + await expect(QuoteStore.open({ rootDir, policy: policy({ enabled: false }) })).resolves.toBeNull(); + }); + + it("returns null when the database directory cannot be created", async () => { + const fileInsteadOfDirectory = path.join(rootDir, "plain-file"); + fs.writeFileSync(fileInsteadOfDirectory, "not a directory"); + await expect(QuoteStore.open({ + rootDir: path.join(fileInsteadOfDirectory, "child"), + policy: policy(), + })).resolves.toBeNull(); + }); + + it("ignores invalid writes, normalizes unsafe paths, and substitutes invalid timestamps", async () => { + const source = path.join(rootDir, "media.extension-that-is-far-too-long"); + fs.writeFileSync(source, "content"); + const store = await open(); + await store.put({ + accountId: " ", conversationId: "user", messageId: "media", + direction: "inbound", body: "[文件]", sourceMediaPath: source, createdAt: Number.NaN, + }); + const record = store.find(" ", "user", "media"); + expect(record?.createdAt).toBeGreaterThan(0); + expect(record?.mediaPath).toContain(`${path.sep}default-`); + expect(path.extname(record!.mediaPath!)).toBe(""); + + await store.put({ + accountId: "account", conversationId: "user", messageId: "", + direction: "inbound", body: "ignored", createdAt: Date.now(), + }); + await store.put({ + accountId: "account", conversationId: "user", messageId: "empty", + direction: "inbound", body: "", createdAt: Date.now(), + }); + expect(store.find("account", "user", "")).toBeNull(); + expect(store.find("account", "user", "empty")).toBeNull(); + + store.close(); + await store.put({ + accountId: "account", conversationId: "user", messageId: "closed", + direction: "inbound", body: "ignored", createdAt: Date.now(), + }); + expect(store.find("account", "user", "closed")).toBeNull(); + store.deleteAccount("account"); + store.runGc(); + }); +}); + +describe("global quote store lifecycle", () => { + it("shares one store across active accounts and closes after the last account stops", async () => { + process.env.OPENCLAW_STATE_DIR = rootDir; + const first = await initializeQuoteStore({}, "account-a"); + const second = await initializeQuoteStore({}, "account-b"); + expect(first).toBe(second); + expect(getQuoteStore()).toBe(first); + + deactivateQuoteStoreAccount("account-a"); + expect(getQuoteStore()).toBe(first); + deactivateQuoteStoreAccount("account-b"); + expect(getQuoteStore()).toBeNull(); + }); + + it("honors explicit disablement", async () => { + process.env.OPENCLAW_STATE_DIR = rootDir; + await expect(initializeQuoteStore({ + channels: { "openclaw-weixin": { quoteCache: { enabled: false } } }, + }, "account")).resolves.toBeNull(); + expect(getQuoteStore()).toBeNull(); + }); + + it("closes an already-active store when the feature is disabled", async () => { + process.env.OPENCLAW_STATE_DIR = rootDir; + expect(await initializeQuoteStore({}, "account")).not.toBeNull(); + await expect(initializeQuoteStore({ + channels: { "openclaw-weixin": { quoteCache: { enabled: false } } }, + }, "account")).resolves.toBeNull(); + expect(getQuoteStore()).toBeNull(); + }); + + it("deletes an account through the active singleton", async () => { + process.env.OPENCLAW_STATE_DIR = rootDir; + const store = await initializeQuoteStore({}, "account"); + await store!.put({ + accountId: "account", conversationId: "user", messageId: "1", + direction: "inbound", body: "hello", createdAt: Date.now(), + }); + deleteQuoteCacheForAccount("account"); + expect(store!.find("account", "user", "1")).toBeNull(); + }); + + it("deletes an account from a closed on-disk database", async () => { + process.env.OPENCLAW_STATE_DIR = rootDir; + const managedRoot = path.join(rootDir, "openclaw-weixin"); + const store = await QuoteStore.open({ rootDir: managedRoot, policy: policy() }); + expect(store).not.toBeNull(); + await store!.put({ + accountId: "account", conversationId: "user", messageId: "1", + direction: "inbound", body: "hello", createdAt: Date.now(), + }); + store!.close(); + + deleteQuoteCacheForAccount("account"); + const reopened = await QuoteStore.open({ rootDir: managedRoot, policy: policy() }); + expect(reopened?.find("account", "user", "1")).toBeNull(); + reopened?.close(); + }); +}); + +describe("resolveQuoteCachePolicy", () => { + it("uses safe defaults and honors explicit disablement", () => { + expect(resolveQuoteCachePolicy({})).toMatchObject({ + enabled: true, + maxMessagesPerAccount: 10_000, + maxMediaBytesPerAccount: 256 * 1024 * 1024, + }); + expect(resolveQuoteCachePolicy({ + channels: { "openclaw-weixin": { quoteCache: { enabled: false } } }, + })).toMatchObject({ enabled: false }); + }); + + it("accepts positive limits and replaces invalid values with defaults", () => { + expect(resolveQuoteCachePolicy({ + channels: { "openclaw-weixin": { quoteCache: { + retentionDays: 2, + maxMessagesPerAccount: 3.9, + mediaRetentionDays: 4, + maxMediaBytesPerAccount: 5, + maxSingleMediaBytes: 6, + } } }, + })).toMatchObject({ + retentionMs: 2 * 24 * 60 * 60 * 1000, + maxMessagesPerAccount: 3, + mediaRetentionMs: 4 * 24 * 60 * 60 * 1000, + maxMediaBytesPerAccount: 5, + maxSingleMediaBytes: 6, + }); + expect(resolveQuoteCachePolicy({ + channels: { "openclaw-weixin": { quoteCache: { + retentionDays: Number.NaN, + maxMessagesPerAccount: 0, + mediaRetentionDays: -1, + maxMediaBytesPerAccount: Number.POSITIVE_INFINITY, + maxSingleMediaBytes: 0, + } } }, + })).toMatchObject({ + maxMessagesPerAccount: 10_000, + maxMediaBytesPerAccount: 256 * 1024 * 1024, + maxSingleMediaBytes: 25 * 1024 * 1024, + }); + }); +}); diff --git a/src/messaging/quote-store.ts b/src/messaging/quote-store.ts new file mode 100644 index 0000000..1bca775 --- /dev/null +++ b/src/messaging/quote-store.ts @@ -0,0 +1,570 @@ +import crypto from "node:crypto"; +import fs from "node:fs"; +import { createRequire } from "node:module"; +import path from "node:path"; + +import type { OpenClawConfig } from "openclaw/plugin-sdk/core"; + +import { resolveStateDir } from "../storage/state-dir.js"; +import { logger } from "../util/logger.js"; + +const DEFAULT_RETENTION_DAYS = 30; +const DEFAULT_MAX_MESSAGES_PER_ACCOUNT = 10_000; +const DEFAULT_MEDIA_RETENTION_DAYS = 7; +const DEFAULT_MAX_MEDIA_BYTES_PER_ACCOUNT = 256 * 1024 * 1024; +const DEFAULT_MAX_SINGLE_MEDIA_BYTES = 25 * 1024 * 1024; +const GC_WRITE_INTERVAL = 100; +const GC_TIMER_MS = 60 * 60 * 1000; + +type SqliteStatement = { + run: (...params: unknown[]) => unknown; + get: (...params: unknown[]) => unknown; + all: (...params: unknown[]) => unknown[]; +}; + +type SqliteDatabase = { + exec: (sql: string) => void; + prepare: (sql: string) => SqliteStatement; + close: () => void; +}; + +type DatabaseSyncConstructor = new (filePath: string) => SqliteDatabase; + +export type QuoteCachePolicy = { + enabled: boolean; + retentionMs: number; + maxMessagesPerAccount: number; + mediaRetentionMs: number; + maxMediaBytesPerAccount: number; + maxSingleMediaBytes: number; +}; + +export type QuoteMessageRecord = { + accountId: string; + conversationId: string; + messageId: string; + direction: "inbound" | "outbound"; + body: string; + mediaPath?: string; + mediaMime?: string; + mediaName?: string; + mediaSize?: number; + createdAt: number; +}; + +export type QuoteMessageInput = Omit & { + sourceMediaPath?: string; +}; + +type QuoteCacheConfig = { + enabled?: boolean; + retentionDays?: number; + maxMessagesPerAccount?: number; + mediaRetentionDays?: number; + maxMediaBytesPerAccount?: number; + maxSingleMediaBytes?: number; +}; + +type WeixinChannelConfig = { + quoteCache?: QuoteCacheConfig; +}; + +type StoredRow = { + account_id: string; + conversation_id: string; + message_id: string; + direction: "inbound" | "outbound"; + body: string; + media_path: string | null; + media_mime: string | null; + media_name: string | null; + media_size: number | null; + created_at: number; +}; + +type MediaRow = { + media_path: string; + media_size: number; + newest_at: number; +}; + +function positiveNumber(value: unknown, fallback: number): number { + return typeof value === "number" && Number.isFinite(value) && value > 0 ? value : fallback; +} + +function positiveInteger(value: unknown, fallback: number): number { + return Math.max(1, Math.floor(positiveNumber(value, fallback))); +} + +export function resolveQuoteCachePolicy(cfg: OpenClawConfig): QuoteCachePolicy { + const section = cfg.channels?.["openclaw-weixin"] as WeixinChannelConfig | undefined; + const quote = section?.quoteCache; + return { + enabled: quote?.enabled !== false, + retentionMs: + positiveNumber(quote?.retentionDays, DEFAULT_RETENTION_DAYS) * 24 * 60 * 60 * 1000, + maxMessagesPerAccount: positiveInteger( + quote?.maxMessagesPerAccount, + DEFAULT_MAX_MESSAGES_PER_ACCOUNT, + ), + mediaRetentionMs: + positiveNumber(quote?.mediaRetentionDays, DEFAULT_MEDIA_RETENTION_DAYS) * + 24 * + 60 * + 60 * + 1000, + maxMediaBytesPerAccount: positiveInteger( + quote?.maxMediaBytesPerAccount, + DEFAULT_MAX_MEDIA_BYTES_PER_ACCOUNT, + ), + maxSingleMediaBytes: positiveInteger( + quote?.maxSingleMediaBytes, + DEFAULT_MAX_SINGLE_MEDIA_BYTES, + ), + }; +} + +function safePathSegment(raw: string): string { + const safe = raw.trim().toLowerCase().replace(/[^a-z0-9._-]+/g, "_").replace(/\.\.+/g, "_"); + return safe && safe !== "." && safe !== ".." ? safe : "default"; +} + +function accountMediaDirName(accountId: string): string { + const readable = safePathSegment(accountId).slice(0, 48); + const digest = crypto.createHash("sha256").update(accountId, "utf8").digest("hex").slice(0, 16); + return `${readable}-${digest}`; +} + +function safeMediaExtension(filePath: string): string { + const ext = path.extname(filePath).toLowerCase(); + return /^\.[a-z0-9]{1,16}$/.test(ext) ? ext : ""; +} + +function asStoredRow(value: unknown): StoredRow | null { + if (!value || typeof value !== "object") return null; + return value as StoredRow; +} + +export class QuoteStore { + private readonly db: SqliteDatabase; + private readonly rootDir: string; + private readonly mediaRoot: string; + private policy: QuoteCachePolicy; + private writesSinceGc = 0; + private gcRunning = false; + private gcRequested = false; + private gcScheduled = false; + private closed = false; + private readonly gcTimer: ReturnType; + + private constructor(db: SqliteDatabase, rootDir: string, policy: QuoteCachePolicy) { + this.db = db; + this.rootDir = rootDir; + this.mediaRoot = path.join(rootDir, "ref-media"); + this.policy = policy; + this.initializeSchema(); + this.gcTimer = setInterval(() => this.requestGc(), GC_TIMER_MS); + this.gcTimer.unref(); + this.runGc(); + } + + static async open(params: { + policy: QuoteCachePolicy; + rootDir?: string; + }): Promise { + if (!params.policy.enabled) return null; + try { + const sqlite = (await import("node:sqlite")) as unknown as { + DatabaseSync: DatabaseSyncConstructor; + }; + const rootDir = params.rootDir ?? path.join(resolveStateDir(), "openclaw-weixin"); + fs.mkdirSync(rootDir, { recursive: true, mode: 0o700 }); + try { + fs.chmodSync(rootDir, 0o700); + } catch { + // best-effort on filesystems without POSIX permissions + } + const dbPath = path.join(rootDir, "ref-messages.sqlite"); + const db = new sqlite.DatabaseSync(dbPath); + try { + fs.chmodSync(dbPath, 0o600); + } catch { + // best-effort + } + return new QuoteStore(db, rootDir, params.policy); + } catch (err) { + logger.warn( + `quote cache disabled: node:sqlite is unavailable or the database could not be opened: ${String(err)}`, + ); + return null; + } + } + + updatePolicy(policy: QuoteCachePolicy): void { + this.policy = policy; + this.requestGc(); + } + + private initializeSchema(): void { + this.db.exec(` + PRAGMA journal_mode = WAL; + PRAGMA synchronous = NORMAL; + PRAGMA busy_timeout = 5000; + CREATE TABLE IF NOT EXISTS quote_messages ( + account_id TEXT NOT NULL, + conversation_id TEXT NOT NULL, + message_id TEXT NOT NULL, + direction TEXT NOT NULL, + body TEXT NOT NULL DEFAULT '', + media_path TEXT, + media_mime TEXT, + media_name TEXT, + media_size INTEGER, + created_at INTEGER NOT NULL, + PRIMARY KEY (account_id, conversation_id, message_id) + ); + CREATE INDEX IF NOT EXISTS quote_messages_account_created + ON quote_messages (account_id, created_at); + CREATE INDEX IF NOT EXISTS quote_messages_media_path + ON quote_messages (media_path); + `); + } + + find(accountId: string, conversationId: string, messageId: string): QuoteMessageRecord | null { + if (this.closed || !messageId) return null; + const row = asStoredRow( + this.db + .prepare(` + SELECT account_id, conversation_id, message_id, direction, body, + media_path, media_mime, media_name, media_size, created_at + FROM quote_messages + WHERE account_id = ? AND conversation_id = ? AND message_id = ? + `) + .get(accountId, conversationId, messageId), + ); + if (!row) return null; + if (Date.now() - row.created_at > this.policy.retentionMs) { + this.deleteMessage(accountId, conversationId, messageId); + return null; + } + return { + accountId: row.account_id, + conversationId: row.conversation_id, + messageId: row.message_id, + direction: row.direction, + body: row.body, + ...(row.media_path ? { mediaPath: row.media_path } : {}), + ...(row.media_mime ? { mediaMime: row.media_mime } : {}), + ...(row.media_name ? { mediaName: row.media_name } : {}), + ...(typeof row.media_size === "number" ? { mediaSize: row.media_size } : {}), + createdAt: row.created_at, + }; + } + + async put(input: QuoteMessageInput): Promise { + if (this.closed || !input.messageId || (!input.body && !input.sourceMediaPath)) return; + const media = input.sourceMediaPath + ? await this.cacheMedia(input.accountId, input.sourceMediaPath, input.mediaMime) + : null; + this.db + .prepare(` + INSERT INTO quote_messages ( + account_id, conversation_id, message_id, direction, body, + media_path, media_mime, media_name, media_size, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(account_id, conversation_id, message_id) DO UPDATE SET + direction = excluded.direction, + body = excluded.body, + media_path = COALESCE(excluded.media_path, quote_messages.media_path), + media_mime = COALESCE(excluded.media_mime, quote_messages.media_mime), + media_name = COALESCE(excluded.media_name, quote_messages.media_name), + media_size = COALESCE(excluded.media_size, quote_messages.media_size), + created_at = excluded.created_at + `) + .run( + input.accountId, + input.conversationId, + input.messageId, + input.direction, + input.body, + media?.path ?? null, + input.mediaMime ?? media?.mime ?? null, + input.mediaName ?? media?.name ?? null, + media?.size ?? null, + Number.isFinite(input.createdAt) && input.createdAt > 0 ? input.createdAt : Date.now(), + ); + + this.writesSinceGc += 1; + if (media) { + this.enforceMediaBudget(input.accountId); + } + if (this.writesSinceGc >= GC_WRITE_INTERVAL) { + this.writesSinceGc = 0; + this.requestGc(); + } + } + + deleteAccount(accountId: string): void { + if (this.closed) return; + this.db.prepare("DELETE FROM quote_messages WHERE account_id = ?").run(accountId); + const accountMediaDir = path.join(this.mediaRoot, accountMediaDirName(accountId)); + try { + fs.rmSync(accountMediaDir, { recursive: true, force: true }); + } catch (err) { + logger.warn(`quote cache: failed to remove media for account=${accountId}: ${String(err)}`); + } + } + + private deleteMessage(accountId: string, conversationId: string, messageId: string): void { + const existing = this.findMediaPath(accountId, conversationId, messageId); + this.db + .prepare( + "DELETE FROM quote_messages WHERE account_id = ? AND conversation_id = ? AND message_id = ?", + ) + .run(accountId, conversationId, messageId); + if (existing) this.removeMediaIfOrphaned(existing); + } + + private findMediaPath(accountId: string, conversationId: string, messageId: string): string | null { + const row = this.db + .prepare( + "SELECT media_path FROM quote_messages WHERE account_id = ? AND conversation_id = ? AND message_id = ?", + ) + .get(accountId, conversationId, messageId) as { media_path?: unknown } | undefined; + return typeof row?.media_path === "string" ? row.media_path : null; + } + + private async cacheMedia( + accountId: string, + sourcePath: string, + mime?: string, + ): Promise<{ path: string; mime?: string; name: string; size: number } | null> { + try { + const stat = await fs.promises.stat(sourcePath); + if (!stat.isFile() || stat.size > this.policy.maxSingleMediaBytes) return null; + const data = await fs.promises.readFile(sourcePath); + const hash = crypto.createHash("sha256").update(data).digest("hex"); + const accountDir = path.join(this.mediaRoot, accountMediaDirName(accountId)); + await fs.promises.mkdir(accountDir, { recursive: true, mode: 0o700 }); + const destination = path.join(accountDir, `${hash}${safeMediaExtension(sourcePath)}`); + try { + await fs.promises.writeFile(destination, data, { flag: "wx", mode: 0o600 }); + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "EEXIST") throw err; + } + return { + path: destination, + ...(mime ? { mime } : {}), + name: path.basename(sourcePath), + size: stat.size, + }; + } catch (err) { + logger.warn(`quote cache: failed to cache media path=${sourcePath}: ${String(err)}`); + return null; + } + } + + private requestGc(): void { + if (this.closed) return; + if (this.gcRunning) { + this.gcRequested = true; + return; + } + if (this.gcScheduled) return; + this.gcScheduled = true; + queueMicrotask(() => { + this.gcScheduled = false; + this.runGc(); + }); + } + + runGc(now = Date.now()): void { + if (this.closed || this.gcRunning) { + this.gcRequested = true; + return; + } + this.gcRunning = true; + try { + const messageCutoff = now - this.policy.retentionMs; + this.db.prepare("DELETE FROM quote_messages WHERE created_at < ?").run(messageCutoff); + + const accounts = this.db + .prepare("SELECT DISTINCT account_id FROM quote_messages") + .all() as Array<{ account_id: string }>; + for (const { account_id: accountId } of accounts) { + this.db + .prepare(` + DELETE FROM quote_messages + WHERE rowid IN ( + SELECT rowid FROM quote_messages + WHERE account_id = ? + ORDER BY created_at DESC, rowid DESC + LIMIT -1 OFFSET ? + ) + `) + .run(accountId, this.policy.maxMessagesPerAccount); + this.expireOldMedia(accountId, now - this.policy.mediaRetentionMs); + this.enforceMediaBudget(accountId); + } + this.pruneOrphanMediaFiles(); + } catch (err) { + logger.warn(`quote cache GC failed: ${String(err)}`); + } finally { + this.gcRunning = false; + if (this.gcRequested) { + this.gcRequested = false; + this.requestGc(); + } + } + } + + private listMedia(accountId: string): MediaRow[] { + return this.db + .prepare(` + SELECT media_path, MAX(COALESCE(media_size, 0)) AS media_size, + MAX(created_at) AS newest_at + FROM quote_messages + WHERE account_id = ? AND media_path IS NOT NULL + GROUP BY media_path + ORDER BY newest_at ASC + `) + .all(accountId) as MediaRow[]; + } + + private expireOldMedia(accountId: string, cutoff: number): void { + for (const media of this.listMedia(accountId)) { + if (media.newest_at >= cutoff) continue; + this.detachMedia(media.media_path); + } + } + + private enforceMediaBudget(accountId: string): void { + const rows = this.listMedia(accountId); + let total = rows.reduce((sum, row) => sum + Math.max(0, row.media_size), 0); + for (const row of rows) { + if (total <= this.policy.maxMediaBytesPerAccount) break; + this.detachMedia(row.media_path); + total -= Math.max(0, row.media_size); + } + } + + private detachMedia(mediaPath: string): void { + this.db + .prepare("UPDATE quote_messages SET media_path = NULL WHERE media_path = ?") + .run(mediaPath); + this.removeMediaIfOrphaned(mediaPath); + } + + private removeMediaIfOrphaned(mediaPath: string): void { + const row = this.db + .prepare("SELECT 1 AS found FROM quote_messages WHERE media_path = ? LIMIT 1") + .get(mediaPath) as { found?: number } | undefined; + if (row?.found) return; + try { + fs.unlinkSync(mediaPath); + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "ENOENT") { + logger.warn(`quote cache: failed to remove media path=${mediaPath}: ${String(err)}`); + } + } + } + + private pruneOrphanMediaFiles(): void { + if (!fs.existsSync(this.mediaRoot)) return; + const known = new Set( + (this.db + .prepare("SELECT DISTINCT media_path FROM quote_messages WHERE media_path IS NOT NULL") + .all() as Array<{ media_path: string }>).map((row) => row.media_path), + ); + const visit = (dir: string) => { + for (const entry of fs.readdirSync(dir, { withFileTypes: true })) { + const fullPath = path.join(dir, entry.name); + if (entry.isDirectory()) visit(fullPath); + else if (entry.isFile() && !known.has(fullPath)) { + try { + fs.unlinkSync(fullPath); + } catch { + // retry during the next GC pass + } + } + } + }; + visit(this.mediaRoot); + } + + close(): void { + if (this.closed) return; + this.closed = true; + clearInterval(this.gcTimer); + this.db.close(); + } +} + +let activeStore: QuoteStore | null = null; +let initialization: Promise | null = null; +const activeAccounts = new Set(); + +export async function initializeQuoteStore( + cfg: OpenClawConfig, + accountId?: string, +): Promise { + if (accountId) activeAccounts.add(accountId); + const policy = resolveQuoteCachePolicy(cfg); + if (!policy.enabled) { + activeStore?.close(); + activeStore = null; + initialization = null; + return null; + } + if (activeStore) { + activeStore.updatePolicy(policy); + return activeStore; + } + if (!initialization) { + initialization = QuoteStore.open({ policy }).then((store) => { + activeStore = store; + return store; + }); + } + return initialization; +} + +export function getQuoteStore(): QuoteStore | null { + return activeStore; +} + +export function closeQuoteStore(): void { + activeStore?.close(); + activeStore = null; + initialization = null; + activeAccounts.clear(); +} + +export function deleteQuoteCacheForAccount(accountId: string): void { + if (activeStore) { + activeStore.deleteAccount(accountId); + return; + } + + const rootDir = path.join(resolveStateDir(), "openclaw-weixin"); + const dbPath = path.join(rootDir, "ref-messages.sqlite"); + if (!fs.existsSync(dbPath)) return; + try { + const require = createRequire(import.meta.url); + const sqlite = require("node:sqlite") as { DatabaseSync: DatabaseSyncConstructor }; + const db = new sqlite.DatabaseSync(dbPath); + db.prepare("DELETE FROM quote_messages WHERE account_id = ?").run(accountId); + db.close(); + fs.rmSync(path.join(rootDir, "ref-media", accountMediaDirName(accountId)), { + recursive: true, + force: true, + }); + } catch (err) { + logger.warn(`quote cache: failed to delete account=${accountId}: ${String(err)}`); + } +} + +export function deactivateQuoteStoreAccount(accountId: string): void { + activeAccounts.delete(accountId); + if (activeAccounts.size === 0) closeQuoteStore(); +} diff --git a/src/messaging/reply-progress-sender.ts b/src/messaging/reply-progress-sender.ts index f1e060b..5ad90e8 100644 --- a/src/messaging/reply-progress-sender.ts +++ b/src/messaging/reply-progress-sender.ts @@ -12,6 +12,7 @@ export type WeixinReplyProgressSenderDeps = { opts: WeixinApiOptions & { contextToken?: string; runId?: string; + accountId?: string; }; }; @@ -44,7 +45,7 @@ export class WeixinReplyProgressSender { this.runId = deps.runId; this.to = deps.to; this.accountId = deps.accountId; - this.opts = { ...deps.opts, runId: deps.runId }; + this.opts = { ...deps.opts, runId: deps.runId, accountId: deps.accountId }; } get replyOptions() { diff --git a/src/messaging/send-media.test.ts b/src/messaging/send-media.test.ts index 98ca68c..e7d337f 100644 --- a/src/messaging/send-media.test.ts +++ b/src/messaging/send-media.test.ts @@ -85,6 +85,8 @@ describe("sendWeixinMediaFile", () => { fileName: "doc.pdf", uploaded: fakeUploaded, opts: baseParams.opts, + filePath: "/tmp/doc.pdf", + mediaMime: "application/pdf", }); }); diff --git a/src/messaging/send-media.ts b/src/messaging/send-media.ts index da3ecbd..0ecdf83 100644 --- a/src/messaging/send-media.ts +++ b/src/messaging/send-media.ts @@ -18,7 +18,7 @@ export async function sendWeixinMediaFile(params: { filePath: string; to: string; text: string; - opts: WeixinApiOptions & { contextToken?: string; runId?: string }; + opts: WeixinApiOptions & { contextToken?: string; runId?: string; accountId?: string }; cdnBaseUrl: string; }): Promise<{ messageId: string }> { const { filePath, to, text, opts, cdnBaseUrl } = params; @@ -36,7 +36,7 @@ export async function sendWeixinMediaFile(params: { logger.info( `[weixin] sendWeixinMediaFile: video upload done filekey=${uploaded.filekey} size=${uploaded.fileSize}`, ); - return sendVideoMessageWeixin({ to, text, uploaded, opts }); + return sendVideoMessageWeixin({ to, text, uploaded, opts, filePath, mediaMime: mime }); } if (mime.startsWith("image/")) { @@ -50,7 +50,7 @@ export async function sendWeixinMediaFile(params: { logger.info( `[weixin] sendWeixinMediaFile: image upload done filekey=${uploaded.filekey} size=${uploaded.fileSize}`, ); - return sendImageMessageWeixin({ to, text, uploaded, opts }); + return sendImageMessageWeixin({ to, text, uploaded, opts, filePath, mediaMime: mime }); } // File attachment: pdf, doc, zip, etc. @@ -68,5 +68,5 @@ export async function sendWeixinMediaFile(params: { logger.info( `[weixin] sendWeixinMediaFile: file upload done filekey=${uploaded.filekey} size=${uploaded.fileSize}`, ); - return sendFileMessageWeixin({ to, text, fileName, uploaded, opts }); + return sendFileMessageWeixin({ to, text, fileName, uploaded, opts, filePath, mediaMime: mime }); } diff --git a/src/messaging/send.test.ts b/src/messaging/send.test.ts index c255c05..ed6f957 100644 --- a/src/messaging/send.test.ts +++ b/src/messaging/send.test.ts @@ -9,7 +9,8 @@ vi.mock("../util/logger.js", () => ({ }, })); -const { mockSendMessageApi } = vi.hoisted(() => ({ +const { mockQuotePut, mockSendMessageApi } = vi.hoisted(() => ({ + mockQuotePut: vi.fn(), mockSendMessageApi: vi.fn(), })); @@ -17,6 +18,10 @@ vi.mock("../api/api.js", () => ({ sendMessage: mockSendMessageApi, })); +vi.mock("./quote-store.js", () => ({ + getQuoteStore: () => ({ put: mockQuotePut }), +})); + vi.mock("node:crypto", () => ({ default: { randomBytes: vi.fn(() => Buffer.from("deadbeef", "hex")), @@ -105,6 +110,34 @@ describe("sendMessageWeixin", () => { }), ).rejects.toThrow("api fail"); }); + + it("stores outbound text under the lossless server ID", async () => { + mockSendMessageApi.mockResolvedValueOnce({ message_id: "18446744073709551615" }); + const result = await sendMessageWeixin({ + to: "user1", + text: "hello", + opts: { baseUrl: "https://api.com", accountId: "account1" }, + }); + expect(result.serverMessageId).toBe("18446744073709551615"); + expect(result.messageId).not.toBe(result.serverMessageId); + expect(mockQuotePut).toHaveBeenCalledWith(expect.objectContaining({ + accountId: "account1", + conversationId: "user1", + messageId: "18446744073709551615", + body: "hello", + direction: "outbound", + })); + }); + + it("does not turn a successful send into a failure when caching fails", async () => { + mockSendMessageApi.mockResolvedValueOnce({ message_id: "99" }); + mockQuotePut.mockRejectedValueOnce(new Error("disk full")); + await expect(sendMessageWeixin({ + to: "user1", + text: "hello", + opts: { baseUrl: "https://api.com", accountId: "account1" }, + })).resolves.toMatchObject({ serverMessageId: "99" }); + }); }); describe("sendMessageItemWeixin", () => { @@ -135,6 +168,40 @@ describe("sendMessageItemWeixin", () => { }, ]); }); + + it("caches a structured text item and preserves an explicit client ID", async () => { + mockSendMessageApi.mockResolvedValueOnce({ message_id: "123" }); + const result = await sendMessageItemWeixin({ + to: "user1", + item: { type: MessageItemType.TEXT, text_item: { text: "structured text" } }, + opts: { baseUrl: "https://api.com", accountId: "account1" }, + clientId: "explicit-client-id", + }); + expect(result).toEqual({ messageId: "explicit-client-id", serverMessageId: "123" }); + expect(mockQuotePut).toHaveBeenCalledWith(expect.objectContaining({ + messageId: "123", + body: "structured text", + })); + }); + + it("returns the client ID when a structured item has no server ID", async () => { + mockSendMessageApi.mockResolvedValueOnce({}); + const result = await sendMessageItemWeixin({ + to: "user1", + item: { type: MessageItemType.TEXT, text_item: {} }, + opts: { baseUrl: "https://api.com" }, + }); + expect(result.serverMessageId).toBeUndefined(); + }); + + it("rethrows structured-item API errors", async () => { + mockSendMessageApi.mockRejectedValueOnce(new Error("structured fail")); + await expect(sendMessageItemWeixin({ + to: "user1", + item: { type: MessageItemType.TEXT, text_item: { text: "hello" } }, + opts: { baseUrl: "https://api.com" }, + })).rejects.toThrow("structured fail"); + }); }); function makeUploadedFileInfo(overrides?: Partial): UploadedFileInfo { @@ -213,6 +280,25 @@ describe("sendImageMessageWeixin", () => { }), ).rejects.toThrow("cdn fail"); }); + + it("stores managed metadata for an outbound image", async () => { + mockSendMessageApi.mockResolvedValueOnce({ message_id: "image-server-id" }); + await sendImageMessageWeixin({ + to: "user1", + text: "", + uploaded: makeUploadedFileInfo(), + opts: { baseUrl: "https://api.com", accountId: "account1" }, + filePath: "/tmp/photo.png", + mediaMime: "image/png", + }); + expect(mockQuotePut).toHaveBeenCalledWith(expect.objectContaining({ + messageId: "image-server-id", + body: "[图片]", + sourceMediaPath: "/tmp/photo.png", + mediaMime: "image/png", + mediaName: "photo.png", + })); + }); }); describe("sendVideoMessageWeixin", () => { diff --git a/src/messaging/send.ts b/src/messaging/send.ts index 9c8379e..0bc12e0 100644 --- a/src/messaging/send.ts +++ b/src/messaging/send.ts @@ -7,14 +7,48 @@ import { generateId } from "../util/random.js"; import type { MessageItem, SendMessageReq } from "../api/types.js"; import { MessageItemType, MessageState, MessageType } from "../api/types.js"; import type { UploadedFileInfo } from "../cdn/upload.js"; +import { getMediaLabel } from "./inbound.js"; +import { getQuoteStore } from "./quote-store.js"; export { StreamingMarkdownFilter } from "./markdown-filter.js"; type WeixinMessageSendOptions = WeixinApiOptions & { contextToken?: string; runId?: string; + accountId?: string; }; +type WeixinSendResult = { messageId: string; serverMessageId?: string }; + +async function cacheOutboundMessage(params: { + opts: WeixinMessageSendOptions; + to: string; + serverMessageId?: string; + body: string; + sourceMediaPath?: string; + mediaMime?: string; + mediaName?: string; +}): Promise { + if (!params.opts.accountId || !params.serverMessageId) return; + try { + await getQuoteStore()?.put({ + accountId: params.opts.accountId, + conversationId: params.to, + messageId: params.serverMessageId, + direction: "outbound", + body: params.body, + ...(params.sourceMediaPath ? { sourceMediaPath: params.sourceMediaPath } : {}), + ...(params.mediaMime ? { mediaMime: params.mediaMime } : {}), + ...(params.mediaName ? { mediaName: params.mediaName } : {}), + createdAt: Date.now(), + }); + } catch (err) { + logger.warn( + `quote cache: failed to save outbound message id=${params.serverMessageId}: ${String(err)}`, + ); + } +} + function generateClientId(): string { return generateId("openclaw-weixin"); } @@ -70,7 +104,7 @@ export async function sendMessageWeixin(params: { to: string; text: string; opts: WeixinMessageSendOptions; -}): Promise<{ messageId: string }> { +}): Promise { const { to, text, opts } = params; if (!opts.contextToken) { logger.warn(`sendMessageWeixin: contextToken missing for to=${to}, sending without context`); @@ -84,17 +118,22 @@ export async function sendMessageWeixin(params: { clientId, }); try { - await sendMessageApi({ + const response = await sendMessageApi({ baseUrl: opts.baseUrl, token: opts.token, timeoutMs: opts.timeoutMs, body: req, }); + const serverMessageId = response?.message_id; + await cacheOutboundMessage({ opts, to, serverMessageId, body: text }); + return { + messageId: clientId, + ...(serverMessageId ? { serverMessageId } : {}), + }; } catch (err) { logger.error(`sendMessageWeixin: failed to=${to} clientId=${clientId} err=${String(err)}`); throw err; } - return { messageId: clientId }; } /** Send a single structured MessageItem downstream. */ @@ -104,7 +143,7 @@ export async function sendMessageItemWeixin(params: { opts: WeixinMessageSendOptions; clientId?: string; label?: string; -}): Promise<{ messageId: string }> { +}): Promise { const { to, item, opts } = params; if (!opts.contextToken) { logger.warn(`sendMessageItemWeixin: contextToken missing for to=${to}, sending without context`); @@ -123,19 +162,25 @@ export async function sendMessageItemWeixin(params: { }, }; try { - await sendMessageApi({ + const response = await sendMessageApi({ baseUrl: opts.baseUrl, token: opts.token, timeoutMs: opts.timeoutMs, body: req, }); + const serverMessageId = response?.message_id; + const itemText = item.type === MessageItemType.TEXT ? item.text_item?.text ?? "" : ""; + if (itemText) await cacheOutboundMessage({ opts, to, serverMessageId, body: itemText }); + return { + messageId: clientId, + ...(serverMessageId ? { serverMessageId } : {}), + }; } catch (err) { logger.error( `${params.label ?? "sendMessageItemWeixin"}: failed to=${to} clientId=${clientId} err=${String(err)}`, ); throw err; } - return { messageId: clientId }; } /** @@ -148,7 +193,10 @@ async function sendMediaItems(params: { mediaItem: MessageItem; opts: WeixinMessageSendOptions; label: string; -}): Promise<{ messageId: string }> { + sourceMediaPath?: string; + mediaMime?: string; + mediaName?: string; +}): Promise { const { to, text, mediaItem, opts, label } = params; const runId = opts.runId; @@ -159,6 +207,7 @@ async function sendMediaItems(params: { items.push(mediaItem); let lastClientId = ""; + let lastServerMessageId: string | undefined; for (const item of items) { lastClientId = generateClientId(); const req: SendMessageReq = { @@ -174,12 +223,31 @@ async function sendMediaItems(params: { }, }; try { - await sendMessageApi({ + const response = await sendMessageApi({ baseUrl: opts.baseUrl, token: opts.token, timeoutMs: opts.timeoutMs, body: req, }); + lastServerMessageId = response?.message_id; + if (item.type === MessageItemType.TEXT) { + await cacheOutboundMessage({ + opts, + to, + serverMessageId: lastServerMessageId, + body: item.text_item?.text ?? text, + }); + } else { + await cacheOutboundMessage({ + opts, + to, + serverMessageId: lastServerMessageId, + body: getMediaLabel(item.type), + sourceMediaPath: params.sourceMediaPath, + mediaMime: params.mediaMime, + mediaName: params.mediaName, + }); + } } catch (err) { logger.error( `${label}: failed to=${to} clientId=${lastClientId} err=${String(err)}`, @@ -189,7 +257,10 @@ async function sendMediaItems(params: { } logger.info(`${label}: success to=${to} clientId=${lastClientId}`); - return { messageId: lastClientId }; + return { + messageId: lastClientId, + ...(lastServerMessageId ? { serverMessageId: lastServerMessageId } : {}), + }; } /** @@ -206,7 +277,9 @@ export async function sendImageMessageWeixin(params: { text: string; uploaded: UploadedFileInfo; opts: WeixinMessageSendOptions; -}): Promise<{ messageId: string }> { + filePath?: string; + mediaMime?: string; +}): Promise { const { to, text, uploaded, opts } = params; if (!opts.contextToken) { logger.warn(`sendImageMessageWeixin: contextToken missing for to=${to}, sending without context`); @@ -227,7 +300,16 @@ export async function sendImageMessageWeixin(params: { }, }; - return sendMediaItems({ to, text, mediaItem: imageItem, opts, label: "sendImageMessageWeixin" }); + return sendMediaItems({ + to, + text, + mediaItem: imageItem, + opts, + label: "sendImageMessageWeixin", + sourceMediaPath: params.filePath, + mediaMime: params.mediaMime, + mediaName: params.filePath?.split(/[\\/]/).pop(), + }); } /** @@ -240,7 +322,9 @@ export async function sendVideoMessageWeixin(params: { text: string; uploaded: UploadedFileInfo; opts: WeixinMessageSendOptions; -}): Promise<{ messageId: string }> { + filePath?: string; + mediaMime?: string; +}): Promise { const { to, text, uploaded, opts } = params; if (!opts.contextToken) { logger.warn(`sendVideoMessageWeixin: contextToken missing for to=${to}, sending without context`); @@ -258,7 +342,16 @@ export async function sendVideoMessageWeixin(params: { }, }; - return sendMediaItems({ to, text, mediaItem: videoItem, opts, label: "sendVideoMessageWeixin" }); + return sendMediaItems({ + to, + text, + mediaItem: videoItem, + opts, + label: "sendVideoMessageWeixin", + sourceMediaPath: params.filePath, + mediaMime: params.mediaMime, + mediaName: params.filePath?.split(/[\\/]/).pop(), + }); } /** @@ -272,7 +365,9 @@ export async function sendFileMessageWeixin(params: { fileName: string; uploaded: UploadedFileInfo; opts: WeixinMessageSendOptions; -}): Promise<{ messageId: string }> { + filePath?: string; + mediaMime?: string; +}): Promise { const { to, text, fileName, uploaded, opts } = params; if (!opts.contextToken) { logger.warn(`sendFileMessageWeixin: contextToken missing for to=${to}, sending without context`); @@ -290,5 +385,14 @@ export async function sendFileMessageWeixin(params: { }, }; - return sendMediaItems({ to, text, mediaItem: fileItem, opts, label: "sendFileMessageWeixin" }); + return sendMediaItems({ + to, + text, + mediaItem: fileItem, + opts, + label: "sendFileMessageWeixin", + sourceMediaPath: params.filePath, + mediaMime: params.mediaMime, + mediaName: fileName, + }); } diff --git a/src/messaging/slash-commands.ts b/src/messaging/slash-commands.ts index 9d2160e..0c43894 100644 --- a/src/messaging/slash-commands.ts +++ b/src/messaging/slash-commands.ts @@ -28,10 +28,11 @@ export interface SlashCommandContext { /** 发送回复消息 */ async function sendReply(ctx: SlashCommandContext, text: string): Promise { - const opts: WeixinApiOptions & { contextToken?: string } = { + const opts: WeixinApiOptions & { contextToken?: string; accountId?: string } = { baseUrl: ctx.baseUrl, token: ctx.token, contextToken: ctx.contextToken, + accountId: ctx.accountId, }; await sendMessageWeixin({ to: ctx.to, text, opts }); } From e94ab0dd8189fb87a6e66070e2374ff7e95a079e Mon Sep 17 00:00:00 2001 From: scotthuang Date: Wed, 2 Sep 2026 14:30:35 +0800 Subject: [PATCH 2/5] fix: keep quoted media in managed inbound storage --- src/media/media-download.ts | 13 ++-- src/messaging/inbound.test.ts | 10 +++ src/messaging/inbound.ts | 14 ++++ src/messaging/process-message.ts | 3 +- src/messaging/quote-store.test.ts | 124 +++++++++++++++++++++++++----- src/messaging/quote-store.ts | 121 ++++++++++++++++++++++++----- 6 files changed, 238 insertions(+), 47 deletions(-) diff --git a/src/media/media-download.ts b/src/media/media-download.ts index 0ad7b19..43e3036 100644 --- a/src/media/media-download.ts +++ b/src/media/media-download.ts @@ -29,12 +29,15 @@ export async function downloadMediaFromItem( deps: { cdnBaseUrl: string; saveMedia: SaveMediaFn; + /** OpenClaw media-store subdirectory; defaults to the transient inbound bucket. */ + mediaSubdir?: string; log: (msg: string) => void; errLog: (msg: string) => void; label: string; }, ): Promise { const { cdnBaseUrl, saveMedia, log, errLog, label } = deps; + const mediaSubdir = deps.mediaSubdir ?? "inbound"; const result: WeixinInboundMediaOpts = {}; if (item.type === MessageItemType.IMAGE) { @@ -61,7 +64,7 @@ export async function downloadMediaFromItem( `${label} image-plain`, img.media.full_url, ); - const saved = await saveMedia(buf, undefined, "inbound", WEIXIN_MEDIA_MAX_BYTES); + const saved = await saveMedia(buf, undefined, mediaSubdir, WEIXIN_MEDIA_MAX_BYTES); result.decryptedPicPath = saved.path; logger.debug(`${label} image saved: ${saved.path}`); } catch (err) { @@ -83,12 +86,12 @@ export async function downloadMediaFromItem( logger.debug(`${label} voice: decrypted ${silkBuf.length} bytes, attempting silk transcode`); const wavBuf = await silkToWav(silkBuf); if (wavBuf) { - const saved = await saveMedia(wavBuf, "audio/wav", "inbound", WEIXIN_MEDIA_MAX_BYTES); + const saved = await saveMedia(wavBuf, "audio/wav", mediaSubdir, WEIXIN_MEDIA_MAX_BYTES); result.decryptedVoicePath = saved.path; result.voiceMediaType = "audio/wav"; logger.debug(`${label} voice: saved WAV to ${saved.path}`); } else { - const saved = await saveMedia(silkBuf, "audio/silk", "inbound", WEIXIN_MEDIA_MAX_BYTES); + const saved = await saveMedia(silkBuf, "audio/silk", mediaSubdir, WEIXIN_MEDIA_MAX_BYTES); result.decryptedVoicePath = saved.path; result.voiceMediaType = "audio/silk"; logger.debug(`${label} voice: silk transcode unavailable, saved raw SILK to ${saved.path}`); @@ -113,7 +116,7 @@ export async function downloadMediaFromItem( const saved = await saveMedia( buf, mime, - "inbound", + mediaSubdir, WEIXIN_MEDIA_MAX_BYTES, fileItem.file_name ?? undefined, ); @@ -136,7 +139,7 @@ export async function downloadMediaFromItem( `${label} video`, videoItem.media.full_url, ); - const saved = await saveMedia(buf, "video/mp4", "inbound", WEIXIN_MEDIA_MAX_BYTES); + const saved = await saveMedia(buf, "video/mp4", mediaSubdir, WEIXIN_MEDIA_MAX_BYTES); result.decryptedVideoPath = saved.path; logger.debug(`${label} video: saved to ${saved.path}`); } catch (err) { diff --git a/src/messaging/inbound.test.ts b/src/messaging/inbound.test.ts index 0f6ce18..b4bab47 100644 --- a/src/messaging/inbound.test.ts +++ b/src/messaging/inbound.test.ts @@ -434,11 +434,21 @@ describe("stored quote resolution", () => { find: () => ({ accountId: "acc", conversationId: "user1", messageId: "9007199254740993123", direction: "inbound", body: "[图片]", mediaPath: quotedPath, mediaMime: "image/png", + mediaName: "quoted-original.png", createdAt: Date.now(), }), }); expect(ctx.MediaPaths).toEqual(["/tmp/current.pdf", quotedPath]); expect(ctx.MediaTypes).toEqual(["application/octet-stream", "image/png"]); + expect(ctx.media).toEqual([ + { path: "/tmp/current.pdf", contentType: "application/octet-stream" }, + { + path: quotedPath, + contentType: "image/png", + fileName: "quoted-original.png", + messageId: "9007199254740993123", + }, + ]); } finally { fs.rmSync(tempDir, { recursive: true, force: true }); } diff --git a/src/messaging/inbound.ts b/src/messaging/inbound.ts index f1d8033..cb75703 100644 --- a/src/messaging/inbound.ts +++ b/src/messaging/inbound.ts @@ -160,6 +160,12 @@ export type WeixinMsgContext = { MediaType?: string; MediaPaths?: string[]; MediaTypes?: string[]; + media?: Array<{ + path?: string; + contentType?: string; + fileName?: string; + messageId?: string; + }>; ReplyToId?: string; ReplyToBody?: string; ReplyToQuoteText?: string; @@ -372,6 +378,14 @@ export function resolveStoredQuoteContext( ctx.MediaTypes ?? (ctx.MediaPath ? [ctx.MediaType ?? "application/octet-stream"] : []); ctx.MediaPaths = [...currentPaths, record.mediaPath]; ctx.MediaTypes = [...currentTypes, record.mediaMime ?? "application/octet-stream"]; + ctx.media = ctx.MediaPaths.map((mediaPath, index) => ({ + path: mediaPath, + contentType: ctx.MediaTypes?.[index] ?? "application/octet-stream", + ...(index === ctx.MediaPaths!.length - 1 && record.mediaName + ? { fileName: record.mediaName } + : {}), + ...(index === ctx.MediaPaths!.length - 1 ? { messageId: referenceId } : {}), + })); } /** Extract the context_token from an inbound WeixinMsgContext. */ diff --git a/src/messaging/process-message.ts b/src/messaging/process-message.ts index e7ddf9a..aa3355f 100644 --- a/src/messaging/process-message.ts +++ b/src/messaging/process-message.ts @@ -36,7 +36,7 @@ import { sendWeixinMediaFile } from "./send-media.js"; import { StreamingMarkdownFilter } from "./markdown-filter.js"; import { sendMessageWeixin } from "./send.js"; import { WeixinReplyProgressSender } from "./reply-progress-sender.js"; -import { getQuoteStore } from "./quote-store.js"; +import { getActiveQuoteMediaSubdir, getQuoteStore } from "./quote-store.js"; import { handleSlashCommand } from "./slash-commands.js"; const MEDIA_OUTBOUND_TEMP_DIR = path.join(resolvePreferredOpenClawTmpDir(), "weixin/media/outbound-temp"); @@ -151,6 +151,7 @@ export async function processOneMessage( const downloaded = await downloadMediaFromItem(mediaItem, { cdnBaseUrl: deps.cdnBaseUrl, saveMedia: deps.channelRuntime.media.saveMediaBuffer, + mediaSubdir: getActiveQuoteMediaSubdir(deps.accountId), log: deps.log, errLog: deps.errLog, label, diff --git a/src/messaging/quote-store.test.ts b/src/messaging/quote-store.test.ts index 9bec2ab..3369ce8 100644 --- a/src/messaging/quote-store.test.ts +++ b/src/messaging/quote-store.test.ts @@ -10,8 +10,10 @@ import { closeQuoteStore, deactivateQuoteStoreAccount, deleteQuoteCacheForAccount, + getActiveQuoteMediaSubdir, getQuoteStore, initializeQuoteStore, + resolveQuoteMediaSubdir, resolveQuoteCachePolicy, } from "./quote-store.js"; @@ -25,6 +27,7 @@ vi.mock("../util/logger.js", () => ({ })); let rootDir: string; +let mediaRoot: string; let stores: QuoteStore[]; function policy(overrides: Partial = {}): QuoteCachePolicy { @@ -40,14 +43,21 @@ function policy(overrides: Partial = {}): QuoteCachePolicy { } async function open(overrides: Partial = {}): Promise { - const store = await QuoteStore.open({ rootDir, policy: policy(overrides) }); + const store = await QuoteStore.open({ rootDir, mediaRoot, policy: policy(overrides) }); expect(store).not.toBeNull(); stores.push(store!); return store!; } +function managedMediaPath(accountId: string, fileName: string): string { + const accountDir = path.join(mediaRoot, path.basename(resolveQuoteMediaSubdir(accountId))); + fs.mkdirSync(accountDir, { recursive: true }); + return path.join(accountDir, fileName); +} + beforeEach(() => { rootDir = fs.mkdtempSync(path.join(os.tmpdir(), "weixin-quote-store-")); + mediaRoot = path.join(rootDir, "managed-media"); stores = []; }); @@ -107,10 +117,10 @@ describe("QuoteStore", () => { expect(store.find("account", "user", "2")).toBeNull(); }); - it("copies media into managed storage, deduplicates it, and deletes it with the account", async () => { - const source = path.join(rootDir, "source.png"); - fs.writeFileSync(source, "same-content"); + it("registers one managed media copy and deletes it with the account", async () => { const store = await open(); + const source = managedMediaPath("account", "source.png"); + fs.writeFileSync(source, "same-content"); for (const messageId of ["1", "2"]) { await store.put({ accountId: "account", @@ -126,7 +136,7 @@ describe("QuoteStore", () => { const first = store.find("account", "user", "1"); const second = store.find("account", "user", "2"); expect(first?.mediaPath).toBe(second?.mediaPath); - expect(first?.mediaPath).not.toBe(source); + expect(first?.mediaPath).toBe(source); expect(fs.readFileSync(first!.mediaPath!, "utf8")).toBe("same-content"); const managedPath = first!.mediaPath!; @@ -136,10 +146,10 @@ describe("QuoteStore", () => { }); it("does not let sanitized account names share a media directory", async () => { - const source = path.join(rootDir, "source.png"); - fs.writeFileSync(source, "same-content"); const store = await open(); for (const accountId of ["a/b", "a_b"]) { + const source = managedMediaPath(accountId, "source.png"); + fs.writeFileSync(source, "same-content"); await store.put({ accountId, conversationId: "user", @@ -159,9 +169,9 @@ describe("QuoteStore", () => { }); it("keeps message metadata but skips oversized media", async () => { - const source = path.join(rootDir, "large.bin"); - fs.writeFileSync(source, "12345"); const store = await open({ maxSingleMediaBytes: 4 }); + const source = managedMediaPath("account", "large.bin"); + fs.writeFileSync(source, "12345"); await store.put({ accountId: "account", conversationId: "user", @@ -182,11 +192,11 @@ describe("QuoteStore", () => { }); it("enforces the per-account media byte budget oldest-first", async () => { - const firstSource = path.join(rootDir, "first.bin"); - const secondSource = path.join(rootDir, "second.bin"); + const store = await open({ maxMediaBytesPerAccount: 4 }); + const firstSource = managedMediaPath("account", "first.bin"); + const secondSource = managedMediaPath("account", "second.bin"); fs.writeFileSync(firstSource, "1111"); fs.writeFileSync(secondSource, "2222"); - const store = await open({ maxMediaBytesPerAccount: 4 }); const now = Date.now(); await store.put({ accountId: "account", conversationId: "user", messageId: "1", @@ -201,10 +211,10 @@ describe("QuoteStore", () => { }); it("expires managed media without discarding the message body", async () => { - const source = path.join(rootDir, "old.mp3"); + const store = await open({ mediaRetentionMs: 10 }); + const source = managedMediaPath("account", "old.mp3"); fs.writeFileSync(source, "voice"); const now = Date.now(); - const store = await open({ mediaRetentionMs: 10 }); await store.put({ accountId: "account", conversationId: "user", @@ -226,10 +236,10 @@ describe("QuoteStore", () => { }); it("removes an expired message but preserves media shared by a newer record", async () => { - const source = path.join(rootDir, "shared.png"); + const store = await open({ retentionMs: 100 }); + const source = managedMediaPath("account", "shared.png"); fs.writeFileSync(source, "shared"); const now = Date.now(); - const store = await open({ retentionMs: 100 }); await store.put({ accountId: "account", conversationId: "user", messageId: "old", direction: "inbound", body: "old", sourceMediaPath: source, createdAt: now - 1000, @@ -256,7 +266,7 @@ describe("QuoteStore", () => { }); expect(store.find("account", "user", "missing")?.mediaPath).toBeUndefined(); - const orphanDir = path.join(rootDir, "ref-media", "nested"); + const orphanDir = path.join(mediaRoot, "nested"); fs.mkdirSync(orphanDir, { recursive: true }); const orphan = path.join(orphanDir, "orphan.bin"); fs.writeFileSync(orphan, "orphan"); @@ -292,10 +302,10 @@ describe("QuoteStore", () => { })).resolves.toBeNull(); }); - it("ignores invalid writes, normalizes unsafe paths, and substitutes invalid timestamps", async () => { - const source = path.join(rootDir, "media.extension-that-is-far-too-long"); - fs.writeFileSync(source, "content"); + it("ignores invalid writes, isolates unsafe account names, and substitutes invalid timestamps", async () => { const store = await open(); + const source = managedMediaPath(" ", "media"); + fs.writeFileSync(source, "content"); await store.put({ accountId: " ", conversationId: "user", messageId: "media", direction: "inbound", body: "[文件]", sourceMediaPath: source, createdAt: Number.NaN, @@ -325,15 +335,89 @@ describe("QuoteStore", () => { store.deleteAccount("account"); store.runGc(); }); + + it("refuses to claim media outside its account-owned directory", async () => { + const source = path.join(rootDir, "outside.bin"); + fs.writeFileSync(source, "content"); + const store = await open(); + await store.put({ + accountId: "account", conversationId: "user", messageId: "outside", + direction: "inbound", body: "[文件]", sourceMediaPath: source, createdAt: Date.now(), + }); + expect(store.find("account", "user", "outside")?.mediaPath).toBeUndefined(); + expect(fs.existsSync(source)).toBe(true); + }); + + it("moves legacy media into the OpenClaw-managed cache and rewrites its database path", async () => { + const legacyRoot = path.join(rootDir, "ref-media"); + const legacyStore = await QuoteStore.open({ + rootDir, + mediaRoot: legacyRoot, + policy: policy(), + }); + expect(legacyStore).not.toBeNull(); + stores.push(legacyStore!); + const legacyPath = path.join( + legacyRoot, + path.basename(resolveQuoteMediaSubdir("account")), + "legacy.pdf", + ); + fs.mkdirSync(path.dirname(legacyPath), { recursive: true }); + fs.writeFileSync(legacyPath, "pdf"); + await legacyStore!.put({ + accountId: "account", conversationId: "user", messageId: "legacy", + direction: "inbound", body: "[文件]", sourceMediaPath: legacyPath, + mediaName: "report.pdf", createdAt: Date.now(), + }); + await legacyStore!.put({ + accountId: "account", conversationId: "user", messageId: "legacy-shared", + direction: "inbound", body: "[文件]", sourceMediaPath: legacyPath, + mediaName: "report.pdf", createdAt: Date.now(), + }); + + const existingLegacyPath = path.join(path.dirname(legacyPath), "existing.pdf"); + fs.writeFileSync(existingLegacyPath, "existing"); + await legacyStore!.put({ + accountId: "account", conversationId: "user", messageId: "existing", + direction: "inbound", body: "[文件]", sourceMediaPath: existingLegacyPath, + createdAt: Date.now(), + }); + const existingManagedPath = managedMediaPath("account", "existing.pdf"); + fs.writeFileSync(existingManagedPath, "existing"); + + const missingLegacyPath = path.join(path.dirname(legacyPath), "missing.pdf"); + fs.writeFileSync(missingLegacyPath, "missing"); + await legacyStore!.put({ + accountId: "account", conversationId: "user", messageId: "missing-legacy", + direction: "inbound", body: "[文件]", sourceMediaPath: missingLegacyPath, + createdAt: Date.now(), + }); + fs.unlinkSync(missingLegacyPath); + legacyStore!.close(); + + const migratedStore = await open(); + const migrated = migratedStore.find("account", "user", "legacy"); + expect(migrated?.mediaPath).toBe(managedMediaPath("account", "legacy.pdf")); + expect(migrated?.mediaName).toBe("report.pdf"); + expect(fs.existsSync(migrated!.mediaPath!)).toBe(true); + expect(fs.existsSync(legacyPath)).toBe(false); + expect(migratedStore.find("account", "user", "legacy-shared")?.mediaPath) + .toBe(migrated?.mediaPath); + expect(migratedStore.find("account", "user", "existing")?.mediaPath) + .toBe(existingManagedPath); + expect(fs.existsSync(existingLegacyPath)).toBe(false); + }); }); describe("global quote store lifecycle", () => { it("shares one store across active accounts and closes after the last account stops", async () => { process.env.OPENCLAW_STATE_DIR = rootDir; + expect(getActiveQuoteMediaSubdir("account-a")).toBeUndefined(); const first = await initializeQuoteStore({}, "account-a"); const second = await initializeQuoteStore({}, "account-b"); expect(first).toBe(second); expect(getQuoteStore()).toBe(first); + expect(getActiveQuoteMediaSubdir("account-a")).toBe(resolveQuoteMediaSubdir("account-a")); deactivateQuoteStoreAccount("account-a"); expect(getQuoteStore()).toBe(first); diff --git a/src/messaging/quote-store.ts b/src/messaging/quote-store.ts index 1bca775..41b51cb 100644 --- a/src/messaging/quote-store.ts +++ b/src/messaging/quote-store.ts @@ -15,6 +15,7 @@ const DEFAULT_MAX_MEDIA_BYTES_PER_ACCOUNT = 256 * 1024 * 1024; const DEFAULT_MAX_SINGLE_MEDIA_BYTES = 25 * 1024 * 1024; const GC_WRITE_INTERVAL = 100; const GC_TIMER_MS = 60 * 60 * 1000; +const QUOTE_MEDIA_SUBDIR = "inbound/openclaw-weixin-quotes"; type SqliteStatement = { run: (...params: unknown[]) => unknown; @@ -135,9 +136,13 @@ function accountMediaDirName(accountId: string): string { return `${readable}-${digest}`; } -function safeMediaExtension(filePath: string): string { - const ext = path.extname(filePath).toLowerCase(); - return /^\.[a-z0-9]{1,16}$/.test(ext) ? ext : ""; +/** OpenClaw media-store subdirectory owned and garbage-collected by this plugin. */ +export function resolveQuoteMediaSubdir(accountId: string): string { + return path.posix.join(QUOTE_MEDIA_SUBDIR, accountMediaDirName(accountId)); +} + +function resolveQuoteMediaRoot(): string { + return path.join(resolveStateDir(), "media", ...QUOTE_MEDIA_SUBDIR.split("/")); } function asStoredRow(value: unknown): StoredRow | null { @@ -157,12 +162,18 @@ export class QuoteStore { private closed = false; private readonly gcTimer: ReturnType; - private constructor(db: SqliteDatabase, rootDir: string, policy: QuoteCachePolicy) { + private constructor( + db: SqliteDatabase, + rootDir: string, + mediaRoot: string, + policy: QuoteCachePolicy, + ) { this.db = db; this.rootDir = rootDir; - this.mediaRoot = path.join(rootDir, "ref-media"); + this.mediaRoot = mediaRoot; this.policy = policy; this.initializeSchema(); + this.migrateLegacyMedia(); this.gcTimer = setInterval(() => this.requestGc(), GC_TIMER_MS); this.gcTimer.unref(); this.runGc(); @@ -171,6 +182,7 @@ export class QuoteStore { static async open(params: { policy: QuoteCachePolicy; rootDir?: string; + mediaRoot?: string; }): Promise { if (!params.policy.enabled) return null; try { @@ -178,7 +190,9 @@ export class QuoteStore { DatabaseSync: DatabaseSyncConstructor; }; const rootDir = params.rootDir ?? path.join(resolveStateDir(), "openclaw-weixin"); + const mediaRoot = params.mediaRoot ?? resolveQuoteMediaRoot(); fs.mkdirSync(rootDir, { recursive: true, mode: 0o700 }); + fs.mkdirSync(mediaRoot, { recursive: true, mode: 0o700 }); try { fs.chmodSync(rootDir, 0o700); } catch { @@ -191,7 +205,7 @@ export class QuoteStore { } catch { // best-effort } - return new QuoteStore(db, rootDir, params.policy); + return new QuoteStore(db, rootDir, mediaRoot, params.policy); } catch (err) { logger.warn( `quote cache disabled: node:sqlite is unavailable or the database could not be opened: ${String(err)}`, @@ -230,6 +244,64 @@ export class QuoteStore { `); } + /** Move media written by pre-managed-root builds without duplicating file contents. */ + private migrateLegacyMedia(): void { + const legacyRoots = [ + path.join(this.rootDir, "ref-media"), + // A short-lived local build accidentally omitted the media/ segment. + path.join(path.dirname(this.rootDir), ...QUOTE_MEDIA_SUBDIR.split("/")), + ]; + for (const legacyRoot of new Set(legacyRoots)) { + this.migrateLegacyMediaRoot(legacyRoot); + } + } + + private migrateLegacyMediaRoot(legacyRoot: string): void { + if (path.resolve(legacyRoot) === path.resolve(this.mediaRoot) || !fs.existsSync(legacyRoot)) { + return; + } + const rows = this.db + .prepare(` + SELECT account_id, media_path + FROM quote_messages + WHERE media_path IS NOT NULL + `) + .all() as Array<{ account_id: string; media_path: string }>; + const canonicalLegacyRoot = fs.realpathSync(legacyRoot); + const migrated = new Map(); + for (const row of rows) { + if (migrated.has(row.media_path)) continue; + if (!fs.existsSync(row.media_path)) continue; + const canonicalLegacyPath = fs.realpathSync(row.media_path); + const relative = path.relative(canonicalLegacyRoot, canonicalLegacyPath); + if (!relative || relative.startsWith("..") || path.isAbsolute(relative)) continue; + const destinationDir = path.join(this.mediaRoot, accountMediaDirName(row.account_id)); + const destination = path.join(destinationDir, path.basename(row.media_path)); + try { + fs.mkdirSync(destinationDir, { recursive: true, mode: 0o700 }); + if (!fs.existsSync(destination)) { + try { + fs.renameSync(row.media_path, destination); + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== "EXDEV") throw err; + fs.copyFileSync(row.media_path, destination, fs.constants.COPYFILE_EXCL); + fs.unlinkSync(row.media_path); + } + } else { + fs.unlinkSync(row.media_path); + } + migrated.set(row.media_path, destination); + } catch (err) { + logger.warn(`quote cache: failed to migrate media path=${row.media_path}: ${String(err)}`); + } + } + for (const [legacyPath, managedPath] of migrated) { + this.db + .prepare("UPDATE quote_messages SET media_path = ? WHERE media_path = ?") + .run(managedPath, legacyPath); + } + } + find(accountId: string, conversationId: string, messageId: string): QuoteMessageRecord | null { if (this.closed || !messageId) return null; const row = asStoredRow( @@ -264,7 +336,7 @@ export class QuoteStore { async put(input: QuoteMessageInput): Promise { if (this.closed || !input.messageId || (!input.body && !input.sourceMediaPath)) return; const media = input.sourceMediaPath - ? await this.cacheMedia(input.accountId, input.sourceMediaPath, input.mediaMime) + ? await this.registerManagedMedia(input.accountId, input.sourceMediaPath, input.mediaMime) : null; this.db .prepare(` @@ -334,32 +406,30 @@ export class QuoteStore { return typeof row?.media_path === "string" ? row.media_path : null; } - private async cacheMedia( + private async registerManagedMedia( accountId: string, sourcePath: string, mime?: string, ): Promise<{ path: string; mime?: string; name: string; size: number } | null> { try { - const stat = await fs.promises.stat(sourcePath); - if (!stat.isFile() || stat.size > this.policy.maxSingleMediaBytes) return null; - const data = await fs.promises.readFile(sourcePath); - const hash = crypto.createHash("sha256").update(data).digest("hex"); const accountDir = path.join(this.mediaRoot, accountMediaDirName(accountId)); - await fs.promises.mkdir(accountDir, { recursive: true, mode: 0o700 }); - const destination = path.join(accountDir, `${hash}${safeMediaExtension(sourcePath)}`); - try { - await fs.promises.writeFile(destination, data, { flag: "wx", mode: 0o600 }); - } catch (err) { - if ((err as NodeJS.ErrnoException).code !== "EEXIST") throw err; + const canonicalRoot = await fs.promises.realpath(accountDir).catch(() => path.resolve(accountDir)); + const canonicalSource = await fs.promises.realpath(sourcePath); + const relative = path.relative(canonicalRoot, canonicalSource); + if (!relative || relative.startsWith("..") || path.isAbsolute(relative)) { + logger.warn(`quote cache: refusing unmanaged media path=${sourcePath}`); + return null; } + const stat = await fs.promises.stat(canonicalSource); + if (!stat.isFile() || stat.size > this.policy.maxSingleMediaBytes) return null; return { - path: destination, + path: path.resolve(sourcePath), ...(mime ? { mime } : {}), - name: path.basename(sourcePath), + name: path.basename(canonicalSource), size: stat.size, }; } catch (err) { - logger.warn(`quote cache: failed to cache media path=${sourcePath}: ${String(err)}`); + logger.warn(`quote cache: failed to register media path=${sourcePath}: ${String(err)}`); return null; } } @@ -533,6 +603,10 @@ export function getQuoteStore(): QuoteStore | null { return activeStore; } +export function getActiveQuoteMediaSubdir(accountId: string): string | undefined { + return activeStore ? resolveQuoteMediaSubdir(accountId) : undefined; +} + export function closeQuoteStore(): void { activeStore?.close(); activeStore = null; @@ -547,6 +621,7 @@ export function deleteQuoteCacheForAccount(accountId: string): void { } const rootDir = path.join(resolveStateDir(), "openclaw-weixin"); + const mediaRoot = resolveQuoteMediaRoot(); const dbPath = path.join(rootDir, "ref-messages.sqlite"); if (!fs.existsSync(dbPath)) return; try { @@ -555,6 +630,10 @@ export function deleteQuoteCacheForAccount(accountId: string): void { const db = new sqlite.DatabaseSync(dbPath); db.prepare("DELETE FROM quote_messages WHERE account_id = ?").run(accountId); db.close(); + fs.rmSync(path.join(mediaRoot, accountMediaDirName(accountId)), { + recursive: true, + force: true, + }); fs.rmSync(path.join(rootDir, "ref-media", accountMediaDirName(accountId)), { recursive: true, force: true, From e52535aec08fc3cb6da92fdce8b83c4fbb6d71d6 Mon Sep 17 00:00:00 2001 From: scotthuang Date: Wed, 2 Sep 2026 15:33:20 +0800 Subject: [PATCH 3/5] fix: expose quoted attachment tool access --- src/messaging/inbound.test.ts | 37 +++++++++++++++++++++++++++++++++++ src/messaging/inbound.ts | 17 ++++++++++++++++ 2 files changed, 54 insertions(+) diff --git a/src/messaging/inbound.test.ts b/src/messaging/inbound.test.ts index b4bab47..143879c 100644 --- a/src/messaging/inbound.test.ts +++ b/src/messaging/inbound.test.ts @@ -430,6 +430,7 @@ describe("stored quote resolution", () => { try { const msg = quotedMessage(); const ctx = weixinMessageToMsgContext(msg, "acc", { decryptedFilePath: "/tmp/current.pdf" }); + ctx.ChannelPromptContext = ["existing channel context"]; resolveStoredQuoteContext(ctx, msg, "acc", { find: () => ({ accountId: "acc", conversationId: "user1", messageId: "9007199254740993123", @@ -449,6 +450,21 @@ describe("stored quote resolution", () => { messageId: "9007199254740993123", }, ]); + expect(ctx.ChannelPromptContext).toEqual([ + "existing channel context", + [ + "Quoted attachment tool access:", + JSON.stringify({ + message_id: "9007199254740993123", + original_filename: "quoted-original.png", + managed_source_path: quotedPath, + workspace_directory: "media/inbound/", + }), + "The attachment is staged into the agent workspace under media/inbound/. " + + "If automatic extraction fails and the user asks about its contents, use the available " + + "file/PDF tools to locate it by original_filename and read it.", + ].join("\n"), + ]); } finally { fs.rmSync(tempDir, { recursive: true, force: true }); } @@ -466,6 +482,27 @@ describe("stored quote resolution", () => { }); expect(ctx.ReplyToBody).toBe("[引用的视频已过期]"); }); + + it("adds a tool-access hint and falls back to the managed basename", () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "weixin-inbound-quote-hint-")); + const quotedPath = path.join(tempDir, "managed-document.pdf"); + fs.writeFileSync(quotedPath, "pdf"); + try { + const msg = quotedMessage(); + const ctx = weixinMessageToMsgContext(msg, "acc"); + resolveStoredQuoteContext(ctx, msg, "acc", { + find: () => ({ + accountId: "acc", conversationId: "user1", messageId: "9007199254740993123", + direction: "inbound", body: "[文件]", mediaPath: quotedPath, + mediaMime: "application/pdf", createdAt: Date.now(), + }), + }); + expect(ctx.ChannelPromptContext?.[0]).toContain('"original_filename":"managed-document.pdf"'); + expect(ctx.ChannelPromptContext?.[0]).toContain('"workspace_directory":"media/inbound/"'); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); + } + }); }); describe("getWeixinMessageId", () => { diff --git a/src/messaging/inbound.ts b/src/messaging/inbound.ts index cb75703..f7dc743 100644 --- a/src/messaging/inbound.ts +++ b/src/messaging/inbound.ts @@ -166,6 +166,7 @@ export type WeixinMsgContext = { fileName?: string; messageId?: string; }>; + ChannelPromptContext?: string[]; ReplyToId?: string; ReplyToBody?: string; ReplyToQuoteText?: string; @@ -386,6 +387,22 @@ export function resolveStoredQuoteContext( : {}), ...(index === ctx.MediaPaths!.length - 1 ? { messageId: referenceId } : {}), })); + const quotedAttachment = { + message_id: referenceId, + original_filename: record.mediaName ?? path.basename(record.mediaPath), + managed_source_path: record.mediaPath, + workspace_directory: "media/inbound/", + }; + ctx.ChannelPromptContext = [ + ...(ctx.ChannelPromptContext ?? []), + [ + "Quoted attachment tool access:", + JSON.stringify(quotedAttachment), + "The attachment is staged into the agent workspace under media/inbound/. " + + "If automatic extraction fails and the user asks about its contents, use the available " + + "file/PDF tools to locate it by original_filename and read it.", + ].join("\n"), + ]; } /** Extract the context_token from an inbound WeixinMsgContext. */ From 102d548c25dc379ae989988d3858a152bd767230 Mon Sep 17 00:00:00 2001 From: scotthuang Date: Wed, 2 Sep 2026 16:01:55 +0800 Subject: [PATCH 4/5] docs: clarify quoted media storage and tool access --- CHANGELOG.md | 2 +- CHANGELOG.zh_CN.md | 2 +- README.md | 17 ++++++++++------- README.zh_CN.md | 10 ++++++---- 4 files changed, 18 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 718c6f1..e0d74ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ This project follows the [Keep a Changelog](https://keepachangelog.com/) format. ### Added -- **Quote reconstruction for newer WeChat clients:** Losslessly parse message IDs and resolve ID-only `svr_id` text and partial quotes through a per-account, per-conversation SQLite side store. Images, video, voice, and attachments are copied into managed storage with time, count, byte-budget, and single-file eviction limits. +- **Quote reconstruction for newer WeChat clients:** Losslessly parse message IDs and resolve ID-only `svr_id` text and partial quotes through a per-account, per-conversation SQLite side store. Images, video, voice, and attachments are written directly into plugin-owned OpenClaw managed storage, using one retained file with time, count, byte-budget, and single-file eviction limits. Restored attachments include tool-access hints so agents can locate and read the file when automatic extraction fails. - **Compatible degradation:** Disable quote caching when `node:sqlite` is unavailable or `quoteCache.enabled=false`, with no in-memory fallback. Cache failures never interrupt normal message delivery. ## [2.4.7] - 2026-08-31 diff --git a/CHANGELOG.zh_CN.md b/CHANGELOG.zh_CN.md index 668f5e1..6a65b6d 100644 --- a/CHANGELOG.zh_CN.md +++ b/CHANGELOG.zh_CN.md @@ -8,7 +8,7 @@ ### 新增 -- **新版微信引用消息还原：** 对消息 ID 做无损解析，并用按账号、会话隔离的 SQLite 旁路存储还原只携带 `svr_id` 的文本引用和局部引用。图片、视频、语音及附件会复制到受管目录，支持按时间、条数、空间和单文件大小淘汰。 +- **新版微信引用消息还原：** 对消息 ID 做无损解析，并用按账号、会话隔离的 SQLite 旁路存储还原只携带 `svr_id` 的文本引用和局部引用。图片、视频、语音及附件首次直接写入插件专属的 OpenClaw 受管目录，以单份文件支持按时间、条数、空间和单文件大小淘汰；还原引用附件时会提供工具访问提示，便于 agent 在自动抽取失败后定位并读取文件。 - **兼容降级：** `node:sqlite` 不可用或 `quoteCache.enabled=false` 时关闭引用缓存，不启用内存替代方案；缓存异常不会影响正常消息收发。 ## [2.4.7] - 2026-08-31 diff --git a/README.md b/README.md index f853a0b..3f12a90 100644 --- a/README.md +++ b/README.md @@ -114,13 +114,16 @@ added in a future version if needed. ## Local quote cache Newer WeChat clients may send only a server message ID for a quoted message. The -plugin therefore stores text and media metadata in SQLite and copies images, -video, voice, and attachments into managed storage. Records are isolated by -account and conversation. By default, text is retained for 30 days with a limit -of 10,000 messages per account; media is retained for 7 days with a 256 MiB -per-account budget and a 25 MiB single-file limit. Cleanup runs at startup, -hourly, every 100 writes, whenever the media budget is exceeded, and when an -account is deleted. +plugin therefore stores text and media metadata in SQLite and writes images, +video, voice, and attachments directly into plugin-owned OpenClaw managed +storage, so current delivery and later quotes reuse one file. When a quoted +attachment is restored, the agent also receives its original filename, managed +source path, and workspace `media/inbound/` hint, allowing file/PDF tools to read +it if automatic extraction fails. Records are isolated by account and +conversation. By default, text is retained for 30 days with a limit of 10,000 +messages per account; media is retained for 7 days with a 256 MiB per-account +budget and a 25 MiB single-file limit. Cleanup runs at startup, hourly, every 100 +writes, whenever the media budget is exceeded, and when an account is deleted. If `node:sqlite` is unavailable, the plugin logs a warning and disables this feature; it does not fall back to an in-memory cache. You can also disable it or diff --git a/README.zh_CN.md b/README.zh_CN.md index 6a60af5..4393973 100644 --- a/README.zh_CN.md +++ b/README.zh_CN.md @@ -110,10 +110,12 @@ openclaw config set session.dmScope per-account-channel-peer ## 引用消息本地缓存 新版微信的引用消息可能只携带服务端消息 ID。插件默认使用 SQLite 保存文本和媒体元数据， -并将图片、视频、语音及附件复制到受管目录，从而在后续引用时还原原消息。缓存按账号和会话 -隔离；文本默认保留 30 天且每账号最多 10,000 条，媒体默认保留 7 天、每账号最多 256 MiB、 -单文件最多 25 MiB。淘汰在启动时、每小时、每写入 100 条及媒体超出空间上限时触发；删除 -账号时会同步删除其引用缓存。 +并将图片、视频、语音及附件首次直接写入插件专属的 OpenClaw 受管目录，让当前消息和后续引用 +复用同一份文件。命中引用附件时，还会向 agent 提供原文件名、受管源路径及工作区 +`media/inbound/` 提示；自动抽取失败时，agent 可调用文件/PDF 工具读取。缓存按账号和会话隔离； +文本默认保留 30 天且每账号最多 10,000 条，媒体默认保留 7 天、每账号最多 256 MiB、单文件 +最多 25 MiB。淘汰在启动时、每小时、每写入 100 条及媒体超出空间上限时触发；删除账号时 +会同步删除其引用缓存。 如当前 Node.js 不提供 `node:sqlite`，插件会记录警告并自动关闭此功能，不使用内存缓存降级。 也可以显式关闭或调整限制： From 2645786ab5f866bc3ca96775f032dd6e9adc53d7 Mon Sep 17 00:00:00 2001 From: scotthuang Date: Wed, 2 Sep 2026 16:10:07 +0800 Subject: [PATCH 5/5] docs: explain quote cache configuration --- README.md | 53 +++++++++++++++++++++++++++++++++++++++++-------- README.zh_CN.md | 49 ++++++++++++++++++++++++++++++++++++++------- 2 files changed, 87 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 3f12a90..eb8e49d 100644 --- a/README.md +++ b/README.md @@ -125,20 +125,32 @@ messages per account; media is retained for 7 days with a 256 MiB per-account budget and a 25 MiB single-file limit. Cleanup runs at startup, hourly, every 100 writes, whenever the media budget is exceeded, and when an account is deleted. -If `node:sqlite` is unavailable, the plugin logs a warning and disables this -feature; it does not fall back to an in-memory cache. You can also disable it or -change the limits explicitly: +The quote cache is **enabled by default**. No configuration is required after +upgrading: this preserves the quoted-message experience that existed before +newer WeChat clients stopped including the quoted body. Configure it under +`channels.openclaw-weixin.quoteCache` only when you need different limits. + +| Option | Default | Description | +| --- | ---: | --- | +| `enabled` | `true` | Enables local quote reconstruction. | +| `retentionDays` | `30` | Number of days to retain text and message metadata. | +| `maxMessagesPerAccount` | `10000` | Maximum message records retained per account. | +| `mediaRetentionDays` | `7` | Number of days to retain quoted media files. | +| `maxMediaBytesPerAccount` | `268435456` | Maximum retained media per account, in bytes (256 MiB). | +| `maxSingleMediaBytes` | `26214400` | Maximum size of one retained media file, in bytes (25 MiB). Larger files are still delivered normally but are not retained for later quote reconstruction. | + +Example with custom limits while keeping the feature enabled: ```json { "channels": { "openclaw-weixin": { "quoteCache": { - "enabled": false, - "retentionDays": 30, - "maxMessagesPerAccount": 10000, - "mediaRetentionDays": 7, - "maxMediaBytesPerAccount": 268435456, + "enabled": true, + "retentionDays": 14, + "maxMessagesPerAccount": 5000, + "mediaRetentionDays": 3, + "maxMediaBytesPerAccount": 134217728, "maxSingleMediaBytes": 26214400 } } @@ -146,6 +158,31 @@ change the limits explicitly: } ``` +To disable local quote storage explicitly: + +```json +{ + "channels": { + "openclaw-weixin": { + "quoteCache": { + "enabled": false + } + } + } +} +``` + +Restart the gateway after changing the configuration: + +```bash +openclaw gateway restart +``` + +If `node:sqlite` is unavailable or the database cannot be opened, the plugin +logs a warning and disables this feature automatically. It does not fall back to +an in-memory cache, and quote-cache failures do not interrupt normal message +delivery. + ## Backend API Protocol This plugin communicates with the backend gateway via HTTP JSON API. Developers integrating with their own backend need to implement the following interfaces. diff --git a/README.zh_CN.md b/README.zh_CN.md index 4393973..043d260 100644 --- a/README.zh_CN.md +++ b/README.zh_CN.md @@ -117,19 +117,31 @@ openclaw config set session.dmScope per-account-channel-peer 最多 25 MiB。淘汰在启动时、每小时、每写入 100 条及媒体超出空间上限时触发；删除账号时 会同步删除其引用缓存。 -如当前 Node.js 不提供 `node:sqlite`，插件会记录警告并自动关闭此功能，不使用内存缓存降级。 -也可以显式关闭或调整限制： +引用缓存**默认开启**，升级后无需增加任何配置。这样可以在新版微信客户端不再携带引用正文后， +继续保持此前的引用消息体验。只有需要调整缓存限制时，才需要配置 +`channels.openclaw-weixin.quoteCache`。 + +| 配置项 | 默认值 | 说明 | +| --- | ---: | --- | +| `enabled` | `true` | 是否启用本地引用消息还原。 | +| `retentionDays` | `30` | 文本及消息元数据的保留天数。 | +| `maxMessagesPerAccount` | `10000` | 每个账号最多保留的消息记录数。 | +| `mediaRetentionDays` | `7` | 引用媒体文件的保留天数。 | +| `maxMediaBytesPerAccount` | `268435456` | 每个账号最多保留的媒体总大小，单位为字节（256 MiB）。 | +| `maxSingleMediaBytes` | `26214400` | 单个可保留媒体文件的最大大小，单位为字节（25 MiB）。超出后仍正常投递当前消息，但不会为后续引用保留文件。 | + +保持功能开启并自定义限制的示例： ```json { "channels": { "openclaw-weixin": { "quoteCache": { - "enabled": false, - "retentionDays": 30, - "maxMessagesPerAccount": 10000, - "mediaRetentionDays": 7, - "maxMediaBytesPerAccount": 268435456, + "enabled": true, + "retentionDays": 14, + "maxMessagesPerAccount": 5000, + "mediaRetentionDays": 3, + "maxMediaBytesPerAccount": 134217728, "maxSingleMediaBytes": 26214400 } } @@ -137,6 +149,29 @@ openclaw config set session.dmScope per-account-channel-peer } ``` +如需明确关闭本地引用存储，只需配置： + +```json +{ + "channels": { + "openclaw-weixin": { + "quoteCache": { + "enabled": false + } + } + } +} +``` + +修改配置后重启 gateway： + +```bash +openclaw gateway restart +``` + +如当前 Node.js 不提供 `node:sqlite`，或数据库无法打开，插件会记录警告并自动关闭此功能。 +插件不会使用内存缓存降级，引用缓存异常也不会中断正常消息收发。 + ## 后端 API 协议 本插件通过 HTTP JSON API 与后端网关通信。二次开发者若需对接自有后端，需实现以下接口。