From a4eb490a5f0b069af261a2f0b903078407e1ba0f Mon Sep 17 00:00:00 2001 From: 8x22b <062340513224@student.polsri.ac.id> Date: Tue, 7 Jul 2026 01:25:41 +0300 Subject: [PATCH 1/2] fix(bot): coalesce Telegram-split messages into a single prompt Telegram delivers long text (and a single paste) as several consecutive message updates, which the bot dispatched as separate prompts - one agent run per chunk. Buffer plain-text prompts per chat for MESSAGE_MERGE_WINDOW_MS (default 1500; 0 disables). Each new chunk restarts the window; when it elapses, buffered texts are joined and dispatched as one prompt. Commands and active interaction flows (question/task/rename/catalog) are unaffected. --- .env.example | 6 + README.md | 3 +- src/bot/handlers/message-merger.ts | 90 +++++++++++++++ src/bot/routers/message-router.ts | 9 +- src/config.ts | 21 ++++ tests/bot/handlers/message-merger.test.ts | 133 ++++++++++++++++++++++ tests/helpers/reset-singleton-state.ts | 3 + 7 files changed, 261 insertions(+), 4 deletions(-) create mode 100644 src/bot/handlers/message-merger.ts create mode 100644 tests/bot/handlers/message-merger.test.ts diff --git a/.env.example b/.env.example index b3916907..bea207af 100644 --- a/.env.example +++ b/.env.example @@ -105,6 +105,12 @@ OPENCODE_MODEL_ID=big-pickle # raw = show assistant replies as plain text # MESSAGE_FORMAT_MODE=markdown +# Merge quick consecutive text messages into one prompt (default: 1500 ms) +# Telegram splits long text into several messages; within this window (in +# milliseconds) they are buffered and sent to OpenCode as a single prompt. +# Set to 0 to disable merging and process every message immediately. +# MESSAGE_MERGE_WINDOW_MS=1500 + # Directory Browser Roots (optional) # Comma-separated list of paths that /open is allowed to browse. # Supports ~ for home directory. Defaults to ~ when not set. diff --git a/README.md b/README.md index 3edd6ee7..ed2acb2d 100644 --- a/README.md +++ b/README.md @@ -232,7 +232,8 @@ When installed via npm, the configuration wizard handles the initial setup. The | `BASH_TOOL_DISPLAY_MAX_LENGTH` | Maximum displayed length for `bash` tool commands in Telegram summaries; longer commands are truncated | No | `128` | | `TRACK_BACKGROUND_SESSIONS` | Track detached/non-current sessions in the current selected project/worktree and send short notifications | No | `true` | | `RESPONSE_STREAM_THROTTLE_MS` | Stream update throttle in milliseconds for assistant, thinking, and tool message edits | No | `1000` | -| `MESSAGE_FORMAT_MODE` | Assistant reply formatting mode: `markdown` (Telegram MarkdownV2) or `raw` | No | `markdown` | +| `MESSAGE_FORMAT_MODE` | Assistant reply formatting mode: `markdown` (Telegram MarkdownV2) or `raw` | No | `markdown` +`MESSAGE_MERGE_WINDOW_MS` | Merge quick consecutive text messages into one OpenCode prompt (ms). Telegram splits long text into several messages; within this window they are buffered and sent together. `0` disables merging | No | `1500` | | | `CODE_FILE_MAX_SIZE_KB` | Max file size (KB) to send as document | No | `100` | | `STT_API_URL` | Whisper-compatible API base URL (enables voice/audio transcription) | No | — | | `STT_API_KEY` | API key for your STT provider | No | — | diff --git a/src/bot/handlers/message-merger.ts b/src/bot/handlers/message-merger.ts new file mode 100644 index 00000000..a0a84b09 --- /dev/null +++ b/src/bot/handlers/message-merger.ts @@ -0,0 +1,90 @@ +import type { Context } from "grammy"; +import { processUserPrompt, type ProcessPromptDeps } from "./prompt.js"; +import { logger } from "../../utils/logger.js"; + +interface PendingPrompt { + texts: string[]; + ctx: Context; + deps: ProcessPromptDeps; + timer: ReturnType; +} + +// Buffered plain-text prompts, keyed by chat id. Telegram delivers one long +// message (or one paste) as several consecutive updates; merging them here +// turns those chunks into a single OpenCode prompt. +const pendingByChat = new Map(); + +function flushPending(chatId: number): void { + const pending = pendingByChat.get(chatId); + if (!pending) { + return; + } + + pendingByChat.delete(chatId); + clearTimeout(pending.timer); + + const { texts, ctx, deps } = pending; + if (texts.length > 1) { + logger.info( + `[Bot] Merging ${texts.length} quick consecutive messages into one prompt (chatId=${chatId}, totalLength=${texts.reduce((sum, part) => sum + part.length, 0)})`, + ); + } else { + logger.debug(`[Bot] Flushing single pending prompt (chatId=${chatId})`); + } + + void processUserPrompt(ctx, texts.join("\n\n"), deps); +} + +/** + * Buffers a plain-text prompt so several quick consecutive messages (e.g. + * Telegram splitting one long message into chunks) are merged into a single + * OpenCode prompt. Each new chunk restarts the wait window; once the window + * elapses with no new chunk, the buffered text is sent at once. + * + * Pass `mergeWindowMs <= 0` to disable merging and process the message + * immediately. + */ +export function queuePromptForMerging( + ctx: Context, + text: string, + deps: ProcessPromptDeps, + mergeWindowMs: number, +): void { + const chatId = ctx.chat!.id; + + if (mergeWindowMs <= 0) { + void processUserPrompt(ctx, text, deps); + return; + } + + const existing = pendingByChat.get(chatId); + if (existing) { + existing.texts.push(text); + existing.ctx = ctx; + clearTimeout(existing.timer); + existing.timer = setTimeout(() => flushPending(chatId), mergeWindowMs); + logger.debug( + `[Bot] Appended message to pending prompt (chatId=${chatId}, parts=${existing.texts.length})`, + ); + return; + } + + const timer = setTimeout(() => flushPending(chatId), mergeWindowMs); + pendingByChat.set(chatId, { texts: [text], ctx, deps, timer }); + logger.debug( + `[Bot] Started prompt merge window (chatId=${chatId}, mergeWindowMs=${mergeWindowMs})`, + ); +} + +/** Immediately flush any buffered prompt for the chat (e.g. when a command arrives). */ +export function flushPendingPrompt(chatId: number): void { + flushPending(chatId); +} + +/** Test helper: clears all buffered prompts and their timers. */ +export function __resetMessageMergerForTests(): void { + for (const pending of pendingByChat.values()) { + clearTimeout(pending.timer); + } + pendingByChat.clear(); +} diff --git a/src/bot/routers/message-router.ts b/src/bot/routers/message-router.ts index 443a34cc..707892e4 100644 --- a/src/bot/routers/message-router.ts +++ b/src/bot/routers/message-router.ts @@ -1,4 +1,5 @@ import type { Bot, Context } from "grammy"; +import { config } from "../../config.js"; import { interactionManager } from "../../app/managers/interaction-manager.js"; import { questionManager } from "../../app/managers/question-manager.js"; import { t } from "../../i18n/index.js"; @@ -21,7 +22,7 @@ import { import { handleDocumentMessage } from "../handlers/document-handler.js"; import { createMediaGroupAttachmentMiddleware } from "../handlers/media-group-handler.js"; import { handlePhotoMessage } from "../handlers/photo-handler.js"; -import { processUserPrompt } from "../handlers/prompt.js"; +import { queuePromptForMerging } from "../handlers/message-merger.js"; import { handleCatalogTextArguments } from "../handlers/text-message-handler.js"; import { handleVoiceMessage } from "../handlers/voice-handler.js"; import { unknownCommandMiddleware } from "../middleware/unknown-command.js"; @@ -190,8 +191,10 @@ export function registerMessageRouter(bot: Bot, deps: MessageRouterDeps return; } - await processUserPrompt(ctx, text, promptDeps); + queuePromptForMerging(ctx, text, promptDeps, config.bot.messageMergeWindowMs); - logger.debug("[Bot] message:text handler completed (prompt sent in background)"); + logger.debug( + `[Bot] message:text handler completed (merge window=${config.bot.messageMergeWindowMs}ms)`, + ); }); } diff --git a/src/config.ts b/src/config.ts index 12051fc6..ce74ac6f 100644 --- a/src/config.ts +++ b/src/config.ts @@ -33,6 +33,22 @@ function getOptionalPositiveIntEnvVar(key: string, defaultValue: number): number return parsedValue; } +// Like getOptionalPositiveIntEnvVar, but also accepts 0 (used to disable a feature). +function getOptionalNonNegativeIntEnvVar(key: string, defaultValue: number): number { + const value = getEnvVar(key, false); + + if (!value) { + return defaultValue; + } + + const parsedValue = Number.parseInt(value, 10); + if (Number.isNaN(parsedValue) || parsedValue < 0) { + return defaultValue; + } + + return parsedValue; +} + function getOptionalLocaleEnvVar(key: string, defaultValue: Locale): Locale { const value = getEnvVar(key, false); return normalizeLocale(value, defaultValue); @@ -167,6 +183,11 @@ export const config = { locale: getOptionalLocaleEnvVar("BOT_LOCALE", "en"), trackBackgroundSessions: getOptionalBooleanEnvVar("TRACK_BACKGROUND_SESSIONS", true), messageFormatMode: getOptionalMessageFormatModeEnvVar("MESSAGE_FORMAT_MODE", "markdown"), + // Telegram splits long outgoing text into several messages, and clients can + // also deliver a single paste as multiple updates. Queue quick consecutive + // text prompts within this window (ms) and send them as one OpenCode prompt. + // 0 disables merging and processes every message immediately. + messageMergeWindowMs: getOptionalNonNegativeIntEnvVar("MESSAGE_MERGE_WINDOW_MS", 1500), }, files: { maxFileSizeKb: parseInt(getEnvVar("CODE_FILE_MAX_SIZE_KB", false) || "100", 10), diff --git a/tests/bot/handlers/message-merger.test.ts b/tests/bot/handlers/message-merger.test.ts new file mode 100644 index 00000000..dd89568b --- /dev/null +++ b/tests/bot/handlers/message-merger.test.ts @@ -0,0 +1,133 @@ +import { beforeEach, afterEach, describe, expect, it, vi } from "vitest"; +import type { Context } from "grammy"; + +const processUserPromptMock = vi.hoisted(() => vi.fn()); + +vi.mock("../../../src/bot/handlers/prompt.js", () => ({ + processUserPrompt: processUserPromptMock, +})); + +vi.mock("../../../src/utils/logger.js", () => ({ + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +import { + queuePromptForMerging, + flushPendingPrompt, + __resetMessageMergerForTests, +} from "../../../src/bot/handlers/message-merger.js"; + +const DEPS = { bot: {} as never, ensureEventSubscription: vi.fn() }; + +function makeContext(chatId: number): Context { + return { chat: { id: chatId } } as unknown as Context; +} + +describe("message-merger", () => { + beforeEach(() => { + __resetMessageMergerForTests(); + processUserPromptMock.mockReset(); + vi.useFakeTimers(); + }); + + afterEach(() => { + __resetMessageMergerForTests(); + vi.useRealTimers(); + }); + + it("processes immediately when merge window is disabled (<=0)", () => { + const ctx = makeContext(1); + + queuePromptForMerging(ctx, "hello", DEPS, 0); + + expect(processUserPromptMock).toHaveBeenCalledTimes(1); + expect(processUserPromptMock).toHaveBeenCalledWith(ctx, "hello", DEPS); + }); + + it("flushes a single buffered message after the window elapses", () => { + const ctx = makeContext(1); + + queuePromptForMerging(ctx, "hello", DEPS, 1500); + + expect(processUserPromptMock).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(1500); + + expect(processUserPromptMock).toHaveBeenCalledTimes(1); + expect(processUserPromptMock).toHaveBeenCalledWith(ctx, "hello", DEPS); + }); + + it("merges quick consecutive messages into one prompt", () => { + const ctx = makeContext(1); + + queuePromptForMerging(ctx, "part 1", DEPS, 1500); + vi.advanceTimersByTime(1000); + queuePromptForMerging(ctx, "part 2", DEPS, 1500); + + // Window restarted by the second chunk, no flush yet. + expect(processUserPromptMock).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(1499); + expect(processUserPromptMock).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(1); + + expect(processUserPromptMock).toHaveBeenCalledTimes(1); + expect(processUserPromptMock).toHaveBeenCalledWith(ctx, "part 1\n\npart 2", DEPS); + }); + + it("flushes separately when messages are farther apart than the window", () => { + const ctx = makeContext(1); + + queuePromptForMerging(ctx, "first", DEPS, 1500); + vi.advanceTimersByTime(1500); + expect(processUserPromptMock).toHaveBeenCalledTimes(1); + + queuePromptForMerging(ctx, "second", DEPS, 1500); + vi.advanceTimersByTime(1500); + + expect(processUserPromptMock).toHaveBeenCalledTimes(2); + expect(processUserPromptMock).toHaveBeenNthCalledWith(1, ctx, "first", DEPS); + expect(processUserPromptMock).toHaveBeenNthCalledWith(2, ctx, "second", DEPS); + }); + + it("buffers per chat independently", () => { + const ctxA = makeContext(1); + const ctxB = makeContext(2); + + queuePromptForMerging(ctxA, "a1", DEPS, 1500); + queuePromptForMerging(ctxB, "b1", DEPS, 1500); + + vi.advanceTimersByTime(1500); + + expect(processUserPromptMock).toHaveBeenCalledTimes(2); + expect(processUserPromptMock).toHaveBeenCalledWith(ctxA, "a1", DEPS); + expect(processUserPromptMock).toHaveBeenCalledWith(ctxB, "b1", DEPS); + }); + + it("flushPendingPrompt flushes immediately and clears the buffer", () => { + const ctx = makeContext(1); + + queuePromptForMerging(ctx, "buffered", DEPS, 1500); + flushPendingPrompt(1); + + expect(processUserPromptMock).toHaveBeenCalledTimes(1); + expect(processUserPromptMock).toHaveBeenCalledWith(ctx, "buffered", DEPS); + + // A late timer fire must not double-flush. + vi.advanceTimersByTime(1500); + expect(processUserPromptMock).toHaveBeenCalledTimes(1); + }); + + it("uses the latest ctx when merging", () => { + const ctx1 = makeContext(1); + const ctx2 = makeContext(1); + + queuePromptForMerging(ctx1, "first", DEPS, 1500); + queuePromptForMerging(ctx2, "second", DEPS, 1500); + vi.advanceTimersByTime(1500); + + expect(processUserPromptMock).toHaveBeenCalledTimes(1); + expect(processUserPromptMock).toHaveBeenCalledWith(ctx2, "first\n\nsecond", DEPS); + }); +}); diff --git a/tests/helpers/reset-singleton-state.ts b/tests/helpers/reset-singleton-state.ts index c4f573b6..72b28f3e 100644 --- a/tests/helpers/reset-singleton-state.ts +++ b/tests/helpers/reset-singleton-state.ts @@ -55,6 +55,7 @@ export async function resetSingletonState(): Promise { { pinnedMessageManager }, { stopEventListening }, { __resetSessionDirectoryCacheForTests }, + { __resetMessageMergerForTests }, loggerModule, ] = await Promise.all([ import("../../src/app/managers/question-manager.js"), @@ -66,6 +67,7 @@ export async function resetSingletonState(): Promise { import("../../src/bot/pinned/pinned-message-manager.js"), import("../../src/opencode/events.js"), import("../../src/app/services/session-cache-service.js"), + import("../../src/bot/handlers/message-merger.js"), import("../../src/utils/logger.js"), ]); @@ -75,6 +77,7 @@ export async function resetSingletonState(): Promise { renameManager.clear(); interactionManager.clear("test_reset"); summaryAggregator.clear(); + __resetMessageMergerForTests(); const aggregator = summaryAggregator as unknown as SummaryAggregatorPrivateState; aggregator.onCompleteCallback = null; From 428b71b5492641c96af79ef227a57c676b4da380 Mon Sep 17 00:00:00 2001 From: 8x22b <062340513224@student.polsri.ac.id> Date: Sat, 11 Jul 2026 22:29:08 +0300 Subject: [PATCH 2/2] fix(bot): refine Telegram split-message merging --- .env.example | 6 +- README.md | 4 +- src/bot/handlers/document-handler.ts | 3 + src/bot/handlers/media-group-handler.ts | 3 + src/bot/handlers/message-merger.ts | 21 +++--- src/bot/handlers/photo-handler.ts | 3 + src/bot/handlers/voice-handler.ts | 3 + src/bot/routers/command-router.ts | 8 +++ tests/bot/handlers/document.test.ts | 10 +++ tests/bot/handlers/media-group.test.ts | 10 +++ tests/bot/handlers/message-merger.test.ts | 78 +++++++++++++++++------ tests/bot/handlers/photo-handler.test.ts | 10 +++ tests/bot/handlers/voice.test.ts | 7 ++ tests/bot/routers/command-router.test.ts | 23 ++++++- 14 files changed, 157 insertions(+), 32 deletions(-) diff --git a/.env.example b/.env.example index bea207af..8b27467b 100644 --- a/.env.example +++ b/.env.example @@ -105,9 +105,9 @@ OPENCODE_MODEL_ID=big-pickle # raw = show assistant replies as plain text # MESSAGE_FORMAT_MODE=markdown -# Merge quick consecutive text messages into one prompt (default: 1500 ms) -# Telegram splits long text into several messages; within this window (in -# milliseconds) they are buffered and sent to OpenCode as a single prompt. +# Merge Telegram-split long text messages into one prompt (default: 1500 ms) +# Text near Telegram's message length limit is buffered for this window (in +# milliseconds); any immediately following chunks are sent as one prompt. # Set to 0 to disable merging and process every message immediately. # MESSAGE_MERGE_WINDOW_MS=1500 diff --git a/README.md b/README.md index ed2acb2d..8587c151 100644 --- a/README.md +++ b/README.md @@ -232,8 +232,8 @@ When installed via npm, the configuration wizard handles the initial setup. The | `BASH_TOOL_DISPLAY_MAX_LENGTH` | Maximum displayed length for `bash` tool commands in Telegram summaries; longer commands are truncated | No | `128` | | `TRACK_BACKGROUND_SESSIONS` | Track detached/non-current sessions in the current selected project/worktree and send short notifications | No | `true` | | `RESPONSE_STREAM_THROTTLE_MS` | Stream update throttle in milliseconds for assistant, thinking, and tool message edits | No | `1000` | -| `MESSAGE_FORMAT_MODE` | Assistant reply formatting mode: `markdown` (Telegram MarkdownV2) or `raw` | No | `markdown` -`MESSAGE_MERGE_WINDOW_MS` | Merge quick consecutive text messages into one OpenCode prompt (ms). Telegram splits long text into several messages; within this window they are buffered and sent together. `0` disables merging | No | `1500` | | +| `MESSAGE_FORMAT_MODE` | Assistant reply formatting mode: `markdown` (Telegram MarkdownV2) or `raw` | No | `markdown` | +| `MESSAGE_MERGE_WINDOW_MS` | Merge Telegram-split long text messages into one prompt after this wait window (ms); `0` disables merging | No | `1500` | | `CODE_FILE_MAX_SIZE_KB` | Max file size (KB) to send as document | No | `100` | | `STT_API_URL` | Whisper-compatible API base URL (enables voice/audio transcription) | No | — | | `STT_API_KEY` | API key for your STT provider | No | — | diff --git a/src/bot/handlers/document-handler.ts b/src/bot/handlers/document-handler.ts index e53882df..14bdfe0a 100644 --- a/src/bot/handlers/document-handler.ts +++ b/src/bot/handlers/document-handler.ts @@ -12,6 +12,7 @@ import { getStoredModel } from "../../app/services/model-selection-service.js"; import { logger } from "../../utils/logger.js"; import { t } from "../../i18n/index.js"; import type { FilePartInput, Model } from "@opencode-ai/sdk/v2"; +import { flushPendingPrompt } from "./message-merger.js"; export interface DocumentHandlerDeps extends ProcessPromptDeps { downloadFile?: ( @@ -45,6 +46,8 @@ export async function handleDocumentMessage( return; } + flushPendingPrompt(ctx.chat!.id); + const caption = ctx.message.caption || ""; const mimeType = doc.mime_type || ""; const filename = doc.file_name || "document"; diff --git a/src/bot/handlers/media-group-handler.ts b/src/bot/handlers/media-group-handler.ts index 44a678fa..6abd6034 100644 --- a/src/bot/handlers/media-group-handler.ts +++ b/src/bot/handlers/media-group-handler.ts @@ -12,6 +12,7 @@ import { toDataUri, } from "../../app/services/file-download-service.js"; import { processUserPrompt, type ProcessPromptDeps } from "./prompt.js"; +import { flushPendingPrompt } from "./message-merger.js"; const DEFAULT_MEDIA_GROUP_DEBOUNCE_MS = 1_000; @@ -112,6 +113,8 @@ export class MediaGroupAttachmentHandler { return; } + flushPendingPrompt(chatId); + const key = this.getBatchKey(chatId, mediaGroupId); const existingBatch = this.batches.get(key); diff --git a/src/bot/handlers/message-merger.ts b/src/bot/handlers/message-merger.ts index a0a84b09..ffeffadf 100644 --- a/src/bot/handlers/message-merger.ts +++ b/src/bot/handlers/message-merger.ts @@ -2,6 +2,8 @@ import type { Context } from "grammy"; import { processUserPrompt, type ProcessPromptDeps } from "./prompt.js"; import { logger } from "../../utils/logger.js"; +const TELEGRAM_SPLIT_CHUNK_MIN_LENGTH = 4000; + interface PendingPrompt { texts: string[]; ctx: Context; @@ -32,14 +34,15 @@ function flushPending(chatId: number): void { logger.debug(`[Bot] Flushing single pending prompt (chatId=${chatId})`); } - void processUserPrompt(ctx, texts.join("\n\n"), deps); + void processUserPrompt(ctx, texts.join("\n\n"), deps).catch((err) => { + logger.error(`[Bot] Failed to process merged prompt (chatId=${chatId})`, err); + }); } /** - * Buffers a plain-text prompt so several quick consecutive messages (e.g. - * Telegram splitting one long message into chunks) are merged into a single - * OpenCode prompt. Each new chunk restarts the wait window; once the window - * elapses with no new chunk, the buffered text is sent at once. + * Buffers a near-limit plain-text prompt so Telegram-split chunks are merged + * into a single OpenCode prompt. Short messages are processed immediately + * unless they follow a buffered chunk. Each new chunk restarts the wait window. * * Pass `mergeWindowMs <= 0` to disable merging and process the message * immediately. @@ -51,13 +54,15 @@ export function queuePromptForMerging( mergeWindowMs: number, ): void { const chatId = ctx.chat!.id; + const existing = pendingByChat.get(chatId); - if (mergeWindowMs <= 0) { - void processUserPrompt(ctx, text, deps); + if (mergeWindowMs <= 0 || (!existing && text.length < TELEGRAM_SPLIT_CHUNK_MIN_LENGTH)) { + void processUserPrompt(ctx, text, deps).catch((err) => { + logger.error(`[Bot] Failed to process prompt (chatId=${chatId})`, err); + }); return; } - const existing = pendingByChat.get(chatId); if (existing) { existing.texts.push(text); existing.ctx = ctx; diff --git a/src/bot/handlers/photo-handler.ts b/src/bot/handlers/photo-handler.ts index 98eeeb1b..a6b0a2a6 100644 --- a/src/bot/handlers/photo-handler.ts +++ b/src/bot/handlers/photo-handler.ts @@ -5,6 +5,7 @@ import { getModelCapabilities, supportsInput } from "../../app/services/model-ca import { getStoredModel } from "../../app/services/model-selection-service.js"; import { t } from "../../i18n/index.js"; import { logger } from "../../utils/logger.js"; +import { flushPendingPrompt } from "./message-merger.js"; import { processUserPrompt, type ProcessPromptDeps } from "./prompt.js"; export interface PhotoHandlerDeps extends ProcessPromptDeps { @@ -31,6 +32,8 @@ export async function handlePhotoMessage(ctx: Context, deps: PhotoHandlerDeps): return; } + flushPendingPrompt(ctx.chat!.id); + const caption = ctx.message.caption || ""; const largestPhoto = photos[photos.length - 1]; const downloadFile = deps.downloadFile ?? downloadTelegramFile; diff --git a/src/bot/handlers/voice-handler.ts b/src/bot/handlers/voice-handler.ts index 720aca0f..b0f403bb 100644 --- a/src/bot/handlers/voice-handler.ts +++ b/src/bot/handlers/voice-handler.ts @@ -13,6 +13,7 @@ import { type SttResult, } from "../../app/services/stt-service.js"; import { processUserPrompt, type ProcessPromptDeps } from "./prompt.js"; +import { flushPendingPrompt } from "./message-merger.js"; import { logger } from "../../utils/logger.js"; import { t } from "../../i18n/index.js"; import { buildTelegramFileUrl } from "../../app/services/file-download-service.js"; @@ -191,6 +192,8 @@ export async function handleVoiceMessage(ctx: Context, deps: VoiceMessageDeps): return; } + flushPendingPrompt(ctx.chat!.id); + // Check if STT is configured if (!sttConfigured()) { await ctx.reply(t("stt.not_configured")); diff --git a/src/bot/routers/command-router.ts b/src/bot/routers/command-router.ts index 481ba110..4e2faf1c 100644 --- a/src/bot/routers/command-router.ts +++ b/src/bot/routers/command-router.ts @@ -23,6 +23,7 @@ import { helpCommand } from "../commands/help-command.js"; import { statusCommand } from "../commands/status-command.js"; import { BOT_COMMANDS } from "../commands/definitions.js"; import { logger } from "../../utils/logger.js"; +import { flushPendingPrompt } from "../handlers/message-merger.js"; interface CommandRouterDeps { ensureEventSubscription: (directory: string) => Promise; @@ -60,6 +61,13 @@ export async function ensureCommandsInitialized(ctx: Context, next: NextFunction } export function registerCommandRouter(bot: Bot, deps: CommandRouterDeps): void { + bot.use(async (ctx, next) => { + if (ctx.chat && ctx.message?.text?.startsWith("/")) { + flushPendingPrompt(ctx.chat.id); + } + await next(); + }); + bot.command("start", startCommand); bot.command("help", helpCommand); bot.command("status", statusCommand); diff --git a/tests/bot/handlers/document.test.ts b/tests/bot/handlers/document.test.ts index 637dbf07..bfeea486 100644 --- a/tests/bot/handlers/document.test.ts +++ b/tests/bot/handlers/document.test.ts @@ -1,5 +1,13 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { Context } from "grammy"; + +const flushPendingPromptMock = vi.hoisted(() => vi.fn()); + +vi.mock("../../../src/bot/handlers/message-merger.js", () => ({ + flushPendingPrompt: flushPendingPromptMock, + __resetMessageMergerForTests: vi.fn(), +})); + import { handleDocumentMessage, type DocumentHandlerDeps, @@ -73,6 +81,7 @@ function createDocumentDeps(overrides: Partial = {}): { describe("bot/handlers/document", () => { beforeEach(() => { vi.restoreAllMocks(); + flushPendingPromptMock.mockClear(); }); describe("text files", () => { @@ -82,6 +91,7 @@ describe("bot/handlers/document", () => { await handleDocumentMessage(ctx, deps); + expect(flushPendingPromptMock).toHaveBeenCalledWith(777); expect(replyMock).toHaveBeenCalledWith(t("bot.file_downloading")); expect(downloadMock).toHaveBeenCalled(); expect(processPromptMock).toHaveBeenCalledWith( diff --git a/tests/bot/handlers/media-group.test.ts b/tests/bot/handlers/media-group.test.ts index f212a0f1..2c0cd49e 100644 --- a/tests/bot/handlers/media-group.test.ts +++ b/tests/bot/handlers/media-group.test.ts @@ -1,5 +1,13 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { Context, NextFunction } from "grammy"; + +const flushPendingPromptMock = vi.hoisted(() => vi.fn()); + +vi.mock("../../../src/bot/handlers/message-merger.js", () => ({ + flushPendingPrompt: flushPendingPromptMock, + __resetMessageMergerForTests: vi.fn(), +})); + import { MediaGroupAttachmentHandler, type MediaGroupHandlerDeps, @@ -126,6 +134,7 @@ async function addToHandler(handler: MediaGroupAttachmentHandler, ctx: Context): describe("bot/handlers/media-group", () => { beforeEach(() => { vi.restoreAllMocks(); + flushPendingPromptMock.mockClear(); }); afterEach(() => { @@ -153,6 +162,7 @@ describe("bot/handlers/media-group", () => { await addToHandler(handler, second.ctx); await handler.flushAll(); + expect(flushPendingPromptMock).toHaveBeenCalledWith(777); expect(first.replyMock).toHaveBeenCalledWith(t("bot.files_downloading")); expect(processPromptMock).toHaveBeenCalledTimes(1); expect(processPromptMock).toHaveBeenCalledWith( diff --git a/tests/bot/handlers/message-merger.test.ts b/tests/bot/handlers/message-merger.test.ts index dd89568b..a4ff9f33 100644 --- a/tests/bot/handlers/message-merger.test.ts +++ b/tests/bot/handlers/message-merger.test.ts @@ -2,13 +2,14 @@ import { beforeEach, afterEach, describe, expect, it, vi } from "vitest"; import type { Context } from "grammy"; const processUserPromptMock = vi.hoisted(() => vi.fn()); +const loggerErrorMock = vi.hoisted(() => vi.fn()); vi.mock("../../../src/bot/handlers/prompt.js", () => ({ processUserPrompt: processUserPromptMock, })); vi.mock("../../../src/utils/logger.js", () => ({ - logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: loggerErrorMock }, })); import { @@ -18,6 +19,7 @@ import { } from "../../../src/bot/handlers/message-merger.js"; const DEPS = { bot: {} as never, ensureEventSubscription: vi.fn() }; +const LARGE_TEXT = "x".repeat(4000); function makeContext(chatId: number): Context { return { chat: { id: chatId } } as unknown as Context; @@ -26,7 +28,8 @@ function makeContext(chatId: number): Context { describe("message-merger", () => { beforeEach(() => { __resetMessageMergerForTests(); - processUserPromptMock.mockReset(); + processUserPromptMock.mockReset().mockResolvedValue(true); + loggerErrorMock.mockReset(); vi.useFakeTimers(); }); @@ -44,23 +47,31 @@ describe("message-merger", () => { expect(processUserPromptMock).toHaveBeenCalledWith(ctx, "hello", DEPS); }); - it("flushes a single buffered message after the window elapses", () => { + it("processes short messages immediately when merging is enabled", () => { const ctx = makeContext(1); queuePromptForMerging(ctx, "hello", DEPS, 1500); + expect(processUserPromptMock).toHaveBeenCalledWith(ctx, "hello", DEPS); + }); + + it("flushes a single buffered message after the window elapses", () => { + const ctx = makeContext(1); + + queuePromptForMerging(ctx, LARGE_TEXT, DEPS, 1500); + expect(processUserPromptMock).not.toHaveBeenCalled(); vi.advanceTimersByTime(1500); expect(processUserPromptMock).toHaveBeenCalledTimes(1); - expect(processUserPromptMock).toHaveBeenCalledWith(ctx, "hello", DEPS); + expect(processUserPromptMock).toHaveBeenCalledWith(ctx, LARGE_TEXT, DEPS); }); it("merges quick consecutive messages into one prompt", () => { const ctx = makeContext(1); - queuePromptForMerging(ctx, "part 1", DEPS, 1500); + queuePromptForMerging(ctx, LARGE_TEXT, DEPS, 1500); vi.advanceTimersByTime(1000); queuePromptForMerging(ctx, "part 2", DEPS, 1500); @@ -73,46 +84,51 @@ describe("message-merger", () => { vi.advanceTimersByTime(1); expect(processUserPromptMock).toHaveBeenCalledTimes(1); - expect(processUserPromptMock).toHaveBeenCalledWith(ctx, "part 1\n\npart 2", DEPS); + expect(processUserPromptMock).toHaveBeenCalledWith(ctx, `${LARGE_TEXT}\n\npart 2`, DEPS); }); it("flushes separately when messages are farther apart than the window", () => { const ctx = makeContext(1); - queuePromptForMerging(ctx, "first", DEPS, 1500); + const first = "a".repeat(4000); + const second = "b".repeat(4000); + + queuePromptForMerging(ctx, first, DEPS, 1500); vi.advanceTimersByTime(1500); expect(processUserPromptMock).toHaveBeenCalledTimes(1); - queuePromptForMerging(ctx, "second", DEPS, 1500); + queuePromptForMerging(ctx, second, DEPS, 1500); vi.advanceTimersByTime(1500); expect(processUserPromptMock).toHaveBeenCalledTimes(2); - expect(processUserPromptMock).toHaveBeenNthCalledWith(1, ctx, "first", DEPS); - expect(processUserPromptMock).toHaveBeenNthCalledWith(2, ctx, "second", DEPS); + expect(processUserPromptMock).toHaveBeenNthCalledWith(1, ctx, first, DEPS); + expect(processUserPromptMock).toHaveBeenNthCalledWith(2, ctx, second, DEPS); }); it("buffers per chat independently", () => { const ctxA = makeContext(1); const ctxB = makeContext(2); - queuePromptForMerging(ctxA, "a1", DEPS, 1500); - queuePromptForMerging(ctxB, "b1", DEPS, 1500); + const textA = "a".repeat(4000); + const textB = "b".repeat(4000); + queuePromptForMerging(ctxA, textA, DEPS, 1500); + queuePromptForMerging(ctxB, textB, DEPS, 1500); vi.advanceTimersByTime(1500); expect(processUserPromptMock).toHaveBeenCalledTimes(2); - expect(processUserPromptMock).toHaveBeenCalledWith(ctxA, "a1", DEPS); - expect(processUserPromptMock).toHaveBeenCalledWith(ctxB, "b1", DEPS); + expect(processUserPromptMock).toHaveBeenCalledWith(ctxA, textA, DEPS); + expect(processUserPromptMock).toHaveBeenCalledWith(ctxB, textB, DEPS); }); it("flushPendingPrompt flushes immediately and clears the buffer", () => { const ctx = makeContext(1); - queuePromptForMerging(ctx, "buffered", DEPS, 1500); + queuePromptForMerging(ctx, LARGE_TEXT, DEPS, 1500); flushPendingPrompt(1); expect(processUserPromptMock).toHaveBeenCalledTimes(1); - expect(processUserPromptMock).toHaveBeenCalledWith(ctx, "buffered", DEPS); + expect(processUserPromptMock).toHaveBeenCalledWith(ctx, LARGE_TEXT, DEPS); // A late timer fire must not double-flush. vi.advanceTimersByTime(1500); @@ -123,11 +139,37 @@ describe("message-merger", () => { const ctx1 = makeContext(1); const ctx2 = makeContext(1); - queuePromptForMerging(ctx1, "first", DEPS, 1500); + queuePromptForMerging(ctx1, LARGE_TEXT, DEPS, 1500); queuePromptForMerging(ctx2, "second", DEPS, 1500); vi.advanceTimersByTime(1500); expect(processUserPromptMock).toHaveBeenCalledTimes(1); - expect(processUserPromptMock).toHaveBeenCalledWith(ctx2, "first\n\nsecond", DEPS); + expect(processUserPromptMock).toHaveBeenCalledWith(ctx2, `${LARGE_TEXT}\n\nsecond`, DEPS); + }); + + it("logs rejected immediate prompt processing", async () => { + const error = new Error("prompt failed"); + processUserPromptMock.mockRejectedValueOnce(error); + + queuePromptForMerging(makeContext(1), "hello", DEPS, 1500); + await Promise.resolve(); + + expect(loggerErrorMock).toHaveBeenCalledWith( + "[Bot] Failed to process prompt (chatId=1)", + error, + ); + }); + + it("logs rejected merged prompt processing", async () => { + const error = new Error("merged prompt failed"); + processUserPromptMock.mockRejectedValueOnce(error); + + queuePromptForMerging(makeContext(1), LARGE_TEXT, DEPS, 1500); + await vi.advanceTimersByTimeAsync(1500); + + expect(loggerErrorMock).toHaveBeenCalledWith( + "[Bot] Failed to process merged prompt (chatId=1)", + error, + ); }); }); diff --git a/tests/bot/handlers/photo-handler.test.ts b/tests/bot/handlers/photo-handler.test.ts index 54d504ca..07ba1176 100644 --- a/tests/bot/handlers/photo-handler.test.ts +++ b/tests/bot/handlers/photo-handler.test.ts @@ -1,5 +1,13 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { Context } from "grammy"; + +const flushPendingPromptMock = vi.hoisted(() => vi.fn()); + +vi.mock("../../../src/bot/handlers/message-merger.js", () => ({ + flushPendingPrompt: flushPendingPromptMock, + __resetMessageMergerForTests: vi.fn(), +})); + import { handlePhotoMessage, type PhotoHandlerDeps } from "../../../src/bot/handlers/photo-handler.js"; import { t } from "../../../src/i18n/index.js"; @@ -49,6 +57,7 @@ function createDeps(overrides: Partial = {}): { describe("bot/handlers/photo-handler", () => { beforeEach(() => { vi.restoreAllMocks(); + flushPendingPromptMock.mockClear(); }); it("downloads the largest photo and sends it as a file part", async () => { @@ -57,6 +66,7 @@ describe("bot/handlers/photo-handler", () => { await handlePhotoMessage(ctx, deps); + expect(flushPendingPromptMock).toHaveBeenCalledWith(777); expect(replyMock).toHaveBeenCalledWith(t("bot.photo_downloading")); expect(downloadMock).toHaveBeenCalledWith(ctx.api, "large-photo"); expect(processPromptMock).toHaveBeenCalledWith( diff --git a/tests/bot/handlers/voice.test.ts b/tests/bot/handlers/voice.test.ts index db428403..0ebf8ab6 100644 --- a/tests/bot/handlers/voice.test.ts +++ b/tests/bot/handlers/voice.test.ts @@ -6,6 +6,7 @@ import { t } from "../../../src/i18n/index.js"; const mocked = vi.hoisted(() => ({ getTtsModeMock: vi.fn(), + flushPendingPromptMock: vi.fn(), })); vi.mock("../../../src/app/stores/settings-store.js", () => ({ @@ -21,6 +22,11 @@ vi.mock("../../../src/utils/logger.js", () => ({ }, })); +vi.mock("../../../src/bot/handlers/message-merger.js", () => ({ + flushPendingPrompt: mocked.flushPendingPromptMock, + __resetMessageMergerForTests: vi.fn(), +})); + async function loadVoiceModule() { vi.resetModules(); return import("../../../src/bot/handlers/voice-handler.js"); @@ -148,6 +154,7 @@ describe("bot/handlers/voice-handler", () => { await handleVoiceMessage(ctx, deps); + expect(mocked.flushPendingPromptMock).toHaveBeenCalledWith(777); expect(replyMock).toHaveBeenCalledWith(t("stt.recognizing")); expect(processPromptMock).toHaveBeenCalledWith(ctx, "run tests", deps, [], { responseMode: "text_only", diff --git a/tests/bot/routers/command-router.test.ts b/tests/bot/routers/command-router.test.ts index 2580fe41..61fb2653 100644 --- a/tests/bot/routers/command-router.test.ts +++ b/tests/bot/routers/command-router.test.ts @@ -1,5 +1,13 @@ import { describe, expect, it, vi } from "vitest"; import type { Context, NextFunction } from "grammy"; + +const flushPendingPromptMock = vi.hoisted(() => vi.fn()); + +vi.mock("../../../src/bot/handlers/message-merger.js", () => ({ + flushPendingPrompt: flushPendingPromptMock, + __resetMessageMergerForTests: vi.fn(), +})); + import { ensureCommandsInitialized, registerCommandRouter, @@ -9,7 +17,7 @@ import { config } from "../../../src/config.js"; describe("bot/routers/command-router", () => { it("registers bot slash command handlers", () => { - const bot = { command: vi.fn() }; + const bot = { command: vi.fn(), use: vi.fn() }; registerCommandRouter(bot as never, { ensureEventSubscription: vi.fn() }); @@ -38,6 +46,19 @@ describe("bot/routers/command-router", () => { ]); }); + it("flushes a pending prompt before routing a command", async () => { + const bot = { command: vi.fn(), use: vi.fn() }; + const next = vi.fn(); + registerCommandRouter(bot as never, { ensureEventSubscription: vi.fn() }); + const middleware = bot.use.mock.calls[0][0]; + const ctx = { chat: { id: 123 }, message: { text: "/new" } } as unknown as Context; + + await middleware(ctx, next); + + expect(flushPendingPromptMock).toHaveBeenCalledWith(123); + expect(next).toHaveBeenCalledOnce(); + }); + it("initializes commands for the authorized chat", async () => { const next: NextFunction = vi.fn(); const ctx = {