diff --git a/src/telegram/bridge.test.ts b/src/telegram/bridge.test.ts index 3097be6..85f378d 100644 --- a/src/telegram/bridge.test.ts +++ b/src/telegram/bridge.test.ts @@ -55,7 +55,7 @@ class FakeScheduler implements SchedulerLike { } } -function harness() { +function harness(overrides: { debounceMs?: number } = {}) { const botInfo: UserFromGetMe = { id: 1, is_bot: true, first_name: "sec", username: "sec_bot", can_join_groups: false, can_read_all_group_messages: false, supports_inline_queries: false, can_connect_to_business: false, has_main_web_app: false, @@ -82,6 +82,7 @@ function harness() { scheduler, outbound, logger, + debounceMs: overrides.debounceMs ?? 5, }); const texts = () => calls.filter((c) => c.method === "sendMessage").map((c) => c.payload.text as string); return { bot, calls, host, scheduler, texts }; @@ -142,15 +143,46 @@ describe("registerBridge", () => { expect(h.texts().slice(1)).toEqual(["Cancelled job abc12.", "No job with id abc12.", "Usage: /cancel ", "No scheduled jobs."]); }); - it("forwards text to the host and reacts when busy", async () => { + it("forwards text to the host after the debounce window and reacts when busy", async () => { const h = harness(); await h.bot.handleUpdate(textUpdate("what time is it")); - expect(h.host.sent).toEqual([{ text: "what time is it", images: undefined }]); + await vi.waitFor(() => expect(h.host.sent).toEqual([{ text: "what time is it", images: undefined }])); expect(h.calls.find((c) => c.method === "setMessageReaction")).toBeUndefined(); h.host.busy = true; await h.bot.handleUpdate(textUpdate("also this")); expect(h.calls.find((c) => c.method === "setMessageReaction")).toBeDefined(); - expect(h.host.sent).toHaveLength(2); + await vi.waitFor(() => expect(h.host.sent).toHaveLength(2)); + }); + + it("merges a forward-with-comment pair into a single agent turn", async () => { + const h = harness({ debounceMs: 30 }); + // Comment arrives first. + await h.bot.handleUpdate(textUpdate("look at this")); + // Then the forwarded message from Alice, within the window. + const fwd = textUpdate("hello everyone"); + (fwd.message as { forward_origin?: unknown }).forward_origin = { + type: "user", + date: 0, + sender_user: { id: 77, is_bot: false, first_name: "Alice", username: "alice" }, + }; + await h.bot.handleUpdate(fwd); + // Nothing has been delivered yet. + expect(h.host.sent).toEqual([]); + await vi.waitFor(() => + expect(h.host.sent).toEqual([ + { text: "look at this\n\n[Forwarded from @alice]\nhello everyone", images: undefined }, + ]), + ); + }); + + it("drops buffered messages when /new starts a fresh session", async () => { + const h = harness({ debounceMs: 100 }); + await h.bot.handleUpdate(textUpdate("hold this thought")); + await h.bot.handleUpdate(textUpdate("/new")); + // Wait past the debounce window; nothing should reach the host. + await new Promise((r) => setTimeout(r, 150)); + expect(h.host.sent).toEqual([]); + expect(h.host.newSessions).toBe(1); }); it("rejects unknown commands without bothering the agent", async () => { diff --git a/src/telegram/bridge.ts b/src/telegram/bridge.ts index a02a081..bdd43fb 100644 --- a/src/telegram/bridge.ts +++ b/src/telegram/bridge.ts @@ -7,9 +7,13 @@ import type { Logger } from "../log.js"; import type { Scheduler } from "../scheduler/scheduler.js"; import type { Transcriber } from "../stt/openai.js"; import { describeJob } from "../tools/schedule.js"; +import { InputDebouncer } from "./debouncer.js"; import { normalizeMessage } from "./inbound.js"; import type { Outbound } from "./outbound.js"; +/** How long we wait after the last message before delivering the merged batch to the agent. */ +const DEFAULT_DEBOUNCE_MS = 600; + export type HostLike = Pick; export type SchedulerLike = Pick; @@ -22,6 +26,8 @@ export interface BridgeDeps { transcriber?: Transcriber; logger: Logger; fetchFn?: typeof fetch; + /** Debounce window for merging rapid-succession messages (e.g. forward + comment). Defaults to 600ms. */ + debounceMs?: number; } export const BOT_COMMANDS = [ @@ -36,6 +42,15 @@ export const BOT_COMMANDS = [ export function registerBridge(deps: BridgeDeps): void { const { bot, config, host, scheduler, outbound, logger } = deps; const warned = new Set(); + const debouncer = new InputDebouncer({ + windowMs: deps.debounceMs ?? DEFAULT_DEBOUNCE_MS, + onFlush: (input) => { + void host.send(input).catch((err) => { + logger.error("agent send failed", err); + void outbound.sendPlain(`⚠️ ${errorText(err)}`).catch(() => undefined); + }); + }, + }); bot.use(async (ctx, next) => { const from = ctx.from?.id; @@ -58,6 +73,7 @@ export function registerBridge(deps: BridgeDeps): void { }); bot.command("new", async (ctx) => { + debouncer.discard(); await host.newSession(); await ctx.reply("New session started."); }); @@ -109,10 +125,7 @@ export function registerBridge(deps: BridgeDeps): void { return; } if (host.busy) await ctx.react("👀").catch(() => undefined); - void host.send(result.input).catch((err) => { - logger.error("agent send failed", err); - void outbound.sendPlain(`⚠️ ${errorText(err)}`).catch(() => undefined); - }); + debouncer.add(result.input); }); let queue: Promise = Promise.resolve(); diff --git a/src/telegram/debouncer.test.ts b/src/telegram/debouncer.test.ts new file mode 100644 index 0000000..d586a98 --- /dev/null +++ b/src/telegram/debouncer.test.ts @@ -0,0 +1,92 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { AgentInput } from "../agent/events.js"; +import { InputDebouncer, merge } from "./debouncer.js"; + +describe("InputDebouncer", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("flushes a single buffered input after the window", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 200, onFlush: (i) => flushed.push(i) }); + d.add({ text: "hi" }); + vi.advanceTimersByTime(199); + expect(flushed).toEqual([]); + vi.advanceTimersByTime(1); + expect(flushed).toEqual([{ text: "hi", images: undefined }]); + }); + + it("merges inputs that arrive within the window and resets the timer on each add", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 200, onFlush: (i) => flushed.push(i) }); + d.add({ text: "my comment" }); + vi.advanceTimersByTime(150); + d.add({ text: "[Forwarded from Alice]: hello" }); + vi.advanceTimersByTime(150); + expect(flushed).toEqual([]); + vi.advanceTimersByTime(50); + expect(flushed).toEqual([{ text: "my comment\n\n[Forwarded from Alice]: hello", images: undefined }]); + }); + + it("concatenates images across buffered inputs", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 100, onFlush: (i) => flushed.push(i) }); + d.add({ text: "one", images: [{ data: "a", mimeType: "image/jpeg" }] }); + d.add({ text: "two", images: [{ data: "b", mimeType: "image/jpeg" }] }); + vi.advanceTimersByTime(100); + expect(flushed).toEqual([ + { + text: "one\n\ntwo", + images: [ + { data: "a", mimeType: "image/jpeg" }, + { data: "b", mimeType: "image/jpeg" }, + ], + }, + ]); + }); + + it("does not fire when the buffer is empty", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 50, onFlush: (i) => flushed.push(i) }); + d.flushNow(); + vi.advanceTimersByTime(1000); + expect(flushed).toEqual([]); + }); + + it("flushNow delivers immediately and cancels the pending timer", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 500, onFlush: (i) => flushed.push(i) }); + d.add({ text: "urgent" }); + d.flushNow(); + expect(flushed).toEqual([{ text: "urgent", images: undefined }]); + vi.advanceTimersByTime(1000); + expect(flushed).toHaveLength(1); + }); + + it("discard drops buffered items without delivering", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 100, onFlush: (i) => flushed.push(i) }); + d.add({ text: "gone" }); + d.discard(); + vi.advanceTimersByTime(1000); + expect(flushed).toEqual([]); + }); +}); + +describe("merge", () => { + it("joins texts with a blank line and preserves image order", () => { + expect( + merge([ + { text: "a", images: [{ data: "x", mimeType: "image/jpeg" }] }, + { text: "b" }, + ]), + ).toEqual({ text: "a\n\nb", images: [{ data: "x", mimeType: "image/jpeg" }] }); + }); + + it("skips empty text parts", () => { + expect(merge([{ text: "", images: [{ data: "x", mimeType: "image/jpeg" }] }, { text: "b" }])).toEqual({ + text: "b", + images: [{ data: "x", mimeType: "image/jpeg" }], + }); + }); +}); diff --git a/src/telegram/debouncer.ts b/src/telegram/debouncer.ts new file mode 100644 index 0000000..98ddace --- /dev/null +++ b/src/telegram/debouncer.ts @@ -0,0 +1,58 @@ +import type { AgentImage, AgentInput } from "../agent/events.js"; + +export interface InputDebouncerOptions { + windowMs: number; + onFlush: (input: AgentInput) => void; +} + +/** + * Buffers AgentInputs arriving in quick succession and flushes a single merged + * input once the window goes quiet. Telegram sends "forward with a comment" as + * two separate messages (the comment, then the forwarded content); this stitches + * them together so the agent sees one turn. + */ +export class InputDebouncer { + private buffer: AgentInput[] = []; + private timer: NodeJS.Timeout | undefined; + + constructor(private readonly opts: InputDebouncerOptions) {} + + add(input: AgentInput): void { + this.buffer.push(input); + if (this.timer) clearTimeout(this.timer); + this.timer = setTimeout(() => this.flush(), this.opts.windowMs); + this.timer.unref?.(); + } + + /** Flushes immediately (e.g. on shutdown or when the caller wants to force delivery). */ + flushNow(): void { + if (this.timer) clearTimeout(this.timer); + this.timer = undefined; + this.flush(); + } + + /** Drops the buffered items without delivering them. */ + discard(): void { + if (this.timer) clearTimeout(this.timer); + this.timer = undefined; + this.buffer = []; + } + + private flush(): void { + this.timer = undefined; + if (this.buffer.length === 0) return; + const merged = merge(this.buffer); + this.buffer = []; + this.opts.onFlush(merged); + } +} + +export function merge(inputs: AgentInput[]): AgentInput { + const parts: string[] = []; + const images: AgentImage[] = []; + for (const input of inputs) { + if (input.text) parts.push(input.text); + if (input.images) images.push(...input.images); + } + return { text: parts.join("\n\n"), images: images.length > 0 ? images : undefined }; +} diff --git a/src/telegram/inbound.test.ts b/src/telegram/inbound.test.ts index 2138e82..5cc7e5f 100644 --- a/src/telegram/inbound.test.ts +++ b/src/telegram/inbound.test.ts @@ -91,6 +91,49 @@ describe("normalizeMessage", () => { expect(r.input).toEqual({ text: "see pic", images: undefined }); expect(r.notes[0]).toMatch(/Could not download photo: unknown file missing/); }); + + it("labels forwarded text with the origin", async () => { + const r = await normalizeMessage( + { text: "look at this", forward_origin: { type: "user", sender_user: { first_name: "Alice", username: "alice" } } }, + deps(), + ); + expect(r.input).toEqual({ text: "[Forwarded from @alice]\nlook at this", images: undefined }); + }); + + it("labels forwards from users without a username by name", async () => { + const r = await normalizeMessage( + { text: "hi", forward_origin: { type: "user", sender_user: { first_name: "Alice", last_name: "Smith" } } }, + deps(), + ); + expect(r.input?.text).toBe("[Forwarded from Alice Smith]\nhi"); + }); + + it("labels channel forwards by the channel title", async () => { + const r = await normalizeMessage( + { text: "news", forward_origin: { type: "channel", chat: { title: "Daily News" } } }, + deps(), + ); + expect(r.input?.text).toBe("[Forwarded from Daily News]\nnews"); + }); + + it("labels hidden-user forwards by their exposed name", async () => { + const r = await normalizeMessage( + { text: "secret", forward_origin: { type: "hidden_user", sender_user_name: "Someone" } }, + deps(), + ); + expect(r.input?.text).toBe("[Forwarded from Someone]\nsecret"); + }); + + it("labels forwarded photos even when there is no caption", async () => { + const r = await normalizeMessage( + { + photo: [{ file_id: "p2", width: 1, height: 1 }], + forward_origin: { type: "user", sender_user: { first_name: "Alice" } }, + }, + deps(), + ); + expect(r.input?.text).toBe("[Forwarded from Alice]\n(photo attached)"); + }); }); describe("saveToInbox", () => { diff --git a/src/telegram/inbound.ts b/src/telegram/inbound.ts index 6fddd19..d865ad3 100644 --- a/src/telegram/inbound.ts +++ b/src/telegram/inbound.ts @@ -3,6 +3,12 @@ import path from "node:path"; import type { AgentImage, AgentInput } from "../agent/events.js"; import type { Transcriber } from "../stt/openai.js"; +export type ForwardOrigin = + | { type: "user"; sender_user: { first_name: string; last_name?: string; username?: string } } + | { type: "hidden_user"; sender_user_name: string } + | { type: "chat"; sender_chat: { title?: string; username?: string }; author_signature?: string } + | { type: "channel"; chat: { title?: string; username?: string }; author_signature?: string }; + export interface InboundMessage { text?: string; caption?: string; @@ -10,6 +16,7 @@ export interface InboundMessage { voice?: { file_id: string; mime_type?: string; duration?: number }; audio?: { file_id: string; mime_type?: string; file_name?: string }; document?: { file_id: string; file_name?: string; mime_type?: string }; + forward_origin?: ForwardOrigin; } export interface InboundDeps { @@ -75,7 +82,11 @@ export async function normalizeMessage(msg: InboundMessage, deps: InboundDeps): } if (parts.length === 0 && images.length > 0) parts.push("(photo attached)"); - const text = parts.join("\n\n"); + let text = parts.join("\n\n"); + if (msg.forward_origin && (text || images.length > 0)) { + const label = describeForwardOrigin(msg.forward_origin); + text = text ? `[Forwarded from ${label}]\n${text}` : `[Forwarded from ${label}]`; + } return { input: text ? { text, images: images.length > 0 ? images : undefined } : null, transcript, @@ -83,6 +94,27 @@ export async function normalizeMessage(msg: InboundMessage, deps: InboundDeps): }; } +export function describeForwardOrigin(origin: ForwardOrigin): string { + switch (origin.type) { + case "user": { + const u = origin.sender_user; + if (u.username) return `@${u.username}`; + const name = [u.first_name, u.last_name].filter(Boolean).join(" "); + return name || "unknown user"; + } + case "hidden_user": + return origin.sender_user_name || "hidden user"; + case "chat": { + const c = origin.sender_chat; + return c.title || (c.username ? `@${c.username}` : "chat"); + } + case "channel": { + const c = origin.chat; + return c.title || (c.username ? `@${c.username}` : "channel"); + } + } +} + const MAX_STEM = 80; const MAX_EXT = 16;