From ad65d93b2ed3036a390985c3131558e8699f1741 Mon Sep 17 00:00:00 2001 From: Hong Jiarong Date: Wed, 8 Jul 2026 16:40:39 +0800 Subject: [PATCH] feat: add message burst debouncing (MessageBatcher) - Create MessageBatcher: debounce 600ms, adaptive delay for long chunks, max 8 messages/4000 chars before immediate flush, per (chatId, sender) key - Integrate into trigger: text messages enqueued, slash commands and file/image bypass batching, locked projects respond immediately - Add unit tests (8) for batcher lifecycle and edge cases --- hub/src/feishu/messageBatcher.ts | 125 ++++++++++ hub/src/feishu/trigger.ts | 327 +++++++++++++++---------- hub/test/integration/trigger.test.ts | 76 ++++-- hub/test/unit/feishu-reactions.test.ts | 1 + hub/test/unit/message-batcher.test.ts | 114 +++++++++ 5 files changed, 492 insertions(+), 151 deletions(-) create mode 100644 hub/src/feishu/messageBatcher.ts create mode 100644 hub/test/unit/message-batcher.test.ts diff --git a/hub/src/feishu/messageBatcher.ts b/hub/src/feishu/messageBatcher.ts new file mode 100644 index 0000000..59afb37 --- /dev/null +++ b/hub/src/feishu/messageBatcher.ts @@ -0,0 +1,125 @@ +export interface MessageBatcherOptions { + /** Default: 600. */ + readonly debounceMs?: number; + /** Default: 8. */ + readonly maxMessages?: number; + /** Default: 4000. */ + readonly maxChars?: number; + /** Default: 3500. */ + readonly splitThreshold?: number; + /** Default: 1200. */ + readonly extendedDebounceMs?: number; +} + +export type MessageBatcherCallback = (mergedText: string, key?: string) => Promise; + +interface PendingBatch { + readonly texts: string[]; + charCount: number; + timer: ReturnType | null; +} + +const DEFAULT_OPTIONS = { + debounceMs: 600, + maxMessages: 8, + maxChars: 4000, + splitThreshold: 3500, + extendedDebounceMs: 1200, +} as const; + +export class MessageBatcher { + private readonly debounceMs: number; + private readonly maxMessages: number; + private readonly maxChars: number; + private readonly splitThreshold: number; + private readonly extendedDebounceMs: number; + private readonly pending = new Map(); + private readonly chains = new Map>(); + + constructor( + private readonly onFlush: MessageBatcherCallback, + options: MessageBatcherOptions = {}, + ) { + this.debounceMs = options.debounceMs ?? DEFAULT_OPTIONS.debounceMs; + this.maxMessages = options.maxMessages ?? DEFAULT_OPTIONS.maxMessages; + this.maxChars = options.maxChars ?? DEFAULT_OPTIONS.maxChars; + this.splitThreshold = options.splitThreshold ?? DEFAULT_OPTIONS.splitThreshold; + this.extendedDebounceMs = options.extendedDebounceMs ?? DEFAULT_OPTIONS.extendedDebounceMs; + } + + async enqueue(chatId: string, senderOpenId: string, text: string): Promise { + const key = messageBatchKey(chatId, senderOpenId); + await this.runSerial(key, async () => { + const existing = this.pending.get(key); + const batch = + existing ?? + { + texts: [], + charCount: 0, + timer: null, + }; + + if (existing === undefined) { + this.pending.set(key, batch); + } + + batch.texts.push(text); + batch.charCount += text.length + (batch.texts.length === 1 ? 0 : 1); + + if (batch.texts.length >= this.maxMessages || batch.charCount >= this.maxChars) { + await this.flushPendingBatch(key, batch); + return; + } + + this.scheduleFlush(key, batch, text.length >= this.splitThreshold ? this.extendedDebounceMs : this.debounceMs); + }); + } + + async flushNow(key: string): Promise { + await this.runSerial(key, async () => { + const batch = this.pending.get(key); + if (batch === undefined) return; + await this.flushPendingBatch(key, batch); + }); + } + + async flushAll(): Promise { + const keys = [...this.pending.keys()]; + await Promise.all(keys.map((key) => this.flushNow(key))); + } + + private scheduleFlush(key: string, batch: PendingBatch, delayMs: number): void { + if (batch.timer !== null) { + clearTimeout(batch.timer); + } + batch.timer = setTimeout(() => { + void this.flushNow(key).catch(() => undefined); + }, delayMs); + (batch.timer as { unref?: () => void }).unref?.(); + } + + private async flushPendingBatch(key: string, batch: PendingBatch): Promise { + if (batch.timer !== null) { + clearTimeout(batch.timer); + batch.timer = null; + } + if (this.pending.get(key) !== batch) return; + this.pending.delete(key); + await this.onFlush(batch.texts.join("\n"), key); + } + + private runSerial(key: string, operation: () => Promise): Promise { + const previous = this.chains.get(key) ?? Promise.resolve(); + const next = previous.catch(() => undefined).then(operation); + this.chains.set(key, next); + return next.finally(() => { + if (this.chains.get(key) === next) { + this.chains.delete(key); + } + }); + } +} + +export function messageBatchKey(chatId: string, senderOpenId: string): string { + return `${chatId}:${senderOpenId}`; +} diff --git a/hub/src/feishu/trigger.ts b/hub/src/feishu/trigger.ts index eff9aa7..7c96cbd 100644 --- a/hub/src/feishu/trigger.ts +++ b/hub/src/feishu/trigger.ts @@ -20,6 +20,7 @@ import { writeAudit } from "../audit.js"; import { PatchableTextStream } from "./textStream.js"; import { createFileDeliveryMcpServer } from "./fileDeliveryTool.js"; import { readFeishuContext } from "./read.js"; +import { MessageBatcher, messageBatchKey, type MessageBatcherOptions } from "./messageBatcher.js"; interface TriggerDeps { readonly prisma: PrismaClient; @@ -27,6 +28,21 @@ interface TriggerDeps { readonly logger: FastifyBaseLogger; readonly runAgent?: (req: RunRequest) => Promise; readonly authorizer?: PermissionAuthorizer | undefined; + readonly messageBatcherOptions?: MessageBatcherOptions | undefined; +} + +interface TriggerActor { + readonly feishuOpenId: string; + readonly chatId: string; +} + +interface TriggerRunContext { + readonly msg: MessageReceiveEvent["message"]; + readonly rt: FeishuRuntime; + readonly chatId: string; + readonly projectId: string; + readonly senderOpenId: string; + readonly actor: TriggerActor; } /** @@ -36,145 +52,29 @@ interface TriggerDeps { */ export function makeTriggerHandler(deps: TriggerDeps) { const authorizer = deps.authorizer ?? createPermissionAuthorizer(deps.prisma); - return async (event: MessageReceiveEvent, rt: FeishuRuntime): Promise => { - const runAgent = deps.runAgent ?? defaultRunAgent; - const msg = event.message; - const sender = event.sender; - void sender; // sender→user mapping is OPEN (principal sub-typology) - - // Dedup: the lark ws client redelivers at-least-once. A duplicate event - // would spawn a second AgentRun — the lock would refuse it, but a FAILED - // run row + a busy reply would leak. Record the event_id up front; if it - // already exists, this is a redelivery → stop. - const eventId = feishuEventId(event); - if (eventId !== undefined && eventId !== "") { - const existing = await deps.prisma.feishuEventReceipt.findUnique({ - where: { eventId }, - select: { id: true }, - }); - if (existing !== null) { - deps.logger.debug({ eventId }, "feishu trigger: duplicate event, skipping"); - return; - } - await deps.prisma.feishuEventReceipt.create({ - data: { - eventId, - eventType: feishuEventType(event), - messageId: msg.message_id, - }, - }); - } - - // Only react to text, file, or image messages in group chats. - if (msg.message_type !== "text" && msg.message_type !== "file" && msg.message_type !== "image") return; - const chatId = msg.chat_id; - if (chatId === "") return; - - // ADR-0001: resolve chat → project. Unknown chat ⇒ not a project group. - const binding = await deps.prisma.projectGroupBinding.findUnique({ - where: { chatId }, - select: { projectId: true }, - }); - if (binding === null) { - deps.logger.debug({ chatId }, "feishu trigger: chat not bound to any project"); + const runAgent = deps.runAgent ?? defaultRunAgent; + const batchContexts = new Map(); + const messageBatcher = new MessageBatcher(async (mergedText, key) => { + if (key === undefined) { + deps.logger.warn("feishu trigger: message batch flushed without a key"); return; } - const projectId = binding.projectId; - // ADR-0019: triggerAgent is authorized through actor → principal-set - // resolution, then grant/settings composition. - // A missing open_id is treated as an untrusted event: deny explicitly - // rather than skipping the check (fail closed). - const senderOpenId = event.sender.sender_id.open_id ?? ""; - if (senderOpenId === "") { - deps.logger.warn({ projectId, chatId }, "feishu trigger: sender has no open_id, denying"); - await writeAudit(deps.prisma, { projectId, action: "trigger.denied", metadata: { reason: "missing open_id" } }); - await sendText(rt, chatId, "无法识别发送者,拒绝触发。"); + const context = batchContexts.get(key); + if (context === undefined) { + deps.logger.warn({ key }, "feishu trigger: message batch flushed without trigger context"); return; } - const actor = { feishuOpenId: senderOpenId, chatId }; - const triggerDecision = await authorizer.can({ - actor, - action: "agent.trigger", - resource: { type: "PROJECT", id: projectId }, - }); - if (!triggerDecision.allowed) { - deps.logger.info({ projectId, senderOpenId, reason: triggerDecision.reason }, "feishu trigger: permission denied"); - await writeAudit(deps.prisma, { - projectId, - ...(triggerDecision.actorUserId !== undefined ? { actorUserId: triggerDecision.actorUserId } : {}), - action: "trigger.denied", - metadata: { decision: auditDecision(triggerDecision) }, - }); - await sendText(rt, chatId, "无权限触发。"); - return; + batchContexts.delete(key); + try { + await startAgentRun(context, mergedText); + } catch (e) { + deps.logger.error({ key, err: e instanceof Error ? e.message : String(e) }, "feishu trigger: batched run setup failed"); + throw e; } + }, deps.messageBatcherOptions); - let cleanPrompt: string | null = null; - let downloadedFiles: string[] = []; - if (msg.message_type === "text") { - cleanPrompt = extractPrompt(msg); - } else if (msg.message_type === "file" || msg.message_type === "image") { - // Download the file/image to the workspace, build a prompt around it. - const project = await deps.prisma.project.findUnique({ - where: { id: projectId }, - select: { workspaceDir: true }, - }); - if (project === null) return; - try { - const content = FileMessageContentSchema.parse(JSON.parse(msg.content)); - const key = content.file_key ?? content.image_key ?? ""; - if (key !== "") { - const fileName = `${msg.message_type}-${Date.now()}.${msg.message_type === "image" ? "png" : "bin"}`; - const savePath = `${project.workspaceDir}/.cph/inbox/${fileName}`; - await downloadMessageFile(rt, msg.message_id, key, savePath); - downloadedFiles.push(savePath); - cleanPrompt = `用户发来一个文件: ${savePath}`; - } - } catch { return; } - } - if (cleanPrompt === null) return; - // Slash commands: session management, not agent runs. These bypass the - // lock — they don't create a run, so they can't conflict with one. - if (cleanPrompt.startsWith("/")) { - const cmd = cleanPrompt.split(/\s+/)[0]; - switch (cmd) { - case "/new": { - await deps.prisma.agentSession.updateMany({ - where: { projectId, archivedAt: null }, - data: { archivedAt: new Date() }, - }); - await sendText(rt, chatId, "已开新会话,下次 @bot 将从头开始。"); - return; - } - case "/resume": { - const latest = await deps.prisma.agentSession.findFirst({ - where: { projectId, archivedAt: { not: null } }, - orderBy: { archivedAt: "desc" }, - select: { id: true }, - }); - if (latest === null) { - await sendText(rt, chatId, "没有可恢复的会话。"); - return; - } - await deps.prisma.agentSession.update({ - where: { id: latest.id }, - data: { archivedAt: null }, - }); - await sendText(rt, chatId, "已恢复上一个会话。"); - return; - } - case "/reset": { - await deps.prisma.agentSession.updateMany({ - where: { projectId, archivedAt: null }, - data: { archivedAt: new Date() }, - }); - await sendText(rt, chatId, "已重置,下次 @bot 将从头开始。"); - return; - } - default: - break; - } - } + async function startAgentRun(context: TriggerRunContext, cleanPrompt: string): Promise { + const { msg, rt, chatId, projectId, senderOpenId, actor } = context; // ADR-0002: if locked by another run, reply busy. const existing = await currentLockRunId(deps.prisma, projectId); @@ -302,7 +202,6 @@ export function makeTriggerHandler(deps: TriggerDeps) { return; } - const projectCtx: ProjectContext = { projectId, boundChatId: chatId, // ADR-0003: Feishu reads stay in this chat @@ -408,6 +307,166 @@ export function makeTriggerHandler(deps: TriggerDeps) { deps.logger.warn({ runId: run.id, err: e instanceof Error ? e.message : String(e) }, "trigger: could not release lock"); } }); + } + + return async (event: MessageReceiveEvent, rt: FeishuRuntime): Promise => { + const msg = event.message; + const sender = event.sender; + void sender; // sender→user mapping is OPEN (principal sub-typology) + + // Dedup: the lark ws client redelivers at-least-once. A duplicate event + // would spawn a second AgentRun — the lock would refuse it, but a FAILED + // run row + a busy reply would leak. Record the event_id up front; if it + // already exists, this is a redelivery → stop. + const eventId = feishuEventId(event); + if (eventId !== undefined && eventId !== "") { + const existing = await deps.prisma.feishuEventReceipt.findUnique({ + where: { eventId }, + select: { id: true }, + }); + if (existing !== null) { + deps.logger.debug({ eventId }, "feishu trigger: duplicate event, skipping"); + return; + } + await deps.prisma.feishuEventReceipt.create({ + data: { + eventId, + eventType: feishuEventType(event), + messageId: msg.message_id, + }, + }); + } + + // Only react to text, file, or image messages in group chats. + if (msg.message_type !== "text" && msg.message_type !== "file" && msg.message_type !== "image") return; + const chatId = msg.chat_id; + if (chatId === "") return; + + // ADR-0001: resolve chat → project. Unknown chat ⇒ not a project group. + const binding = await deps.prisma.projectGroupBinding.findUnique({ + where: { chatId }, + select: { projectId: true }, + }); + if (binding === null) { + deps.logger.debug({ chatId }, "feishu trigger: chat not bound to any project"); + return; + } + const projectId = binding.projectId; + // ADR-0019: triggerAgent is authorized through actor → principal-set + // resolution, then grant/settings composition. + // A missing open_id is treated as an untrusted event: deny explicitly + // rather than skipping the check (fail closed). + const senderOpenId = event.sender.sender_id.open_id ?? ""; + if (senderOpenId === "") { + deps.logger.warn({ projectId, chatId }, "feishu trigger: sender has no open_id, denying"); + await writeAudit(deps.prisma, { projectId, action: "trigger.denied", metadata: { reason: "missing open_id" } }); + await sendText(rt, chatId, "无法识别发送者,拒绝触发。"); + return; + } + const actor = { feishuOpenId: senderOpenId, chatId }; + const triggerDecision = await authorizer.can({ + actor, + action: "agent.trigger", + resource: { type: "PROJECT", id: projectId }, + }); + if (!triggerDecision.allowed) { + deps.logger.info({ projectId, senderOpenId, reason: triggerDecision.reason }, "feishu trigger: permission denied"); + await writeAudit(deps.prisma, { + projectId, + ...(triggerDecision.actorUserId !== undefined ? { actorUserId: triggerDecision.actorUserId } : {}), + action: "trigger.denied", + metadata: { decision: auditDecision(triggerDecision) }, + }); + await sendText(rt, chatId, "无权限触发。"); + return; + } + + let cleanPrompt: string | null = null; + if (msg.message_type === "text") { + cleanPrompt = extractPrompt(msg); + } else if (msg.message_type === "file" || msg.message_type === "image") { + // Download the file/image to the workspace, build a prompt around it. + const project = await deps.prisma.project.findUnique({ + where: { id: projectId }, + select: { workspaceDir: true }, + }); + if (project === null) return; + try { + const content = FileMessageContentSchema.parse(JSON.parse(msg.content)); + const key = content.file_key ?? content.image_key ?? ""; + if (key !== "") { + const fileName = `${msg.message_type}-${Date.now()}.${msg.message_type === "image" ? "png" : "bin"}`; + const savePath = `${project.workspaceDir}/.cph/inbox/${fileName}`; + await downloadMessageFile(rt, msg.message_id, key, savePath); + cleanPrompt = `用户发来一个文件: ${savePath}`; + } + } catch { return; } + } + if (cleanPrompt === null) return; + const runContext: TriggerRunContext = { msg, rt, chatId, projectId, senderOpenId, actor }; + + // Slash commands: session management, not agent runs. These bypass the + // batcher. Session commands bypass the lock because they don't create runs; + // role/unknown slash prompts still run immediately through the normal lock. + if (cleanPrompt.startsWith("/")) { + const cmd = cleanPrompt.split(/\s+/)[0]; + switch (cmd) { + case "/new": { + await deps.prisma.agentSession.updateMany({ + where: { projectId, archivedAt: null }, + data: { archivedAt: new Date() }, + }); + await sendText(rt, chatId, "已开新会话,下次 @bot 将从头开始。"); + return; + } + case "/resume": { + const latest = await deps.prisma.agentSession.findFirst({ + where: { projectId, archivedAt: { not: null } }, + orderBy: { archivedAt: "desc" }, + select: { id: true }, + }); + if (latest === null) { + await sendText(rt, chatId, "没有可恢复的会话。"); + return; + } + await deps.prisma.agentSession.update({ + where: { id: latest.id }, + data: { archivedAt: null }, + }); + await sendText(rt, chatId, "已恢复上一个会话。"); + return; + } + case "/reset": { + await deps.prisma.agentSession.updateMany({ + where: { projectId, archivedAt: null }, + data: { archivedAt: new Date() }, + }); + await sendText(rt, chatId, "已重置,下次 @bot 将从头开始。"); + return; + } + default: + break; + } + await startAgentRun(runContext, cleanPrompt); + return; + } + + if (msg.message_type === "text") { + const existing = await currentLockRunId(deps.prisma, projectId); + if (existing !== null) { + await reactToMessage(rt, msg.message_id, "OnIt"); + await sendText(rt, chatId, "项目正在处理中,请稍候。"); + return; + } + const key = messageBatchKey(chatId, senderOpenId); + if (!batchContexts.has(key)) { + batchContexts.set(key, runContext); + } + await messageBatcher.enqueue(chatId, senderOpenId, cleanPrompt); + return; + } + + await startAgentRun(runContext, cleanPrompt); }; } diff --git a/hub/test/integration/trigger.test.ts b/hub/test/integration/trigger.test.ts index fe6f6ee..e616c81 100644 --- a/hub/test/integration/trigger.test.ts +++ b/hub/test/integration/trigger.test.ts @@ -96,7 +96,7 @@ describe("trigger full lifecycle (integration)", () => { it("creates a run, acquires + releases the lock, sends status card", async () => { await seedProject("proj-1", "chat-1"); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); await trigger(makeEvent("chat-1", "@_user_1 写教案"), rt); @@ -125,9 +125,51 @@ describe("trigger full lifecycle (integration)", () => { expect(runAgentCalls[0]?.maxTurns).toBe(7); }); + it("batches quick text messages from the same chat and sender into one run", async () => { + await seedProject("proj-1b", "chat-1b"); + const trigger = makeTriggerHandler({ + prisma, + settings, + logger: silentLogger, + runAgent, + messageBatcherOptions: { debounceMs: 10_000, maxMessages: 2 }, + }); + + await trigger(makeEvent("chat-1b", "@_user_1 第一段"), rt); + expect(await prisma.agentRun.findMany()).toHaveLength(0); + + await trigger(makeEvent("chat-1b", "@_user_1 第二段"), rt); + + await vi.waitFor(async () => { + const runs = await prisma.agentRun.findMany(); + expect(runs).toHaveLength(1); + expect(runs[0]?.status).toBe("COMPLETED"); + expect(runs[0]?.prompt).toBe("第一段\n第二段"); + }); + expect(runAgentCalls).toHaveLength(1); + expect(runAgentCalls[0]?.prompt).toBe("第一段\n第二段"); + }); + + it("/new bypasses message batching", async () => { + await seedProject("proj-1c", "chat-1c"); + const trigger = makeTriggerHandler({ + prisma, + settings, + logger: silentLogger, + runAgent, + messageBatcherOptions: { debounceMs: 10_000 }, + }); + + await trigger(makeEvent("chat-1c", "@_user_1 /new"), rt); + + expect(rt.sentTexts).toContain("已开新会话,下次 @bot 将从头开始。"); + expect(runAgentCalls).toHaveLength(0); + expect(await prisma.agentRun.findMany()).toHaveLength(0); + }); + it("rejects a sender without edit grant (ADR-0004)", async () => { await seedProject("proj-2", "chat-2", { role: "READ" }); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); await trigger(makeEvent("chat-2", "@_user_1 写教案"), rt); @@ -147,7 +189,7 @@ describe("trigger full lifecycle (integration)", () => { data: { projectId: "proj-3", runId: existingRun.id }, }); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); await trigger(makeEvent("chat-3", "@_user_1 写教案"), rt); expect(rt.sentTexts).toContain("项目正在处理中,请稍候。"); @@ -159,7 +201,7 @@ describe("trigger full lifecycle (integration)", () => { it("ignores messages from unbound chats (ADR-0001)", async () => { await seedProject("proj-4", "chat-4"); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); await trigger(makeEvent("chat-UNKNOWN", "@_user_1 写教案"), rt); @@ -171,7 +213,7 @@ describe("trigger full lifecycle (integration)", () => { it("ignores messages without @bot mention", async () => { await seedProject("proj-5", "chat-5"); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); const event: MessageReceiveEvent = { message: { @@ -193,7 +235,7 @@ describe("trigger full lifecycle (integration)", () => { it("/new archives current session (no run created)", async () => { await seedProject("proj-6", "chat-6"); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); // First @bot creates a session + run. await trigger(makeEvent("chat-6", "@_user_1 写教案"), rt); @@ -216,7 +258,7 @@ describe("trigger full lifecycle (integration)", () => { it("/resume un-archives the most recent session", async () => { await seedProject("proj-7", "chat-7"); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); // Create + archive a session via /new. await trigger(makeEvent("chat-7", "@_user_1 写教案"), rt); @@ -236,7 +278,7 @@ describe("trigger full lifecycle (integration)", () => { it("/reset archives current session", async () => { await seedProject("proj-8", "chat-8"); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); await trigger(makeEvent("chat-8", "@_user_1 写教案"), rt); await vi.waitFor(async () => { @@ -255,7 +297,7 @@ describe("trigger full lifecycle (integration)", () => { it("unknown slash command falls through to agent", async () => { await seedProject("proj-9", "chat-9"); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); await trigger(makeEvent("chat-9", "@_user_1 /unknown"), rt); // Should create a run (falls through as a normal prompt). @@ -272,7 +314,7 @@ describe("trigger full lifecycle (integration)", () => { await prisma.roleTriggerGrant.create({ data: { projectId: "proj-10", roleId: "review", principalType: "USER", principalId: "ou_other" }, }); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); await trigger(makeEvent("chat-10", "@_user_1 /review 看看这节"), rt); @@ -287,7 +329,7 @@ describe("trigger full lifecycle (integration)", () => { await prisma.roleTriggerGrant.create({ data: { projectId: "proj-11", roleId: "review", principalType: "USER", principalId: "ou_test_user" }, }); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); await trigger(makeEvent("chat-11", "@_user_1 /review 看看这节"), rt); @@ -301,7 +343,7 @@ describe("trigger full lifecycle (integration)", () => { it("extractRole: /draft sets roleId=draft, strips command from prompt", async () => { await seedProject("proj-12", "chat-12"); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); await trigger(makeEvent("chat-12", "@_user_1 /draft 写第三单元"), rt); @@ -315,7 +357,7 @@ describe("trigger full lifecycle (integration)", () => { it("dedups a redelivered event by event_id (no second run)", async () => { await seedProject("proj-13", "chat-13"); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); // First delivery: processes normally. await trigger(makeEvent("chat-13", "@_user_1 写教案", "ou_test_user", "evt-dedup-1"), rt); @@ -338,7 +380,7 @@ describe("trigger full lifecycle (integration)", () => { it("dedups flattened websocket events by top-level event_id", async () => { await seedProject("proj-13b", "chat-13b"); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); const event: MessageReceiveEvent = { ...makeEvent("chat-13b", "@_user_1 写教案"), event_id: "evt-flat-1", @@ -376,7 +418,7 @@ describe("trigger full lifecycle (integration)", () => { sender: { id: "ou_teacher_2", sender_type: "user" }, body: { content: JSON.stringify({ text: "上一条需求: 把第二题改成探究题。" }) }, }); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); const baseEvent = makeEvent("chat-13c", "@_user_1 继续这个改法"); const event: MessageReceiveEvent = { ...baseEvent, @@ -418,7 +460,7 @@ describe("trigger full lifecycle (integration)", () => { it("writes audit entries across the run lifecycle", async () => { await seedProject("proj-14", "chat-14"); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); await trigger(makeEvent("chat-14", "@_user_1 写教案"), rt); await vi.waitFor(async () => { @@ -433,7 +475,7 @@ describe("trigger full lifecycle (integration)", () => { it("persists AgentMessage rows for the run (A-3 structured history)", async () => { await seedProject("proj-15", "chat-15"); - const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent }); + const trigger = makeTriggerHandler({ prisma, settings, logger: silentLogger, runAgent, messageBatcherOptions: { maxMessages: 1 } }); await trigger(makeEvent("chat-15", "@_user_1 写教案"), rt); await vi.waitFor(async () => { diff --git a/hub/test/unit/feishu-reactions.test.ts b/hub/test/unit/feishu-reactions.test.ts index b171e21..f10abc8 100644 --- a/hub/test/unit/feishu-reactions.test.ts +++ b/hub/test/unit/feishu-reactions.test.ts @@ -103,6 +103,7 @@ async function triggerWithRunAgent( logger: rt.logger, authorizer: allowAllAuthorizer(), runAgent, + messageBatcherOptions: { maxMessages: 1 }, }); await trigger(makeEvent(), rt); diff --git a/hub/test/unit/message-batcher.test.ts b/hub/test/unit/message-batcher.test.ts new file mode 100644 index 0000000..f913be3 --- /dev/null +++ b/hub/test/unit/message-batcher.test.ts @@ -0,0 +1,114 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { MessageBatcher, messageBatchKey } from "../../src/feishu/messageBatcher.js"; + +describe("MessageBatcher", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("flushes a single message after the debounce period", async () => { + const flushed: string[] = []; + const batcher = new MessageBatcher(async (text) => { + flushed.push(text); + }, { debounceMs: 25 }); + + await batcher.enqueue("chat-1", "sender-1", "hello"); + + expect(flushed).toEqual([]); + await vi.advanceTimersByTimeAsync(24); + expect(flushed).toEqual([]); + await vi.advanceTimersByTimeAsync(1); + expect(flushed).toEqual(["hello"]); + }); + + it("merges two quick messages with a newline", async () => { + const flushed: string[] = []; + const batcher = new MessageBatcher(async (text) => { + flushed.push(text); + }, { debounceMs: 25 }); + + await batcher.enqueue("chat-1", "sender-1", "first"); + await batcher.enqueue("chat-1", "sender-1", "second"); + await vi.advanceTimersByTimeAsync(25); + + expect(flushed).toEqual(["first\nsecond"]); + }); + + it("flushes immediately when the max message count is reached", async () => { + const flushed: string[] = []; + const batcher = new MessageBatcher(async (text) => { + flushed.push(text); + }, { debounceMs: 1000, maxMessages: 2 }); + + await batcher.enqueue("chat-1", "sender-1", "first"); + expect(flushed).toEqual([]); + await batcher.enqueue("chat-1", "sender-1", "second"); + + expect(flushed).toEqual(["first\nsecond"]); + }); + + it("flushes immediately when the max character count is reached", async () => { + const flushed: string[] = []; + const batcher = new MessageBatcher(async (text) => { + flushed.push(text); + }, { debounceMs: 1000, maxChars: 5 }); + + await batcher.enqueue("chat-1", "sender-1", "abc"); + expect(flushed).toEqual([]); + await batcher.enqueue("chat-1", "sender-1", "d"); + + expect(flushed).toEqual(["abc\nd"]); + }); + + it("uses the extended debounce for a chunk near the split threshold", async () => { + const flushed: string[] = []; + const batcher = new MessageBatcher(async (text) => { + flushed.push(text); + }, { debounceMs: 10, splitThreshold: 5, extendedDebounceMs: 50 }); + + await batcher.enqueue("chat-1", "sender-1", "12345"); + await vi.advanceTimersByTimeAsync(49); + expect(flushed).toEqual([]); + await vi.advanceTimersByTimeAsync(1); + + expect(flushed).toEqual(["12345"]); + }); + + it("keeps different chat/sender pairs in separate batches", async () => { + const flushed: string[] = []; + const batcher = new MessageBatcher(async (text) => { + flushed.push(text); + }, { debounceMs: 25 }); + + await batcher.enqueue("chat-1", "sender-1", "one"); + await batcher.enqueue("chat-1", "sender-2", "two"); + await batcher.enqueue("chat-2", "sender-1", "three"); + await vi.advanceTimersByTimeAsync(25); + + expect(flushed.sort()).toEqual(["one", "three", "two"]); + }); + + it("flushAll flushes every pending batch", async () => { + const flushed: string[] = []; + const batcher = new MessageBatcher(async (text) => { + flushed.push(text); + }, { debounceMs: 1000 }); + + await batcher.enqueue("chat-1", "sender-1", "one"); + await batcher.enqueue("chat-2", "sender-2", "two"); + await batcher.flushAll(); + + expect(flushed.sort()).toEqual(["one", "two"]); + }); + + it("builds keys from chat and sender open ids", () => { + expect(messageBatchKey("chat-1", "ou-1")).toBe("chat-1:ou-1"); + }); + + // Slash-command bypass is owned by trigger.ts, not MessageBatcher. Existing + // trigger integration tests cover /new, /resume, /reset as immediate paths. +});